diff --git a/History/2026-08-24-interrupt-eta-implemented.md b/History/2026-08-24-interrupt-eta-implemented.md new file mode 100644 index 0000000..4370e33 --- /dev/null +++ b/History/2026-08-24-interrupt-eta-implemented.md @@ -0,0 +1,49 @@ +# Реализовано: TTL-фикс + прерывание с сохранением + ETA + UI-таблица (v0.0.63) + +_2026-08-24. По плану `2026-08-24-implementation-plan-for-flash.md`. Код написан (роль Flash)._ + +## Что сделано + +### session.py — TTL-фикс + отмена +- `touch(sid)`, `pause_ttl(sid)`, `resume_ttl(sid)` — продление/пауза/возобновление TTL. +- В сессии `"cancel": threading.Event()`; `request_cancel(sid)`, `get_cancel_event(sid)`. + +### scanner.py — прогресс и мягкая остановка LLM +- `class CancelRequested`. +- `scan_llm_ner(..., cancel_event, file_progress)`: отмена ТОЛЬКО между файлами (текущий добирается); + события `file_start{chars,chunks}`, `file_chunk{chunks_done,chunks_total}`, `file_done{elapsed}`. + +### obfuscator.py — частичный результат +- `obfuscate()` возвращает `(zip, csv, meta)`: `meta=None` (штатно) или + `{"cancelled": True, "processed", "total"}` (прерывание). +- Трекинг `llm_done` (файлы с завершённым LLM). При отмене в LLM — в результат идут + ТОЛЬКО файлы из `llm_done` (их обфускация корректна); при отмене в фазе замены — + стоп после текущего файла. Отмена в replace НЕ проверяется, если LLM уже прерван + (файлы из llm_done добираются). +- Событие `extract_done{total_chars, per_file}` (объём текста для ETA). + +### api_bp.py — API +- `POST /api/cancel/`. +- `process_stream`: `pause_ttl` в начале / `resume_ttl` в `finally`; передача `cancel_event` + и `file_progress` в воркер; события `extract_done/file_start/file_chunk/file_done/cancelled`; + heartbeat `llm` расширен: `eta_sec, done_chars, total_chars`; per-file ETA в `file_chunk`. +- При закрытии вкладки (`disconnect`) ставится и `cancel`, и `cancel_event` (воркер останавливается). +- legacy `process()` — фикс 3-значного возврата + TTL. + +### index.html — UI +- Кнопка «⏹ Прервать» + модал-подтверждение (объясняет, что сохранится). +- Таблица 3 секций: «✓ Обработанные» (факт время) / «▶ Текущий файл» (прошло / ~осталось) / + «○ Ожидают» (~оценка из скорости текущего файла). +- Live-блок: «осталось ~X» (глобальная ETA), текущий файл с per-file ETA. +- SSE-обработчики: extract_done/file_start/file_chunk/file_done/cancelled; done различает + пропущенные (до extract_done) и готовые (после). + +## Проверка +- `py_compile` всех .py — OK; `node --check` (JS из index.html) — OK; `get_errors` — нет. +- Локальные тесты логики отмены (FakeLLM): штатно=4 файла; отмена в LLM=3 сохранено; + отмена до LLM=0..1 (текущий добирается); отмена после завершения=полный. +- Смоук-тест: приложение создаётся, все роуты включая `/api/cancel/` зарегистрированы. + +## ВАЖНО +- Версия 0.0.63. **Код не задеплоен** — нужен редеплой на кластере и прогон реальных тестов + (долгий прогон + прерывание). diff --git a/drhider/obfuscator.py b/drhider/obfuscator.py index cd3ad40..1b9385f 100644 --- a/drhider/obfuscator.py +++ b/drhider/obfuscator.py @@ -88,8 +88,9 @@ class TwoPassObfuscator: def obfuscate( self, files: List[Tuple[str, bytes, str]], - progress_cb: Optional[Callable[[str, int, int, str], None]] = None - ) -> Tuple[bytes, str]: + progress_cb: Optional[Callable[[str, int, int, str], None]] = None, + cancel_event=None, file_progress=None + ) -> Tuple[bytes, str, Optional[dict]]: """Обфусцировать список файлов. Все форматы → Markdown → замена → результат. @@ -99,16 +100,34 @@ class TwoPassObfuscator: progress_cb: Опциональный коллбек (phase, idx, total, fname, elapsed), где phase ∈ {"start", "done"}, idx — 0-based индекс, elapsed — время обработки конкретного файла (сек). + cancel_event: threading.Event — мягкая остановка. Проверяется между + файлами LLM и между файлами замены; текущий файл добирается. + file_progress: callable(event, fname, **fields) — прогресс по файлам/чанкам. + События: extract_done{total_chars}, file_start{chars,chunks,idx}, + file_chunk{chunks_done,chunks_total,idx}, file_done{elapsed,idx}. Returns: - (zip_bytes, csv_string): - zip_bytes — ZIP-архив с обфусцированными .md файлами + mapping.csv + (zip_bytes, csv_string, meta): + zip_bytes — ZIP-архив с обфусцированными .md файлами (без mapping.csv) csv_string — содержимое mapping.csv как строка + meta — None при штатном завершении; {"cancelled": True, "processed", "total"} + при прерывании (частичный результат) """ # ── Предобработка: распаковать ZIP + уникализировать имена ── files = extractor.expand_zips(files) files = _dedupe_file_names(files) + # имя файла (внутр. ключ all_texts) -> idx в files (для событий file_progress) + name_to_idx = {fname: i for i, (fname, _, _) in enumerate(files)} + llm_done: set = set() # файлы, чей LLM-анализ ПОЛНОСТЬЮ завершён + + def _fp(event, fname, **fields): + if event == "file_done": + llm_done.add(fname) + if file_progress: + fields["idx"] = name_to_idx.get(fname) + file_progress(event, fname, **fields) + try: # ── Проход 1: сбор сущностей ── # Извлекаем Markdown из каждого файла @@ -142,9 +161,23 @@ class TwoPassObfuscator: scanner.scan_regex(text, self._mapping, self._counters) file_times[i] += time.time() - t0 + # Объём извлечённого текста — для оценки времени (глобальной ETA) + total_chars = sum(len(t) for t in all_texts.values()) + if file_progress: + file_progress("extract_done", None, total_chars=total_chars, + per_file={fname: len(t) for fname, t in all_texts.items()}) + # LLM-сканирование (получает уже найденное regex'ом чтобы не дублировать) + cancelled_llm = False if self._llm_client: - scanner.scan_llm_ner(all_texts, self._mapping, self._llm_client, self._counters) + try: + scanner.scan_llm_ner( + all_texts, self._mapping, self._llm_client, self._counters, + cancel_event=cancel_event, file_progress=_fp, + ) + except scanner.CancelRequested: + log.info("obfuscate: LLM cancelled, llm_done=%d", len(llm_done)) + cancelled_llm = True # Предсортировать ключи один раз (по убыванию длины) self._sorted_keys = sorted( @@ -154,6 +187,7 @@ class TwoPassObfuscator: self._compiled_re = replacer._build_combined_re(self._sorted_keys) # ── Проход 2: замена сущностей ── + cancelled = cancelled_llm results: List[Tuple[str, bytes]] = [] for i, (in_fname, content, ctype) in enumerate(files): @@ -166,6 +200,17 @@ class TwoPassObfuscator: progress_cb("done", i, total, display_name, 0.0) continue + # При прерывании в LLM — в результат идут ТОЛЬКО файлы с завершённым LLM + if cancelled_llm and fname not in llm_done: + continue + + # Мягкая остановка в фазе замены (между файлами). + # НЕ проверяем, если LLM уже прерван: файлы из llm_done добираем (сохраняем всё готовое). + if not cancelled_llm and cancel_event is not None and cancel_event.is_set(): + log.info("obfuscate: cancelled in replace phase, processed=%d", len(results)) + cancelled = True + break + t0 = time.time() obf_content = content # По умолчанию — без изменений @@ -198,7 +243,11 @@ class TwoPassObfuscator: csv_str = builder.build_mapping_csv(self._mapping) zip_data = builder.build_zip(results, csv_str) - return zip_data, csv_str + meta = None + if cancelled: + expected_total = total - len(skipped) + meta = {"cancelled": True, "processed": len(results), "total": expected_total} + return zip_data, csv_str, meta finally: # Очистка состояния (обфускатор может использоваться повторно) @@ -214,8 +263,9 @@ class TwoPassObfuscator: def obfuscate_files( files: List[Tuple[str, bytes, str]], llm_client=None, - progress_cb: Optional[Callable[[str, int, int, str], None]] = None -) -> Tuple[bytes, str]: + progress_cb: Optional[Callable[[str, int, int, str], None]] = None, + cancel_event=None, file_progress=None +) -> Tuple[bytes, str, Optional[dict]]: """Обфусцировать список файлов — удобная функция. Создаёт экземпляр TwoPassObfuscator и вызывает .obfuscate(). @@ -224,9 +274,12 @@ def obfuscate_files( files: [(filename, content_bytes, content_type), ...] llm_client: Опциональный LLM-клиент progress_cb: Опциональный коллбек (phase, idx, total, fname) + cancel_event: threading.Event — мягкая остановка (см. TwoPassObfuscator.obfuscate) + file_progress: callable(event, fname, **fields) — прогресс по файлам/чанкам Returns: - (zip_bytes, csv_string) + (zip_bytes, csv_string, meta) """ obf = TwoPassObfuscator(llm_client=llm_client) - return obf.obfuscate(files, progress_cb=progress_cb) + return obf.obfuscate(files, progress_cb=progress_cb, + cancel_event=cancel_event, file_progress=file_progress) diff --git a/drhider/scanner.py b/drhider/scanner.py index 08801e0..283b005 100644 --- a/drhider/scanner.py +++ b/drhider/scanner.py @@ -12,6 +12,7 @@ import re import json import random +import time import logging from concurrent.futures import ThreadPoolExecutor from typing import Dict, List, Optional @@ -111,6 +112,9 @@ def scan_regex(text: str, mapping: Dict[str, str], counters: Dict[str, int]) -> # LLM-NER: пофайловый чанкинг (Sonnet-схема) # ═══════════════════════════════════════════════════════════════════════════ +class CancelRequested(Exception): + """Обработка прервана пользователем (мягкая остановка между файлами LLM).""" + # Приоритет границ для чанков (от предпочтительных к жёстким) _CHUNK_BOUNDARIES = ['\n\n', '\n', '. ', '? ', '! ', '; ', ', ', ' '] _CHUNK_SIZE = 6000 # символов на чанк (≈1500-2000 токенов RU — NER-качество) @@ -216,11 +220,12 @@ def _call_llm(text: str, llm_client) -> List[dict]: def scan_llm_ner(all_texts: Dict[str, str], mapping: Dict[str, str], - llm_client, counters: Dict[str, int]) -> None: + llm_client, counters: Dict[str, int], + cancel_event=None, file_progress=None) -> None: """Универсальное LLM-обнаружение приватных данных (пофайлово, целиком). Каждый файл обрабатывается ПОЛНОСТЬЮ: текст разбивается на чанки - (_CHUNK_SIZE=6000, overlap=_CHUNK_OVERLAP), чанки обрабатываются параллельно + (_CHUNK_SIZE=6000, overlap=_CHUNK_OVERLAP), чанки обрабатываются последовательно (_LLM_CONCURRENCY), сущности дедуплицируются и верифицируются против полного текста файла (отсечка галлюцинаций). regex-найденное не дублируется. @@ -229,6 +234,8 @@ def scan_llm_ner(all_texts: Dict[str, str], mapping: Dict[str, str], mapping: Словарь замен (мутабельный, пополняется) llm_client: Объект с методом .complete(prompt) -> str counters: Глобальные счётчики токенов (мутабельный) + cancel_event: threading.Event — мягкая остановка между файлами (текущий добирается до конца) + file_progress: callable(event, fname, **fields) — прогресс по файлам/чанкам """ already_found = list(mapping.keys()) @@ -238,10 +245,19 @@ def scan_llm_ner(all_texts: Dict[str, str], mapping: Dict[str, str], ) for fname, full_text in file_items: + # Мягкая остановка ТОЛЬКО между файлами — текущий файл добирается до конца + if cancel_event is not None and cancel_event.is_set(): + raise CancelRequested() + if not full_text or full_text.startswith("[DOC binary"): continue chunks = split_into_chunks(full_text) + if file_progress: + file_progress("file_start", fname, chars=len(full_text), chunks=len(chunks)) + + t0 = time.time() + # Параллельные вызовы LLM по чанкам этого файла if len(chunks) > 1 and _LLM_CONCURRENCY > 1: with ThreadPoolExecutor(max_workers=_LLM_CONCURRENCY) as _ex: @@ -252,8 +268,11 @@ def scan_llm_ner(all_texts: Dict[str, str], mapping: Dict[str, str], all_ent.extend(fut.result()) else: all_ent = [] - for c in chunks: + for k, c in enumerate(chunks): all_ent.extend(_call_llm(_build_llm_prompt(c, mapping.keys()), llm_client)) + if file_progress: + file_progress("file_chunk", fname, + chunks_done=k + 1, chunks_total=len(chunks)) # Дедуп + верификация + добавление в mapping seen: set = set() @@ -270,3 +289,6 @@ def scan_llm_ner(all_texts: Dict[str, str], mapping: Dict[str, str], continue ent_type = ent.get("type", "").strip().lower().replace(" ", "_") mapping[val] = _next_token(ent_type, counters) + + if file_progress: + file_progress("file_done", fname, elapsed=round(time.time() - t0, 2)) diff --git a/site/app.py b/site/app.py index 53bd4af..2a6558a 100644 --- a/site/app.py +++ b/site/app.py @@ -21,7 +21,7 @@ if _sys_path_root not in sys.path: sys.path.insert(0, _sys_path_root) # Версия приложения (меняется при изменениях) -VERSION = "0.0.62" +VERSION = "0.0.63" def setup_logging(): diff --git a/site/routes/api_bp.py b/site/routes/api_bp.py index 52ae7ef..61214b5 100644 --- a/site/routes/api_bp.py +++ b/site/routes/api_bp.py @@ -13,6 +13,7 @@ import io import json import queue import threading +import time import zipfile import traceback import logging @@ -23,7 +24,8 @@ from flask import Blueprint, request, send_file, jsonify, Response, stream_with_ from drhider import obfuscate_files, LLMClient from session import (create_session, add_file, get_files, store_result, get_result, store_csv, get_csv, cleanup, file_count, - MAX_FILE_BYTES) + MAX_FILE_BYTES, pause_ttl, resume_ttl, + request_cancel, get_cancel_event) api_bp = Blueprint("api", __name__, url_prefix="/api") log = logging.getLogger("routes.api_bp") @@ -142,6 +144,20 @@ def session_files(sid): }) +@api_bp.route("/cancel/", methods=["POST"]) +def cancel(sid): + """Запросить мягкое прерывание обработки сессии. + + Воркер останавливается на ближайшей границе файла (текущий добирается), + собирает частичный результат (готовые файлы + mapping) и шлёт SSE-событие `cancelled`. + """ + if request_cancel(sid): + log.info("cancel: requested sid=%s", sid) + return jsonify({"ok": True}), 200 + log.warning("cancel: session not found sid=%s", sid) + return jsonify({"ok": False, "error": "Session not found"}), 404 + + @api_bp.route("/process_stream/", methods=["GET"]) def process_stream(sid): """SSE: process all session files, streaming per-file progress. @@ -149,7 +165,7 @@ def process_stream(sid): Все файлы обрабатываются ЕДИНЫМ вызовом obfuscate_files (общий mapping, согласованные токены). Обработка идёт в отдельном потоке; прогресс передаётся через очередь. Разрыв соединения клиента корректно - перехватывается и останавливает генератор. + перехватывается и останавливает генератор (и воркер — через cancel_event). """ files = get_files(sid) if files is None: @@ -159,30 +175,60 @@ def process_stream(sid): all_files = [(fname, content, "") for fname, content in files] log.info("process_stream: start sid=%s files=%d", sid, len(all_files)) + # Сессия живёт, пока идёт обработка (TTL возобновляется в finally генератора) + pause_ttl(sid) def generate(): llm = LLMClient() q = queue.Queue() - cancel = threading.Event() + cancel = threading.Event() # локальный: разрыв клиента (стоп heartbeat) + cancel_event = get_cancel_event(sid) # из сессии: мягкая отмена (кнопка «Прервать») + + # Состояние для глобальной ETA (символы) + eta = {"total_chars": 0, "done_chars": 0, "cur_chars": 0, "cur_total": 0, "cur_done": 0} + _TOKENS_PER_CHAR = 8.0 # эмпирический коэффициент символов -> токенов LLM def progress(phase, idx, total_, name, elapsed): q.put(("progress", phase, idx, name, total_, elapsed)) + # Состояние для per-file ETA (чанки) + fstate = {"t0": 0.0, "total": 0, "done": 0} + + def file_progress(event, fname, **fields): + if event == "file_start": + fstate["t0"] = time.time() + fstate["total"] = fields.get("chunks", 0) + fstate["done"] = 0 + elif event == "file_chunk": + fstate["done"] = fields.get("chunks_done", 0) + elapsed = time.time() - fstate["t0"] + rate = fstate["done"] / elapsed if elapsed > 0 else 0 + rem = fstate["total"] - fstate["done"] + if rate > 0: + fields["eta_sec"] = max(0, round(rem / rate)) + q.put(("file", event, fname, fields)) + def worker(): log.info("worker: start sid=%s files=%d", sid, len(all_files)) t0 = datetime.utcnow() try: - zip_data, csv_str = obfuscate_files( - all_files, llm_client=llm, progress_cb=progress + zip_data, csv_str, meta = obfuscate_files( + all_files, llm_client=llm, progress_cb=progress, + cancel_event=cancel_event, file_progress=file_progress, ) stats = { "tokens": llm.tokens_total, "llm_sec": round(llm.llm_sec, 1), } dt = (datetime.utcnow() - t0).total_seconds() - log.info("worker: done sid=%s in %.1fs tokens=%d llm_sec=%.1f zip_len=%d", - sid, dt, llm.tokens_total, llm.llm_sec, len(zip_data)) - q.put(("result", zip_data, csv_str, stats)) + if meta and meta.get("cancelled"): + log.info("worker: cancelled sid=%s in %.1fs processed=%d/%d", + sid, dt, meta.get("processed", 0), meta.get("total", 0)) + q.put(("cancelled", zip_data, csv_str, stats, meta)) + else: + log.info("worker: done sid=%s in %.1fs tokens=%d llm_sec=%.1f zip_len=%d", + sid, dt, llm.tokens_total, llm.llm_sec, len(zip_data)) + q.put(("result", zip_data, csv_str, stats)) except Exception as e: log.error("worker: exception sid=%s: %r\n%s", sid, e, traceback.format_exc()) q.put(("error", repr(e))) @@ -190,70 +236,133 @@ def process_stream(sid): threading.Thread(target=worker, daemon=True).start() log.debug("process_stream: worker thread started sid=%s", sid) - while True: - try: - evt = q.get(timeout=1) - except queue.Empty: - if cancel.is_set(): - log.info("process_stream: cancelled sid=%s (generator exit)", sid) - return - # Heartbeat: живая статистика LLM (для таймера в UI) + try: + while True: try: - yield ( - f"event: llm\n" - f"data: {json.dumps({'active': llm.llm_active, 'elapsed': round(llm.llm_elapsed_now(), 1), 'tokens': llm.tokens_total})}\n\n" - ) - except _disconnect_exceptions() as e: - log.warning("process_stream: disconnect during heartbeat sid=%s err=%r", sid, e) - cancel.set() - return - continue + evt = q.get(timeout=1) + except queue.Empty: + if cancel.is_set(): + log.info("process_stream: cancelled sid=%s (generator exit)", sid) + return + # Heartbeat: живая статистика LLM + глобальная ETA + tokens = llm.tokens_total + elapsed = llm.llm_elapsed_now() + eta_sec = None + if eta["total_chars"] > 0 and tokens > 0 and elapsed > 0: + est_total_tokens = eta["total_chars"] * _TOKENS_PER_CHAR + rate = tokens / elapsed + if rate > 0: + eta_sec = max(0, int((est_total_tokens - tokens) / rate)) + done_chars = eta["done_chars"] + if eta["cur_total"] > 0: + done_chars += int(eta["cur_chars"] * (eta["cur_done"] / eta["cur_total"])) + try: + yield ( + f"event: llm\n" + f"data: {json.dumps({'active': llm.llm_active, 'elapsed': round(elapsed, 1), 'tokens': tokens, 'eta_sec': eta_sec, 'done_chars': done_chars, 'total_chars': eta['total_chars']})}\n\n" + ) + except _disconnect_exceptions() as e: + log.warning("process_stream: disconnect during heartbeat sid=%s err=%r", sid, e) + cancel.set() + if cancel_event: + cancel_event.set() + return + continue - kind = evt[0] + kind = evt[0] - if kind == "progress": - _, phase, idx, name, total_, elapsed = evt - log.debug("process_stream: event=%s idx=%d name=%r elapsed=%s sid=%s", - phase, idx, name, elapsed, sid) - try: - yield ( - f"event: {phase}\n" - f"data: {json.dumps({'idx': idx, 'name': name, 'total': total_, 'elapsed': elapsed})}\n\n" - ) - except _disconnect_exceptions() as e: - log.warning("process_stream: disconnect on progress sid=%s phase=%s err=%r", sid, phase, e) - cancel.set() + if kind == "progress": + _, phase, idx, name, total_, elapsed = evt + log.debug("process_stream: event=%s idx=%d name=%r elapsed=%s sid=%s", + phase, idx, name, elapsed, sid) + try: + yield ( + f"event: {phase}\n" + f"data: {json.dumps({'idx': idx, 'name': name, 'total': total_, 'elapsed': elapsed})}\n\n" + ) + except _disconnect_exceptions() as e: + log.warning("process_stream: disconnect on progress sid=%s phase=%s err=%r", sid, phase, e) + cancel.set() + if cancel_event: + cancel_event.set() + return + + elif kind == "file": + _, event, fname, fields = evt + # Обновляем состояние для глобальной ETA + if event == "extract_done": + eta["total_chars"] = fields.get("total_chars", 0) + elif event == "file_start": + eta["cur_chars"] = fields.get("chars", 0) + eta["cur_total"] = fields.get("chunks", 0) + eta["cur_done"] = 0 + elif event == "file_chunk": + eta["cur_done"] = fields.get("chunks_done", 0) + elif event == "file_done": + eta["done_chars"] += eta["cur_chars"] + eta["cur_chars"] = 0 + eta["cur_total"] = 0 + eta["cur_done"] = 0 + try: + yield ( + f"event: {event}\n" + f"data: {json.dumps({'name': fname, **fields})}\n\n" + ) + except _disconnect_exceptions() as e: + log.warning("process_stream: disconnect on file event sid=%s err=%r", sid, e) + cancel.set() + if cancel_event: + cancel_event.set() + return + + elif kind == "result": + _, zip_data, csv_str, stats = evt + log.info("process_stream: result sid=%s, storing result", sid) + store_result(sid, zip_data) + if csv_str: + store_csv(sid, csv_str) + count = 0 + with zipfile.ZipFile(io.BytesIO(zip_data)) as zf: + count = len([n for n in zf.namelist() if n != "mapping.csv"]) + log.info("process_stream: complete sid=%s count=%d stats=%r", sid, count, stats) + try: + yield ( + f"event: complete\n" + f"data: {json.dumps({'total': count, **stats})}\n\n" + ) + except _disconnect_exceptions() as e: + log.warning("process_stream: disconnect on complete sid=%s err=%r", sid, e) + return return - elif kind == "result": - _, zip_data, csv_str, stats = evt - log.info("process_stream: result sid=%s, storing result", sid) - store_result(sid, zip_data) - if csv_str: - store_csv(sid, csv_str) - count = 0 - with zipfile.ZipFile(io.BytesIO(zip_data)) as zf: - count = len([n for n in zf.namelist() if n != "mapping.csv"]) - log.info("process_stream: complete sid=%s count=%d stats=%r", sid, count, stats) - try: - yield ( - f"event: complete\n" - f"data: {json.dumps({'total': count, **stats})}\n\n" - ) - except _disconnect_exceptions() as e: - log.warning("process_stream: disconnect on complete sid=%s err=%r", sid, e) + elif kind == "cancelled": + _, zip_data, csv_str, stats, meta = evt + log.info("process_stream: cancelled sid=%s, storing partial result", sid) + store_result(sid, zip_data) + if csv_str: + store_csv(sid, csv_str) + try: + yield ( + f"event: cancelled\n" + f"data: {json.dumps({'saved': meta.get('processed', 0), 'total': meta.get('total', 0), **stats})}\n\n" + ) + except _disconnect_exceptions() as e: + log.warning("process_stream: disconnect on cancelled sid=%s err=%r", sid, e) + return return - return - elif kind == "error": - _, msg = evt - log.error("process_stream: error event sid=%s msg=%r", sid, msg) - try: - yield f"event: error\ndata: {json.dumps({'error': msg})}\n\n" - except _disconnect_exceptions() as e: - log.warning("process_stream: disconnect on error sid=%s err=%r", sid, e) + elif kind == "error": + _, msg = evt + log.error("process_stream: error event sid=%s msg=%r", sid, msg) + try: + yield f"event: error\ndata: {json.dumps({'error': msg})}\n\n" + except _disconnect_exceptions() as e: + log.warning("process_stream: disconnect on error sid=%s err=%r", sid, e) + return return - return + finally: + resume_ttl(sid) + log.debug("process_stream: generator exit sid=%s, TTL resumed", sid) return Response( stream_with_context(generate()), @@ -273,7 +382,11 @@ def process(sid): try: llm = LLMClient() all_files = [(fname, content, "") for fname, content in files] - zip_data, csv_str = obfuscate_files(all_files, llm_client=llm) + pause_ttl(sid) + try: + zip_data, csv_str, _ = obfuscate_files(all_files, llm_client=llm) + finally: + resume_ttl(sid) store_result(sid, zip_data) if csv_str: store_csv(sid, csv_str) diff --git a/site/session.py b/site/session.py index db2f7d0..3fe5139 100644 --- a/site/session.py +++ b/site/session.py @@ -42,6 +42,58 @@ def _start_timer(sid: str): return timer +def touch(sid: str): + """Продлить жизнь сессии: перезапустить TTL-таймер (если сессия существует).""" + with _lock: + s = _sessions.get(sid) + if not s: + return + if s.get("timer"): + s["timer"].cancel() + s["timer"] = _start_timer(sid) + + +def pause_ttl(sid: str): + """Приостановить TTL сессии (во время обработки): сессия живёт, пока идёт воркер.""" + with _lock: + s = _sessions.get(sid) + if s and s.get("timer"): + s["timer"].cancel() + s["timer"] = None + + +def resume_ttl(sid: str): + """Возобновить TTL сессии (после завершения обработки): результат доступен ещё TTL.""" + with _lock: + s = _sessions.get(sid) + if not s: + return + if s.get("timer"): + s["timer"].cancel() + s["timer"] = _start_timer(sid) + + +def request_cancel(sid: str) -> bool: + """Запросить мягкое прерывание обработки сессии. + + Returns: + True если сессия существует и отмена запрошена, False если нет. + """ + with _lock: + s = _sessions.get(sid) + if not s: + return False + s["cancel"].set() + return True + + +def get_cancel_event(sid: str) -> Optional[threading.Event]: + """Получить событие отмены сессии (или None, если сессии нет).""" + with _lock: + s = _sessions.get(sid) + return s["cancel"] if s else None + + # ═══════════════════════════════════════════════════════════════════════════ # API # ═══════════════════════════════════════════════════════════════════════════ @@ -57,6 +109,7 @@ def create_session() -> str: _sessions[sid] = { "files": [], "result": None, + "cancel": threading.Event(), "timer": _start_timer(sid), } return sid diff --git a/site/templates/index.html b/site/templates/index.html index de6d61b..88ddedf 100644 --- a/site/templates/index.html +++ b/site/templates/index.html @@ -62,6 +62,12 @@ .row-over td { background: rgba(220,38,38,.10) !important; color:#c0392b; } .row-over .num-cell { color:#c0392b; } .row-over .name-cell { color:#c0392b; } + .row-current td { background: rgba(37,99,235,.12) !important; color:#1d4ed8; font-weight: 600; } + .grp-row td { + background: var(--brand-gray); font-size: 12px; font-weight: 700; + letter-spacing: .03em; text-transform: uppercase; color: var(--muted); + padding: 4px 8px; border: none; + } .remove-btn { cursor: pointer; color: #f87171; background: none; border: none; padding: 2px 4px; font-size: 14px; line-height: 1; @@ -115,6 +121,7 @@ font-size: 18px; font-weight: 700; width: fit-content; } .live-llm.show { display: inline-block; } + .live-eta { margin-top: 8px; font-size: 14px; color: #1d4ed8; font-weight: 600; } @@ -153,6 +160,7 @@
0 файлов +
@@ -165,6 +173,7 @@
—
0.0 с
🤖 ИИ обрабатывает… 0.0 с
+
📊 Итоги обработки
@@ -196,6 +205,13 @@ const db = document.getElementById('dlBtns'); let sf = []; let fileMeta = new Map(); // имя -> {size, mtime} для дедупа/суффиксов let overNames = new Set(); // имена файлов сверх лимита (не участвуют в обфускации) +// ── Состояние обработки (3-секционная таблица: готово / текущий / ожидают) ── +let procPhase = 'idle'; // 'idle' | 'processing' +let procState = {}; // sfIdx -> {st, elapsed, eta, est, chars, t0} +let procNameIdx = {}; // имя (с бэка) -> sfIdx +let procExtractDone = false; // true после extract_done (далее done = реальная готовность) +let procRefresh = null; // setInterval перерисовки таблицы +let ptimer = null, liveRefresh = null; // таймеры статуса и live-блока const MAX_FILE_BYTES = 50 * 1024 * 1024; // 50 МБ на один файл const MAX_SESSION_BYTES = 500 * 1024 * 1024; // 500 МБ суммарно на сессию const VM_UPLOAD_URL = 'https://contracts.kube5s.ru/drhider-upload/'; // ВМ-буфер: PUT больших файлов (шлюз кластера их рвёт) @@ -208,10 +224,18 @@ let activeXHR = null; // активный XHR function resetAll() { if (activeES) { activeES.close(); activeES = null; } if (activeXHR) { activeXHR.abort(); activeXHR = null; } + if (ptimer) { clearInterval(ptimer); ptimer = null; } + if (liveRefresh) { clearInterval(liveRefresh); liveRefresh = null; } + if (procRefresh) { clearInterval(procRefresh); procRefresh = null; } currentSid = ''; sf = []; fileMeta = new Map(); overNames = new Set(); + procPhase = 'idle'; + procState = {}; + procNameIdx = {}; + procExtractDone = false; + document.getElementById('cancelBtn').style.display = 'none'; fi.value = ''; rr(); st.className = ''; @@ -227,7 +251,13 @@ window.addEventListener('beforeunload', () => resetAll()); function fs(b) { return b < 1024 ? b + ' B' : b < 1048576 ? (b / 1024).toFixed(1) + ' KB' : (b / 1048576).toFixed(1) + ' MB'; } +function fmtSec(s) { + s = Math.max(0, Math.round(s)); + return s >= 60 ? Math.floor(s / 60) + 'м ' + (s % 60) + 'с' : s + 'с'; +} + function rr() { + if (procPhase === 'processing') { renderProcTable(); return; } if (sf.length === 0) { fl.innerHTML = 'Нет выбранных файлов'; } else { fl.innerHTML = sf.map((f, i) => { @@ -244,6 +274,91 @@ function rr() { ub.disabled = cntMain === 0; } +// ═══ 3-секционная таблица во время обработки (готово / текущий / ожидают) ═══ +function procRow(i, stTxt) { + const f = sf[i]; + const over = overNames.has(f.name); + const cls = (procState[i] && procState[i].st === 'current') ? ' class="row-current"' + : (over ? ' class="row-over"' : ''); + return '' + f.name + '' + fs(f.size) + '' + stTxt + ''; +} + +function renderProcTable() { + const groups = { done: [], current: [], pending: [] }; + for (let i = 0; i < sf.length; i++) { + const st = procState[i] ? procState[i].st : 'pending'; + if (st === 'done' || st === 'skipped') groups.done.push(i); + else if (st === 'current') groups.current.push(i); + else groups.pending.push(i); + } + // Оценка скорости из текущего файла (сек/символ) — для «ожидающих» + let rate = null; + for (const i of groups.current) { + const p = procState[i]; + const cur = (performance.now() - p.t0) / 1000; + const total = cur + (p.eta != null ? p.eta : 0); + if (p.chars > 0 && total > 0) rate = total / p.chars; + } + const rows = []; + if (groups.done.length) { + rows.push('✓ Обработанные (' + groups.done.length + ')'); + for (const i of groups.done) { + const p = procState[i]; + const txt = p.st === 'skipped' + ? 'пропущен' + : '✓ ' + (p.elapsed ? p.elapsed.toFixed(1) : '0.0') + 'с'; + rows.push(procRow(i, txt)); + } + } + if (groups.current.length) { + rows.push('▶ Текущий файл'); + for (const i of groups.current) { + const p = procState[i]; + const cur = ((performance.now() - p.t0) / 1000).toFixed(1); + const eta = (p.eta != null) ? ' / ~' + fmtSec(p.eta) : ''; + rows.push(procRow(i, '⏳ ' + cur + 'с' + eta + '')); + } + } + if (groups.pending.length) { + rows.push('○ Ожидают обработки (' + groups.pending.length + ')'); + for (const i of groups.pending) { + const p = procState[i] || {}; + if (rate && !p.est && p.chars > 0) p.est = p.chars * rate; + let txt; + if (overNames.has(sf[i].name)) txt = '🔥 не учитывается'; + else if (p.st === 'analyzed') txt = 'анализ ✓'; + else if (p.est != null) txt = '~' + fmtSec(p.est) + ''; + else txt = '—'; + rows.push(procRow(i, txt)); + } + } + fl.innerHTML = rows.join(''); + const cntDone = groups.done.length; + fc.textContent = 'Готово ' + cntDone + ' / ' + sf.length + ' файлов'; +} + +function confirmCancel() { + document.getElementById('cancelModal').style.display = 'flex'; +} + +function doCancel() { + document.getElementById('cancelModal').style.display = 'none'; + if (!currentSid) return; + fetch('/api/cancel/' + currentSid, { method: 'POST' }).catch(() => {}); + st.textContent = '⏹ Остановка… завершаем текущий файл'; +} + +function finishProcUI() { + if (activeES) { activeES.close(); activeES = null; } + if (ptimer) { clearInterval(ptimer); ptimer = null; } + if (liveRefresh) { clearInterval(liveRefresh); liveRefresh = null; } + if (procRefresh) { clearInterval(procRefresh); procRefresh = null; } + document.getElementById('liveBlock').classList.remove('show'); + document.getElementById('cancelBtn').style.display = 'none'; + procPhase = 'idle'; + rr(); +} + function rm(i) { const nm = sf[i].name; overNames.delete(nm); fileMeta.delete(nm); sf.splice(i, 1); const d = new DataTransfer(); sf.forEach(f => d.items.add(f)); fi.files = d.files; rr(); } window.addEventListener('load', () => { sf = []; fileMeta = new Map(); fi.value = ''; rr(); /* прогрев upstream-соединения */ fetch('/health').catch(() => {}); }); @@ -524,12 +639,24 @@ async function uploadFiles() { // Фаза 2: обработка (SSE — прогресс по каждому файлу) // Сброс загрузочных статусов — теперь этап обработки (только отправленные) for (let k = 0; k < total; k++) ss(sendIdx[k], '⏳'); + // Инициализация 3-секционной таблицы (готово / текущий / ожидают) + procPhase = 'processing'; + procState = {}; + procNameIdx = {}; + procExtractDone = false; + for (let k = 0; k < total; k++) { + procState[sendIdx[k]] = { st: 'pending', elapsed: 0, eta: null, est: null, chars: 0, t0: 0 }; + } + document.getElementById('cancelBtn').style.display = 'inline-block'; + if (procRefresh) clearInterval(procRefresh); + procRefresh = setInterval(rr, 1000); // тикающий рендер (elapsed текущего файла) + rr(); const fileTimers = {}; // idx -> performance.now() const fileIntervals = {}; // idx -> setInterval id st.className = 'status progress'; st.textContent = 'Обработка (этап 2/2)…'; const t0 = performance.now(); - const ptimer = setInterval(() => { + ptimer = setInterval(() => { const sec = Math.round((performance.now() - t0) / 1000); st.textContent = 'Обработка (этап 2/2)… ' + sec + 'с'; }, 1000); @@ -540,12 +667,14 @@ async function uploadFiles() { const liveFileEl = document.getElementById('liveFile'); const liveLlm = document.getElementById('liveLlm'); const liveLlmTime = document.getElementById('liveLlmTime'); + const liveEta = document.getElementById('liveEta'); let currentLiveFile = ''; // последнее имя файла из start liveLlm.classList.remove('show'); + liveEta.textContent = ''; liveTimerEl.textContent = '0.0 с'; liveFileEl.textContent = 'Подготовка…'; liveBlock.classList.add('show'); - const liveRefresh = setInterval(() => { + liveRefresh = setInterval(() => { const sec = ((performance.now() - t0) / 1000).toFixed(1); liveTimerEl.textContent = sec + ' с'; }, 200); @@ -556,29 +685,53 @@ async function uploadFiles() { activeES.addEventListener('start', function(e) { const d = JSON.parse(e.data); const idx = sendIdx[d.idx]; // индекс в sf для вывода прогресса - // Таймер тикает ТОЛЬКО у текущего файла; таймеры остальных останавливаем - for (const k in fileIntervals) { - clearInterval(fileIntervals[k]); - delete fileIntervals[k]; - } - fileTimers[idx] = performance.now(); - fileIntervals[idx] = setInterval(function() { - const elapsed = ((performance.now() - fileTimers[idx]) / 1000).toFixed(1); - ss(idx, '⏳ ' + elapsed + 'с'); - }, 200); - // Показываем текущий файл и его размер в live-блоке + procNameIdx[d.name] = idx; + const p = procState[idx]; + if (p) { p.st = 'pending'; p.elapsed = 0; } const f = sf[idx]; const sz = f ? fs(f.size) : ''; currentLiveFile = 'Файл ' + (d.idx + 1) + '/' + total + ': ' + d.name + (sz ? ' (' + sz + ')' : ''); liveFileEl.textContent = currentLiveFile; }); + activeES.addEventListener('extract_done', function(e) { + const d = JSON.parse(e.data); + procExtractDone = true; + // per-file символы -> оценка времени для ожидающих файлов + if (d.per_file) { + for (const nm in d.per_file) { + const idx = procNameIdx[nm]; + if (idx != null && procState[idx]) procState[idx].chars = d.per_file[nm]; + } + } + }); + activeES.addEventListener('file_start', function(e) { + const d = JSON.parse(e.data); + const idx = sendIdx[d.idx]; + const p = procState[idx]; + if (p) { p.st = 'current'; p.t0 = performance.now(); p.eta = null; } + }); + activeES.addEventListener('file_chunk', function(e) { + const d = JSON.parse(e.data); + const idx = sendIdx[d.idx]; + const p = procState[idx]; + if (p && typeof d.eta_sec === 'number') p.eta = d.eta_sec; + liveFileEl.textContent = 'Файл ' + (d.idx + 1) + '/' + total + ': ' + d.name + + (typeof d.eta_sec === 'number' ? ' — осталось ~' + fmtSec(d.eta_sec) : ''); + }); + activeES.addEventListener('file_done', function(e) { + const d = JSON.parse(e.data); + const idx = sendIdx[d.idx]; + const p = procState[idx]; + if (p) { p.st = 'analyzed'; } + }); activeES.addEventListener('llm', function(e) { const d = JSON.parse(e.data); if (d.active) { liveLlmTime.textContent = d.elapsed + ' с'; liveLlm.classList.add('show'); - // LLM анализирует ВСЕ файлы вместе — не показываем имя конкретного liveFileEl.textContent = '🤖 ИИ анализирует все файлы…'; + if (typeof d.eta_sec === 'number' && d.eta_sec >= 0) liveEta.textContent = '⏳ осталось ~' + fmtSec(d.eta_sec); + else liveEta.textContent = ''; } else { liveLlm.classList.remove('show'); if (currentLiveFile) liveFileEl.textContent = currentLiveFile; @@ -587,19 +740,41 @@ async function uploadFiles() { activeES.addEventListener('done', function(e) { const d = JSON.parse(e.data); const idx = sendIdx[d.idx]; - clearInterval(fileIntervals[idx]); + if (!procExtractDone) { + // done на этапе извлечения = битый/пропущенный файл + const p0 = procState[idx]; + if (p0) p0.st = 'skipped'; + return; + } // Время на КОНКРЕТНЫЙ файл приходит с бэка (extract_text + замена, без общего LLM) const sec = (typeof d.elapsed === 'number' && d.elapsed > 0) ? d.elapsed.toFixed(1) : ((performance.now() - (fileTimers[idx] || performance.now())) / 1000).toFixed(1); - ss(idx, '✓ ' + sec + 'с'); + const p = procState[idx]; + if (p) { p.st = 'done'; p.elapsed = d.elapsed > 0 ? d.elapsed : parseFloat(sec); } + else procState[idx] = { st: 'done', elapsed: parseFloat(sec), eta: null, est: null, chars: 0, t0: 0 }; + }); + activeES.addEventListener('cancelled', function(e) { + const d = JSON.parse(e.data); + finishProcUI(); + const totalSec = ((performance.now() - t0) / 1000).toFixed(1); + st.className = 'status done'; + st.textContent = '⏹ Остановлено. Сохранено ' + d.saved + ' из ' + d.total + ' файлов + таблица замен (общее ' + totalSec + 'с)'; + const sb = document.getElementById('statsBlock'); + document.getElementById('stTotalTime').textContent = totalSec + ' с'; + if (d.llm_sec > 0) { + document.getElementById('stLlmTime').textContent = d.llm_sec + ' с'; + document.getElementById('stLlmTokens').textContent = d.tokens > 0 ? d.tokens : '—'; + } else { + document.getElementById('stLlmTime').textContent = '—'; + document.getElementById('stLlmTokens').textContent = '—'; + } + sb.classList.add('show'); + db.classList.add('show'); + resolve(); }); activeES.addEventListener('complete', function(e) { - activeES.close(); - activeES = null; - clearInterval(ptimer); - clearInterval(liveRefresh); - liveBlock.classList.remove('show'); + finishProcUI(); const d = JSON.parse(e.data); const totalSec = ((performance.now() - t0) / 1000).toFixed(1); st.className = 'status done'; @@ -619,18 +794,12 @@ async function uploadFiles() { resolve(); }); activeES.onerror = function() { - activeES.close(); - activeES = null; - clearInterval(ptimer); - clearInterval(liveRefresh); - liveBlock.classList.remove('show'); + finishProcUI(); reject(new Error('SSE connection failed')); }; }); } catch (err) { - clearInterval(ptimer); - clearInterval(liveRefresh); - liveBlock.classList.remove('show'); + finishProcUI(); st.className = 'status error'; st.textContent = 'Ошибка: ' + err.message; } @@ -686,5 +855,19 @@ function downloadCsv() {
+