"""Depozitul de documente pentru indexarea RAG a lui Maria. Fisiere text (.txt/.md/.xml) sub STATE_DIR/documents/, un singur nivel (fara subdirectoare), ca numele afisat in dashboard sa fie neambiguu si sa poata servi direct ca parametru de request fara riscuri de traversare de cale. Cand acelasi document exista in mai multe formate (`d406_saft_knowledge.xml` si `d406_saft_knowledge.md`), la indexare intra UNUL SINGUR, cel cu formatul cel mai bogat: .xml > .md > .txt. Sursele .xml sunt structurate pe probleme (``, ``) si de regula mai noi decat exporturile .md. Celalalt ramane pe disc si se vede in dashboard, marcat ca umbrit — altfel acelasi raspuns ar aparea de doua ori in rezultatele RAG. """ from __future__ import annotations import pathlib import re import config _SAFE_NAME = re.compile(r"^[A-Za-z0-9._-]{1,200}$") # Extensiile acceptate in depozit. Sursa unica de adevar pentru sync.py si indexer.py; # dashboard-ul (discord-bridge/dashboard/api.py) tine o copie, fiindca nu importa # module din acest proiect (coliziune de nume pe `config`) — tine-le sincronizate. DOC_EXTENSIONS = (".txt", ".md", ".xml") # Mai mic = preferat cand acelasi nume de baza exista in mai multe formate. _FORMAT_RANK = {".xml": 0, ".md": 1, ".txt": 2} def validate_name(name: str) -> str: if not name or not _SAFE_NAME.match(name) or ".." in name or "/" in name: raise ValueError(f"nume de document invalid: {name!r}") if not name.endswith(DOC_EXTENSIONS): raise ValueError("doar fisiere " + ", ".join(DOC_EXTENSIONS)) return name def list_documents() -> list[dict]: """Tot ce e in depozit. `shadowed_by` spune de ce un fisier nu ajunge in index.""" config.DOCS_DIR.mkdir(parents=True, exist_ok=True) files = [ f for f in sorted(config.DOCS_DIR.glob("*")) if f.is_file() and f.suffix in DOC_EXTENSIONS ] # castigatorul per nume de baza: formatul cel mai bogat winners: dict[str, str] = {} for f in files: best = winners.get(f.stem) if best is None or _FORMAT_RANK[f.suffix] < _FORMAT_RANK[pathlib.Path(best).suffix]: winners[f.stem] = f.name out = [] for f in files: st = f.stat() winner = winners[f.stem] out.append({ "name": f.name, "size": st.st_size, "mtime": st.st_mtime, "shadowed_by": None if winner == f.name else winner, }) return out def documents_for_index() -> list[dict]: """Doar documentele care ajung efectiv in index (fara cele umbrite).""" return [d for d in list_documents() if not d["shadowed_by"]] def read_document(name: str) -> str: name = validate_name(name) return (config.DOCS_DIR / name).read_text(encoding="utf-8") def write_document(name: str, content: str) -> None: name = validate_name(name) config.DOCS_DIR.mkdir(parents=True, exist_ok=True) (config.DOCS_DIR / name).write_text(content, encoding="utf-8") def delete_document(name: str) -> bool: name = validate_name(name) path = config.DOCS_DIR / name if not path.exists(): return False path.unlink() return True