dd5553e327
Move the /tmp prototype (Baileys bridge + RAG consumer) into git as a proper sibling project to discord-bridge/: own systemd --user units (whatsapp bridge, rag consumer, dashboard, periodic Drive sync timer), a filesystem document store with a stdlib control dashboard (start/ stop/restart, document CRUD, reindex, Google Drive sync via rclone), and an idempotent ops/install.sh following the same conventions. Co-Authored-By: Claude Agent <noreply@anthropic.com>
54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
"""Depozitul de documente pentru indexarea RAG a lui Maria.
|
|
|
|
Fisiere text (.txt/.md) 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.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
import config
|
|
|
|
_SAFE_NAME = re.compile(r"^[A-Za-z0-9._-]{1,200}$")
|
|
|
|
|
|
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((".txt", ".md")):
|
|
raise ValueError("doar fisiere .txt sau .md")
|
|
return name
|
|
|
|
|
|
def list_documents() -> list[dict]:
|
|
config.DOCS_DIR.mkdir(parents=True, exist_ok=True)
|
|
out = []
|
|
for f in sorted(config.DOCS_DIR.glob("*")):
|
|
if not f.is_file() or f.suffix not in (".txt", ".md"):
|
|
continue
|
|
st = f.stat()
|
|
out.append({"name": f.name, "size": st.st_size, "mtime": st.st_mtime})
|
|
return out
|
|
|
|
|
|
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
|