expression_processor.py - master - eDuArDG/expression-processor

expression_processor.py
433 строки19 KB
eDuArDG
Create: expression_processor.py
11 июн 2026, 03:16
11 июн 2026, 03:169bd6d97
import os
import time
import json
import re
import tkinter as tk
from tkinter import filedialog, messagebox
from pathlib import Path
import zipfile
import io


def get_line_context(line):
    stripped = line.lstrip()
    return "json" if stripped.startswith('{') else "assignment"


def find_atomic_blocks(line):
    blocks = []
    i = 0
    n = len(line)
    while i < n:
        if line[i] == '{':
            j = i + 1
            while j < n:
                if line[j] == '{':
                    break
                if line[j] == '}':
                    blocks.append(line[i:j + 1])
                    i = j
                    break
                j += 1
            else:
                i += 1
                continue
            i += 1
        else:
            i += 1
    return blocks


def parse_block_to_csv_row(block, id_name, include_id_column=True):
    id_num = id_name[1:] if id_name.startswith('@') else id_name

    op_match = re.search(r'"op"\s*:\s*"([^"]*)"', block)
    op = op_match.group(1) if op_match else ''

    fo = ''
    fo_match = re.search(r'"fO"\s*:\s*(?:"([^"]*)"|(@\d+)|(\d+))', block)
    if fo_match:
        if fo_match.group(1):
            fo = fo_match.group(1)
        elif fo_match.group(2):
            fo = fo_match.group(2)
        elif fo_match.group(3):
            fo = fo_match.group(3)

    so = ''
    so_match = re.search(r'"sO"\s*:\s*(?:"([^"]*)"|(@\d+)|(\d+))', block)
    if so_match:
        if so_match.group(1):
            so = so_match.group(1)
        elif so_match.group(2):
            so = so_match.group(2)
        elif so_match.group(3):
            so = so_match.group(3)

    if not fo:
        od_match = re.search(r'"od"\s*:\s*(?:"([^"]*)"|(@\d+)|(\d+))', block)
        if od_match:
            if od_match.group(1):
                fo = od_match.group(1)
            elif od_match.group(2):
                fo = od_match.group(2)
            elif od_match.group(3):
                fo = od_match.group(3)
            so = ''

    if include_id_column:
        return f"{id_num};{op};{fo};{so}"
    else:
        return f"{op};{fo};{so}"


def _create_empty_outputs(output_dir, generate_statistics, generate_parallel_plan, include_id_column, output_filename):
    with open(os.path.join(output_dir, 'structure.txt'), 'w', encoding='utf-8') as out:
        out.write("Блоков не найдено\n")
    if generate_statistics:
        with open(os.path.join(output_dir, 'statistics.txt'), 'w', encoding='utf-8') as out:
            out.write("Блоков не найдено\n")
    if generate_parallel_plan:
        with open(os.path.join(output_dir, 'parallel_plan(height=0_quasi-width=0).csv'), 'w', encoding='utf-8') as out:
            out.write("0;0\n")


def process_files_from_folder(input_path, output_base_dir, min_occurrences=2, output_filename="structure.txt",
                              generate_statistics=True, generate_parallel_plan=True, include_id_column=True,
                              zip_threshold=1, zip_archive=True):
    path = Path(input_path)
    files_to_process = []
    if path.is_file():
        folder_name = path.stem
    else:
        folder_name = path.name

    output_dir = os.path.join(output_base_dir, folder_name)
    os.makedirs(output_dir, exist_ok=True)
    print(f"Выходные файлы будут сохранены в: {output_dir}")

    if path.is_file():
        if path.suffix.lower() == '.txt':
            files_to_process = [path]
            print(f"Обрабатывается один файл: {path.name}")
        else:
            print(f"Файл '{path.name}' не является .txt файлом!")
            _create_empty_outputs(output_dir, generate_statistics, generate_parallel_plan, include_id_column,
                                  output_filename)
            return
    elif path.is_dir():
        files_to_process = [f for f in path.iterdir() if f.is_file() and f.suffix.lower() == '.txt']
        print(f"Обрабатывается папка '{input_path}', найдено {len(files_to_process)} .txt файлов")
        if not files_to_process:
            print(f"В папке '{input_path}' нет .txt файлов!")
            _create_empty_outputs(output_dir, generate_statistics, generate_parallel_plan, include_id_column,
                                  output_filename)
            return
    else:
        print(f"Путь '{input_path}' не найден!")
        _create_empty_outputs(output_dir, generate_statistics, generate_parallel_plan, include_id_column,
                              output_filename)
        return

    all_lines = []
    original_lines = []
    for file_path in sorted(files_to_process):
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                lines = f.read().splitlines()
                all_lines.extend(lines)
                original_lines.extend(lines)
                print(f"Загружено {len(lines)} строк из {file_path.name}")
        except Exception as e:
            print(f"Пропущен файл {file_path}: {e}")

    if not all_lines:
        print("Нет данных для обработки — все файлы пусты.")
        _create_empty_outputs(output_dir, generate_statistics, generate_parallel_plan, include_id_column,
                              output_filename)
        return

    working_lines = all_lines[:]
    id_counter = 1
    block_to_id = {}
    block_metadata = {}
    generations = []
    pass_times = []
    max_width = 0
    total_passes = 0
    total_start = time.time()
    pass_number = 0

    while True:
        pass_number += 1
        pass_start = time.time()
        block_occurrences = {}
        for idx, line in enumerate(working_lines):
            context = get_line_context(original_lines[idx])
            for block in find_atomic_blocks(line):
                if block not in block_occurrences:
                    block_occurrences[block] = []
                block_occurrences[block].append((idx, context))

        new_gen = []
        for block, positions in block_occurrences.items():
            if len(positions) >= min_occurrences and block not in block_to_id:
                contexts = {ctx for (_, ctx) in positions}
                id_name = f"@{id_counter}"
                id_counter += 1
                block_to_id[block] = id_name
                block_metadata[block] = {
                    'count': len(positions),
                    'contexts': contexts,
                    'level': pass_number
                }
                new_gen.append((block, id_name))

        pass_end = time.time()
        pass_times.append(pass_end - pass_start)

        if not new_gen:
            pass_times.pop()
            break

        generations.append(new_gen)
        current_width = len(new_gen)
        if current_width > max_width:
            max_width = current_width
        total_passes = pass_number

        for block, id_name in sorted(new_gen, key=lambda x: len(x[0]), reverse=True):
            for i in range(len(working_lines)):
                working_lines[i] = working_lines[i].replace(block, id_name)

    total_time = time.time() - total_start

    # Формируем группы уровней
    if generations:
        levels_dict = {}
        for level_idx, gen in enumerate(generations):
            level_num = level_idx + 1
            levels_dict[level_num] = [{'block': b, 'id_name': i} for b, i in gen]

        zip_groups = []
        current_group_levels = []
        current_group_count = 0
        current_start_level = 1
        sorted_levels = sorted(levels_dict.keys())
        total_levels = len(sorted_levels)

        for i, level_num in enumerate(sorted_levels):
            level_count = len(levels_dict[level_num])
            current_group_levels.append(level_num)
            current_group_count += level_count

            if current_group_count >= zip_threshold and (i + 1) < total_levels:
                zip_groups.append({
                    'start': current_start_level,
                    'end': current_group_levels[-1],
                    'levels': current_group_levels.copy()
                })
                current_group_levels = []
                current_group_count = 0
                current_start_level = sorted_levels[i + 1] if (i + 1) < total_levels else level_num

        if current_group_levels:
            zip_groups.append({
                'start': current_start_level,
                'end': current_group_levels[-1],
                'levels': current_group_levels
            })

        for group in zip_groups:
            ext = "zip" if zip_archive else "csv"
            filename = f"expressions({group['start']}-{group['end']}).{ext}"
            filepath = os.path.join(output_dir, filename)

            # Сначала формируем содержимое CSV
            csv_lines = []
            for level_num in group['levels']:
                for item in levels_dict[level_num]:
                    row = parse_block_to_csv_row(item['block'], item['id_name'], include_id_column=include_id_column)
                    csv_lines.append(f"{row}\n")
            csv_content = "".join(csv_lines)

            if zip_archive:
                with zipfile.ZipFile(filepath, 'w', zipfile.ZIP_DEFLATED) as zf:
                    csv_inside_name = f"expressions({group['start']}-{group['end']}).csv"
                    zf.writestr(csv_inside_name, csv_content)
            else:
                with open(filepath, 'w', encoding='utf-8') as f:
                    f.write(csv_content)

    # statistics.txt
    if generate_statistics:
        with open(os.path.join(output_dir, 'statistics.txt'), 'w', encoding='utf-8') as f:
            if generations:
                all_items = [(b, block_to_id[b]) for gen in generations for (b, _) in gen]
                all_items.sort(key=lambda x: int(x[1][1:]))
                for block, id_name in all_items:
                    m = block_metadata[block]
                    ctx = m['contexts']
                    typ = "uw" if 'assignment' in ctx and 'json' in ctx else ("w" if 'assignment' in ctx else "u")
                    f.write(f"{id_name} - {m['count']} - {typ} - level={m['level']}\n")
            else:
                f.write("Блоков не найдено\n")

    # parallel_plan.csv
    if generate_parallel_plan:
        parallel_plan_filename = f"parallel_plan(height={total_passes}_quasi-width={max_width}).csv"
        with open(os.path.join(output_dir, parallel_plan_filename), 'w', encoding='utf-8') as f:
            if generations:
                for i, gen in enumerate(generations, 1):
                    f.write(f"{i};{len(gen)}\n")
            else:
                f.write("0;0\n")

    # Выходной файл (structure.txt)
    with open(os.path.join(output_dir, output_filename), 'w', encoding='utf-8') as f:
        for line in working_lines:
            f.write(line + '\n')

    print(f"\nВРЕМЯ:  ")
    for i, t in enumerate(pass_times, 1):
        print(f"  Проход {i}: {t:.4f} сек ")
    print(f"  Всего: {len(pass_times)} проходов, общее время: {total_time:.4f} сек ")

    print(f"\nСозданы файлы в папке: {output_dir} ")
    print(f"  - structure.txt ({os.path.getsize(os.path.join(output_dir, 'structure.txt'))} байт)  ")
    if generate_statistics:
        print(f"  - statistics.txt ({os.path.getsize(os.path.join(output_dir, 'statistics.txt'))} байт)  ")
    if generate_parallel_plan:
        print(f"  - parallel_plan(...) ({os.path.getsize(os.path.join(output_dir, parallel_plan_filename))} байт)  ")

    # Вывод информации о выражениях
    if generations:
        expr_count = 0
        for f_name in os.listdir(output_dir):
            if f_name.startswith("expressions(") and (f_name.endswith(".zip") or f_name.endswith(".csv")):
                expr_count += 1
                f_type = "ZIP" if zip_archive else "CSV"
                print(f"  - {f_name} ({f_type}, {os.path.getsize(os.path.join(output_dir, f_name))} байт)  ")
        print(f"  Всего файлов выражений: {expr_count} ")


class ProcessingGUI:
    def __init__(self, root):
        self.root = root
        self.root.title("Обработка файлов выражений")
        self.root.geometry("560x590")
        self.root.resizable(False, False)

        self.input_path = tk.StringVar()
        self.output_base_dir = tk.StringVar(value=os.path.join(os.path.expanduser("~"), "Desktop"))
        self.min_occurrences = tk.IntVar(value=1)
        self.zip_threshold = tk.IntVar(value=1)
        self.var_zip_archive = tk.BooleanVar(value=True)

        self.create_widgets()

    def create_widgets(self):
        title_label = tk.Label(self.root, text="Настройки обработки", font=("Arial", 14, "bold"))
        title_label.pack(pady=10)

        frame_path = tk.Frame(self.root)
        frame_path.pack(pady=5, padx=20, fill='x')
        tk.Label(frame_path, text="Входной путь:  ").pack(side='left')
        tk.Entry(frame_path, textvariable=self.input_path, width=30).pack(side='left', padx=5)
        tk.Button(frame_path, text="Обзор...", command=self.browse_input).pack(side='left')

        frame_output = tk.Frame(self.root)
        frame_output.pack(pady=5, padx=20, fill='x')
        tk.Label(frame_output, text="Папка для результатов:  ").pack(side='left')
        tk.Entry(frame_output, textvariable=self.output_base_dir, width=30).pack(side='left', padx=5)
        tk.Button(frame_output, text="Обзор...", command=self.browse_output).pack(side='left')

        frame_min = tk.Frame(self.root)
        frame_min.pack(pady=5, padx=20, fill='x')
        tk.Label(frame_min, text="Мин. вхождений выражений:  ").pack(side='left')
        tk.Spinbox(frame_min, from_=1, to=100, textvariable=self.min_occurrences, width=5).pack(side='left', padx=5)

        frame_zip_thr = tk.Frame(self.root)
        frame_zip_thr.pack(pady=5, padx=20, fill='x')
        tk.Label(frame_zip_thr, text="Порог строк в файле таблицы выражений:  ").pack(side='left')
        tk.Spinbox(frame_zip_thr, from_=1, to=10000, increment=1, textvariable=self.zip_threshold, width=5).pack(
            side='left', padx=5)

        self.var_statistics = tk.BooleanVar(value=True)
        self.var_parallel = tk.BooleanVar(value=True)
        self.var_id_column = tk.BooleanVar(value=True)

        frame_options = tk.LabelFrame(self.root, text="Опции вывода", padx=10, pady=10)
        frame_options.pack(pady=10, padx=20, fill='x')

        tk.Checkbutton(frame_options, text="Формировать statistics.txt", variable=self.var_statistics).pack(anchor='w')
        tk.Checkbutton(frame_options, text="Формировать parallel_plan.csv", variable=self.var_parallel).pack(anchor='w')
        tk.Checkbutton(frame_options, text="Добавлять столбец ID в expressions (в файле/ZIP)",
                       variable=self.var_id_column).pack(anchor='w')

        tk.Checkbutton(frame_options, text="Архивировать файлы expressions в ZIP", variable=self.var_zip_archive).pack(
            anchor='w')

        tk.Label(frame_options, text="(Выходной файл: structure.txt, выражения: CSV или ZIP)", font=("Arial", 8),
                 fg="gray").pack(anchor='w')

        frame_buttons = tk.Frame(self.root)
        frame_buttons.pack(pady=20)
        tk.Button(frame_buttons, text="Запустить обработку", command=self.run_processing, bg="#4CAF50", fg="white",
                  width=20).pack(side='left', padx=10)
        tk.Button(frame_buttons, text="Выход", command=self.root.quit, bg="#f44336", fg="white", width=10).pack(
            side='left', padx=10)

        self.status_label = tk.Label(self.root, text="  ", fg="blue")
        self.status_label.pack(pady=5)

    def browse_input(self):
        path = filedialog.askopenfilename(title="Выберите файл или папку",
                                          filetypes=[("Text files", "*.txt"), ("All files", "*.*")])
        if path:
            self.input_path.set(path)

    def browse_output(self):
        path = filedialog.askdirectory(title="Выберите папку для сохранения результатов")
        if path:
            self.output_base_dir.set(path)

    def run_processing(self):
        input_path = self.input_path.get()
        if not input_path:
            messagebox.showwarning("Предупреждение", "Пожалуйста, выберите входной файл или папку!")
            return
        if not os.path.exists(input_path):
            messagebox.showerror("Ошибка", f"Путь '{input_path}' не найден!")
            return

        output_base_dir = self.output_base_dir.get()
        if not output_base_dir:
            messagebox.showwarning("Предупреждение", "Пожалуйста, выберите папку для результатов!")
            return

        self.status_label.config(text="Обработка... Пожалуйста, подождите.")
        self.root.update()

        try:
            process_files_from_folder(
                input_path=input_path,
                output_base_dir=output_base_dir,
                min_occurrences=self.min_occurrences.get(),
                generate_statistics=self.var_statistics.get(),
                generate_parallel_plan=self.var_parallel.get(),
                include_id_column=self.var_id_column.get(),
                zip_threshold=self.zip_threshold.get(),
                zip_archive=self.var_zip_archive.get()
            )
            self.status_label.config(text="Обработка завершена успешно!")
            messagebox.showinfo("Готово", "Обработка завершена успешно!\nПроверьте созданные файлы.")
        except Exception as e:
            self.status_label.config(text="Ошибка обработки!")
            messagebox.showerror("Ошибка", f"Произошла ошибка:\n{str(e)}")


if __name__ == "__main__":
    root = tk.Tk()
    app = ProcessingGUI(root)
    root.mainloop()