Backup automatico script del 2026-08-23 12:04

services/loogle-mcp/app/context/gitea_link.py

@@ -0,0 +1,154 @@
+# -*- coding: utf-8 -*-
+"""Collegamento progetti MCP ↔ repository Gitea."""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from ..knowledge import gitea
+
+LOGGER = logging.getLogger("loogle_mcp.context.gitea_link")
+
+README_CANDIDATES = ("README.md", "readme.md", "Readme.md", "README.MD")
+DOCS_DIR = "docs"
+MAX_README_CHARS = 12_000
+MAX_DOC_FILES = 25
+
+
+def normalize_gitea_repo(repo: Optional[str]) -> Optional[str]:
+    if repo is None:
+        return None
+    cleaned = repo.strip()
+    if not cleaned:
+        return None
+    owner, name = gitea.parse_repo(cleaned)
+    return f"{owner}/{name}"
+
+
+def verify_repo_access(username: str, repo: str) -> None:
+    owner, name = gitea.parse_repo(repo)
+    gitea._request("GET", f"/repos/{owner}/{name}", username=username)
+
+
+def fetch_readme(username: str, repo: str) -> Optional[dict]:
+    for path in README_CANDIDATES:
+        try:
+            data = gitea.get_file(repo, path, username=username)
+        except FileNotFoundError:
+            continue
+        except Exception as exc:
+            LOGGER.warning("README %s/%s non leggibile: %s", repo, path, exc)
+            continue
+        if data.get("type") != "file":
+            continue
+        content = (data.get("content") or "").strip()
+        if not content:
+            continue
+        truncated = len(content) > MAX_README_CHARS
+        return {
+            "path": path,
+            "sha": data.get("sha"),
+            "html_url": data.get("html_url"),
+            "content": content[:MAX_README_CHARS],
+            "truncated": truncated,
+        }
+    return None
+
+
+def fetch_docs_index(username: str, repo: str) -> list[dict]:
+    try:
+        data = gitea.get_file(repo, DOCS_DIR, username=username)
+    except FileNotFoundError:
+        return []
+    except Exception as exc:
+        LOGGER.warning("Directory docs/ non leggibile per %s: %s", repo, exc)
+        return []
+
+    if data.get("type") != "dir":
+        return []
+
+    entries = []
+    for item in data.get("entries") or []:
+        if item.get("type") != "file":
+            continue
+        path = item.get("path") or ""
+        if not path.lower().endswith((".md", ".txt", ".rst")):
+            continue
+        entries.append(
+            {
+                "path": path,
+                "size": item.get("size"),
+            }
+        )
+        if len(entries) >= MAX_DOC_FILES:
+            break
+    return entries
+
+
+def project_enrichment(username: str, gitea_repo: str) -> dict:
+    repo = normalize_gitea_repo(gitea_repo)
+    if not repo:
+        return {"linked": False}
+
+    base = {
+        "linked": True,
+        "repo": repo,
+        "html_url": f"{gitea.public_base_url()}/{repo}",
+    }
+
+    if not gitea.is_configured(username):
+        return {
+            **base,
+            "available": False,
+            "error": "Gitea non configurato per questo utente",
+        }
+
+    try:
+        verify_repo_access(username, repo)
+        readme = fetch_readme(username, repo)
+        docs = fetch_docs_index(username, repo)
+        open_issues = None
+        try:
+            issues = gitea.list_issues(repo, state="open", limit=1, username=username)
+            open_issues = issues.get("count", 0)
+        except Exception:
+            pass
+        return {
+            **base,
+            "available": True,
+            "readme": readme,
+            "docs_files": docs,
+            "open_issues": open_issues,
+        }
+    except Exception as exc:
+        return {
+            **base,
+            "available": False,
+            "error": str(exc),
+        }
+
+
+def readme_seed_block(repo: str, readme: dict) -> str:
+    path = readme.get("path") or "README.md"
+    content = readme.get("content") or ""
+    truncated_note = "\n\n*(README troncato — usa get_file per il testo completo)*" if readme.get("truncated") else ""
+    return (
+        f"<!-- seed:gitea {repo} {path} -->\n\n"
+        f"## Sorgente Gitea: `{repo}`\n\n"
+        f"Contenuto iniziale da `{path}`.\n\n"
+        f"{content.rstrip()}{truncated_note}\n"
+    )
+
+
+def seed_context_from_readme(username: str, repo: str, context_md: str) -> tuple[str, bool]:
+    """Importa README nel context se non già presente un seed Gitea."""
+    if "<!-- seed:gitea " in context_md:
+        return context_md, False
+    readme = fetch_readme(username, repo)
+    if not readme:
+        return context_md, False
+    block = readme_seed_block(repo, readme)
+    if context_md.strip():
+        return context_md.rstrip() + "\n\n---\n\n" + block, True
+    return block, True

services/loogle-mcp/app/context/store.py

@@ -0,0 +1,252 @@
+# -*- 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"<!-- gitea_repo: {gitea_repo} -->\n\n" + text
+    return {
+        "uri": uri,
+        "mimeType": "text/markdown",
+        "text": text,
+    }