""" Ingest pipeline - due fasi: analisi + generazione """ import json import re from pathlib import Path from typing import List, Dict, Optional from . import llm, wiki_ops import config SYSTEM_PROMPT_ANALYSIS = """Sei un bibliotecario esperto. Analizza il documento fornito e estrai informazioni strutturate. Rispondi SOLO con JSON valido in questo formato: { "entities": ["nome persona/organizzazione/strumento"], "concepts": ["concetto/tecnica/metodo"], "claims": ["affermazione importante trovata"], "connections": ["riferimento a [[pagina-wiki]] esistente"], "conflicts": ["affermazione che contraddice qualcosa di esistente"] } Non inventare informazioni. Solo cio che e esplicitamente nel documento.""" SYSTEM_PROMPT_GENERATION = """Sei un wiki maintainer. Genera pagine markdown per una wiki basata sull'analisi fornita. Regole: - Una pagina per ogni entity e concept - Usa il formato: # Nome Pagina > Una frase che riassume. ## Tags `entity` o `concept` ## Body Contenuto dettagliato... ## Related - [[pagina-correlata]] ## Sources - [nome source](path) - Per cross-references usa [[nome-pagina]] - Mantieni i nomi delle pagine in minuscolo con trattini -rispondi con JSON: {"pages": [{"name": "nome", "category": "entities|concepts", "content": "contenuto markdown"}]}""" def chunk_text(text: str, max_chars: int = 3000) -> List[str]: """Suddivide testo in chunks se troppo lungo per Gemma.""" if len(text) <= max_chars: return [text] chunks = [] paragraphs = text.split("\n\n") current_chunk = "" for para in paragraphs: if len(current_chunk) + len(para) <= max_chars: current_chunk += "\n\n" + para else: if current_chunk: chunks.append(current_chunk.strip()) current_chunk = para if current_chunk: chunks.append(current_chunk.strip()) return chunks async def run_analysis(source_path: Path) -> Dict: """Fase 1: analizza il source e ritorna struttura.""" content = source_path.read_text(encoding="utf-8") prompt = f"{SYSTEM_PROMPT_ANALYSIS}\n\n## Documento:\n{content}" try: response = await llm.llm_call(prompt) # Estrai JSON dalla risposta json_match = re.search(r'\{.*\}', response, re.DOTALL) if json_match: return json.loads(json_match.group()) except Exception as e: return {"error": str(e), "entities": [], "concepts": [], "claims": [], "connections": [], "conflicts": []} return {"entities": [], "concepts": [], "claims": [], "connections": [], "conflicts": []} async def run_generation(source_path: Path, analysis: Dict) -> List[Dict]: """Fase 2: genera pagine wiki dall'analisi.""" content = source_path.read_text(encoding="utf-8") source_name = source_path.stem prompt = f"""{SYSTEM_PROMPT_GENERATION} ## Source file: {source_path.name} ## Analysis: {json.dumps(analysis, ensure_ascii=False)} ## Source content (per reference): {content[:2000]}... """ try: response = await llm.llm_call(prompt) json_match = re.search(r'\{.*\}', response, re.DOTALL) if json_match: data = json.loads(json_match.group()) return data.get("pages", []) except Exception: pass return [] async def ingest_source(source_path: Path) -> Dict: """Pipeline completa: analisi -> generazione -> scrittura file.""" result = { "source": str(source_path), "analysis": None, "pages_created": [], "pages_updated": [], "error": None } # Fase 1: Analisi analysis = await run_analysis(source_path) result["analysis"] = analysis if "error" in analysis: result["error"] = analysis["error"] return result # Fase 2: Generazione pages = await run_generation(source_path, analysis) result["pages_created"] = [p["name"] for p in pages] # Scrittura file wiki_root = config.WIKI_ROOT source_category = "sources" created_pages = [] for page in pages: page_name = page["name"].lower().replace(" ", "-") category = page.get("category", "concepts") page_content = page["content"] category_dir = wiki_root / "wiki" / category page_path = category_dir / f"{page_name}.md" wiki_ops.write_page(page_path, page_content) created_pages.append(page_name) # Aggiorna index if created_pages: wiki_ops.update_index(wiki_root, created_pages, category) # Append log wiki_ops.append_log(wiki_root, source_path.name, created_pages, []) return result async def ingest_streaming(source_path: Path): """Pipeline con streaming per feedback in tempo reale.""" wiki_root = config.WIKI_ROOT yield f"data: Inizio ingest di {source_path.name}\n\n" # Fase 1: Analisi yield f"data: [1/2] Analisi in corso...\n\n" analysis = await run_analysis(source_path) yield f"data: Analisi completata. Trovati: {len(analysis.get('entities', []))} entities, {len(analysis.get('concepts', []))} concepts\n\n" if "error" in analysis: yield f"data: ERROR: {analysis['error']}\n\n" return # Fase 2: Generazione yield f"data: [2/2] Generazione pagine wiki...\n\n" pages = await run_generation(source_path, analysis) created = [] for page in pages: page_name = page["name"].lower().replace(" ", "-") category = page.get("category", "concepts") page_content = page["content"] category_dir = wiki_root / "wiki" / category page_path = category_dir / f"{page_name}.md" wiki_ops.write_page(page_path, page_content) created.append(page_name) yield f"data: Creata: [[{page_name}]]\n\n" # Aggiorna index e log if created: wiki_ops.update_index(wiki_root, created, "concepts") wiki_ops.append_log(wiki_root, source_path.name, created, []) yield f"data: DONE: {len(created)} pagine create\n\n"