155 lines
4.5 KiB
Python
155 lines
4.5 KiB
Python
# -*- 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
|