"""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)