708 lines
31 KiB
Python
708 lines
31 KiB
Python
#!/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()
|