mirror of
https://github.com/Maethik/gitea-heatmap-backfill.git
synced 2026-07-11 13:21:17 +00:00
backfill Gitea heatmap from Git history
This commit is contained in:
@@ -0,0 +1,329 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user