#!/usr/bin/env python3 """ Gitea Heatmap Backfill ====================== Retroactively populate the Gitea activity heatmap from existing Git history. Reads commit history from bare repositories on disk and inserts matching action records (op_type=5, PUSH_EVENT) into the Gitea SQLite database. Author attribution is resolved by matching commit author emails against the email_address table, so all registered emails (primary + alternatives) are covered. Usage ----- Run from the Docker host (not inside the container): python3 gitea_backfill.py Or with custom paths: DB_PATH=/path/to/gitea.db GITEA_ROOT=/path/to/repositories python3 gitea_backfill.py Requirements ------------ - Python 3.7+ - git available on the host - Read/write access to the Gitea SQLite database - git safe.directory configured if repos are owned by a different user: git config --global --add safe.directory '*' """ import os import sys import sqlite3 import subprocess from pathlib import Path from datetime import datetime, timezone import hashlib # --------------------------------------------------------------------------- # Configuration — override via environment variables or edit directly # --------------------------------------------------------------------------- DB_PATH = os.getenv("DB_PATH", "/etc/komodo/stacks/gitea/data/gitea/gitea.db") GITEA_ROOT = os.getenv("GITEA_ROOT", "/etc/komodo/stacks/gitea/data/git/repositories") BATCH_SIZE = int(os.getenv("BATCH_SIZE", "500")) # --------------------------------------------------------------------------- # Database helpers # --------------------------------------------------------------------------- def create_backup(db_path: str) -> str: """Create an atomic backup of the SQLite database using VACUUM INTO.""" backup_path = db_path.replace(".db", ".backup-before-backfill.db") print(f"[1] Creating backup → {backup_path}") if os.path.exists(backup_path): print(" [!] Backup already exists, skipping.") return backup_path try: conn = sqlite3.connect(db_path) conn.execute(f"VACUUM INTO '{backup_path}'") conn.close() print(" [✓] Backup created.") return backup_path except Exception as exc: print(f" [✗] Backup failed: {exc}") sys.exit(1) def load_user_map(db_path: str) -> dict[str, int]: """ Return a mapping of lowercase email → user_id. Reads from the email_address table so that both the primary address and all alternative addresses registered in Gitea are covered. """ conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute( "SELECT uid, lower_email FROM email_address WHERE is_activated = 1" ) rows = cur.fetchall() conn.close() return {lower_email.strip(): uid for uid, lower_email in rows if lower_email} def load_repo_map(db_path: str) -> dict[str, int]: """Return a mapping of 'owner/repo' (lowercase) → repo_id.""" conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT id, owner_name, name FROM repository") rows = cur.fetchall() conn.close() return { f"{owner}/{name}".lower(): rid for rid, owner, name in rows if owner and name } def load_existing_hashes(cur: sqlite3.Cursor, repo_id: int) -> set[str]: """ Load all backfill content hashes already present for a given repo. A single SELECT per repo replaces one SELECT per commit, making duplicate detection an O(1) in-memory set lookup. """ cur.execute( "SELECT content FROM action" " WHERE op_type = 5 AND repo_id = ? AND content IS NOT NULL", (repo_id,), ) return {row[0] for row in cur.fetchall()} # --------------------------------------------------------------------------- # Repository discovery # --------------------------------------------------------------------------- def find_repos(gitea_root: str) -> list[tuple[str, str]]: """ Walk the Gitea repository root and return all bare Git repositories, excluding wiki repos (*.wiki.git). Returns a list of (absolute_path, 'owner/repo') tuples. """ repos = [] root = Path(gitea_root) for owner_dir in root.iterdir(): if not owner_dir.is_dir(): continue for repo_dir in owner_dir.iterdir(): if not repo_dir.is_dir(): continue if repo_dir.name.endswith(".wiki.git"): continue # Bare repo detection: HEAD file + objects/ directory at root if (repo_dir / "HEAD").exists() and (repo_dir / "objects").is_dir(): repo_name = repo_dir.name.removesuffix(".git") key = f"{owner_dir.name}/{repo_name}".lower() repos.append((str(repo_dir), key)) return repos # --------------------------------------------------------------------------- # Git log extraction # --------------------------------------------------------------------------- def extract_commits(repo_path: str) -> list[dict]: """ Run `git log --all` on a bare repository and return a list of commits. Each commit is a dict with keys: sha, email, iso. """ cmd = ["git", "-C", repo_path, "log", "--all", "--format=%H|%ae|%aI"] try: result = subprocess.run( cmd, capture_output=True, text=True, check=True, timeout=120 ) except subprocess.CalledProcessError as exc: print(f" [✗] git log failed: {exc.stderr.strip()}") return [] except subprocess.TimeoutExpired: print(f" [✗] git log timed out") return [] commits = [] for line in result.stdout.splitlines(): parts = line.split("|") if len(parts) < 3: continue sha, email, iso = parts[0], parts[1].lower().strip(), parts[2].strip() if sha and email and iso: commits.append({"sha": sha, "email": email, "iso": iso}) return commits # --------------------------------------------------------------------------- # Date parsing # --------------------------------------------------------------------------- def iso_to_unix(iso_str: str) -> int | None: """ Parse an ISO 8601 timestamp with any UTC offset and return a Unix timestamp in UTC. Returns None if parsing fails. """ try: dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00")) return int(dt.astimezone(timezone.utc).timestamp()) except ValueError as exc: print(f" [⚠] Cannot parse date '{iso_str}': {exc}") return None # --------------------------------------------------------------------------- # Deduplication key # --------------------------------------------------------------------------- def make_content_hash(sha: str, repo_id: int, user_id: int) -> str: """ Produce a stable, unique identifier for a (commit, repo, user) triple. Stored in the action.content column and used to prevent duplicate rows on repeated runs. """ return hashlib.sha256(f"{sha}:{repo_id}:{user_id}".encode()).hexdigest() # --------------------------------------------------------------------------- # Core backfill logic # --------------------------------------------------------------------------- def backfill(db_path: str, user_map: dict, repo_map: dict, repos: list) -> None: """Insert PUSH_EVENT actions for every matched commit in every repo.""" conn = sqlite3.connect(db_path) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA synchronous=NORMAL") conn.execute("PRAGMA cache_size=-32000") # 32 MB page cache cur = conn.cursor() cur.execute("SELECT COUNT(*) FROM action WHERE op_type = 5") print(f"[2] Existing PUSH_EVENT rows: {cur.fetchone()[0]}\n") inserted = 0 skipped = 0 no_user = 0 no_repo = 0 for repo_path, repo_key in repos: repo_id = repo_map.get(repo_key) if not repo_id: print(f"[⚠] No database entry for repo '{repo_key}' — skipping") no_repo += 1 continue print(f"[→] {repo_key} (id={repo_id})") commits = extract_commits(repo_path) if not commits: print(" No commits found.") continue print(f" {len(commits)} commits found.") existing = load_existing_hashes(cur, repo_id) for commit in commits: user_id = user_map.get(commit["email"]) if not user_id: no_user += 1 continue content_hash = make_content_hash(commit["sha"], repo_id, user_id) if content_hash in existing: skipped += 1 continue created_unix = iso_to_unix(commit["iso"]) if created_unix is None: skipped += 1 continue cur.execute( """ INSERT INTO action (user_id, op_type, act_user_id, repo_id, comment_id, ref_name, is_private, content, created_unix) VALUES (?, 5, ?, ?, NULL, NULL, 0, ?, ?) """, (user_id, user_id, repo_id, content_hash, created_unix), ) existing.add(content_hash) inserted += 1 if inserted % BATCH_SIZE == 0: conn.commit() print(f" [{inserted} inserted so far]") conn.commit() conn.close() width = 44 print(f"\n{'─' * width}") print(f" Backfill complete") print(f"{'─' * width}") print(f" Inserted : {inserted}") print(f" Skipped : {skipped:<6} (already present or bad date)") print(f" No user : {no_user:<6} (email not registered in Gitea)") print(f" No repo : {no_repo:<6} (repo not found in database)") print(f"{'─' * width}") print(f"\nRestart Gitea to refresh the heatmap:") print(f" docker restart gitea\n") # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- def main() -> None: print("Gitea Heatmap Backfill") print("=" * 44) if not os.path.exists(DB_PATH): print(f"[✗] Database not found: {DB_PATH}") sys.exit(1) create_backup(DB_PATH) print() user_map = load_user_map(DB_PATH) print(f"[✓] {len(user_map)} email(s) loaded from email_address table") repo_map = load_repo_map(DB_PATH) print(f"[✓] {len(repo_map)} repo(s) loaded from repository table") repos = find_repos(GITEA_ROOT) print(f"[✓] {len(repos)} bare repo(s) found on disk") if not repos: print("\n[✗] No repositories found. Check GITEA_ROOT.") sys.exit(1) print() backfill(DB_PATH, user_map, repo_map, repos) if __name__ == "__main__": main()