239 lines
8.7 KiB
Python
239 lines
8.7 KiB
Python
"""Références croisées ``external_ref`` des notes — Fusion #215 (ACC-1.3).
|
|
|
|
Modèle figé par le programme Fusion dans ``dracodev-commons``
|
|
(``contrats/external_ref.schema.json``, v0.1.0) : la référence vit chez
|
|
le propriétaire de l'objet source — ici la note ObsiGate, identifiée
|
|
par ``(vault, path)`` — et pointe vers un objet d'une autre application
|
|
(projet ou page Flowdeck, tâche ou liste TikTak, carte Kanbante, note
|
|
ObsiGate) par son URL profonde (``contrats/liens-profonds.md``). Une
|
|
seule source de vérité par objet : on stocke des références, jamais
|
|
des copies du contenu cible.
|
|
|
|
Stockage : JSON par utilisateur dans ``data/external_refs/``, même
|
|
patron que :mod:`backend.history` (pas de base de données, cf.
|
|
``AGENTS.md``). Le répertoire est surchargeable par la variable
|
|
d'environnement ``OBSIGATE_EXTERNAL_REFS_DIR`` (tests).
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import parse_qs, quote, unquote, urlparse
|
|
|
|
logger = logging.getLogger("obsigate.external_refs")
|
|
|
|
|
|
def _refs_dir() -> Path:
|
|
override = os.environ.get("OBSIGATE_EXTERNAL_REFS_DIR")
|
|
directory = Path(override) if override else Path("data/external_refs")
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
return directory
|
|
|
|
|
|
def _refs_file(username: str) -> Path:
|
|
return _refs_dir() / f"{username}_refs.json"
|
|
|
|
|
|
def _read_refs(username: str) -> list[dict[str, Any]]:
|
|
file = _refs_file(username)
|
|
if not file.exists():
|
|
return []
|
|
try:
|
|
data = json.loads(file.read_text(encoding="utf-8"))
|
|
return data if isinstance(data, list) else []
|
|
except Exception as e:
|
|
logger.error(f"Failed to read external refs from {file.name}: {e}")
|
|
return []
|
|
|
|
|
|
def _write_refs(username: str, refs: list[dict[str, Any]]) -> None:
|
|
file = _refs_file(username)
|
|
try:
|
|
tmp = file.with_suffix(".tmp")
|
|
tmp.write_text(json.dumps(refs, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
shutil.move(str(tmp), str(file))
|
|
except Exception as e:
|
|
logger.error(f"Failed to write external refs to {file.name}: {e}")
|
|
|
|
|
|
# ------------------------------------------------------------ parseur d'URL
|
|
|
|
def build_deep_link(target_app: str, target_type: str, target_id: str) -> str | None:
|
|
"""Construit l'URL profonde d'affichage depuis la **configuration**
|
|
courante (#223), ou ``None`` si la forme est inconnue.
|
|
|
|
Les références stockées portent une URL absolue figée au moment du
|
|
collage (voir ``add_ref``) : elle peut nommer l'autre environnement
|
|
(un lien collé en prod, lu en dev). L'affichage ne l'utilise donc
|
|
jamais telle quelle — il reconstruit l'URL depuis
|
|
``(target_app, target_type, target_id)`` et les bases publiques
|
|
configurées (``backend.draco_config``), au moment de la lecture.
|
|
Aucune donnée stockée n'est migrée ni réécrite.
|
|
"""
|
|
from backend.draco_config import app_url
|
|
|
|
if target_app == "tiktak":
|
|
if target_type == "task":
|
|
return f"{app_url('tiktak')}/?task={quote(target_id, safe='')}"
|
|
if target_type == "list":
|
|
return f"{app_url('tiktak')}/?list={quote(target_id, safe='')}"
|
|
return None
|
|
if target_app == "obsigate" and target_type == "note":
|
|
return f"{app_url('obsigate')}/#file={quote(target_id, safe=':')}"
|
|
if target_app == "flowdeck":
|
|
if target_type == "page":
|
|
return f"{app_url('flowdeck')}/pages/{quote(target_id, safe='')}"
|
|
if target_type == "project":
|
|
return f"{app_url('flowdeck')}/collections/{quote(target_id, safe='')}"
|
|
return None
|
|
if target_app == "kanbante" and target_type == "card":
|
|
m = re.fullmatch(r"([^/]+)/([^/#]+)#(\d+)", target_id or "")
|
|
if m:
|
|
owner, repo, num = m.groups()
|
|
return f"{app_url('gitea')}/{owner}/{repo}/issues/{num}"
|
|
return None
|
|
return None
|
|
|
|
|
|
def _avec_url_affichee(ref: dict[str, Any]) -> dict[str, Any]:
|
|
"""Copie de la référence dont ``target_url`` est reconstruite
|
|
depuis la config courante (#223) — le stockage n'est pas touché."""
|
|
construite = build_deep_link(
|
|
str(ref.get("target_app") or ""),
|
|
str(ref.get("target_type") or ""),
|
|
str(ref.get("target_id") or ""),
|
|
)
|
|
if construite is None:
|
|
return ref
|
|
copie = dict(ref)
|
|
copie["target_url"] = construite
|
|
return copie
|
|
|
|
|
|
def parse_deep_link(url: str) -> tuple[str, str, str, str] | None:
|
|
"""Reconnaît une URL profonde des applications dracodev.
|
|
|
|
Formes figées par ``dracodev-commons/contrats/liens-profonds.md``.
|
|
Retourne ``(target_app, target_type, target_id, label_par_défaut)``
|
|
ou ``None`` si l'URL ne correspond à aucune forme du contrat.
|
|
|
|
Hôtes reconnus (#223) : les hôtes **officiels** et les hôtes
|
|
**configurés** de chaque application (``backend.draco_config``) —
|
|
un lien collé depuis l'environnement courant (noms en ``-dev``
|
|
déclarés dans son ``.env``) comme depuis l'autre environnement
|
|
est donc reconnu ; aucun nom d'environnement n'est figé ici.
|
|
"""
|
|
from backend.draco_config import hotes_reconnus
|
|
|
|
try:
|
|
u = urlparse((url or "").strip())
|
|
except ValueError:
|
|
return None
|
|
host = (u.hostname or "").lower()
|
|
app = hotes_reconnus().get(host)
|
|
if app == "tiktak":
|
|
q = parse_qs(u.query)
|
|
if q.get("task"):
|
|
return ("tiktak", "task", q["task"][0], f"TikTak · tâche {q['task'][0]}")
|
|
if q.get("list"):
|
|
return ("tiktak", "list", q["list"][0], f"TikTak · liste {q['list'][0]}")
|
|
return None
|
|
if app == "obsigate":
|
|
frag = u.fragment or ""
|
|
if frag.startswith("file="):
|
|
cible = unquote(frag[5:])
|
|
if ":" in cible:
|
|
nom = cible.rsplit("/", 1)[-1]
|
|
return ("obsigate", "note", cible, f"ObsiGate · {nom}")
|
|
return None
|
|
if app == "flowdeck":
|
|
m = re.fullmatch(r"/pages/(\d+)", u.path or "")
|
|
if m:
|
|
return ("flowdeck", "page", m.group(1), f"Flowdeck · page {m.group(1)}")
|
|
m = re.fullmatch(r"/collections/(\d+)", u.path or "")
|
|
if m:
|
|
return ("flowdeck", "project", m.group(1), f"Flowdeck · projet {m.group(1)}")
|
|
return None
|
|
if app == "gitea":
|
|
m = re.fullmatch(r"/([^/]+)/([^/]+)/issues/(\d+)", u.path or "")
|
|
if m:
|
|
owner, repo, num = m.groups()
|
|
return ("kanbante", "card", f"{owner}/{repo}#{num}", f"Kanbante · {repo} #{num}")
|
|
return None
|
|
return None
|
|
|
|
|
|
# ------------------------------------------------------------ CRUD
|
|
|
|
def list_refs(username: str, vault: str, path: str) -> list[dict[str, Any]]:
|
|
"""Références posées sur une note, dans l'ordre de création.
|
|
|
|
``target_url`` de chaque référence retournée est **reconstruite
|
|
depuis la config courante** (#223, ``_avec_url_affichee``) : la
|
|
valeur stockée, figée au collage, n'est jamais servie telle quelle.
|
|
"""
|
|
if not username:
|
|
return []
|
|
return [
|
|
_avec_url_affichee(r) for r in _read_refs(username)
|
|
if r.get("owner_vault") == vault and r.get("owner_path") == path
|
|
]
|
|
|
|
|
|
def get_ref(username: str, ref_id: str) -> dict[str, Any] | None:
|
|
"""Une référence par son identifiant, ou ``None`` (URL reconstruite, #223)."""
|
|
for r in _read_refs(username):
|
|
if r.get("id") == ref_id:
|
|
return _avec_url_affichee(r)
|
|
return None
|
|
|
|
|
|
def add_ref(
|
|
username: str,
|
|
vault: str,
|
|
path: str,
|
|
target_app: str,
|
|
target_type: str,
|
|
target_id: str,
|
|
target_url: str,
|
|
label: str,
|
|
) -> dict[str, Any]:
|
|
"""Ajoute une référence à une note et la retourne (avec son ``id``)."""
|
|
refs = _read_refs(username)
|
|
ref = {
|
|
"id": uuid.uuid4().hex,
|
|
"owner_type": "obsigate.note",
|
|
"owner_id": f"{vault}:{path}",
|
|
"owner_vault": vault,
|
|
"owner_path": path,
|
|
"target_app": target_app,
|
|
"target_type": target_type,
|
|
"target_id": target_id,
|
|
"target_url": target_url,
|
|
"label": label,
|
|
"domain": None,
|
|
"status": "ok",
|
|
"created_by": "user",
|
|
"created_at": datetime.now(UTC).isoformat(),
|
|
}
|
|
refs.append(ref)
|
|
_write_refs(username, refs)
|
|
return ref
|
|
|
|
|
|
def delete_ref(username: str, ref_id: str) -> dict[str, Any] | None:
|
|
"""Retire une référence ; retourne la référence retirée ou ``None``."""
|
|
refs = _read_refs(username)
|
|
kept = [r for r in refs if r.get("id") != ref_id]
|
|
if len(kept) == len(refs):
|
|
return None
|
|
removed = next(r for r in refs if r.get("id") == ref_id)
|
|
_write_refs(username, kept)
|
|
return _avec_url_affichee(removed)
|