116 lines
4.3 KiB
Python
116 lines
4.3 KiB
Python
"""
|
|
Projection Markdown durable des memory_entries + commit Git automatique.
|
|
SQLite (models.py) reste la source servie en instantane par l'API/MCP ;
|
|
ce module ne fait que projeter et versionner, jamais l'inverse (on ne relit
|
|
jamais le .md pour repondre a une requete).
|
|
"""
|
|
import os
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import Dict, Any, List
|
|
|
|
import httpx
|
|
|
|
from app.models import list_memory_entries
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
GITEA_INTERNAL_URL = "http://172.17.0.1:3232" # jamais l'IP LAN depuis un container
|
|
GITEA_REPO = "bolbol/context-hub"
|
|
GITEA_TOKEN_ENV = "GITEA_TOKEN"
|
|
|
|
TYPE_ORDER = ["decision", "constraint", "best-practice", "common-error", "do-not-use"]
|
|
TYPE_LABELS = {
|
|
"decision": "Decisions",
|
|
"constraint": "Contraintes",
|
|
"best-practice": "Bonnes pratiques",
|
|
"common-error": "Erreurs communes",
|
|
"do-not-use": "A ne pas reessayer",
|
|
}
|
|
|
|
|
|
def _render_entry(e: Dict[str, Any]) -> str:
|
|
tags = ", ".join(e.get("tags") or [])
|
|
tags_line = f" — tags: {tags}" if tags else ""
|
|
return (
|
|
f"- **[{e['id']}]** {e['title']}\n"
|
|
f" {e['body']}\n"
|
|
f" _cree par {e['created_by']} le {e['created_at']}{tags_line}_\n"
|
|
)
|
|
|
|
|
|
def render_scope_markdown(scope: str, entries: List[Dict[str, Any]]) -> str:
|
|
generated_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
lines = [
|
|
f"# Memoire {scope} — genere automatiquement, ne pas editer a la main",
|
|
"",
|
|
"> Source de verite operationnelle : table `memory_entries` dans context-hub (SQLite + API/MCP).",
|
|
"> Ce fichier est une projection durable et versionnee, regeneree a chaque `record_lesson`.",
|
|
f"> Derniere generation : {generated_at}",
|
|
"",
|
|
]
|
|
|
|
by_project: Dict[Any, List[Dict[str, Any]]] = {}
|
|
for e in entries:
|
|
by_project.setdefault(e.get("project"), []).append(e)
|
|
|
|
projects = sorted(by_project.keys(), key=lambda p: (p is not None, p or ""))
|
|
|
|
if not entries:
|
|
lines.append("_Aucune entree active pour ce scope._")
|
|
|
|
for project in projects:
|
|
heading = "## Transverse" if project is None else f"## Projet: {project}"
|
|
lines.append(heading)
|
|
lines.append("")
|
|
by_type: Dict[str, List[Dict[str, Any]]] = {}
|
|
for e in by_project[project]:
|
|
by_type.setdefault(e["type"], []).append(e)
|
|
for type_ in TYPE_ORDER:
|
|
if type_ not in by_type:
|
|
continue
|
|
lines.append(f"### {TYPE_LABELS[type_]}")
|
|
lines.append("")
|
|
for e in sorted(by_type[type_], key=lambda x: x["created_at"], reverse=True):
|
|
lines.append(_render_entry(e))
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
async def push_scope_markdown(scope: str) -> Dict[str, Any]:
|
|
token = os.environ.get(GITEA_TOKEN_ENV)
|
|
if not token:
|
|
logger.warning("GITEA_TOKEN absent, projection %s.md non poussee", scope)
|
|
return {"pushed": False, "reason": "GITEA_TOKEN absent"}
|
|
|
|
entries = await list_memory_entries(scope, status="active")
|
|
content = render_scope_markdown(scope, entries)
|
|
path = f"docs/context-memory/{scope}.md"
|
|
url = f"{GITEA_INTERNAL_URL}/api/v1/repos/{GITEA_REPO}/contents/{path}"
|
|
headers = {"Authorization": f"token {token}"}
|
|
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
get_resp = await client.get(url, headers=headers)
|
|
sha = get_resp.json().get("sha") if get_resp.status_code == 200 else None
|
|
|
|
import base64
|
|
payload = {
|
|
"message": f"memory: regeneration {scope}.md ({len(entries)} entrees actives)",
|
|
"content": base64.b64encode(content.encode()).decode(),
|
|
"branch": "main",
|
|
}
|
|
if sha:
|
|
payload["sha"] = sha
|
|
put_resp = await client.put(url, headers=headers, json=payload)
|
|
else:
|
|
put_resp = await client.post(url, headers=headers, json=payload)
|
|
|
|
if put_resp.status_code not in (200, 201):
|
|
logger.error("Push Gitea echoue pour %s: %s %s", path, put_resp.status_code, put_resp.text[:300])
|
|
return {"pushed": False, "reason": f"HTTP {put_resp.status_code}"}
|
|
|
|
commit_sha = put_resp.json().get("commit", {}).get("sha", "")[:8]
|
|
logger.info("Projection %s poussee, commit %s", path, commit_sha)
|
|
return {"pushed": True, "commit": commit_sha, "path": path}
|