Files
Orchestrator/bahn/project-audit/scripts/gitlab-inventory.py
T
ankn a5f8fb49ab Migrate all repos into monorepo context folders
Bahn: aisupport, Analyse-O2C-C2S, awesome-bahn-mcp-servers, beam-mcp,
      Confluence_Bot, db-planet-mcp-server, O2C-Harness, project-audit,
      Projekt-KIQ-HP, teamlandkarte-mcp
Dhive: Jury-Voting
Privat: CV, NoteGraph (NOTE: NoteGraph needs complete redo after consolidation)
Shared: AI-Orchestrator, OrgMyLife, power_skills_and_more
Shared/references: symphony (read-only)

Bahn repos remain available as independent remotes - this monorepo
pulls them in via subtree, the originals are untouched.
2026-06-30 20:39:52 +02:00

408 lines
14 KiB
Python

#!/usr/bin/env python3
"""
GitLab Project Inventory Script für pathOS / Bestellsystem
===========================================================
Zieht alle Projekte aus der GitLab-Gruppe und erstellt eine strukturierte Übersicht.
Voraussetzungen:
- Python 3.8+
- pip install requests
Verwendung:
python gitlab-inventory.py
Du wirst nach deinem Personal Access Token gefragt (read_api Scope reicht).
"""
import requests
import json
import csv
import os
import sys
from datetime import datetime
from getpass import getpass
from pathlib import Path
# === Konfiguration ===
GITLAB_URL = "https://git.tech.rz.db.de"
GROUP_PATH = "bestellsystem1"
API_BASE = f"{GITLAB_URL}/api/v4"
OUTPUT_DIR = Path(__file__).parent.parent / "data" / "gitlab-inventory"
PER_PAGE = 100
def get_token():
"""Token aus Umgebungsvariable oder interaktiver Eingabe."""
token = os.environ.get("GITLAB_TOKEN")
if not token:
token = getpass("GitLab Personal Access Token (read_api): ")
return token
def api_get(session, endpoint, params=None):
"""Paginierte API-Abfrage mit automatischem Durchblättern."""
if params is None:
params = {}
params["per_page"] = PER_PAGE
all_results = []
page = 1
while True:
params["page"] = page
resp = session.get(f"{API_BASE}{endpoint}", params=params)
resp.raise_for_status()
data = resp.json()
if not data:
break
all_results.extend(data)
total_pages = int(resp.headers.get("X-Total-Pages", 1))
print(f" Seite {page}/{total_pages} geladen ({len(data)} Einträge)")
if page >= total_pages:
break
page += 1
return all_results
def fetch_group_info(session):
"""Gruppeninformationen abrufen."""
print(f"\n📁 Lade Gruppeninfo für '{GROUP_PATH}'...")
resp = session.get(f"{API_BASE}/groups/{requests.utils.quote(GROUP_PATH, safe='')}")
resp.raise_for_status()
return resp.json()
def fetch_all_projects(session):
"""Alle Projekte der Gruppe inkl. Untergruppen."""
print("\n📦 Lade alle Projekte (inkl. Untergruppen)...")
projects = api_get(
session,
f"/groups/{requests.utils.quote(GROUP_PATH, safe='')}/projects",
params={
"include_subgroups": "true",
"with_shared": "false",
"order_by": "name",
"sort": "asc",
},
)
print(f" → {len(projects)} Projekte gefunden")
return projects
def fetch_subgroups(session):
"""Alle Untergruppen der Hauptgruppe."""
print("\n📂 Lade Untergruppen...")
subgroups = api_get(
session,
f"/groups/{requests.utils.quote(GROUP_PATH, safe='')}/subgroups",
params={"all_available": "true"},
)
print(f" → {len(subgroups)} Untergruppen gefunden")
return subgroups
def fetch_project_languages(session, project_id):
"""Programmiersprachen eines Projekts."""
try:
resp = session.get(f"{API_BASE}/projects/{project_id}/languages")
resp.raise_for_status()
return resp.json()
except Exception:
return {}
def fetch_project_readme(session, project_id):
"""README.md eines Projekts (falls vorhanden)."""
try:
resp = session.get(
f"{API_BASE}/projects/{project_id}/repository/files/README.md/raw",
params={"ref": "main"},
)
if resp.status_code == 404:
resp = session.get(
f"{API_BASE}/projects/{project_id}/repository/files/README.md/raw",
params={"ref": "master"},
)
if resp.status_code == 200:
return resp.text[:3000] # Erste 3000 Zeichen
except Exception:
pass
return None
def fetch_repo_tree(session, project_id, path="", ref="main"):
"""Top-Level Verzeichnisstruktur eines Repos."""
try:
resp = session.get(
f"{API_BASE}/projects/{project_id}/repository/tree",
params={"ref": ref, "path": path, "per_page": 100},
)
if resp.status_code == 404 and ref == "main":
return fetch_repo_tree(session, project_id, path, "master")
if resp.status_code == 200:
return resp.json()
except Exception:
pass
return []
def enrich_project(session, project, index, total):
"""Projekt mit zusätzlichen Details anreichern."""
pid = project["id"]
name = project["name"]
print(f" [{index}/{total}] {name}...", end=" ", flush=True)
# Sprachen
languages = fetch_project_languages(session, pid)
project["_languages"] = languages
# README (erste 3000 Zeichen)
readme = fetch_project_readme(session, pid)
project["_readme_excerpt"] = readme
# Top-Level Dateistruktur
tree = fetch_repo_tree(session, pid)
project["_top_level_files"] = [
{"name": f["name"], "type": f["type"]} for f in tree
]
print("✓")
return project
def classify_project(project):
"""Projekt grob klassifizieren basierend auf verfügbaren Daten."""
name = project.get("name", "").lower()
desc = (project.get("description") or "").lower()
topics = [t.lower() for t in project.get("topics", [])]
files = [f["name"].lower() for f in project.get("_top_level_files", [])]
languages = project.get("_languages", {})
tags = []
# Sprach-Tags
if "Java" in languages:
tags.append("java")
if "Kotlin" in languages:
tags.append("kotlin")
if "TypeScript" in languages or "JavaScript" in languages:
tags.append("frontend")
if "Python" in languages:
tags.append("python")
# Build-System
if "pom.xml" in files:
tags.append("maven")
if "build.gradle" in files or "build.gradle.kts" in files:
tags.append("gradle")
if "package.json" in files:
tags.append("npm")
if "Dockerfile" in files or "dockerfile" in files:
tags.append("docker")
if any("helm" in f for f in files) or "Chart.yaml" in files:
tags.append("helm")
# Typ-Erkennung
if any(kw in name or kw in desc for kw in ["lib", "library", "common", "shared", "util"]):
tags.append("library")
if any(kw in name or kw in desc for kw in ["service", "api", "backend"]):
tags.append("service")
if any(kw in name or kw in desc for kw in ["ui", "frontend", "web", "app"]):
tags.append("frontend-app")
if any(kw in name or kw in desc for kw in ["test", "e2e", "integration"]):
tags.append("testing")
if any(kw in name or kw in desc for kw in ["config", "infra", "deploy", "pipeline", "ci"]):
tags.append("infrastructure")
if any(kw in name or kw in desc for kw in ["doc", "wiki", "readme"]):
tags.append("documentation")
# Aktivitätsstatus
last_activity = project.get("last_activity_at", "")
if last_activity:
try:
last = datetime.fromisoformat(last_activity.replace("Z", "+00:00"))
days_ago = (datetime.now(last.tzinfo) - last).days
if days_ago > 365:
tags.append("inactive-1y+")
elif days_ago > 180:
tags.append("inactive-6m+")
except Exception:
pass
if project.get("archived"):
tags.append("archived")
return tags
def save_outputs(projects, subgroups, group_info):
"""Alle Ergebnisse speichern."""
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
# 1. Vollständige JSON-Daten
json_path = OUTPUT_DIR / "projects-full.json"
with open(json_path, "w", encoding="utf-8") as f:
json.dump(projects, f, indent=2, ensure_ascii=False, default=str)
print(f"\n💾 Vollständige Daten: {json_path}")
# 2. Übersichts-CSV
csv_path = OUTPUT_DIR / "projects-overview.csv"
with open(csv_path, "w", encoding="utf-8", newline="") as f:
writer = csv.writer(f, delimiter=";")
writer.writerow([
"Name", "Pfad", "Beschreibung", "Sprachen", "Tags",
"Letzte Aktivität", "Erstellt", "Archiviert",
"Default Branch", "Namespace"
])
for p in projects:
langs = ", ".join(f"{k} ({v:.0f}%)" for k, v in p.get("_languages", {}).items())
tags = ", ".join(p.get("_tags", []))
writer.writerow([
p.get("name", ""),
p.get("path_with_namespace", ""),
(p.get("description") or "")[:200],
langs,
tags,
p.get("last_activity_at", ""),
p.get("created_at", ""),
p.get("archived", False),
p.get("default_branch", ""),
p.get("namespace", {}).get("full_path", ""),
])
print(f"📊 Übersichts-CSV: {csv_path}")
# 3. Markdown-Übersicht
md_path = OUTPUT_DIR / "projects-overview.md"
with open(md_path, "w", encoding="utf-8") as f:
f.write(f"# GitLab Projektübersicht — {group_info.get('name', GROUP_PATH)}\n\n")
f.write(f"> Generiert am {datetime.now().strftime('%Y-%m-%d %H:%M')}\n")
f.write(f"> Quelle: {GITLAB_URL}/{GROUP_PATH}\n")
f.write(f"> Anzahl Projekte: {len(projects)}\n\n")
# Untergruppen
if subgroups:
f.write("## Untergruppen\n\n")
for sg in subgroups:
f.write(f"- **{sg['name']}** (`{sg['full_path']}`) — {sg.get('description', 'Keine Beschreibung')}\n")
f.write("\n")
# Projekte nach Namespace gruppiert
by_namespace = {}
for p in projects:
ns = p.get("namespace", {}).get("full_path", "root")
by_namespace.setdefault(ns, []).append(p)
for ns in sorted(by_namespace.keys()):
f.write(f"## {ns}\n\n")
f.write("| Projekt | Beschreibung | Sprachen | Tags | Letzte Aktivität |\n")
f.write("|---------|-------------|----------|------|------------------|\n")
for p in by_namespace[ns]:
langs = ", ".join(p.get("_languages", {}).keys())
tags = ", ".join(p.get("_tags", []))
desc = (p.get("description") or "—")[:80]
last = (p.get("last_activity_at") or "—")[:10]
f.write(f"| {p['name']} | {desc} | {langs} | {tags} | {last} |\n")
f.write("\n")
print(f"📝 Markdown-Übersicht: {md_path}")
# 4. README-Sammlung
readme_dir = OUTPUT_DIR / "readmes"
readme_dir.mkdir(exist_ok=True)
count = 0
for p in projects:
if p.get("_readme_excerpt"):
safe_name = p["path_with_namespace"].replace("/", "__")
with open(readme_dir / f"{safe_name}.md", "w", encoding="utf-8") as f:
f.write(f"# {p['name']}\n\n")
f.write(f"> Quelle: {GITLAB_URL}/{p['path_with_namespace']}\n\n")
f.write(p["_readme_excerpt"])
count += 1
print(f"📖 READMEs gespeichert: {count} Dateien in {readme_dir}")
# 5. Statistik-Zusammenfassung
stats_path = OUTPUT_DIR / "statistics.md"
with open(stats_path, "w", encoding="utf-8") as f:
f.write("# Projektstatistiken\n\n")
f.write(f"- **Gesamtanzahl Projekte**: {len(projects)}\n")
f.write(f"- **Untergruppen**: {len(subgroups)}\n")
f.write(f"- **Archiviert**: {sum(1 for p in projects if p.get('archived'))}\n")
# Sprachen-Verteilung
lang_count = {}
for p in projects:
for lang in p.get("_languages", {}):
lang_count[lang] = lang_count.get(lang, 0) + 1
f.write("\n## Sprachen-Verteilung\n\n")
for lang, count in sorted(lang_count.items(), key=lambda x: -x[1]):
f.write(f"- {lang}: {count} Projekte\n")
# Tag-Verteilung
tag_count = {}
for p in projects:
for tag in p.get("_tags", []):
tag_count[tag] = tag_count.get(tag, 0) + 1
f.write("\n## Tag-Verteilung\n\n")
for tag, count in sorted(tag_count.items(), key=lambda x: -x[1]):
f.write(f"- {tag}: {count} Projekte\n")
print(f"📈 Statistiken: {stats_path}")
def main():
print("=" * 60)
print(" pathOS / Bestellsystem — GitLab Projekt-Inventar")
print("=" * 60)
print(f"\n GitLab: {GITLAB_URL}")
print(f" Gruppe: {GROUP_PATH}")
token = get_token()
session = requests.Session()
session.headers["PRIVATE-TOKEN"] = token
session.verify = True # Auf False setzen falls selbstsigniertes Zertifikat
try:
# Gruppeninfo
group_info = fetch_group_info(session)
print(f" ✓ Gruppe gefunden: {group_info.get('name')} (ID: {group_info.get('id')})")
# Untergruppen
subgroups = fetch_subgroups(session)
# Alle Projekte
projects = fetch_all_projects(session)
# Projekte anreichern
print(f"\n🔍 Reichere {len(projects)} Projekte mit Details an...")
for i, project in enumerate(projects, 1):
enrich_project(session, project, i, len(projects))
project["_tags"] = classify_project(project)
# Ergebnisse speichern
save_outputs(projects, subgroups, group_info)
print("\n✅ Fertig! Alle Daten wurden gespeichert.")
print(f" Nächster Schritt: Ergebnisse in project-audit/data/gitlab-inventory/ prüfen")
except requests.exceptions.HTTPError as e:
if e.response.status_code == 401:
print("\n❌ Authentifizierung fehlgeschlagen. Bitte Token prüfen (read_api Scope nötig).")
elif e.response.status_code == 404:
print(f"\n❌ Gruppe '{GROUP_PATH}' nicht gefunden. Pfad prüfen.")
else:
print(f"\n❌ HTTP-Fehler: {e}")
sys.exit(1)
except requests.exceptions.ConnectionError:
print(f"\n❌ Verbindung zu {GITLAB_URL} fehlgeschlagen. VPN aktiv?")
sys.exit(1)
if __name__ == "__main__":
main()