Scripts de migration Gogs -> Gitea (depots git, collaborateurs, webhooks, dates de creation)
This commit is contained in:
@@ -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()
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user