From eb68df9dfae34b2d91577262e0a0383a4afad29a Mon Sep 17 00:00:00 2001 From: claude_code Date: Thu, 27 Aug 2026 17:42:34 +0200 Subject: [PATCH] Scripts de migration Gogs -> Gitea (depots git, collaborateurs, webhooks, dates de creation) --- fix_creation_dates.py | 158 +++++++++ migrate_gogs_to_gitea.py | 707 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 865 insertions(+) create mode 100644 fix_creation_dates.py create mode 100644 migrate_gogs_to_gitea.py diff --git a/fix_creation_dates.py b/fix_creation_dates.py new file mode 100644 index 0000000..f6ecc9c --- /dev/null +++ b/fix_creation_dates.py @@ -0,0 +1,158 @@ +#!/usr/bin/env python3 +"""Aligne les dates de création affichées sur Gitea avec les dates d'origine Gogs. + +L'API Gitea n'expose PAS d'endpoint pour modifier "created_at" d'un dépôt existant +(vérifié : absent de EditRepoOption et MigrateRepoOptions). Seule une modification +directe en base de données (colonne repository.created_unix) permet de la corriger. + +Ce script NE SE CONNECTE JAMAIS À LA BASE DE DONNÉES ET N'EXÉCUTE AUCUN SQL. Il : + 1. compare, via les deux API REST (Gogs + Gitea, lecture seule), la date de + création d'origine sur Gogs et la date actuellement affichée sur Gitea ; + 2. écrit un rapport JSON détaillé (dates_report.json) ; + 3. génère un script SQL prêt à l'emploi (fix_dates.sql) que l'UTILISATEUR exécute + lui-même, après avoir vérifié le schéma de sa base et fait une sauvegarde. + +Le nom exact de la colonne (created_unix) est celui utilisé par Gitea depuis +plusieurs versions majeures, mais PEUT VARIER -- vérifier avec +`DESCRIBE repository;` avant d'exécuter le SQL généré. + +Usage : + python3 fix_creation_dates.py # tous les dépôts communs + python3 fix_creation_dates.py --only bachir/monrepo # un seul dépôt (test) +""" +import argparse +import json +import sys +from datetime import datetime, timezone +from pathlib import Path + +from migrate_gogs_to_gitea import ( + GogsClient, GiteaClient, discover_gogs_orgs, load_config, +) + +SCRIPT_DIR = Path(__file__).resolve().parent + + +def parse_iso(s): + # Gogs/Gitea renvoient un ISO8601 avec offset, ex: 2017-06-12T11:22:42+02:00 + return datetime.fromisoformat(s) + + +def build_report(gogs, gitea, cfg, only=None): + candidate_owners = set(gitea.list_all_usernames()) | set(gitea.list_all_orgnames()) | gogs.known_orgs + gogs_repos = gogs.list_all_repos(candidate_owners) + if only: + wanted = set(only) + gogs_repos = [r for r in gogs_repos if r["full_name"] in wanted] + + report = [] + for r in gogs_repos: + owner = r["owner"]["username"] + name = r["name"] + full_name = r["full_name"] + gitea_repo = gitea.get_repo(owner, name) + if gitea_repo is None: + continue # pas encore migré côté Gitea, rien à corriger pour l'instant + + gogs_created = parse_iso(r["created_at"]) + gitea_created = parse_iso(gitea_repo["created_at"]) + delta_created_days = abs((gitea_created - gogs_created).total_seconds()) / 86400 + + gogs_updated = parse_iso(r["updated_at"]) + gitea_updated = parse_iso(gitea_repo["updated_at"]) + delta_updated_days = abs((gitea_updated - gogs_updated).total_seconds()) / 86400 + + report.append({ + "full_name": full_name, + "owner": owner, + "name": name, + "gogs_created_at": r["created_at"], + "gitea_created_at": gitea_repo["created_at"], + "gogs_created_unix": int(gogs_created.timestamp()), + "delta_days": round(delta_created_days, 1), + "needs_fix": delta_created_days > 1, # >1 jour d'écart = à corriger + "gogs_updated_at": r["updated_at"], + "gitea_updated_at": gitea_repo["updated_at"], + "gogs_updated_unix": int(gogs_updated.timestamp()), + "delta_updated_days": round(delta_updated_days, 1), + "needs_fix_updated": delta_updated_days > 1, + }) + return report + + +def write_sql(report, sql_path): + to_fix = [e for e in report if e["needs_fix"] or e["needs_fix_updated"]] + lines = [ + "-- Généré par fix_creation_dates.py -- NE PAS EXÉCUTER SANS AVOIR VÉRIFIÉ :", + "-- 1. Le nom des colonnes cibles : DESCRIBE repository;", + "-- (created_unix / updated_unix sont les noms standards sur les versions", + "-- récentes de Gitea, mais vérifier avant d'exécuter)", + "-- 2. Qu'une sauvegarde de la base (ou au moins de la table repository) existe.", + "-- Le script valide (COMMIT) automatiquement à la fin -- fait pour une exécution", + "-- en une commande (mysql -u ... -p BASE < fix_dates.sql). Ne pas oublier le nom", + "-- de la base en argument (sinon: ERROR 1046 No database selected).", + "--", + f"-- {len(to_fix)} dépôt(s) à corriger sur {len(report)} comparé(s).", + "", + "START TRANSACTION;", + "", + ] + for e in to_fix: + sets = [] + if e["needs_fix"]: + sets.append(f"created_unix = {e['gogs_created_unix']}") + if e["needs_fix_updated"]: + sets.append(f"updated_unix = {e['gogs_updated_unix']}") + lines.append( + f"-- {e['full_name']}: created Gitea={e['gitea_created_at']} -> Gogs={e['gogs_created_at']}" + f" | updated Gitea={e['gitea_updated_at']} -> Gogs={e['gogs_updated_at']}" + ) + lines.append( + "UPDATE repository SET {sets} " + "WHERE lower_name = '{name}' AND owner_id = " + "(SELECT id FROM user WHERE lower_name = '{owner}');".format( + sets=", ".join(sets), + name=e["name"].lower().replace("'", "''"), + owner=e["owner"].lower().replace("'", "''"), + ) + ) + lines.append("") + lines.append("COMMIT;") + Path(sql_path).write_text("\n".join(lines), encoding="utf-8") + + +def main(): + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--only", action="append", help="ne comparer que owner/repo (répétable)") + ap.add_argument("--report", default="dates_report.json") + ap.add_argument("--sql", default="fix_dates.sql") + args = ap.parse_args() + + cfg = load_config() + known_orgs = discover_gogs_orgs(cfg["GOGS_URL"]) + gogs = GogsClient(cfg["GOGS_URL"], cfg["GOGS_TOKEN"], known_orgs=known_orgs) + gitea = GiteaClient(cfg["GITEA_URL"], cfg["GITEA_ADMIN_USERNAME"], cfg["GITEA_ADMIN_PASSWORD"]) + + print("Comparaison des dates création/mise à jour Gogs <-> Gitea (lecture seule)...") + report = build_report(gogs, gitea, cfg, only=args.only) + to_fix = [e for e in report if e["needs_fix"] or e["needs_fix_updated"]] + + print(f"\n{len(report)} dépôt(s) comparé(s), {len(to_fix)} avec au moins un écart > 1 jour.") + for e in to_fix[:20]: + bits = [] + if e["needs_fix"]: + bits.append(f"created {e['gitea_created_at'][:10]}->{e['gogs_created_at'][:10]}") + if e["needs_fix_updated"]: + bits.append(f"updated {e['gitea_updated_at'][:10]}->{e['gogs_updated_at'][:10]}") + print(f" {e['full_name']}: " + ", ".join(bits)) + if len(to_fix) > 20: + print(f" ... et {len(to_fix) - 20} autre(s), voir {args.report}") + + Path(args.report).write_text(json.dumps(report, indent=2, ensure_ascii=False)) + write_sql(report, args.sql) + print(f"\nRapport détaillé : {args.report}") + print(f"Script SQL généré (à VÉRIFIER puis exécuter toi-même) : {args.sql}") + + +if __name__ == "__main__": + main() diff --git a/migrate_gogs_to_gitea.py b/migrate_gogs_to_gitea.py new file mode 100644 index 0000000..4cfc5eb --- /dev/null +++ b/migrate_gogs_to_gitea.py @@ -0,0 +1,707 @@ +#!/usr/bin/env python3 +"""Migration Gogs -> Gitea (figureslibres.io) : dépôts git + collaborateurs + webhooks. + +Explicitement HORS scope : issues, wiki, milestones, labels, releases, pull requests. + +Fonctionnement : + - la liste des dépôts Gogs est reconstituée propriétaire par propriétaire : on part + de la liste des comptes/orgs connus côté Gitea (admin, synchronisés LDAP YunoHost) + puis on interroge /api/v1/users|orgs/{login}/repos sur Gogs pour chacun. + (/api/v1/repos/search a été testé et écarté : même avec un token admin il ne + renvoie que les dépôts publics + ceux du compte du token, pas tout le site) ; + - chaque dépôt est rapatrié via l'API Gitea native /repos/migrate (service=git, + volontairement PAS service=gogs : ce dernier interroge d'abord l'API Gogs et + utilise son "clone_url" public, qui repasse par le mur SSO -- service=git fait + un clone brut sur l'adresse fournie, donc reste bien sur le loopback) ; + - un dépôt déjà présent côté Gitea (ex: migré à la main) N'EST JAMAIS RECRÉÉ NI + ÉCRASÉ -- seuls les collaborateurs / webhooks / topic peuvent être complétés ; + - les contributeurs Gogs sont vérifiés un par un côté Gitea avant d'être ajoutés ; + ceux qui n'existent pas sont listés dans le rapport, jamais créés automatiquement ; + - chaque dépôt migré (ou complété) reçoit le topic Gitea "migrated-from-gogs" pour + pouvoir les repérer facilement dans l'UI Gitea ; + - le statut migré/non-migré de CHAQUE dépôt Gogs examiné est consigné dans le + rapport JSON (champ "status") -- rien n'est écrit dans Gogs lui-même : son API + n'expose pas d'endpoint d'édition de dépôt (PATCH /repos/:owner/:repo -> 404), + et modifier ça via le formulaire web aurait risqué de toucher au champ de + visibilité (privé/public) des dépôts. + +Par défaut le script tourne en DRY-RUN : rien n'est écrit nulle part, seul un plan +détaillé est affiché et enregistré dans un rapport JSON. Il faut passer --apply pour +exécuter réellement les actions. + +Configuration : + 1. cp config.env.example config.env + 2. éditer config.env et renseigner les identifiants des comptes admin Gogs et Gitea + (config.env n'est jamais affiché ni loggé par ce script) + +Note : l'API Gogs de cette instance rejette systématiquement l'authentification HTTP +Basic Auth (401 identique avec de bons identifiants, un mauvais mot de passe ou un +utilisateur inexistant -- Basic Auth semble désactivée côté serveur). L'API Gogs +utilise donc un TOKEN (GOGS_TOKEN, à générer via Gogs -> Settings -> Applications -> +Generate New Token) ; GOGS_ADMIN_USERNAME/PASSWORD ne servent plus qu'à l'étape de +clonage git (auth_username/auth_password transmis à Gitea pour /repos/migrate). + +Exemples : + python3 migrate_gogs_to_gitea.py # dry-run, tous les dépôts + python3 migrate_gogs_to_gitea.py --only bachir/monrepo # dry-run, un seul dépôt + python3 migrate_gogs_to_gitea.py --limit 3 # dry-run, 3 premiers dépôts + python3 migrate_gogs_to_gitea.py --apply --only bachir/monrepo # exécution réelle, ciblée + python3 migrate_gogs_to_gitea.py --apply # exécution réelle, tout + python3 migrate_gogs_to_gitea.py --verify-existing # compare les refs git (branches/tags) + # Gogs vs Gitea pour les dépôts déjà + # présents côté Gitea -- lecture seule, + # ne pousse/écrase jamais rien sur Gitea. +""" +import argparse +import json +import os +import re +import subprocess +import sys +from pathlib import Path +from urllib.parse import quote + +import requests + +SCRIPT_DIR = Path(__file__).resolve().parent +ENV_FILE = SCRIPT_DIR / "config.env" + +DEFAULTS = { + "GOGS_URL": "http://127.0.0.1:17750", + "GITEA_URL": "http://127.0.0.1:6002", + "MIGRATION_TOPIC": "migrated-from-gogs", + "AUTO_CREATE_ORGS": "true", + "WEBHOOK_EXCLUDE_SUBSTR": "figureslibres.io/kanboard", +} + + +def load_config(): + cfg = dict(DEFAULTS) + if ENV_FILE.exists(): + for line in ENV_FILE.read_text().splitlines(): + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + k, v = line.split("=", 1) + cfg[k.strip()] = v.strip().strip('"').strip("'") + for k in ("GOGS_URL", "GITEA_URL", "GOGS_ADMIN_USERNAME", "GOGS_ADMIN_PASSWORD", "GOGS_TOKEN", + "GITEA_ADMIN_USERNAME", "GITEA_ADMIN_PASSWORD", "MIGRATION_TOPIC", "AUTO_CREATE_ORGS", + "WEBHOOK_EXCLUDE_SUBSTR"): + if k in os.environ: + cfg[k] = os.environ[k] + missing = [k for k in ("GOGS_ADMIN_USERNAME", "GOGS_ADMIN_PASSWORD", "GOGS_TOKEN", + "GITEA_ADMIN_USERNAME", "GITEA_ADMIN_PASSWORD") if not cfg.get(k)] + if missing: + sys.exit( + f"Config manquante: {', '.join(missing)}. " + f"Copie config.env.example vers config.env et renseigne ces valeurs." + ) + return cfg + + +_NAV_KEYWORDS = { + "", "explore", "user", "admin", "help", "api", "issues", "assets", "css", "js", "img", + "organizations", "repos", "users", "login", "avatars", +} + + +def discover_gogs_orgs(base_url): + # GET /api/v1/orgs/{login} ne vérifie PAS le type sur cette instance : il renvoie + # 200 pour N'IMPORTE QUEL login existant (user OU org), vérifié empiriquement sur + # "bachir" (un simple utilisateur). On ne peut donc pas s'y fier pour distinguer + # user/org -- on scrape la page publique /explore/organizations à la place, qui + # elle liste les VRAIES organisations. Le préfixe (ROOT_URL) est lu dynamiquement + # sur data-suburl plutôt que codé en dur. + orgs = set() + s = requests.Session() + base = base_url.rstrip("/") + suburl = "" + for page in range(1, 10): + try: + r = s.get(f"{base}/explore/organizations", params={"page": page}, timeout=15) + except requests.RequestException: + break + if r.status_code != 200: + break + if page == 1: + m = re.search(r'data-suburl="([^"]*)"', r.text) + suburl = m.group(1) if m else "" + # un seul segment de chemin après le suburl, pas de point (exclut les assets + # .css/.js/.png et les liens externes), pour ne matcher que de vrais logins. + pattern = re.escape(suburl) + r'/([a-zA-Z0-9_\-]+)"' + found = {m for m in re.findall(pattern, r.text) if m not in _NAV_KEYWORDS} + before = len(orgs) + orgs |= found + if len(orgs) == before: + break + return orgs + + +class GogsClient: + def __init__(self, base_url, token, known_orgs=frozenset()): + self.base = base_url.rstrip("/") + "/api/v1" + self.s = requests.Session() + self.s.headers.update({"Accept": "application/json", "Authorization": f"token {token}"}) + self.known_orgs = set(known_orgs) + + def repos_for_owner(self, login): + # /repos/search ne fait PAS une vraie recherche site-wide même pour un compte + # admin (vérifié empiriquement : il ne renvoie que les dépôts publics + ceux + # possédés/collaborés par le compte du token). On énumère donc propriétaire + # par propriétaire via /users/{login}/repos ou /orgs/{login}/repos. + for kind in ("users", "orgs"): + r = self.s.get(f"{self.base}/{kind}/{login}/repos", timeout=30) + if r.status_code == 200: + return r.json() or [] + if r.status_code != 404: + r.raise_for_status() + return [] + + def list_all_repos(self, candidate_owners): + by_full_name = {} + for owner in candidate_owners: + for r in self.repos_for_owner(owner): + by_full_name[r["full_name"]] = r + return list(by_full_name.values()) + + def is_org(self, login): + return login in self.known_orgs + + def collaborators(self, owner, repo): + r = self.s.get(f"{self.base}/repos/{owner}/{repo}/collaborators", timeout=30) + if r.status_code == 404: + return [] + r.raise_for_status() + return r.json() or [] + + def hooks(self, owner, repo): + r = self.s.get(f"{self.base}/repos/{owner}/{repo}/hooks", timeout=30) + if r.status_code == 404: + return [] + r.raise_for_status() + return r.json() or [] + + +class GiteaClient: + def __init__(self, base_url, username, password): + self.base = base_url.rstrip("/") + "/api/v1" + self.s = requests.Session() + self.s.auth = (username, password) + self.s.headers.update({"Accept": "application/json"}) + + def _list_paginated(self, path): + items, page = [], 1 + while True: + r = self.s.get(f"{self.base}{path}", params={"limit": 50, "page": page}, timeout=30) + r.raise_for_status() + data = r.json() + if not data: + break + items.extend(data) + if 'rel="next"' not in r.headers.get("Link", ""): + break + page += 1 + return items + + def list_all_usernames(self): + # source de vérité pour la découverte des dépôts Gogs : ces comptes sont + # synchronisés via le LDAP YunoHost, donc identiques entre Gogs et Gitea. + return [u["login"] for u in self._list_paginated("/admin/users")] + + def list_all_orgnames(self): + return [o.get("username") or o.get("name") for o in self._list_paginated("/admin/orgs")] + + def user_exists(self, username): + r = self.s.get(f"{self.base}/users/{username}", timeout=30) + if r.status_code not in (200, 404): + r.raise_for_status() + return r.status_code == 200 + + def org_exists(self, org): + r = self.s.get(f"{self.base}/orgs/{org}", timeout=30) + if r.status_code not in (200, 404): + r.raise_for_status() + return r.status_code == 200 + + def create_org(self, username, visibility="private"): + r = self.s.post(f"{self.base}/orgs", json={"username": username, "visibility": visibility}, timeout=30) + r.raise_for_status() + return r.json() + + def repo_exists(self, owner, repo): + r = self.s.get(f"{self.base}/repos/{owner}/{repo}", timeout=30) + if r.status_code not in (200, 404): + r.raise_for_status() + return r.status_code == 200 + + def get_repo(self, owner, repo): + r = self.s.get(f"{self.base}/repos/{owner}/{repo}", timeout=30) + if r.status_code == 404: + return None + r.raise_for_status() + return r.json() + + def migrate_repo(self, clone_addr, repo_owner, repo_name, private, + description, auth_username, auth_password): + payload = { + "clone_addr": clone_addr, + "repo_owner": repo_owner, + "repo_name": repo_name, + "service": "git", + "private": private, + "description": description or "", + "mirror": False, + "issues": False, + "labels": False, + "milestones": False, + "pull_requests": False, + "releases": False, + "wiki": False, + "lfs": True, + "auth_username": auth_username, + "auth_password": auth_password, + } + r = self.s.post(f"{self.base}/repos/migrate", json=payload, timeout=300) + r.raise_for_status() + return r.json() + + def list_collaborators(self, owner, repo): + r = self.s.get(f"{self.base}/repos/{owner}/{repo}/collaborators", timeout=30) + if r.status_code == 404: + return [] + r.raise_for_status() + return [u["login"] for u in r.json()] + + def add_collaborator(self, owner, repo, username, permission): + r = self.s.put( + f"{self.base}/repos/{owner}/{repo}/collaborators/{username}", + json={"permission": permission}, timeout=30, + ) + r.raise_for_status() + + def list_hooks(self, owner, repo): + r = self.s.get(f"{self.base}/repos/{owner}/{repo}/hooks", timeout=30) + if r.status_code == 404: + return [] + r.raise_for_status() + return r.json() or [] + + def create_hook(self, owner, repo, hook_type, config, events, active): + r = self.s.post( + f"{self.base}/repos/{owner}/{repo}/hooks", + json={"type": hook_type, "config": config, "events": events, "active": active}, + timeout=30, + ) + r.raise_for_status() + return r.json() + + def get_topics(self, owner, repo): + r = self.s.get(f"{self.base}/repos/{owner}/{repo}/topics", timeout=30) + if r.status_code == 404: + return [] + r.raise_for_status() + return r.json().get("topics", []) + + def set_topics(self, owner, repo, topics): + r = self.s.put(f"{self.base}/repos/{owner}/{repo}/topics", json={"topics": topics}, timeout=30) + r.raise_for_status() + + +# Gogs collaborator objects n'exposent pas toujours un niveau de permission explicite +# selon la version -- si absent, on suppose "write" (c'est le sens historique de +# "collaborateur" chez Gogs). Vérifié/affiché en dry-run avant toute exécution réelle. +def guess_permission(gogs_collaborator): + perm = gogs_collaborator.get("permission") or gogs_collaborator.get("permissions") + if isinstance(perm, dict): + if perm.get("admin"): + return "admin" + if perm.get("push") or perm.get("write"): + return "write" + return "read" + if isinstance(perm, str) and perm in ("read", "write", "admin"): + return perm + return "write" + + +def is_migrated(entry): + return "migrate_git" in entry["actions"] or "skip_git_migrate_already_exists" in entry["actions"] + + +def build_plan(gogs, gitea, cfg, only=None, limit=None): + candidate_owners = set(gitea.list_all_usernames()) | set(gitea.list_all_orgnames()) | gogs.known_orgs + repos = gogs.list_all_repos(candidate_owners) + if only: + wanted = set(only) + repos = [r for r in repos if r["full_name"] in wanted] + if limit: + repos = repos[:limit] + + plan = [] + for r in repos: + owner = r["owner"]["username"] + name = r["name"] + full_name = r["full_name"] + entry = { + "full_name": full_name, + "owner": owner, + "name": name, + "private": r["private"], + "description": r.get("description", ""), + "clone_addr": f"{cfg['GOGS_URL'].rstrip('/')}/{owner}/{name}.git", + "owner_is_org": gogs.is_org(owner), + "owner_exists_on_gitea": None, + "repo_exists_on_gitea": None, + "actions": [], + "collaborators": [], + "missing_collaborators": [], + "hooks": [], + "excluded_hooks": [], + "warnings": [], + } + + if entry["owner_is_org"]: + entry["owner_exists_on_gitea"] = gitea.org_exists(owner) + if not entry["owner_exists_on_gitea"]: + if cfg.get("AUTO_CREATE_ORGS", "true").lower() == "true": + entry["actions"].append("create_org") + else: + entry["warnings"].append("organisation absente côté Gitea et AUTO_CREATE_ORGS=false -> dépôt ignoré") + entry["status"] = "NON MIGRÉ" + plan.append(entry) + continue + else: + entry["owner_exists_on_gitea"] = gitea.user_exists(owner) + if not entry["owner_exists_on_gitea"]: + entry["warnings"].append( + "utilisateur propriétaire absent côté Gitea -> dépôt ignoré " + "(les comptes utilisateurs ne sont jamais créés automatiquement)" + ) + entry["status"] = "NON MIGRÉ" + plan.append(entry) + continue + + entry["repo_exists_on_gitea"] = gitea.repo_exists(owner, name) + if entry["repo_exists_on_gitea"]: + entry["actions"].append("skip_git_migrate_already_exists") + else: + entry["actions"].append("migrate_git") + + # collaborateurs -- on exclut le compte de service (celui utilisé pour le + # token Gogs/l'auth git), ajouté aux dépôts privés uniquement pour pouvoir + # les découvrir, jamais destiné à rester collaborateur côté Gitea. + existing_target_collabs = gitea.list_collaborators(owner, name) if entry["repo_exists_on_gitea"] else [] + for c in gogs.collaborators(owner, name): + username = c.get("username") or c.get("login") + if not username or username == cfg["GOGS_ADMIN_USERNAME"]: + continue + if not gitea.user_exists(username): + entry["missing_collaborators"].append(username) + continue + perm = guess_permission(c) + already = username in existing_target_collabs + entry["collaborators"].append({"username": username, "permission": perm, "already_present": already}) + if not already: + entry["actions"].append(f"add_collaborator:{username}:{perm}") + + # webhooks -- on exclut ceux pointant vers l'URL kanboard interne (secrets non + # transférables, migration explicitement écartée à la demande de l'utilisateur) + exclude_substr = cfg.get("WEBHOOK_EXCLUDE_SUBSTR", "") + existing_target_hooks = gitea.list_hooks(owner, name) if entry["repo_exists_on_gitea"] else [] + existing_urls = {h.get("config", {}).get("url") for h in existing_target_hooks} + for h in gogs.hooks(owner, name): + config = h.get("config", {}) or {} + url = config.get("url") + if exclude_substr and url and exclude_substr in url: + entry["excluded_hooks"].append(url) + continue + has_secret = bool(config.get("secret")) + hook_info = { + "type": h.get("type", "gogs"), + "url": url, + "events": h.get("events", []), + "active": h.get("active", True), + "already_present": url in existing_urls, + "secret_transferable": has_secret, + } + if not has_secret: + entry["warnings"].append( + f"webhook {url}: secret non exposé par l'API Gogs -> à ressaisir " + f"manuellement côté Gitea si ce hook en utilise un" + ) + entry["hooks"].append(hook_info) + if not hook_info["already_present"]: + entry["actions"].append(f"create_hook:{url}") + + # topic de tag côté Gitea + topics = gitea.get_topics(owner, name) if entry["repo_exists_on_gitea"] else [] + if cfg["MIGRATION_TOPIC"] not in topics: + entry["actions"].append(f"add_topic:{cfg['MIGRATION_TOPIC']}") + + # statut prévisionnel, mis à jour après coup si --apply échoue sur ce dépôt + entry["status"] = "MIGRÉ" if is_migrated(entry) else "NON MIGRÉ" + + plan.append(entry) + return plan + + +def apply_plan(gogs, gitea, cfg, plan): + for entry in plan: + owner, name = entry["owner"], entry["name"] + try: + if "create_org" in entry["actions"]: + gitea.create_org(owner) + print(f" [org créée] {owner}") + + if "migrate_git" in entry["actions"]: + gitea.migrate_repo( + clone_addr=entry["clone_addr"], + repo_owner=owner, + repo_name=name, + private=entry["private"], + description=entry["description"], + auth_username=cfg["GOGS_ADMIN_USERNAME"], + auth_password=cfg["GOGS_ADMIN_PASSWORD"], + ) + print(f" [git migré] {entry['full_name']}") + + for c in entry["collaborators"]: + if not c["already_present"]: + gitea.add_collaborator(owner, name, c["username"], c["permission"]) + print(f" [collaborateur ajouté] {entry['full_name']} <- {c['username']} ({c['permission']})") + + for h in entry["hooks"]: + if not h["already_present"] and h["url"]: + gitea.create_hook(owner, name, h["type"], {"url": h["url"], "content_type": "json"}, + h["events"], h["active"]) + print(f" [webhook créé] {entry['full_name']} -> {h['url']}") + + if any(a.startswith("add_topic:") for a in entry["actions"]): + topics = gitea.get_topics(owner, name) + topic = cfg["MIGRATION_TOPIC"] + if topic not in topics: + gitea.set_topics(owner, name, topics + [topic]) + print(f" [topic ajouté] {entry['full_name']} -> {topic}") + + entry["applied"] = True + except requests.HTTPError as e: + entry["applied"] = False + entry["error"] = f"{e} -- {e.response.text[:300] if e.response is not None else ''}" + print(f" [ERREUR] {entry['full_name']}: {entry['error']}") + except Exception as e: + entry["applied"] = False + entry["error"] = str(e) + print(f" [ERREUR] {entry['full_name']}: {e}") + + entry["status"] = "MIGRÉ" if (entry.get("applied") and is_migrated(entry)) else "NON MIGRÉ" + + +def print_summary(plan, apply_mode): + total = len(plan) + ignored = [e for e in plan if not e["actions"] and e["warnings"]] + to_create = [e for e in plan if "migrate_git" in e["actions"]] + already = [e for e in plan if "skip_git_migrate_already_exists" in e["actions"]] + missing_users = sorted({u for e in plan for u in e["missing_collaborators"]}) + hook_secret_warnings = [e for e in plan for w in e["warnings"] if "secret non exposé" in w] + + print("\n" + "=" * 70) + print(f"{'APPLICATION RÉELLE' if apply_mode else 'DRY-RUN (aucune écriture)'} -- résumé") + print("=" * 70) + print(f"Dépôts examinés : {total}") + print(f" -> à migrer (git) : {len(to_create)}") + print(f" -> déjà présents (skip) : {len(already)}") + print(f" -> ignorés (owner absent): {len(ignored)}") + if missing_users: + print(f"Contributeurs Gogs absents de Gitea ({len(missing_users)}) : {', '.join(missing_users)}") + if hook_secret_warnings: + print(f"Webhooks avec secret non transférable : {len(hook_secret_warnings)} (voir rapport)") + print(f"Statut (voir champ 'status' du rapport) : " + f"{sum(1 for e in plan if e['status'] == 'MIGRÉ')} 'MIGRÉ', " + f"{sum(1 for e in plan if e['status'] == 'NON MIGRÉ')} 'NON MIGRÉ'") + print("=" * 70) + for e in plan: + if e["warnings"] and not e["actions"]: + print(f"IGNORÉ {e['full_name']}: {'; '.join(e['warnings'])}") + + +def _clone_url_with_auth(base_url, owner, name, username, password): + scheme, rest = base_url.rstrip("/").split("://", 1) + return f"{scheme}://{quote(username, safe='')}:{quote(password, safe='')}@{rest}/{owner}/{name}.git" + + +def git_refs(url): + r = subprocess.run( + ["git", "ls-remote", "--heads", "--tags", url], + capture_output=True, text=True, timeout=60, + ) + if r.returncode != 0: + raise RuntimeError(r.stderr.strip() or f"git ls-remote a échoué (code {r.returncode})") + refs = {} + for line in r.stdout.splitlines(): + if not line.strip(): + continue + sha, ref = line.split("\t", 1) + refs[ref] = sha + return refs + + +def push_missing_refs(cfg, owner, name, missing_refs): + # N'ajoute QUE des refs totalement absentes côté Gitea (vérifié par l'appelant + # via missing_in_gitea) -- push sans --force, donc impossible d'écraser quoi que + # ce soit d'existant. Les branches divergentes ("Gitea a avancé") ne sont jamais + # touchées ici, volontairement. + import tempfile + gogs_url = _clone_url_with_auth(cfg["GOGS_URL"], owner, name, cfg["GOGS_ADMIN_USERNAME"], cfg["GOGS_ADMIN_PASSWORD"]) + gitea_url = _clone_url_with_auth(cfg["GITEA_URL"], owner, name, cfg["GITEA_ADMIN_USERNAME"], cfg["GITEA_ADMIN_PASSWORD"]) + results = {} + with tempfile.TemporaryDirectory() as tmp: + subprocess.run(["git", "init", "-q", "--bare", tmp], check=True, timeout=30) + for ref in missing_refs: + try: + subprocess.run( + ["git", "fetch", "-q", gogs_url, f"{ref}:{ref}"], + cwd=tmp, check=True, capture_output=True, text=True, timeout=90, + ) + subprocess.run( + ["git", "push", "-q", gitea_url, f"{ref}:{ref}"], + cwd=tmp, check=True, capture_output=True, text=True, timeout=90, + ) + results[ref] = {"ok": True} + except subprocess.CalledProcessError as ex: + results[ref] = {"ok": False, "error": (ex.stderr or str(ex)).strip()[:300]} + except subprocess.TimeoutExpired: + results[ref] = {"ok": False, "error": "timeout"} + return results + + +def verify_existing(cfg, plan): + # Compare les refs (branches + tags) Gogs vs Gitea pour les dépôts déjà présents + # côté Gitea (migrés à la main). Lecture seule (git ls-remote) : ne clone, ne + # pousse et n'écrase jamais rien -- Gitea reste la référence dans tous les cas. + # L'ajout des refs manquantes (si demandé) se fait séparément via + # apply_missing_refs(), jamais dans cette fonction. + targets = [e for e in plan if "skip_git_migrate_already_exists" in e["actions"]] + results = [] + print(f"\nVérification des refs git (lecture seule) sur {len(targets)} dépôt(s) déjà présents côté Gitea...") + for e in targets: + owner, name = e["owner"], e["name"] + gogs_url = _clone_url_with_auth(cfg["GOGS_URL"], owner, name, cfg["GOGS_ADMIN_USERNAME"], cfg["GOGS_ADMIN_PASSWORD"]) + gitea_url = _clone_url_with_auth(cfg["GITEA_URL"], owner, name, cfg["GITEA_ADMIN_USERNAME"], cfg["GITEA_ADMIN_PASSWORD"]) + entry = {"full_name": e["full_name"], "owner": owner, "name": name} + try: + gogs_refs = git_refs(gogs_url) + gitea_refs = git_refs(gitea_url) + except Exception as ex: + entry["error"] = str(ex) + results.append(entry) + print(f" [ERREUR] {e['full_name']}: {ex}") + continue + + common = set(gogs_refs) & set(gitea_refs) + missing_in_gitea = sorted(set(gogs_refs) - set(gitea_refs)) + extra_in_gitea = sorted(set(gitea_refs) - set(gogs_refs)) + diverged = sorted(ref for ref in common if gogs_refs[ref] != gitea_refs[ref]) + matching = sorted(ref for ref in common if gogs_refs[ref] == gitea_refs[ref]) + + entry.update({ + "gogs_ref_count": len(gogs_refs), + "gitea_ref_count": len(gitea_refs), + "missing_in_gitea": missing_in_gitea, + "extra_in_gitea": extra_in_gitea, + "diverged": [{"ref": r, "gogs_sha": gogs_refs[r], "gitea_sha": gitea_refs[r]} for r in diverged], + "matching_count": len(matching), + }) + + status = "OK" if not missing_in_gitea and not diverged else "ÉCART" + print(f" [{status}] {e['full_name']} -- " + f"{len(matching)} identiques, {len(missing_in_gitea)} manquante(s) sur Gitea, " + f"{len(diverged)} divergente(s) (jamais touchées), {len(extra_in_gitea)} en plus sur Gitea") + if missing_in_gitea: + print(f" manquantes: {', '.join(missing_in_gitea)}") + if diverged: + print(f" divergentes (ignorées, Gitea reste tel quel): {', '.join(diverged)}") + + results.append(entry) + return results + + +def apply_missing_refs(cfg, verify_results): + # Deuxième passe, séparée de verify_existing() : pousse UNIQUEMENT les refs + # identifiées comme totalement absentes lors de la vérification (pas de nouvel + # appel git ls-remote, on pousse exactement ce qui a été montré à l'utilisateur). + for entry in verify_results: + if not entry.get("missing_in_gitea"): + continue + owner, name = entry["owner"], entry["name"] + print(f" {entry['full_name']} -- ajout de {len(entry['missing_in_gitea'])} ref(s)...") + push_results = push_missing_refs(cfg, owner, name, entry["missing_in_gitea"]) + entry["push_results"] = push_results + for ref, res in push_results.items(): + if res["ok"]: + print(f" [ajouté] {ref}") + else: + print(f" [ÉCHEC] {ref}: {res['error']}") + + +def main(): + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--apply", action="store_true", help="exécute réellement (par défaut: dry-run)") + ap.add_argument("--only", action="append", help="ne traiter que owner/repo (répétable)") + ap.add_argument("--limit", type=int, help="limiter aux N premiers dépôts trouvés") + ap.add_argument("--report", default="migration_report.json", help="fichier de rapport JSON") + ap.add_argument("--verify-existing", action="store_true", + help="compare les refs git Gogs/Gitea des dépôts déjà présents (lecture seule)") + ap.add_argument("--verify-report", default="verify_report.json", help="fichier de rapport de vérification") + args = ap.parse_args() + + cfg = load_config() + known_orgs = discover_gogs_orgs(cfg["GOGS_URL"]) + gogs = GogsClient(cfg["GOGS_URL"], cfg["GOGS_TOKEN"], known_orgs=known_orgs) + gitea = GiteaClient(cfg["GITEA_URL"], cfg["GITEA_ADMIN_USERNAME"], cfg["GITEA_ADMIN_PASSWORD"]) + print(f"Organisations Gogs détectées : {sorted(known_orgs) or '(aucune)'}") + + print(f"Découverte des dépôts Gogs ({cfg['GOGS_URL']}) ...") + plan = build_plan(gogs, gitea, cfg, only=args.only, limit=args.limit) + print(f"{len(plan)} dépôt(s) analysé(s).") + + if args.verify_existing: + verify_results = verify_existing(cfg, plan) + n_ok = sum(1 for r in verify_results if not r.get("error") and not r["missing_in_gitea"] and not r["diverged"]) + n_ecart = sum(1 for r in verify_results if not r.get("error") and (r["missing_in_gitea"] or r["diverged"])) + n_err = sum(1 for r in verify_results if r.get("error")) + total_missing = sum(len(r.get("missing_in_gitea", [])) for r in verify_results) + print(f"\n{n_ok} OK, {n_ecart} avec écart, {n_err} en erreur.") + + if args.apply and total_missing: + confirm = input( + f"\nATTENTION: {total_missing} ref(s) (branches/tags) totalement absentes vont être " + f"ajoutées sur Gitea, cible {cfg['GITEA_URL']}. Les branches divergentes ne sont " + f"JAMAIS touchées. Tape 'oui' pour confirmer: " + ) + if confirm.strip().lower() != "oui": + print("Annulé (aucune ref ajoutée).") + else: + apply_missing_refs(cfg, verify_results) + elif args.apply: + print("Aucune ref manquante à ajouter.") + + Path(args.verify_report).write_text(json.dumps(verify_results, indent=2, ensure_ascii=False)) + print(f"Rapport: {args.verify_report}") + return + + if args.apply: + confirm = input( + f"\nATTENTION: exécution RÉELLE sur {len(plan)} dépôt(s), cible {cfg['GITEA_URL']}.\n" + f"Tape 'oui' pour confirmer: " + ) + if confirm.strip().lower() != "oui": + print("Annulé.") + sys.exit(1) + apply_plan(gogs, gitea, cfg, plan) + + print_summary(plan, args.apply) + Path(args.report).write_text(json.dumps(plan, indent=2, ensure_ascii=False)) + print(f"\nRapport détaillé écrit dans: {args.report}") + + +if __name__ == "__main__": + main()