# -*- coding: utf-8 -*- """Filesystem-backed project context store.""" import json import os import re import secrets from datetime import datetime from typing import Optional from . import gitea_link from ..knowledge import gitea SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,62}$") def _context_root() -> str: return os.environ.get("MCP_CONTEXT_ROOT", "/data/context") def _user_root(username: str) -> str: path = os.path.join(_context_root(), username, "projects") os.makedirs(path, exist_ok=True) return path def _project_dir(username: str, project_id: str) -> str: if not SLUG_RE.match(project_id): raise ValueError("ID progetto non valido") return os.path.join(_user_root(username), project_id) def _load_meta(username: str, project_id: str) -> tuple[str, dict]: proj_dir = _project_dir(username, project_id) meta_path = os.path.join(proj_dir, "meta.json") if not os.path.isfile(meta_path): raise FileNotFoundError("Progetto non trovato") meta = json.load(open(meta_path, encoding="utf-8")) return meta_path, meta def _save_meta(meta_path: str, meta: dict) -> None: meta["updated_at"] = datetime.now().astimezone().isoformat(timespec="seconds") with open(meta_path, "w", encoding="utf-8") as f: json.dump(meta, f, ensure_ascii=False, indent=2) def _apply_gitea_link( username: str, meta: dict, gitea_repo: Optional[str], *, verify: bool = True, seed_from_gitea: bool = False, ) -> dict: repo = gitea_link.normalize_gitea_repo(gitea_repo) if verify and repo: if not gitea.is_configured(username): raise RuntimeError( "Gitea non configurato per questo utente — impossibile collegare il repository" ) gitea_link.verify_repo_access(username, repo) if repo: meta["gitea_repo"] = repo else: meta.pop("gitea_repo", None) if seed_from_gitea and repo: proj_dir = _project_dir(username, meta["id"]) ctx_path = os.path.join(proj_dir, "context.md") context_md = open(ctx_path, encoding="utf-8").read() if os.path.isfile(ctx_path) else "" new_md, changed = gitea_link.seed_context_from_readme(username, repo, context_md) if changed: with open(ctx_path, "w", encoding="utf-8") as f: f.write(new_md) meta["gitea_seeded_at"] = datetime.now().astimezone().isoformat(timespec="seconds") return meta def list_projects(username: str, include_archived: bool = False) -> list: root = _user_root(username) projects = [] if not os.path.isdir(root): return projects for name in sorted(os.listdir(root)): meta_path = os.path.join(root, name, "meta.json") if not os.path.isfile(meta_path): continue meta = json.load(open(meta_path, encoding="utf-8")) if meta.get("archived") and not include_archived: continue projects.append(meta) return projects def create_project( username: str, title: str, tags: Optional[list[str]] = None, gitea_repo: Optional[str] = None, seed_from_gitea: bool = False, ) -> dict: slug_base = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-") or "progetto" project_id = slug_base[:40] root = _user_root(username) while os.path.exists(os.path.join(root, project_id)): project_id = f"{slug_base[:32]}-{secrets.token_hex(2)}" proj_dir = _project_dir(username, project_id) os.makedirs(os.path.join(proj_dir, "sessions"), exist_ok=True) os.makedirs(os.path.join(proj_dir, "artifacts"), exist_ok=True) now = datetime.now().astimezone().isoformat(timespec="seconds") meta = { "id": project_id, "title": title.strip(), "tags": tags or [], "status": "active", "archived": False, "created_at": now, "updated_at": now, } repo = gitea_link.normalize_gitea_repo(gitea_repo) if repo: meta["gitea_repo"] = repo with open(os.path.join(proj_dir, "meta.json"), "w", encoding="utf-8") as f: json.dump(meta, f, ensure_ascii=False, indent=2) with open(os.path.join(proj_dir, "context.md"), "w", encoding="utf-8") as f: f.write(f"# {title.strip()}\n\n") if repo: meta = _apply_gitea_link( username, meta, repo, verify=True, seed_from_gitea=seed_from_gitea, ) _save_meta(os.path.join(proj_dir, "meta.json"), meta) return meta def link_project_repo( username: str, project_id: str, gitea_repo: Optional[str] = None, seed_from_gitea: bool = False, ) -> dict: meta_path, meta = _load_meta(username, project_id) meta = _apply_gitea_link( username, meta, gitea_repo, verify=bool(gitea_link.normalize_gitea_repo(gitea_repo)), seed_from_gitea=seed_from_gitea, ) _save_meta(meta_path, meta) return meta def get_project_context( username: str, project_id: str, session_limit: int = 5, include_gitea: bool = True, ) -> dict: proj_dir = _project_dir(username, project_id) meta_path = os.path.join(proj_dir, "meta.json") if not os.path.isfile(meta_path): raise FileNotFoundError("Progetto non trovato") meta = json.load(open(meta_path, encoding="utf-8")) ctx_path = os.path.join(proj_dir, "context.md") context_md = open(ctx_path, encoding="utf-8").read() if os.path.isfile(ctx_path) else "" sessions_dir = os.path.join(proj_dir, "sessions") sessions = [] if os.path.isdir(sessions_dir): files = sorted(os.listdir(sessions_dir), reverse=True)[:session_limit] for fname in files: path = os.path.join(sessions_dir, fname) if os.path.isfile(path): sessions.append({"name": fname, "content": open(path, encoding="utf-8").read()}) result = {"meta": meta, "context_md": context_md, "recent_sessions": sessions} gitea_repo = meta.get("gitea_repo") if include_gitea and gitea_repo: result["gitea"] = gitea_link.project_enrichment(username, gitea_repo) return result def save_context(username: str, project_id: str, content: str, mode: str = "append") -> dict: proj_dir = _project_dir(username, project_id) meta_path = os.path.join(proj_dir, "meta.json") if not os.path.isfile(meta_path): raise FileNotFoundError("Progetto non trovato") ctx_path = os.path.join(proj_dir, "context.md") if mode == "replace": text = content else: existing = open(ctx_path, encoding="utf-8").read() if os.path.isfile(ctx_path) else "" text = existing.rstrip() + "\n\n" + content.strip() + "\n" with open(ctx_path, "w", encoding="utf-8") as f: f.write(text) meta = json.load(open(meta_path, encoding="utf-8")) _save_meta(meta_path, meta) snapshot = datetime.now().strftime("%Y%m%d-%H%M%S") + ".md" with open(os.path.join(proj_dir, "sessions", snapshot), "w", encoding="utf-8") as f: f.write(content) return meta def archive_project(username: str, project_id: str, archived: bool = True) -> dict: meta_path, meta = _load_meta(username, project_id) meta["archived"] = archived meta["status"] = "archived" if archived else "active" _save_meta(meta_path, meta) return meta def list_resources(username: str) -> list: resources = [] for meta in list_projects(username, include_archived=False): desc = f"Contesto progetto {meta['id']}" if meta.get("gitea_repo"): desc += f" (Gitea: {meta['gitea_repo']})" resources.append( { "uri": f"loogle://context/{username}/{meta['id']}", "name": meta["title"], "description": desc, "mimeType": "text/markdown", } ) return resources def read_resource(username: str, uri: str) -> dict: prefix = f"loogle://context/{username}/" if not uri.startswith(prefix): raise FileNotFoundError("Risorsa non trovata") project_id = uri[len(prefix):] data = get_project_context(username, project_id, session_limit=0, include_gitea=False) text = data["context_md"] gitea_repo = data["meta"].get("gitea_repo") if gitea_repo: text = f"\n\n" + text return { "uri": uri, "mimeType": "text/markdown", "text": text, }