feat: Fusion bloc 1 — barre d'apps commune (ACC-1.1, #17) et références croisées external_ref (ACC-1.3, #18)

app/services/external_refs.py

@@ -0,0 +1,127 @@
+"""Références croisées ``external_ref`` — Fusion ACC-1.3 (Flowdeck).
+
+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 une page ou un projet
+(collection) Flowdeck — et pointe vers un objet d'une autre application
+par son URL profonde (``contrats/liens-profonds.md``). Une seule source
+de vérité par objet : des références, jamais des copies.
+"""
+
+from __future__ import annotations
+
+import re
+from urllib.parse import parse_qs, unquote, urlparse
+
+from app.db import get_conn
+
+APP_COLORS = {
+    "obsigate": "#7a5af5",
+    "flowdeck": "#2f6fed",
+    "tiktak": "#e8762c",
+    "kanbante": "#0d9488",
+}
+
+
+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
+
+
+def _row_to_ref(row) -> dict:
+    d = dict(row)
+    d["app_color"] = APP_COLORS.get(d["target_app"], "#888")
+    return d
+
+
+def list_refs(owner_type: str, owner_id: int) -> list[dict]:
+    """Références posées sur un objet Flowdeck, dans l'ordre de création."""
+    with get_conn() as conn:
+        rows = conn.execute(
+            "SELECT * FROM external_refs WHERE owner_type = ? AND owner_id = ? "
+            "ORDER BY created_at, id",
+            (owner_type, owner_id),
+        ).fetchall()
+    return [_row_to_ref(r) for r in rows]
+
+
+def add_ref(
+    user_id: int,
+    owner_type: str,
+    owner_id: int,
+    target_app: str,
+    target_type: str,
+    target_id: str,
+    target_url: str,
+    label: str,
+) -> dict:
+    """Ajoute une référence à un objet et la retourne (avec son ``id``)."""
+    with get_conn() as conn:
+        cur = conn.execute(
+            "INSERT INTO external_refs (user_id, owner_type, owner_id, target_app, "
+            "target_type, target_id, target_url, label) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+            (user_id, owner_type, owner_id, target_app, target_type, target_id,
+             target_url, label),
+        )
+        conn.commit()
+        row = conn.execute(
+            "SELECT * FROM external_refs WHERE id = ?", (cur.lastrowid,)
+        ).fetchone()
+    return _row_to_ref(row)
+
+
+def get_ref(ref_id: int) -> dict | None:
+    """Une référence par son identifiant, ou ``None``."""
+    with get_conn() as conn:
+        row = conn.execute(
+            "SELECT * FROM external_refs WHERE id = ?", (ref_id,)
+        ).fetchone()
+    return _row_to_ref(row) if row else None
+
+
+def delete_ref(ref_id: int) -> bool:
+    """Retire une référence ; ``True`` si elle existait."""
+    with get_conn() as conn:
+        cur = conn.execute("DELETE FROM external_refs WHERE id = ?", (ref_id,))
+        conn.commit()
+        return cur.rowcount > 0