128 lines
4.3 KiB
Python
128 lines
4.3 KiB
Python
"""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
|