flowdeck/app/services/external_refs.py

"""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()
    # v7.82.0 (dev/prod) : les hôtes reconnus viennent de la config
    # (TIKTAK_URL, DRACO_SSO_URL, FLOWDECK_URL/OAUTH_REDIRECT_URI,
    # GITEA_URL/KANBANTE_URL) — un dev colle des URL en `-dev`, la prod
    # des URL officielles. L'hôte officiel reste TOUJOURS reconnu en
    # plus de l'hôte configuré : les références déjà stockées portent
    # des URL absolues et aucune donnée n'est migrée pour un renommage.
    from app.services import draco_urls

    if host in draco_urls.hotes_connus("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 host in draco_urls.hotes_connus("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 host in draco_urls.hotes_connus("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 host in draco_urls.hotes_connus("gitea") | draco_urls.hotes_connus("kanbante"):
        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 url_affichage(target_url: str) -> str:
    """URL d'affichage d'une référence : l'hôte vient de la config.

    Les références **stockées** portent l'URL absolue collée par
    l'utilisateur (``target_url``, jamais migrée ni réécrite en base).
    À l'affichage, si l'hôte stocké est un hôte dracodev connu
    (officiel ou configuré), il est remplacé par l'hôte configuré de
    l'environnement courant — chemin, requête et fragment intacts :
    une réf créée en prod s'ouvre sur le dev quand on la consulte
    depuis le dev, et inversement. Hôte inconnu : URL inchangée.
    """
    try:
        u = urlparse((target_url or "").strip())
    except ValueError:
        return target_url
    host = (u.hostname or "").lower()
    if not host:
        return target_url
    from app.services import draco_urls

    base = None
    for app, url in (
        ("tiktak", draco_urls.tiktak_url()),
        ("obsigate", draco_urls.obsigate_url()),
        ("flowdeck", draco_urls.flowdeck_url()),
        ("gitea", draco_urls.gitea_url()),
        ("kanbante", draco_urls.kanbante_url()),
    ):
        if host in draco_urls.hotes_connus(app):
            base = url
            break
    if base is None:
        return target_url
    return u._replace(scheme=urlparse(base).scheme,
                      netloc=urlparse(base).netloc).geturl()


def _row_to_ref(row) -> dict:
    d = dict(row)
    d["app_color"] = APP_COLORS.get(d["target_app"], "#888")
    # Affichage seulement : la valeur stockée en base n'est pas touchée.
    d["target_url"] = url_affichage(d.get("target_url") or "")
    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