ObsiGate/backend/external_refs.py

"""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, 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 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.
    """
    try:
        u = urlparse((url or "").strip())
    except ValueError:
        return None
    host = (u.hostname or "").lower()
    if host == "tiktak.dracodev.net":
        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 host == "og.dracodev.net":
        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 host == "flowdeck.dracodev.net":
        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 host == "git.dracodev.net":
        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."""
    if not username:
        return []
    return [
        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``."""
    for r in _read_refs(username):
        if r.get("id") == ref_id:
            return 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 removed