mirror of
https://github.com/Maethik/gitea-heatmap-backfill.git
synced 2026-07-11 13:21:17 +00:00
330 lines
11 KiB
Python
330 lines
11 KiB
Python
#!/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, is_deleted, ref_name, is_private,
|
|
content, created_unix)
|
|
VALUES (?, 5, ?, ?, NULL, 0, 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()
|