meldestelle/.junie/scripts/youtrack-sync-kb.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Spiegelt Markdown-Dateien (z. B. aus build/dokka/gfm) in eine YouTrack Knowledge Base.
- Verwendet Umgebungsvariablen YT_URL und YT_TOKEN
- Erwartet den KB-Root-Titel in KB_ROOT_TITLE (z. B. "API & Entwicklerdoku")
- Optional: KB_BC_ROOT (Unterordnername, z. B. "BCs") – wird aktuell nur als Titelpräfix genutzt

Sicherheit:
- Tokens werden niemals geloggt.
- Bei HTTP-Fehlern werden Statuscode und gekürzte Antwort ausgegeben.
"""
import argparse
import json
import os
import sys
import time
from pathlib import Path

try:
  import requests
except ImportError:  # pragma: no cover
  print("[YT] requests fehlt. Bitte 'pip install requests' ausführen.")
  sys.exit(2)

SESSION = requests.Session()
SESSION.headers.update({
  "Accept": "application/json",
  "Content-Type": "application/json",
})

# Globale Variablen, in main() gesetzt
PROJECT_ID = "MP"  # Standardwert, wird in main() überschrieben
KB_ID = None       # Knowledge-Base-ID des Projekts
KB_ROOT_ARTICLE_ID = None  # ID des echten KB-Wurzelartikels (Container)

def yt_url(path: str) -> str:
  base = os.environ.get("YT_URL", "").rstrip("/")
  if not base:
    print("[YT] YT_URL fehlt in Env.")
    sys.exit(2)
  if not path.startswith("/"):
    path = "/" + path
  return base + path


def set_auth():
  token = os.environ.get("YT_TOKEN")
  if not token:
    print("[YT] YT_TOKEN fehlt in Env.")
    sys.exit(2)
  # Bearer Token
  SESSION.headers["Authorization"] = f"Bearer {token}"


def http(method: str, url: str, **kw):
  # Einfaches Retry bei 429/5xx
  for attempt in range(5):
    r = SESSION.request(method, url, timeout=30, **kw)
    if r.status_code in (429, 500, 502, 503, 504):
      wait = (attempt + 1) * 1.5
      print(f"[YT] {r.status_code} → Retry in {wait:.1f}s…")
      time.sleep(wait)
      continue
    return r
  return r

# *** KORRIGIERTE FUNKTION ***
def get_project_by_short_name(name: str):
  """Sucht die interne ID des Projekts anhand des Kürzels (z.B. 'MP').
  Versucht zuerst /api/projects, fällt bei Fehlern auf /api/admin/projects zurück.
  """
  # 1) Primärer, nicht-admin Endpunkt
  url1 = yt_url("/api/projects?fields=id,shortName")
  r1 = http("GET", url1)
  if r1.status_code == 200:
    for proj in r1.json():
      if proj.get("shortName") == name:
        print("[YT] Projekte via /api/projects gefunden.")
        return proj
  else:
    print(f"[YT] Projektliste (\"/api/projects\") fehlgeschlagen: HTTP {r1.status_code} {r1.text[:200]}")

  # 2) Fallback: Admin-Endpunkt (ältere Versionen/Setups)
  url2 = yt_url("/api/admin/projects?fields=id,shortName")
  r2 = http("GET", url2)
  if r2.status_code == 200:
    for proj in r2.json():
      if proj.get("shortName") == name:
        print("[YT] Projekte via /api/admin/projects gefunden.")
        return proj
    return None

  print(f"[YT] Projektliste (\"/api/admin/projects\") fehlgeschlagen: HTTP {r2.status_code} {r2.text[:200]}")
  sys.exit(1)

# Neues Hilfs-API: Knowledge Base des Projekts abfragen

def get_project_knowledge_base(project_id: str):
  """Liest die Knowledge-Base eines Projekts, inkl. Root-Container-Artikel.
  Versucht zuerst /api/projects/{id}, fällt bei Fehler auf /api/admin/projects/{id} zurück.
  """
  # 1) Primärer Endpunkt (minimal fields)
  url1 = yt_url(f"/api/projects/{project_id}?fields=knowledgeBase(id,rootArticle(id,title))")
  r1 = http("GET", url1)
  if r1.status_code == 200:
    data = r1.json()
    kb = data.get("knowledgeBase")
    if kb:
      print("[YT] Projekt-Details via /api/projects/{id} geladen.")
      return kb
  else:
    print(f"[YT] Projekt-Details (\"/api/projects/{project_id}\") fehlgeschlagen: HTTP {r1.status_code} {r1.text[:200]}")

  # 2) Fallback: Admin-Endpunkt (minimal fields)
  url2 = yt_url(f"/api/admin/projects/{project_id}?fields=knowledgeBase(id,rootArticle(id,title))")
  r2 = http("GET", url2)
  if r2.status_code == 200:
    data = r2.json()
    kb = data.get("knowledgeBase")
    if kb:
      print("[YT] Projekt-Details via /api/admin/projects/{id} geladen.")
      return kb
    return None

  print(f"[YT] Projekt-Details (\"/api/admin/projects/{project_id}\") fehlgeschlagen: HTTP {r2.status_code} {r2.text[:200]}")
  sys.exit(1)

# *** KORRIGIERTE FUNKTION ***
def find_article_in_kb_by_title(title: str, parent_id: str | None = None):
  """Sucht einen Artikel anhand des Titels in der KB des Projekts.
  Optional kann parent_id angegeben werden, um die Suche auf direkte Kinder eines Artikels einzuschränken.
  """
  url = yt_url("/api/articles")
  params = {
    "query": f'title: "{title}"',
    "fields": "id,title,knowledgeBase(id),project(id),parent(id)"
  }
  r = http("GET", url, params=params)
  if r.status_code != 200:
    print(f"[YT] Artikelsuche fehlgeschlagen: HTTP {r.status_code} {r.text[:400]}")
    sys.exit(1)

  for art in r.json():
    if art.get("title") != title:
      continue
    if KB_ID:
      kb = art.get("knowledgeBase") or {}
      if kb.get("id") != KB_ID:
        continue
    else:
      proj = (art.get("project") or {}).get("id")
      if proj != PROJECT_ID:
        continue
    if parent_id:
      parent = art.get("parent") or {}
      if parent.get("id") != parent_id:
        continue
    return art
  return None

# *** KORRIGIERTE FUNKTION ***
def create_article(title: str, markdown: str, parent_id: str | None = None):
  """Erstellt einen neuen Artikel in der Knowledge Base des Projekts.
  Fallback: Wenn keine KB existiert (KB_ID is None), wird der Artikel dem Projekt zugeordnet.
  """
  url = yt_url("/api/articles?fields=id,title")
  payload = {
    "title": title,
    "content": markdown,
  }
  if KB_ID:
    payload["knowledgeBase"] = {"id": KB_ID}
  else:
    payload["project"] = {"id": PROJECT_ID}
  if parent_id:
    payload["parent"] = {"id": parent_id}

  r = http("POST", url, data=json.dumps(payload))
  if r.status_code not in (200, 201):
    print(f"[YT] Artikel erstellen fehlgeschlagen: HTTP {r.status_code} {r.text[:400]}")
    sys.exit(1)
  return r.json()

# *** KORRIGIERTE FUNKTION ***
def update_article(article_id: str, markdown: str):
  """Aktualisiert einen bestehenden Artikel in der Knowledge Base des Projekts."""
  url = yt_url(f"/api/articles/{article_id}?fields=id")
  payload = {"content": markdown}

  # YouTrack API verwendet POST für Updates an Artikeln
  r = http("POST", url, data=json.dumps(payload))
  if r.status_code not in (200, 201):
    print(f"[YT] Artikel aktualisieren fehlgeschlagen: HTTP {r.status_code} {r.text[:400]}")
    sys.exit(1)


def build_title_from_path(rel_path: Path, bc_root: str | None) -> str:
  # Beispiel: infrastructure/gateway/index.md → "infrastructure / gateway / index.md"
  parts = list(rel_path.parts)
  title = " / ".join(parts)
  if bc_root:
    title = f"{bc_root} / {title}"
  return title


def load_markdown(path: Path) -> str:
  try:
    text = path.read_text(encoding="utf-8")
  except Exception as e:
    print(f"[YT] Kann Datei nicht lesen: {path}: {e}")
    sys.exit(1)
  return text


def main():
  global PROJECT_ID, KB_ID, KB_ROOT_ARTICLE_ID  # Zugriff auf globale Variablen

  ap = argparse.ArgumentParser(description="Sync Dokka Markdown nach YouTrack KB")
  ap.add_argument("--src", default="build/dokka/gfm", help="Quellverzeichnis (Markdown)")
  args = ap.parse_args()

  kb_root_title = os.environ.get("KB_ROOT_TITLE")
  bc_root = os.environ.get("KB_BC_ROOT")
  if not kb_root_title:
    print("[YT] KB_ROOT_TITLE fehlt in Env.")
    sys.exit(2)

  set_auth()

  src = Path(args.src)
  if not src.exists():
    print(f"[YT] Quelle nicht gefunden: {src} – nichts zu tun.")
    return 0

  # Projekt anhand des Kürzels ermitteln
  project_short_name = os.environ.get("YT_PROJECT", "MP").strip() or "MP"
  project = get_project_by_short_name(project_short_name)
  if not project:
    print(f"[YT] Projekt mit Kürzel '{project_short_name}' nicht gefunden.")
    sys.exit(1)
  PROJECT_ID = project["id"]
  print(f"[YT] Arbeite im Projekt: {project_short_name} (ID: {PROJECT_ID})")

  # Knowledge Base des Projekts abrufen
  kb = get_project_knowledge_base(PROJECT_ID)
  if not kb:
    print("[YT] Hinweis: Dieses Projekt hat (noch) keine Knowledge Base – Synchronisation wird übersprungen.")
    return 0
  KB_ID = kb.get("id")
  root_article = kb.get("rootArticle") or {}
  KB_ROOT_ARTICLE_ID = root_article.get("id")
  print(f"[YT] Verwende Knowledge Base: {KB_ID}; Root-Container: {KB_ROOT_ARTICLE_ID}")

  # Root-Artikel (logische Wurzel unterhalb des KB-Root-Containers) finden/erstellen
  kb_root = find_article_in_kb_by_title(kb_root_title, parent_id=KB_ROOT_ARTICLE_ID)
  if not kb_root:
    print(f"[YT] Root-Artikel '{kb_root_title}' nicht gefunden. Erstelle ihn unterhalb des KB-Root-Containers…")
    kb_root = create_article(kb_root_title, f"# {kb_root_title}\n\nDieser Artikel dient als Wurzel für die automatische KDoc-Synchronisation.", parent_id=KB_ROOT_ARTICLE_ID)
    print(f"[YT] Root-Artikel erstellt: {kb_root_title} (ID: {kb_root['id']})")

  kb_root_id = kb_root["id"]
  print(f"[YT] Verwende KB-Root: {kb_root_title} ({kb_root_id})")

  count = 0
  for md in src.rglob("*.md"):
    rel = md.relative_to(src)
    title = build_title_from_path(rel, bc_root)
    content = load_markdown(md)

    existing = find_article_in_kb_by_title(title, parent_id=kb_root_id)
    if existing:
      update_article(existing["id"], content)
      print(f"[YT] Aktualisiert: {title}")
    else:
      # Erstelle Artikel als Kind des Root-Artikels
      create_article(title, content, parent_id=kb_root_id)
      print(f"[YT] Erstellt: {title}")
    count += 1

  print(f"[YT] Fertig. {count} Artikel synchronisiert.")
  return 0


if __name__ == "__main__":
  sys.exit(main())