From 36c9be48bed164711194be53d5fe87bbfdfe994d Mon Sep 17 00:00:00 2001 From: keboss-m <85340750+keboss-m@users.noreply.github.com> Date: Mon, 1 Jun 2026 19:16:23 +0300 Subject: [PATCH] Add document ingestion pipeline, chat analytics modes, and auth fixes Ingest MD/PDF/DOCX/XLSX into org-scoped documents with classify and RAG indexing. Add compare/timeline chat modes and UI upload. Filter WebSocket progress by user ACL and normalize admin project slugs consistently. Co-authored-by: Cursor --- Dockerfile | 3 + Dockerfile.rag | 6 +- backend/auth/models.py | 10 +- backend/auth/routes.py | 11 +- backend/ingest_worker.py | 133 +++++++++++ backend/main.py | 102 +++++++-- backend/paths.py | 15 ++ backend/queue.py | 291 +++++++++++++++++------- backend/static/app.js | 32 ++- backend/static/index.html | 28 ++- backend/static/styles.css | 9 +- config.yaml | 14 ++ requirements.txt | 6 + src/ingest/__init__.py | 5 + src/ingest/classify.py | 92 ++++++++ src/ingest/extractors/__init__.py | 1 + src/ingest/extractors/docx_extractor.py | 32 +++ src/ingest/extractors/pdf_extractor.py | 63 +++++ src/ingest/extractors/text.py | 20 ++ src/ingest/extractors/xlsx_extractor.py | 74 ++++++ src/ingest/formatter.py | 64 ++++++ src/ingest/models.py | 70 ++++++ src/ingest/router.py | 54 +++++ src/rag/query.py | 82 ++++--- tests/test_auth.py | 37 ++- tests/test_ingest.py | 111 +++++++++ 26 files changed, 1198 insertions(+), 167 deletions(-) create mode 100644 backend/ingest_worker.py create mode 100644 src/ingest/__init__.py create mode 100644 src/ingest/classify.py create mode 100644 src/ingest/extractors/__init__.py create mode 100644 src/ingest/extractors/docx_extractor.py create mode 100644 src/ingest/extractors/pdf_extractor.py create mode 100644 src/ingest/extractors/text.py create mode 100644 src/ingest/extractors/xlsx_extractor.py create mode 100644 src/ingest/formatter.py create mode 100644 src/ingest/models.py create mode 100644 src/ingest/router.py create mode 100644 tests/test_ingest.py diff --git a/Dockerfile b/Dockerfile index 119d28a..dd7d141 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,6 +6,9 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ build-essential \ libsndfile1 \ curl \ + tesseract-ocr \ + tesseract-ocr-rus \ + tesseract-ocr-eng \ && rm -rf /var/lib/apt/lists/* # Рабочая директория diff --git a/Dockerfile.rag b/Dockerfile.rag index 9a5f4af..94e4dcf 100644 --- a/Dockerfile.rag +++ b/Dockerfile.rag @@ -8,4 +8,8 @@ RUN pip install --no-cache-dir --timeout 300 \ python-dotenv>=1.0.0 \ sentence-transformers>=3.0.0 \ bcrypt>=4.0.0 \ - "python-jose[cryptography]" + "python-jose[cryptography]" \ + pymupdf>=1.24.0 \ + openpyxl>=3.1.0 \ + Pillow>=10.0.0 \ + pytesseract>=0.3.10 diff --git a/backend/auth/models.py b/backend/auth/models.py index 154325b..333ecb1 100644 --- a/backend/auth/models.py +++ b/backend/auth/models.py @@ -1,7 +1,7 @@ """Auth data models.""" from dataclasses import dataclass, field -from typing import List, Optional +from typing import Any, Dict, List, Optional ROLES = ("admin", "director", "user") @@ -51,3 +51,11 @@ class UserContext: def can_global_search(self) -> bool: return self.has_all_projects_access + + def can_see_task(self, task: Dict[str, Any]) -> bool: + """Whether realtime/API task updates for this task may be shown to the user.""" + if task.get("org_slug") != self.org_slug: + return False + if self.has_all_projects_access: + return True + return task.get("user_id") == self.user_id diff --git a/backend/auth/routes.py b/backend/auth/routes.py index aeca3f7..3719ec8 100644 --- a/backend/auth/routes.py +++ b/backend/auth/routes.py @@ -15,6 +15,7 @@ from backend.auth.service import ( delete_personal_project, get_user_context, list_accessible_projects, + normalize_project_slug, update_user_projects, user_to_dict, ) @@ -134,11 +135,13 @@ async def admin_list_projects(admin: UserContext = Depends(require_admin)): @admin_router.post("/projects") async def admin_create_project(payload: CreateProjectRequest, admin: UserContext = Depends(require_admin)): - slug = payload.slug.strip().lower() - if not slug: - raise HTTPException(status_code=400, detail="slug обязателен") try: - project = db.create_project(admin.org_id, slug, payload.name, owner_user_id=None, config=load_config()) + slug = normalize_project_slug(payload.slug) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + display_name = payload.name.strip() or slug + try: + project = db.create_project(admin.org_id, slug, display_name, owner_user_id=None, config=load_config()) return {"project": project} except Exception as e: if "UNIQUE" in str(e): diff --git a/backend/ingest_worker.py b/backend/ingest_worker.py new file mode 100644 index 0000000..893bea8 --- /dev/null +++ b/backend/ingest_worker.py @@ -0,0 +1,133 @@ +"""Document ingestion worker pipeline.""" + +import asyncio +import json +import shutil +from datetime import datetime +from pathlib import Path +from typing import Any, Dict + +from backend.paths import org_documents_dir, org_rag_index_dir, write_folder_project_meta +from src.config import load_config, resolve_opencode_credentials +from src.ingest.classify import classify_document +from src.ingest.formatter import format_global_index_document, format_index_document +from src.ingest.router import extract_document +from src.rag.indexer import index_meeting + + +async def process_document_ingest(job: Dict[str, Any], tasks: dict, send_progress): + task_id = job["task_id"] + file_path = Path(job["file_path"]) + org_slug = job["org_slug"] + project_slug = job["project_slug"] + doc_type = job.get("doc_type", "other") + display_name = job.get("display_name", file_path.name) + + tasks[task_id].update({"status": "processing", "message": "Извлечение текста...", "progress": 10}) + await send_progress(task_id, 10, "Извлечение текста...", "processing") + + try: + config = load_config() + ingest_cfg = config.get("ingest", {}) + pdf_ocr = ingest_cfg.get("pdf_ocr", True) + + doc = await asyncio.to_thread( + extract_document, + file_path, + project_slug, + doc_type, + None, + pdf_ocr, + ) + + if not doc.full_text.strip(): + raise ValueError("Не удалось извлечь текст из документа") + + documents_dir = org_documents_dir(org_slug) + output_dir = documents_dir / doc.document_id + await asyncio.to_thread(output_dir.mkdir, parents=True, exist_ok=True) + + original_dest = output_dir / file_path.name + await asyncio.to_thread(shutil.copy2, file_path, original_dest) + await asyncio.to_thread( + (output_dir / "extracted.md").write_text, + doc.full_text, + encoding="utf-8", + ) + await asyncio.to_thread(write_folder_project_meta, output_dir, project_slug) + + tasks[task_id].update({"status": "postprocessing", "message": "Анализ документа...", "progress": 40}) + await send_progress(task_id, 40, "Анализ документа...", "postprocessing") + + metadata = doc.to_metadata_dict() + rag_cfg = config.get("rag", {}) + api_key, base_url = resolve_opencode_credentials(config) + + if api_key and ingest_cfg.get("auto_classify", True): + metadata = await classify_document( + text=doc.full_text, + project=project_slug, + doc_type_hint=doc_type, + api_key=api_key, + base_url=base_url, + model=rag_cfg.get("index_model", "mimo-v2.5-free"), + chunk_size=int(rag_cfg.get("classify_chunk_size", 7000)), + ) + metadata["filename"] = doc.filename + metadata["document_id"] = doc.document_id + + await asyncio.to_thread( + (output_dir / "metadata.json").write_text, + json.dumps(metadata, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + + doc_text = format_index_document(doc, metadata) + index_path = output_dir / "index.txt" + await asyncio.to_thread(index_path.write_text, doc_text, encoding="utf-8") + + result_data = { + "document_id": doc.document_id, + "dir": str(output_dir), + "rel_dir": str(output_dir.relative_to(documents_dir)), + "extracted": str(output_dir / "extracted.md"), + "index": str(index_path), + "project": project_slug, + "doc_type": metadata.get("doc_type", doc_type), + "kind": "document", + } + + if rag_cfg.get("enabled", False) and rag_cfg.get("auto_index", True): + tasks[task_id].update({"message": "Индексация в RAG...", "progress": 75}) + await send_progress(task_id, 75, "Индексация в RAG...", "postprocessing") + global_doc_text = format_global_index_document(doc_text, metadata) + await index_meeting( + doc_text=doc_text, + global_doc_text=global_doc_text, + project_name=project_slug, + working_dir_base=org_rag_index_dir(org_slug), + model=rag_cfg.get("index_model", "mimo-v2.5-free"), + api_key=api_key, + base_url=base_url, + ) + + from backend.queue import _cleanup_upload + await asyncio.to_thread(_cleanup_upload, file_path) + + tasks[task_id].update({ + "status": "completed", + "progress": 100, + "message": "Документ проиндексирован", + "result": result_data, + "finished": datetime.now().isoformat(), + }) + await send_progress(task_id, 100, "Документ проиндексирован", "completed", result=result_data) + + except Exception as e: + error_msg = str(e) + tasks[task_id].update({ + "status": "error", + "message": f"Ошибка: {error_msg}", + "error": error_msg, + }) + await send_progress(task_id, 0, f"Ошибка: {error_msg}", "error", error=error_msg) diff --git a/backend/main.py b/backend/main.py index 477d67e..bae11a4 100644 --- a/backend/main.py +++ b/backend/main.py @@ -16,7 +16,7 @@ from backend.auth.models import UserContext from backend.auth.routes import admin_router, router as auth_router from backend.auth import database as auth_db from backend.auth.service import ensure_project_access, list_accessible_projects -from backend.paths import org_meetings_dir, org_rag_index_dir, resolve_meeting_path +from backend.paths import org_documents_dir, org_meetings_dir, org_rag_index_dir, resolve_document_path, resolve_meeting_path from backend.queue import ( delete_folder, get_all_tasks, @@ -29,6 +29,7 @@ from backend.queue import ( set_progress_callback, start_workers, stop_workers, + tasks, ) sys.path.insert(0, str(Path(__file__).parent.parent)) @@ -55,7 +56,14 @@ class ConnectionManager: self.active_connections = [(ws, user if ws is websocket else u) for ws, u in self.active_connections] async def broadcast(self, message: dict): - for conn, _user in self.active_connections: + """Send only to connections allowed to see the task (org + ACL).""" + task_id = message.get("task_id") + task = tasks.get(task_id) if task_id else None + for conn, user in self.active_connections: + if user is None: + continue + if task is not None and not user.can_see_task(task): + continue try: await conn.send_json(message) except Exception: @@ -77,8 +85,13 @@ async def lifespan(app: FastAPI): queue_cfg = config.get("queue", {}) transcribe_workers = int(queue_cfg.get("transcribe_workers", 2)) postprocess_workers = int(queue_cfg.get("postprocess_workers", 1)) + ingest_workers = int(queue_cfg.get("ingest_workers", 1)) print("🚀 Запуск рабочих процессов...") - start_workers(transcribe_workers=transcribe_workers, postprocess_workers=postprocess_workers) + start_workers( + transcribe_workers=transcribe_workers, + postprocess_workers=postprocess_workers, + ingest_workers=ingest_workers, + ) yield print("🛑 Остановка рабочих процессов...") stop_workers() @@ -107,12 +120,24 @@ async def _list_rag_project_slugs(user: UserContext) -> List[str]: return user.filter_projects(projects) -async def _rag_chat_for_user(user: UserContext, question: str, history: list, project_name: Optional[str], mode: str): +async def _rag_chat_for_user( + user: UserContext, + question: str, + history: list, + project_name: Optional[str], + chat_mode: str = "hybrid", + retrieval_mode: str = "hybrid", +): if project_name: ensure_project_access(user, project_name) elif not user.can_global_search(): raise HTTPException(status_code=403, detail="Глобальный поиск доступен только администратору") + if chat_mode not in ("hybrid", "compare", "timeline"): + chat_mode = "hybrid" + if retrieval_mode not in ("naive", "local", "global", "hybrid"): + retrieval_mode = "hybrid" + config = load_config() rag_cfg = config.get("rag", {}) api_key, base_url = resolve_opencode_credentials(config) @@ -124,7 +149,8 @@ async def _rag_chat_for_user(user: UserContext, question: str, history: list, pr project_name=project_name, base_url=base_url, chat_model=rag_cfg.get("chat_model", "deepseek-v4-flash-free"), - mode=mode, + mode=retrieval_mode, + chat_mode=chat_mode, index_model=rag_cfg.get("index_model", "mimo-v2.5-free"), ) @@ -149,13 +175,16 @@ async def login_page(): async def upload_file( file: UploadFile = File(...), project: str = Form(...), + doc_type: str = Form("other"), user: UserContext = Depends(get_current_user), ): content = await file.read() try: - task_id, _ = await save_upload(content, file.filename or "upload.bin", user, project) + task_id, _ = await save_upload(content, file.filename or "upload.bin", user, project, doc_type) except PermissionError as e: raise HTTPException(status_code=403, detail=str(e)) from e + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from e return { "task_id": task_id, @@ -167,10 +196,38 @@ async def upload_file( } +@app.post("/upload-document") +async def upload_document( + file: UploadFile = File(...), + project: str = Form(...), + doc_type: str = Form("other"), + user: UserContext = Depends(get_current_user), +): + """Явная загрузка документа (MD, PDF, DOCX, XLSX, TXT).""" + content = await file.read() + try: + task_id, _ = await save_upload(content, file.filename or "document.bin", user, project, doc_type) + except PermissionError as e: + raise HTTPException(status_code=403, detail=str(e)) from e + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + + return { + "task_id": task_id, + "file": file.filename, + "project": project, + "doc_type": doc_type, + "status": "queued", + "message": "Документ добавлен в очередь ingest", + "queue": get_queue_info(user), + } + + @app.post("/upload-batch") async def upload_batch( files: List[UploadFile] = File(...), project: str = Form(...), + doc_type: str = Form("other"), user: UserContext = Depends(get_current_user), ): if not files: @@ -180,9 +237,11 @@ async def upload_batch( for file in files: content = await file.read() try: - task_id, _ = await save_upload(content, file.filename or "upload.bin", user, project) + task_id, _ = await save_upload(content, file.filename or "upload.bin", user, project, doc_type) except PermissionError as e: raise HTTPException(status_code=403, detail=str(e)) from e + except ValueError as e: + raise HTTPException(status_code=400, detail=f"{file.filename}: {e}") from e results.append({"task_id": task_id, "file": file.filename, "project": project, "status": "queued"}) return { @@ -298,7 +357,8 @@ async def api_rag_query(payload: dict, user: UserContext = Depends(get_current_u payload.get("question", ""), payload.get("history", []), payload.get("project"), - payload.get("mode", "hybrid"), + chat_mode=payload.get("chat_mode", payload.get("mode", "hybrid")), + retrieval_mode=payload.get("retrieval_mode", "hybrid"), ) return {"answer": result["answer"], "context": result["context"], "project": result["project"]} except HTTPException: @@ -317,7 +377,8 @@ async def api_rag_query_global(payload: dict, user: UserContext = Depends(get_cu payload.get("question", ""), payload.get("history", []), None, - payload.get("mode", "hybrid"), + chat_mode=payload.get("chat_mode", payload.get("mode", "hybrid")), + retrieval_mode=payload.get("retrieval_mode", "hybrid"), ) return {"answer": result["answer"], "context": result["context"], "project": None} except HTTPException: @@ -329,19 +390,29 @@ async def api_rag_query_global(payload: dict, user: UserContext = Depends(get_cu @app.post("/api/rag/index/{folder_name:path}") async def api_rag_index_folder(folder_name: str, user: UserContext = Depends(get_current_user)): try: - folder_path = resolve_meeting_path(user.org_slug, folder_name) + from backend.queue import _folder_project_slug + + if folder_name.startswith("documents/"): + folder_path = resolve_document_path(user.org_slug, folder_name[len("documents/"):]) + base_dir = org_documents_dir(user.org_slug) + index_files = list(folder_path.glob("index.txt")) + txt_files = index_files + else: + rel = folder_name[len("meetings/"):] if folder_name.startswith("meetings/") else folder_name + folder_path = resolve_meeting_path(user.org_slug, rel) + base_dir = org_meetings_dir(user.org_slug) + txt_files = list(folder_path.glob("*.txt")) + if not folder_path.exists(): return {"error": "Folder not found"} - from backend.queue import _folder_project_slug - project = _folder_project_slug(folder_path.name, org_meetings_dir(user.org_slug)) + project = _folder_project_slug(folder_path.name, base_dir) if not project: return {"error": "Project metadata not found"} ensure_project_access(user, project) - txt_files = list(folder_path.glob("*.txt")) if not txt_files: - return {"error": "No .txt protocol found in folder"} + return {"error": "No index.txt or .txt protocol found in folder"} doc_text = txt_files[0].read_text(encoding="utf-8") config = load_config() @@ -376,7 +447,8 @@ async def _handle_rag_query_ws(websocket: WebSocket, msg: dict, user: UserContex msg.get("question", ""), msg.get("history", []), project, - msg.get("mode", "hybrid"), + chat_mode=msg.get("chat_mode", msg.get("mode", "hybrid")), + retrieval_mode=msg.get("retrieval_mode", "hybrid"), ) await websocket.send_json({ "type": "rag_response", diff --git a/backend/paths.py b/backend/paths.py index 3b3b04a..7fb4c40 100644 --- a/backend/paths.py +++ b/backend/paths.py @@ -9,6 +9,7 @@ UPLOAD_ROOT = Path("uploads") PROCESSED_ROOT = Path("processed") RAG_CACHE_DIRNAME = "lightrag_caches" MEETINGS_DIRNAME = "meetings" +DOCUMENTS_DIRNAME = "documents" def org_upload_dir(org_slug: str, user_id: int) -> Path: @@ -29,6 +30,20 @@ def org_rag_index_dir(org_slug: str) -> Path: return path +def org_documents_dir(org_slug: str) -> Path: + path = PROCESSED_ROOT / org_slug / DOCUMENTS_DIRNAME + path.mkdir(parents=True, exist_ok=True) + return path + + +def resolve_document_path(org_slug: str, rel_path: str) -> Path: + base = org_documents_dir(org_slug).resolve() + full = (base / rel_path).resolve() + if not str(full).startswith(str(base)): + raise ValueError("Invalid path") + return full + + def resolve_meeting_path(org_slug: str, rel_path: str) -> Path: """Resolve relative path under org meetings dir; reject traversal.""" base = org_meetings_dir(org_slug).resolve() diff --git a/backend/queue.py b/backend/queue.py index 9c970ca..255dc9d 100644 --- a/backend/queue.py +++ b/backend/queue.py @@ -13,7 +13,8 @@ sys.path.insert(0, str(Path(__file__).parent.parent)) from backend.auth.models import UserContext from backend.auth.service import ensure_project_access -from backend.paths import org_meetings_dir, org_rag_index_dir, org_upload_dir, resolve_meeting_path, write_folder_project_meta +from backend.paths import org_documents_dir, org_meetings_dir, org_rag_index_dir, org_upload_dir, resolve_document_path, resolve_meeting_path, write_folder_project_meta +from src.ingest.router import is_audio_file, is_document_file from src.audio_utils import prepare_audio_input from src.config import load_config, resolve_opencode_credentials from src.document import build_document @@ -31,6 +32,7 @@ tasks: Dict[str, Dict[str, Any]] = {} _progress_callback: Optional[Callable] = None _transcribe_queue: asyncio.Queue = asyncio.Queue() _postprocess_queue: asyncio.Queue = asyncio.Queue() +_ingest_queue: asyncio.Queue = asyncio.Queue() _workers: List[asyncio.Task] = [] @@ -40,11 +42,7 @@ def set_progress_callback(callback: Callable): def _task_visible(task: Dict[str, Any], user: UserContext) -> bool: - if task.get("org_slug") != user.org_slug: - return False - if user.has_all_projects_access: - return True - return task.get("user_id") == user.user_id + return user.can_see_task(task) def _filter_tasks_for_user(user: UserContext) -> List[Dict[str, Any]]: @@ -62,7 +60,8 @@ def _filter_queue_info(user: UserContext) -> Dict[str, Any]: "postprocessing": postprocessing, "pending_transcribe": _transcribe_queue.qsize(), "pending_postprocess": _postprocess_queue.qsize(), - "pending_in_queue": _transcribe_queue.qsize(), + "pending_ingest": _ingest_queue.qsize(), + "pending_in_queue": _transcribe_queue.qsize() + _ingest_queue.qsize(), } @@ -77,6 +76,8 @@ async def _send_progress(task_id: str, progress: int, message: str, status: str, "status": status, "file": task_info.get("file", ""), "project": task_info.get("project_slug", ""), + "org_slug": task_info.get("org_slug"), + "user_id": task_info.get("user_id"), "queue_position": task_info.get("queue_position"), "result": result, "error": error, @@ -345,14 +346,35 @@ async def _postprocess_worker_loop(worker_id: int): print(f"[Postprocess Worker {worker_id} Error] {e}") -def start_workers(transcribe_workers: int = 2, postprocess_workers: int = 1): +async def _ingest_worker_loop(worker_id: int): + from backend.ingest_worker import process_document_ingest + + print(f"[Ingest Worker {worker_id}] запущен") + while True: + try: + job = await _ingest_queue.get() + print(f"[Ingest Worker {worker_id}] задача {job.get('task_id')}") + await process_document_ingest(job, tasks, _send_progress) + _ingest_queue.task_done() + except asyncio.CancelledError: + break + except Exception as e: + print(f"[Ingest Worker {worker_id} Error] {e}") + + +def start_workers(transcribe_workers: int = 2, postprocess_workers: int = 1, ingest_workers: int = 1): global _workers _workers.clear() for i in range(transcribe_workers): _workers.append(asyncio.create_task(_transcribe_worker_loop(i + 1))) + for i in range(ingest_workers): + _workers.append(asyncio.create_task(_ingest_worker_loop(i + 1))) for i in range(postprocess_workers): _workers.append(asyncio.create_task(_postprocess_worker_loop(i + 1))) - print(f"[Queue] transcribe_workers={transcribe_workers}, postprocess_workers={postprocess_workers}") + print( + f"[Queue] transcribe={transcribe_workers}, ingest={ingest_workers}, " + f"postprocess={postprocess_workers}" + ) def stop_workers(): @@ -365,10 +387,17 @@ async def save_upload( filename: str, user: UserContext, project_slug: str, + doc_type: str = "other", ) -> tuple[str, Path]: ensure_project_access(user, project_slug) - slug = project_slug.strip().lower() safe_name = Path(filename).name or "upload.bin" + + if is_document_file(safe_name): + return await save_document_upload(content, safe_name, user, project_slug, doc_type) + if not is_audio_file(safe_name): + raise ValueError(f"Неподдерживаемый формат файла: {safe_name}") + + slug = project_slug.strip().lower() task_id = f"task_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}" task_dir = org_upload_dir(user.org_slug, user.user_id) / task_id task_dir.mkdir(parents=True, exist_ok=True) @@ -383,6 +412,7 @@ async def save_upload( "progress": 0, "message": f"В очереди транскрибации (№{queue_position})", "file": safe_name, + "task_type": "transcribe", "project_slug": slug, "org_slug": user.org_slug, "user_id": user.user_id, @@ -398,6 +428,53 @@ async def save_upload( return task_id, file_path +async def save_document_upload( + content: bytes, + filename: str, + user: UserContext, + project_slug: str, + doc_type: str = "other", +) -> tuple[str, Path]: + slug = project_slug.strip().lower() + task_id = f"doc_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}" + task_dir = org_upload_dir(user.org_slug, user.user_id) / task_id + task_dir.mkdir(parents=True, exist_ok=True) + file_path = task_dir / filename + + await asyncio.to_thread(file_path.write_bytes, content) + + queue_position = _ingest_queue.qsize() + 1 + tasks[task_id] = { + "task_id": task_id, + "status": "queued", + "progress": 0, + "message": f"В очереди ingest (№{queue_position})", + "file": filename, + "task_type": "ingest", + "doc_type": doc_type, + "project_slug": slug, + "org_slug": user.org_slug, + "user_id": user.user_id, + "username": user.username, + "queue_position": queue_position, + "result": None, + "error": None, + "started": datetime.now().isoformat(), + } + + job = { + "task_id": task_id, + "file_path": str(file_path), + "display_name": filename, + "org_slug": user.org_slug, + "project_slug": slug, + "doc_type": doc_type, + } + await _ingest_queue.put(job) + await _send_progress(task_id, 0, f"В очереди ingest (№{queue_position})", "queued") + return task_id, file_path + + def get_queue_info(user: Optional[UserContext] = None) -> Dict[str, Any]: if user: return _filter_queue_info(user) @@ -410,10 +487,126 @@ def get_queue_info(user: Optional[UserContext] = None) -> Dict[str, Any]: "postprocessing": postprocessing, "pending_transcribe": _transcribe_queue.qsize(), "pending_postprocess": _postprocess_queue.qsize(), - "pending_in_queue": _transcribe_queue.qsize(), + "pending_ingest": _ingest_queue.qsize(), + "pending_in_queue": _transcribe_queue.qsize() + _ingest_queue.qsize(), } +def _build_tree_section(base_dir: Path, rel_prefix: str, user: UserContext, kind_default: str) -> List[Dict[str, Any]]: + tree = [] + if not base_dir.exists(): + return tree + + skip_suffixes = ("_segments.json",) + skip_names = {".project.json"} + + for item in sorted(base_dir.iterdir()): + if not item.is_dir(): + continue + project_slug = _folder_project_slug(item.name, base_dir) + if project_slug and not user.can_access_project(project_slug): + continue + + files = [] + for f in sorted(item.iterdir(), key=lambda p: _file_sort_key({"name": p.name})): + if f.is_file() and not f.name.endswith(skip_suffixes) and f.name not in skip_names: + kind = kind_default + if "_summary" in f.name: + kind = "summary" + elif f.name in ("extracted.md", "index.txt", "metadata.json"): + kind = "document" + files.append({ + "name": f.name, + "path": f"{rel_prefix}/{f.relative_to(base_dir).as_posix()}", + "size": f.stat().st_size, + "ext": f.suffix.lower(), + "kind": kind, + }) + meta_path = item / "metadata.json" + doc_type = None + if meta_path.exists(): + try: + doc_type = json.loads(meta_path.read_text(encoding="utf-8")).get("doc_type") + except Exception: + pass + + tree.append({ + "name": item.name, + "path": f"{rel_prefix}/{item.relative_to(base_dir).as_posix()}", + "project": project_slug, + "doc_type": doc_type, + "section": rel_prefix, + "files": files, + "created": datetime.fromtimestamp(item.stat().st_ctime).isoformat(), + }) + return tree + + +def get_processed_tree(user: UserContext) -> List[Dict[str, Any]]: + meetings_dir = org_meetings_dir(user.org_slug) + documents_dir = org_documents_dir(user.org_slug) + tree = _build_tree_section(meetings_dir, "meetings", user, "protocol") + tree.extend(_build_tree_section(documents_dir, "documents", user, "document")) + tree.sort(key=lambda x: x.get("created", ""), reverse=True) + return tree + + +def _resolve_content_path(user: UserContext, rel_path: str) -> Path: + if rel_path.startswith("documents/"): + return resolve_document_path(user.org_slug, rel_path[len("documents/"):]) + if rel_path.startswith("meetings/"): + return resolve_meeting_path(user.org_slug, rel_path[len("meetings/"):]) + full = resolve_meeting_path(user.org_slug, rel_path) + if full.exists(): + return full + return resolve_document_path(user.org_slug, rel_path) + + +def read_file_content(user: UserContext, rel_path: str) -> str: + full_path = _resolve_content_path(user, rel_path) + if not full_path.exists() or not full_path.is_file(): + raise FileNotFoundError(f"Файл не найден: {rel_path}") + + base_dir = full_path.parent.parent + folder = full_path.parent.name + project_slug = _folder_project_slug(folder, base_dir) + if project_slug and not user.can_access_project(project_slug): + raise PermissionError("Нет доступа к этому файлу") + + with open(full_path, "r", encoding="utf-8") as f: + return f.read() + + +def get_download_path(user: UserContext, rel_path: str) -> Path: + full_path = _resolve_content_path(user, rel_path) + if not full_path.exists() or not full_path.is_file(): + raise FileNotFoundError(f"Файл не найден: {rel_path}") + base_dir = full_path.parent.parent + project_slug = _folder_project_slug(full_path.parent.name, base_dir) + if project_slug and not user.can_access_project(project_slug): + raise PermissionError("Нет доступа к этому файлу") + return full_path + + +def delete_folder(user: UserContext, folder_rel: str) -> None: + if folder_rel.startswith("documents/"): + folder_path = resolve_document_path(user.org_slug, folder_rel[len("documents/"):]) + base_dir = org_documents_dir(user.org_slug) + elif folder_rel.startswith("meetings/"): + folder_path = resolve_meeting_path(user.org_slug, folder_rel[len("meetings/"):]) + base_dir = org_meetings_dir(user.org_slug) + else: + folder_path = resolve_meeting_path(user.org_slug, folder_rel) + base_dir = org_meetings_dir(user.org_slug) + + if not folder_path.is_dir(): + raise FileNotFoundError("Folder not found") + project_slug = _folder_project_slug(folder_path.name, base_dir) + if project_slug and not user.can_access_project(project_slug): + raise PermissionError("Нет доступа к этой папке") + shutil.rmtree(folder_path) + + def get_task_status(task_id: str, user: Optional[UserContext] = None) -> Optional[Dict[str, Any]]: task = tasks.get(task_id) if not task: @@ -442,8 +635,8 @@ def _file_sort_key(file_info: Dict[str, Any]) -> tuple: return (4, name) -def _folder_project_slug(folder_name: str, meetings_dir: Path) -> Optional[str]: - meta_path = meetings_dir / folder_name / ".project.json" +def _folder_project_slug(folder_name: str, base_dir: Path) -> Optional[str]: + meta_path = base_dir / folder_name / ".project.json" if meta_path.exists(): try: data = json.loads(meta_path.read_text(encoding="utf-8")) @@ -451,75 +644,3 @@ def _folder_project_slug(folder_name: str, meetings_dir: Path) -> Optional[str]: except Exception: pass return None - - -def get_processed_tree(user: UserContext) -> List[Dict[str, Any]]: - tree = [] - meetings_dir = org_meetings_dir(user.org_slug) - if not meetings_dir.exists(): - return tree - - skip_suffixes = ("_segments.json",) - - for item in sorted(meetings_dir.iterdir()): - if not item.is_dir(): - continue - project_slug = _folder_project_slug(item.name, meetings_dir) - if project_slug and not user.can_access_project(project_slug): - continue - - files = [] - for f in sorted(item.iterdir(), key=lambda p: _file_sort_key({"name": p.name})): - if f.is_file() and not f.name.endswith(skip_suffixes) and f.name != ".project.json": - files.append({ - "name": f.name, - "path": str(f.relative_to(meetings_dir)), - "size": f.stat().st_size, - "ext": f.suffix.lower(), - "kind": "summary" if "_summary" in f.name else "protocol", - }) - tree.append({ - "name": item.name, - "path": str(item.relative_to(meetings_dir)), - "project": project_slug, - "files": files, - "created": datetime.fromtimestamp(item.stat().st_ctime).isoformat(), - }) - return tree - - -def read_file_content(user: UserContext, rel_path: str) -> str: - full_path = resolve_meeting_path(user.org_slug, rel_path) - if not full_path.exists() or not full_path.is_file(): - raise FileNotFoundError(f"Файл не найден: {rel_path}") - - folder = full_path.parent.name - meetings_dir = org_meetings_dir(user.org_slug) - project_slug = _folder_project_slug(folder, meetings_dir) - if project_slug and not user.can_access_project(project_slug): - raise PermissionError("Нет доступа к этому файлу") - - with open(full_path, "r", encoding="utf-8") as f: - return f.read() - - -def get_download_path(user: UserContext, rel_path: str) -> Path: - full_path = resolve_meeting_path(user.org_slug, rel_path) - if not full_path.exists() or not full_path.is_file(): - raise FileNotFoundError(f"Файл не найден: {rel_path}") - folder = full_path.parent.name - meetings_dir = org_meetings_dir(user.org_slug) - project_slug = _folder_project_slug(folder, meetings_dir) - if project_slug and not user.can_access_project(project_slug): - raise PermissionError("Нет доступа к этому файлу") - return full_path - - -def delete_folder(user: UserContext, folder_rel: str) -> None: - folder_path = resolve_meeting_path(user.org_slug, folder_rel) - if not folder_path.is_dir(): - raise FileNotFoundError("Folder not found") - project_slug = _folder_project_slug(folder_path.name, org_meetings_dir(user.org_slug)) - if project_slug and not user.can_access_project(project_slug): - raise PermissionError("Нет доступа к этой папке") - shutil.rmtree(folder_path) diff --git a/backend/static/app.js b/backend/static/app.js index 47553ec..b116b40 100644 --- a/backend/static/app.js +++ b/backend/static/app.js @@ -269,6 +269,7 @@ class TranscriptionApp { if (!files.length) return; const project = document.getElementById('uploadProjectSelect')?.value; + const docType = document.getElementById('uploadDocTypeSelect')?.value || 'other'; if (!project) { this.showToast('Выберите проект перед загрузкой', 'error'); return; @@ -276,6 +277,7 @@ class TranscriptionApp { const formData = new FormData(); formData.append('project', project); + formData.append('doc_type', docType); for (const file of files) { formData.append('files', file); } @@ -328,6 +330,7 @@ class TranscriptionApp { const parts = []; if (queue.processing) parts.push(`транскрибация: ${queue.processing}`); if (queue.pending_transcribe) parts.push(`в очереди ASR: ${queue.pending_transcribe}`); + if (queue.pending_ingest) parts.push(`ingest: ${queue.pending_ingest}`); if (queue.postprocessing) parts.push(`summary/RAG: ${queue.postprocessing}`); if (queue.pending_postprocess) parts.push(`в очереди post: ${queue.pending_postprocess}`); el.textContent = parts.length ? parts.join(' · ') : 'очередь пуста'; @@ -381,7 +384,7 @@ class TranscriptionApp {
${this.escapeHtml(task.file || '')} - ${this.getStatusLabel(task.status)} + ${this.getStatusLabel(task)}
@@ -391,11 +394,13 @@ class TranscriptionApp { `; } - getStatusLabel(status) { + getStatusLabel(task) { + const status = typeof task === 'string' ? task : task.status; + const taskType = typeof task === 'object' ? task.task_type : null; const labels = { - 'queued': 'В очереди', - 'processing': 'Транскрибация', - 'postprocessing': 'Summary/RAG', + 'queued': taskType === 'ingest' ? 'В очереди ingest' : 'В очереди', + 'processing': taskType === 'ingest' ? 'Ingest' : 'Транскрибация', + 'postprocessing': taskType === 'ingest' ? 'Индексация' : 'Summary/RAG', 'completed': 'Готово', 'error': 'Ошибка', }; @@ -416,12 +421,16 @@ class TranscriptionApp { } renderFolder(folder) { + const sectionLabel = folder.section === 'documents' + ? '📚 документ' + : (folder.doc_type ? `📋 ${folder.doc_type}` : '🎙️ протокол'); const files = folder.files.map(file => { const isSummary = file.kind === 'summary' || file.name.includes('_summary'); const isMd = file.ext === '.md'; const isDocx = file.ext === '.docx'; const isTxt = file.ext === '.txt'; - const icon = isSummary ? '📋' : isMd ? '📝' : isDocx ? '📄' : isTxt ? '📃' : '📎'; + const isJson = file.ext === '.json'; + const icon = isSummary ? '📋' : isMd ? '📝' : isDocx ? '📄' : isTxt ? '📃' : isJson ? '🏷️' : '📎'; const downloadUrl = this.downloadUrl(file.path); const cssClass = isSummary ? 'file-item file-summary' : 'file-item'; @@ -445,11 +454,12 @@ class TranscriptionApp { }).join(''); return ` -
+
▼ 📁 ${this.escapeHtml(folder.name)} + ${sectionLabel} ${this.formatDate(folder.created)} 🗑️
@@ -668,6 +678,7 @@ class TranscriptionApp { const select = document.getElementById('chatProjectSelect'); const project = select.value; + const chatMode = document.getElementById('chatModeSelect')?.value || 'hybrid'; if (!project && !this.user?.all_projects_access) { this.showToast('Выберите проект для поиска', 'error'); @@ -683,8 +694,9 @@ class TranscriptionApp { action: action, question: question, project: project || undefined, - history: this.chatHistory.slice(-6), // последние 6 пар - mode: 'hybrid', + history: this.chatHistory.slice(-6), + chat_mode: chatMode, + retrieval_mode: 'hybrid', }); } @@ -759,7 +771,7 @@ class TranscriptionApp { const container = document.getElementById('chatMessages'); container.innerHTML = `
-

Здравствуйте! Я помогу найти информацию в протоколах совещаний.

+

Здравствуйте! Я помогу найти информацию в протоколах и документах проекта.

Выберите проект или оставьте «Все проекты» для глобального поиска.

`; diff --git a/backend/static/index.html b/backend/static/index.html index 32a6d0e..ca4995d 100644 --- a/backend/static/index.html +++ b/backend/static/index.html @@ -16,7 +16,7 @@

🎙️ Транскрибация совещаний

-

Загрузите аудио или видео файл для получения протокола

+

Загрузите аудио/видео для протокола или документы для базы знаний

@@ -47,6 +47,18 @@ Проект +
@@ -56,9 +68,10 @@

Перетащите файлы сюда или выберите

-

Поддерживаются: MP4, WEBM, AVI, MKV, MOV, WAV, MP3, M4A, OGG, FLAC

+

Аудио/видео: MP4, WEBM, AVI, MKV, MOV, WAV, MP3, M4A, OGG, FLAC

+

Документы: MD, TXT, DOCX, PDF, XLSX, CSV

- +
@@ -95,17 +108,22 @@

🤖 Чат с базой знаний

+
-

Здравствуйте! Я помогу найти информацию в протоколах совещаний.

+

Здравствуйте! Я помогу найти информацию в протоколах и документах проекта.

Выберите проект для поиска.

- +
diff --git a/backend/static/styles.css b/backend/static/styles.css index d499cc2..8544bd0 100644 --- a/backend/static/styles.css +++ b/backend/static/styles.css @@ -540,7 +540,8 @@ header h1 { gap: 10px; } -#chatProjectSelect { +#chatProjectSelect, +#chatModeSelect { background: var(--bg-hover); color: var(--text-primary); border: 1px solid var(--border); @@ -836,6 +837,12 @@ header h1 { cursor: pointer; } +.folder-badge { + font-size: 0.75rem; + color: var(--text-muted, #888); + margin-left: 0.5rem; +} + .upload-controls { margin-bottom: 12px; } diff --git a/config.yaml b/config.yaml index a38ae9f..43b3a7a 100644 --- a/config.yaml +++ b/config.yaml @@ -70,6 +70,20 @@ rag: queue: transcribe_workers: 2 postprocess_workers: 1 + ingest_workers: 1 + +# Ingest документов (MD, PDF, DOCX, XLSX, TXT) +ingest: + auto_classify: true + pdf_ocr: true + doc_types: + - meeting + - specification + - estimate + - contract + - report + - correspondence + - other # Авторизация и multi-tenant (org + projects) auth: diff --git a/requirements.txt b/requirements.txt index 0d04d28..655eb71 100644 --- a/requirements.txt +++ b/requirements.txt @@ -16,3 +16,9 @@ openai>=1.0.0 python-dotenv>=1.0.0 sentence-transformers>=3.0.0 numpy>=1.24.0 + +# Document ingestion +pymupdf>=1.24.0 +openpyxl>=3.1.0 +Pillow>=10.0.0 +pytesseract>=0.3.10 diff --git a/src/ingest/__init__.py b/src/ingest/__init__.py new file mode 100644 index 0000000..1e1e5a1 --- /dev/null +++ b/src/ingest/__init__.py @@ -0,0 +1,5 @@ +"""Knowledge base document ingestion.""" + +from src.ingest.router import extract_document, is_audio_file, is_document_file + +__all__ = ["extract_document", "is_audio_file", "is_document_file"] diff --git a/src/ingest/classify.py b/src/ingest/classify.py new file mode 100644 index 0000000..41ce664 --- /dev/null +++ b/src/ingest/classify.py @@ -0,0 +1,92 @@ +"""Universal document classification via LLM.""" + +import json +import re +from typing import Any, Dict, List + +from openai import AsyncOpenAI + +from src.ingest.models import DOC_TYPES +from src.rag.parser import split_text_chunks, CHUNK_OVERLAP + +CLASSIFY_DOCUMENT_PROMPT = """ +Проанализируй документ проекта "{project}" (тип: {doc_type_hint}) и извлеки структурированные данные. +Верни ТОЛЬКО JSON: +{{ + "project": "{project}", + "doc_type": "one of: meeting, specification, estimate, contract, report, correspondence, other", + "title": "краткое название", + "topic": "основная тема", + "date": "YYYY-MM-DD или null", + "summary": "краткое содержание в 2-4 предложениях", + "key_decisions": ["..."], + "action_items": [{{"task": "...", "assignee": "..."}}], + "entities": ["ключевые сущности, объекты, системы"] +}} + +Текст документа: +--- +{text} +--- +""" + + +def _empty_metadata(project: str, doc_type: str) -> Dict[str, Any]: + return { + "project": project, + "doc_type": doc_type, + "title": "", + "topic": "", + "date": None, + "summary": "", + "key_decisions": [], + "action_items": [], + "entities": [], + } + + +def _parse_json(content: str, project: str, doc_type: str) -> Dict[str, Any]: + content = content.strip() + match = re.search(r"\{.*\}", content, re.DOTALL) + if match: + content = match.group(0) + try: + data = json.loads(content) + base = _empty_metadata(project, doc_type) + for key in base: + if key in data and data[key] is not None: + base[key] = data[key] + if base.get("doc_type") not in DOC_TYPES: + base["doc_type"] = doc_type + return base + except json.JSONDecodeError: + return _empty_metadata(project, doc_type) + + +async def classify_document( + text: str, + project: str, + doc_type_hint: str, + api_key: str, + base_url: str = "https://opencode.ai/zen/v1", + model: str = "mimo-v2.5-free", + chunk_size: int = 7000, +) -> Dict[str, Any]: + if not text.strip(): + return _empty_metadata(project, doc_type_hint) + + client = AsyncOpenAI(base_url=base_url, api_key=api_key) + chunks = split_text_chunks(text, chunk_size, CHUNK_OVERLAP) + prompt = CLASSIFY_DOCUMENT_PROMPT.format( + project=project, + doc_type_hint=doc_type_hint, + text=chunks[0], + ) + response = await client.chat.completions.create( + model=model, + messages=[{"role": "user", "content": prompt}], + temperature=0.2, + max_tokens=2048, + ) + content = response.choices[0].message.content or "" + return _parse_json(content, project, doc_type_hint) diff --git a/src/ingest/extractors/__init__.py b/src/ingest/extractors/__init__.py new file mode 100644 index 0000000..ecdecff --- /dev/null +++ b/src/ingest/extractors/__init__.py @@ -0,0 +1 @@ +"""Extractors package.""" diff --git a/src/ingest/extractors/docx_extractor.py b/src/ingest/extractors/docx_extractor.py new file mode 100644 index 0000000..3252b2d --- /dev/null +++ b/src/ingest/extractors/docx_extractor.py @@ -0,0 +1,32 @@ +"""DOCX extractor.""" + +from pathlib import Path + +from docx import Document + +from src.ingest.models import DocumentChunk, NormalizedDocument + + +def extract_docx(path: Path, document_id: str, project: str, doc_type: str) -> NormalizedDocument: + doc = Document(str(path)) + parts: list[str] = [] + for para in doc.paragraphs: + line = para.text.strip() + if line: + parts.append(line) + for table in doc.tables: + for row in table.rows: + cells = [cell.text.strip() for cell in row.cells if cell.text.strip()] + if cells: + parts.append(" | ".join(cells)) + full_text = "\n".join(parts).strip() + return NormalizedDocument( + document_id=document_id, + filename=path.name, + doc_type=doc_type, + project=project, + full_text=full_text, + chunks=[DocumentChunk(text=full_text, source=path.name)] if full_text else [], + metadata={"format": "docx", "paragraphs": len(doc.paragraphs), "tables": len(doc.tables)}, + mime_hint="application/vnd.openxmlformats-officedocument.wordprocessingml.document", + ) diff --git a/src/ingest/extractors/pdf_extractor.py b/src/ingest/extractors/pdf_extractor.py new file mode 100644 index 0000000..567f152 --- /dev/null +++ b/src/ingest/extractors/pdf_extractor.py @@ -0,0 +1,63 @@ +"""PDF extractor with optional OCR fallback.""" + +from pathlib import Path +from typing import List + +from src.ingest.models import DocumentChunk, NormalizedDocument + + +def _extract_with_pymupdf(path: Path) -> tuple[str, List[DocumentChunk]]: + import fitz # pymupdf + + doc = fitz.open(str(path)) + chunks: List[DocumentChunk] = [] + parts: List[str] = [] + for page_num, page in enumerate(doc, start=1): + text = page.get_text("text").strip() + if text: + parts.append(f"--- Страница {page_num} ---\n{text}") + chunks.append(DocumentChunk(text=text, source=path.name, page=page_num)) + doc.close() + return "\n\n".join(parts).strip(), chunks + + +def _extract_with_ocr(path: Path) -> tuple[str, List[DocumentChunk]]: + try: + import fitz + import pytesseract + from PIL import Image + import io + except ImportError: + return "", [] + + doc = fitz.open(str(path)) + chunks: List[DocumentChunk] = [] + parts: List[str] = [] + for page_num, page in enumerate(doc, start=1): + pix = page.get_pixmap(dpi=200) + img = Image.open(io.BytesIO(pix.tobytes("png"))) + text = pytesseract.image_to_string(img, lang="rus+eng").strip() + if text: + parts.append(f"--- Страница {page_num} (OCR) ---\n{text}") + chunks.append(DocumentChunk(text=text, source=path.name, page=page_num)) + doc.close() + return "\n\n".join(parts).strip(), chunks + + +def extract_pdf(path: Path, document_id: str, project: str, doc_type: str, use_ocr: bool = True) -> NormalizedDocument: + full_text, chunks = _extract_with_pymupdf(path) + ocr_used = False + if not full_text and use_ocr: + full_text, chunks = _extract_with_ocr(path) + ocr_used = bool(full_text) + + return NormalizedDocument( + document_id=document_id, + filename=path.name, + doc_type=doc_type, + project=project, + full_text=full_text, + chunks=chunks or ([DocumentChunk(text=full_text, source=path.name)] if full_text else []), + metadata={"format": "pdf", "ocr_used": ocr_used, "pages": len(chunks)}, + mime_hint="application/pdf", + ) diff --git a/src/ingest/extractors/text.py b/src/ingest/extractors/text.py new file mode 100644 index 0000000..44b7ebd --- /dev/null +++ b/src/ingest/extractors/text.py @@ -0,0 +1,20 @@ +"""Text and markdown extractors.""" + +from pathlib import Path + +from src.ingest.models import DocumentChunk, NormalizedDocument + + +def extract_text(path: Path, document_id: str, project: str, doc_type: str) -> NormalizedDocument: + text = path.read_text(encoding="utf-8", errors="replace").strip() + suffix = path.suffix.lower() + return NormalizedDocument( + document_id=document_id, + filename=path.name, + doc_type=doc_type, + project=project, + full_text=text, + chunks=[DocumentChunk(text=text, source=path.name)], + metadata={"format": suffix.lstrip(".")}, + mime_hint="text/plain" if suffix == ".txt" else "text/markdown", + ) diff --git a/src/ingest/extractors/xlsx_extractor.py b/src/ingest/extractors/xlsx_extractor.py new file mode 100644 index 0000000..b05dc0b --- /dev/null +++ b/src/ingest/extractors/xlsx_extractor.py @@ -0,0 +1,74 @@ +"""Excel / CSV extractor.""" + +from pathlib import Path +from typing import List + +from src.ingest.models import DocumentChunk, NormalizedDocument + +MAX_ROWS_PREVIEW = 50 + + +def extract_xlsx(path: Path, document_id: str, project: str, doc_type: str) -> NormalizedDocument: + from openpyxl import load_workbook + + wb = load_workbook(str(path), read_only=True, data_only=True) + chunks: List[DocumentChunk] = [] + parts: List[str] = [] + + for sheet_name in wb.sheetnames: + ws = wb[sheet_name] + rows: List[str] = [] + row_count = 0 + for row in ws.iter_rows(values_only=True): + row_count += 1 + cells = [str(c).strip() for c in row if c is not None and str(c).strip()] + if not cells: + continue + line = " | ".join(cells) + if len(rows) < MAX_ROWS_PREVIEW: + rows.append(line) + if rows: + sheet_text = f"=== Лист: {sheet_name} (строк: {row_count}) ===\n" + "\n".join(rows) + if row_count > MAX_ROWS_PREVIEW: + sheet_text += f"\n... (показаны первые {MAX_ROWS_PREVIEW} непустых строк)" + parts.append(sheet_text) + chunks.append(DocumentChunk(text=sheet_text, source=path.name, sheet=sheet_name)) + + wb.close() + full_text = "\n\n".join(parts).strip() + return NormalizedDocument( + document_id=document_id, + filename=path.name, + doc_type=doc_type, + project=project, + full_text=full_text, + chunks=chunks, + metadata={"format": "xlsx", "sheets": len(chunks)}, + mime_hint="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + ) + + +def extract_csv(path: Path, document_id: str, project: str, doc_type: str) -> NormalizedDocument: + import csv + + rows: List[str] = [] + with open(path, "r", encoding="utf-8", errors="replace", newline="") as f: + reader = csv.reader(f) + for i, row in enumerate(reader): + if i >= MAX_ROWS_PREVIEW: + rows.append(f"... (обрезано после {MAX_ROWS_PREVIEW} строк)") + break + cells = [c.strip() for c in row if c.strip()] + if cells: + rows.append(" | ".join(cells)) + full_text = "\n".join(rows).strip() + return NormalizedDocument( + document_id=document_id, + filename=path.name, + doc_type=doc_type, + project=project, + full_text=full_text, + chunks=[DocumentChunk(text=full_text, source=path.name, sheet="csv")] if full_text else [], + metadata={"format": "csv"}, + mime_hint="text/csv", + ) diff --git a/src/ingest/formatter.py b/src/ingest/formatter.py new file mode 100644 index 0000000..46c7c1b --- /dev/null +++ b/src/ingest/formatter.py @@ -0,0 +1,64 @@ +"""Format normalized documents for RAG indexing.""" + +from typing import Any, Dict, List + +from src.ingest.models import NormalizedDocument + + +def format_index_document(doc: NormalizedDocument, metadata: Dict[str, Any] | None = None) -> str: + meta = metadata or doc.metadata or {} + lines = [ + f"=== ДОКУМЕНТ ({meta.get('doc_type', doc.doc_type)}) ===", + f"ID: {doc.document_id}", + f"Проект: {doc.project}", + f"Файл: {doc.filename}", + f"Тип: {meta.get('doc_type', doc.doc_type)}", + ] + if meta.get("title"): + lines.append(f"Название: {meta['title']}") + if meta.get("topic"): + lines.append(f"Тема: {meta['topic']}") + if meta.get("date"): + lines.append(f"Дата: {meta['date']}") + if meta.get("summary"): + lines.append(f"Summary: {meta['summary']}") + + decisions = meta.get("key_decisions") or [] + if decisions: + lines.append("Решения:") + for i, item in enumerate(decisions, 1): + lines.append(f" {i}. {item}") + + actions = meta.get("action_items") or [] + if actions: + lines.append("Action items:") + for item in actions: + if isinstance(item, dict): + lines.append(f" - {item.get('task', item)} ({item.get('assignee', '—')})") + else: + lines.append(f" - {item}") + + lines.extend(["", "--- Содержание ---", ""]) + if doc.chunks: + for chunk in doc.chunks: + prefix = [] + if chunk.page: + prefix.append(f"[стр. {chunk.page}]") + if chunk.sheet: + prefix.append(f"[лист: {chunk.sheet}]") + header = " ".join(prefix) + if header: + lines.append(header) + lines.append(chunk.text) + lines.append("") + else: + lines.append(doc.full_text) + return "\n".join(lines).strip() + + +def format_global_index_document(doc_text: str, metadata: Dict[str, Any]) -> str: + project = metadata.get("project", "unknown") + header = f"=== ДОКУМЕНТ (Проект: {project}) ===\n" + header += f"Тип: {metadata.get('doc_type', 'other')}\n" + header += f"Файл: {metadata.get('filename', metadata.get('source', 'unknown'))}\n\n" + return header + doc_text diff --git a/src/ingest/models.py b/src/ingest/models.py new file mode 100644 index 0000000..8163bd9 --- /dev/null +++ b/src/ingest/models.py @@ -0,0 +1,70 @@ +"""Normalized document model for knowledge base ingestion.""" + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + + +DOC_TYPES = ( + "meeting", + "specification", + "estimate", + "contract", + "report", + "correspondence", + "other", +) + +SUPPORTED_DOCUMENT_EXTENSIONS = { + ".txt", + ".md", + ".markdown", + ".docx", + ".doc", + ".pdf", + ".xlsx", + ".xls", + ".csv", +} + +AUDIO_EXTENSIONS = { + ".mp4", + ".webm", + ".avi", + ".mkv", + ".mov", + ".wav", + ".mp3", + ".m4a", + ".ogg", + ".flac", +} + + +@dataclass +class DocumentChunk: + text: str + source: str = "" + page: Optional[int] = None + sheet: Optional[str] = None + + +@dataclass +class NormalizedDocument: + document_id: str + filename: str + doc_type: str + project: str + full_text: str + chunks: List[DocumentChunk] = field(default_factory=list) + metadata: Dict[str, Any] = field(default_factory=dict) + mime_hint: str = "" + + def to_metadata_dict(self) -> Dict[str, Any]: + return { + "document_id": self.document_id, + "filename": self.filename, + "doc_type": self.doc_type, + "project": self.project, + "mime_hint": self.mime_hint, + **self.metadata, + } diff --git a/src/ingest/router.py b/src/ingest/router.py new file mode 100644 index 0000000..de49b59 --- /dev/null +++ b/src/ingest/router.py @@ -0,0 +1,54 @@ +"""Route files to format-specific extractors.""" + +import uuid +from pathlib import Path + +from src.ingest.extractors.docx_extractor import extract_docx +from src.ingest.extractors.pdf_extractor import extract_pdf +from src.ingest.extractors.text import extract_text +from src.ingest.extractors.xlsx_extractor import extract_csv, extract_xlsx +from src.ingest.models import AUDIO_EXTENSIONS, SUPPORTED_DOCUMENT_EXTENSIONS, NormalizedDocument + + +def is_audio_file(filename: str) -> bool: + return Path(filename).suffix.lower() in AUDIO_EXTENSIONS + + +def is_document_file(filename: str) -> bool: + return Path(filename).suffix.lower() in SUPPORTED_DOCUMENT_EXTENSIONS + + +def resolve_upload_kind(filename: str) -> str: + """Return 'document', 'audio', or raise ValueError.""" + if is_document_file(filename): + return "document" + if is_audio_file(filename): + return "audio" + raise ValueError(f"Неподдерживаемый формат файла: {filename}") + + +def extract_document( + path: Path, + project: str, + doc_type: str = "other", + document_id: str | None = None, + pdf_ocr: bool = True, +) -> NormalizedDocument: + if not path.exists(): + raise FileNotFoundError(f"Файл не найден: {path}") + + doc_id = document_id or f"doc_{uuid.uuid4().hex[:12]}" + suffix = path.suffix.lower() + + if suffix in (".txt", ".md", ".markdown"): + return extract_text(path, doc_id, project, doc_type) + if suffix in (".docx", ".doc"): + return extract_docx(path, doc_id, project, doc_type) + if suffix == ".pdf": + return extract_pdf(path, doc_id, project, doc_type, use_ocr=pdf_ocr) + if suffix in (".xlsx", ".xls"): + return extract_xlsx(path, doc_id, project, doc_type) + if suffix == ".csv": + return extract_csv(path, doc_id, project, doc_type) + + raise ValueError(f"Неподдерживаемый формат документа: {suffix}") diff --git a/src/rag/query.py b/src/rag/query.py index 8e94a75..5ecea1e 100644 --- a/src/rag/query.py +++ b/src/rag/query.py @@ -1,6 +1,5 @@ -"""Запросы к RAG и генерация ответов чат-бота через DeepSeek.""" +"""Запросы к RAG и генерация ответов чат-бота.""" -import os from pathlib import Path from typing import Any, Dict, List, Optional @@ -9,6 +8,32 @@ from openai import AsyncOpenAI from src.rag.indexer import get_global_rag, get_project_rag +CHAT_MODES = { + "hybrid": { + "system": ( + "Ты — ассистент по базе знаний строительной компании. " + "Отвечай на основе контекста. Указывай источники (файл, страница, лист). " + "Если данных нет — так и скажи." + ), + "instruction": "Ответь на вопрос:", + }, + "compare": { + "system": ( + "Ты — аналитик. Сопоставь информацию из разных документов в контексте. " + "Найди совпадения, расхождения, противоречия. Структурируй ответ по пунктам." + ), + "instruction": "Сопоставь и сравни информацию по запросу:", + }, + "timeline": { + "system": ( + "Ты — аналитик хронологии проекта. " + "Восстанови timeline событий, решений, изменений по датам из контекста. " + "Сортируй по времени, указывай источник каждого события." + ), + "instruction": "Построй хронологию / timeline по запросу:", + }, +} + async def retrieve_context( question: str, @@ -19,17 +44,6 @@ async def retrieve_context( base_url: str = "https://opencode.ai/zen/v1", index_model: str = "mimo-v2.5-free", ) -> str: - """Извлекает релевантный контекст из LightRAG. - - Args: - question: вопрос пользователя. - working_dir_base: базовая директория индексов. - project_name: если None — ищет в глобальном индексе. - mode: режим поиска LightRAG (naive, local, global, hybrid). - - Returns: - Строка с найденным контекстом. - """ if project_name: rag = await get_project_rag( project_name, @@ -46,8 +60,7 @@ async def retrieve_context( base_url=base_url, ) - # only_need_context=True — возвращает только найденные фрагменты без генерации ответа - param = QueryParam(mode=mode, only_need_context=True) + param = QueryParam(mode=mode if mode in ("naive", "local", "global", "hybrid") else "hybrid", only_need_context=True) context = await rag.aquery(question, param=param) return context if context else "" @@ -59,20 +72,8 @@ async def generate_chat_response( api_key: str, base_url: str = "https://opencode.ai/zen/v1", model: str = "deepseek-v4-flash-free", + chat_mode: str = "hybrid", ) -> str: - """Генерирует ответ чат-бота через DeepSeek (или другую модель). - - Args: - question: вопрос пользователя. - context: контекст из RAG. - history: список {"question": ..., "answer": ...} предыдущих сообщений. - api_key: API ключ. - base_url: base URL OpenCode. - model: модель для чата. - - Returns: - Ответ ассистента. - """ if not api_key: raise ValueError( "OPENCODE_API_KEY не задан. Укажите rag.opencode_api_key в config.yaml " @@ -80,25 +81,19 @@ async def generate_chat_response( ) client = AsyncOpenAI(base_url=base_url, api_key=api_key) - system_prompt = ( - "Ты — ассистент по протоколам совещаний строительной компании. " - "Отвечай на основе предоставленного контекста из протоколов. " - "Если в контексте нет ответа — так и скажи, не выдумывай. " - "Отвечай кратко и по делу." - ) - - messages = [{"role": "system", "content": system_prompt}] + mode_cfg = CHAT_MODES.get(chat_mode, CHAT_MODES["hybrid"]) + messages = [{"role": "system", "content": mode_cfg["system"]}] for h in history: messages.append({"role": "user", "content": h["question"]}) messages.append({"role": "assistant", "content": h["answer"]}) - user_prompt = f"""Контекст из протоколов совещаний: + user_prompt = f"""Контекст из базы знаний: --- {context} --- -Вопрос: {question} +{mode_cfg["instruction"]} {question} """ messages.append({"role": "user", "content": user_prompt}) @@ -122,17 +117,14 @@ async def rag_chat( chat_model: str = "deepseek-v4-flash-free", mode: str = "hybrid", index_model: str = "mimo-v2.5-free", + chat_mode: str = "hybrid", ) -> Dict[str, Any]: - """Полный цикл RAG-чата: retrieval + generation. - - Returns: - {"answer": str, "context": str, "project": str | None} - """ + retrieval_mode = mode if mode in ("naive", "local", "global", "hybrid") else "hybrid" context = await retrieve_context( question=question, working_dir_base=working_dir_base, project_name=project_name, - mode=mode, + mode=retrieval_mode, api_key=api_key, base_url=base_url, index_model=index_model, @@ -145,10 +137,12 @@ async def rag_chat( api_key=api_key, base_url=base_url, model=chat_model, + chat_mode=chat_mode, ) return { "answer": answer, "context": context, "project": project_name, + "chat_mode": chat_mode, } diff --git a/tests/test_auth.py b/tests/test_auth.py index 6e15608..5814ed7 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -9,6 +9,7 @@ from fastapi import FastAPI from fastapi.testclient import TestClient from backend.auth.database import bootstrap_from_config, init_db +from backend.auth.models import UserContext from backend.auth.routes import admin_router, router as auth_router @@ -119,7 +120,7 @@ class AuthTestCase(unittest.TestCase): projects = self.client.get("/api/auth/projects", headers=d_headers).json() slugs = {p["slug"] for p in projects["projects"]} - self.assertEqual(slugs, {"2026", "gp-merakom"}) + self.assertEqual(slugs, {"2026", "gp-merakom", "org-gp"}) global_query = self.client.post( "/api/rag/query-global", @@ -165,6 +166,40 @@ class AuthTestCase(unittest.TestCase): slugs = {p["slug"] for p in projects["projects"]} self.assertEqual(slugs, {"my-gp"}) + def test_admin_create_project_normalizes_slug(self): + admin_login = self.client.post( + "/api/auth/login", + json={"org_slug": "merakom", "username": "admin", "password": "admin123"}, + ).json() + headers = {"Authorization": f"Bearer {admin_login['access_token']}"} + + created = self.client.post( + "/api/admin/projects", + headers=headers, + json={"slug": "Org-GP!!", "name": "ГП org"}, + ) + self.assertEqual(created.status_code, 200) + self.assertEqual(created.json()["project"]["slug"], "org-gp") + + def test_task_progress_visible_only_to_owner(self): + task = {"org_slug": "merakom", "user_id": 2, "file": "secret.mp3"} + owner = UserContext( + user_id=2, username="worker", role="user", + org_id=1, org_slug="merakom", org_name="Test", + ) + other = UserContext( + user_id=3, username="other", role="user", + org_id=1, org_slug="merakom", org_name="Test", + ) + admin = UserContext( + user_id=1, username="admin", role="admin", + org_id=1, org_slug="merakom", org_name="Test", + ) + + self.assertTrue(owner.can_see_task(task)) + self.assertFalse(other.can_see_task(task)) + self.assertTrue(admin.can_see_task(task)) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_ingest.py b/tests/test_ingest.py new file mode 100644 index 0000000..f1a48b9 --- /dev/null +++ b/tests/test_ingest.py @@ -0,0 +1,111 @@ +"""Tests for document ingestion pipeline.""" + +import json +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +from src.ingest.classify import _parse_json +from src.ingest.formatter import format_index_document +from src.ingest.models import NormalizedDocument, DocumentChunk +from src.ingest.router import extract_document, is_audio_file, is_document_file, resolve_upload_kind + + +def test_is_audio_file(): + assert is_audio_file("meeting.mp4") is True + assert is_audio_file("notes.pdf") is False + + +def test_is_document_file(): + assert is_document_file("spec.pdf") is True + assert is_document_file("audio.wav") is False + assert is_document_file("data.xlsx") is True + + +def test_extract_text_md(): + with tempfile.TemporaryDirectory() as tmp: + md = Path(tmp) / "note.md" + md.write_text("# Заголовок\n\nТекст документа.", encoding="utf-8") + doc = extract_document(md, "test-project", "specification") + assert doc.full_text + assert "Текст документа" in doc.full_text + assert doc.project == "test-project" + assert doc.doc_type == "specification" + + +def test_extract_csv(): + with tempfile.TemporaryDirectory() as tmp: + csv_path = Path(tmp) / "data.csv" + csv_path.write_text("col1,col2\na,b\n", encoding="utf-8") + doc = extract_document(csv_path, "gp-2026", "estimate") + assert "col1" in doc.full_text + + +def test_format_index_document(): + doc = NormalizedDocument( + document_id="doc_test", + filename="test.md", + doc_type="report", + project="2026", + full_text="Содержание отчёта", + chunks=[DocumentChunk(text="Содержание отчёта", source="test.md")], + ) + metadata = { + "title": "Отчёт Q1", + "topic": "Финансы", + "summary": "Кратко", + "key_decisions": ["Утвердить бюджет"], + } + text = format_index_document(doc, metadata) + assert "Отчёт Q1" in text + assert "Утвердить бюджет" in text + assert "Содержание отчёта" in text + + +def test_parse_classify_json(): + raw = """```json + {"project": "2026", "doc_type": "contract", "title": "Договор", "topic": "Субподряд"} + ```""" + meta = _parse_json(raw, "2026", "other") + assert meta["doc_type"] == "contract" + assert meta["title"] == "Договор" + + +def test_resolve_upload_kind(): + assert resolve_upload_kind("spec.pdf") == "document" + assert resolve_upload_kind("call.mp3") == "audio" + try: + resolve_upload_kind("image.png") + assert False, "expected ValueError" + except ValueError: + pass + + +if __name__ == "__main__": + class IngestTestCase(unittest.TestCase): + def test_audio(self): + test_is_audio_file() + + def test_document(self): + test_is_document_file() + + def test_parse(self): + test_parse_classify_json() + + def test_format(self): + test_format_index_document() + + def test_extract_md(self): + test_extract_text_md() + + def test_extract_csv(self): + test_extract_csv() + + def test_route(self): + test_resolve_upload_kind() + + unittest.main(verbosity=2)