Gogs2GiteaMigration/fix_creation_dates.py

#!/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()