ROMFASTSQL/proxmox/lxc171-claude-agent/maria-whatsapp-bridge/rag/store.py

"""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.

DOUA DIRECTOARE, un singur spatiu de nume:
  documents/        oglinda Google Drive. `rclone sync` STERGE de aici tot ce nu
                    exista in Drive — nu pune nimic de mana.
  documents-local/  documente care nu vin din Drive (dictionare proprii, ce se
                    adauga din dashboard). Sincronizarea nu se atinge de ele.
Cand acelasi nume apare in ambele, castiga cel din Drive: e sursa comuna a
echipei, iar copia locala ar putea fi o versiune veche uitata acolo.

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
(`<mesaj_eroare>`, `<rezolvare>`) 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 _all_files() -> list[pathlib.Path]:
    """Fisierele din ambele directoare. La acelasi nume, cel din Drive castiga."""
    vazute: dict[str, pathlib.Path] = {}
    for d in (config.DOCS_LOCAL_DIR, config.DOCS_DIR):  # Drive ultimul = suprascrie
        if not d.is_dir():
            continue
        for f in sorted(d.glob("*")):
            if f.is_file() and f.suffix in DOC_EXTENSIONS:
                vazute[f.name] = f
    return [vazute[n] for n in sorted(vazute)]


def path_for(name: str) -> pathlib.Path | None:
    """Unde e documentul, in oricare din cele doua directoare."""
    name = validate_name(name)
    for d in (config.DOCS_DIR, config.DOCS_LOCAL_DIR):
        p = d / name
        if p.is_file():
            return p
    return None


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 = _all_files()
    # 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,
            "local": f.parent == config.DOCS_LOCAL_DIR,
            "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:
    path = path_for(name)
    if path is None:
        raise FileNotFoundError(name)
    return path.read_text(encoding="utf-8")


def write_document(name: str, content: str) -> None:
    """Scrie in documents-local/, NU in oglinda Drive.

    Un document scris in `documents/` dispare la urmatorul `rclone sync` — asta
    se intampla pana acum cu tot ce se adauga din dashboard.
    """
    name = validate_name(name)
    config.DOCS_LOCAL_DIR.mkdir(parents=True, exist_ok=True)
    (config.DOCS_LOCAL_DIR / name).write_text(content, encoding="utf-8")


def delete_document(name: str) -> bool:
    """Sterge documentul de unde se afla. Cel din Drive revine la sincronizare."""
    path = path_for(name)
    if path is None:
        return False
    path.unlink()
    return True