Login

ian / Forgejo Heatmap Backfill

Last active 1 hour ago

Like 0

Very lightly altered version of the gitea heatmap backfil found below (just adjusted the sqlite schema and nothing else, used to fix heatmap for recently added repos on forgejo 16.0.4) https://raw.githubusercontent.com/Maethik/gitea-heatmap-backfill/main/gitea_backfill.py

git forgejo
forgejo_heatmap_backfill.py Raw
1#!/usr/bin/env python3
2"""
3Gitea Heatmap Backfill
4======================
5Retroactively populate the Gitea activity heatmap from existing Git history.
6
7Reads commit history from bare repositories on disk and inserts matching
8action records (op_type=5, PUSH_EVENT) into the Gitea SQLite database.
9
10Author attribution is resolved by matching commit author emails against
11the email_address table, so all registered emails (primary + alternatives)
12are covered.
13
14Usage
15-----
16Run from the Docker host (not inside the container):
17
18 python3 gitea_backfill.py
19
20Or with custom paths:
21
22 DB_PATH=/path/to/gitea.db GITEA_ROOT=/path/to/repositories python3 gitea_backfill.py
23
24Requirements
25------------
26- Python 3.7+
27- git available on the host
28- Read/write access to the Gitea SQLite database
29- git safe.directory configured if repos are owned by a different user:
30 git config --global --add safe.directory '*'
31"""
32
33import os
34import sys
35import sqlite3
36import subprocess
37from pathlib import Path
38from datetime import datetime, timezone
39import hashlib
40
41# ---------------------------------------------------------------------------
42# Configuration — override via environment variables or edit directly
43# ---------------------------------------------------------------------------
44
45DB_PATH = os.getenv("DB_PATH", "/etc/komodo/stacks/gitea/data/gitea/gitea.db")
46GITEA_ROOT = os.getenv("GITEA_ROOT", "/etc/komodo/stacks/gitea/data/git/repositories")
47BATCH_SIZE = int(os.getenv("BATCH_SIZE", "500"))
48
49
50# ---------------------------------------------------------------------------
51# Database helpers
52# ---------------------------------------------------------------------------
53
54def create_backup(db_path: str) -> str:
55 """Create an atomic backup of the SQLite database using VACUUM INTO."""
56 backup_path = db_path.replace(".db", ".backup-before-backfill.db")
57 print(f"[1] Creating backup → {backup_path}")
58 if os.path.exists(backup_path):
59 print(" [!] Backup already exists, skipping.")
60 return backup_path
61 try:
62 conn = sqlite3.connect(db_path)
63 conn.execute(f"VACUUM INTO '{backup_path}'")
64 conn.close()
65 print(" [✓] Backup created.")
66 return backup_path
67 except Exception as exc:
68 print(f" [✗] Backup failed: {exc}")
69 sys.exit(1)
70
71
72def load_user_map(db_path: str) -> dict[str, int]:
73 """
74 Return a mapping of lowercase email → user_id.
75
76 Reads from the email_address table so that both the primary address
77 and all alternative addresses registered in Gitea are covered.
78 """
79 conn = sqlite3.connect(db_path)
80 cur = conn.cursor()
81 cur.execute(
82 "SELECT uid, lower_email FROM email_address WHERE is_activated = 1"
83 )
84 rows = cur.fetchall()
85 conn.close()
86 return {lower_email.strip(): uid for uid, lower_email in rows if lower_email}
87
88
89def load_repo_map(db_path: str) -> dict[str, int]:
90 """Return a mapping of 'owner/repo' (lowercase) → repo_id."""
91 conn = sqlite3.connect(db_path)
92 cur = conn.cursor()
93 cur.execute("SELECT id, owner_name, name FROM repository")
94 rows = cur.fetchall()
95 conn.close()
96 return {
97 f"{owner}/{name}".lower(): rid
98 for rid, owner, name in rows
99 if owner and name
100 }
101
102
103def load_existing_hashes(cur: sqlite3.Cursor, repo_id: int) -> set[str]:
104 """
105 Load all backfill content hashes already present for a given repo.
106
107 A single SELECT per repo replaces one SELECT per commit, making
108 duplicate detection an O(1) in-memory set lookup.
109 """
110 cur.execute(
111 "SELECT content FROM action"
112 " WHERE op_type = 5 AND repo_id = ? AND content IS NOT NULL",
113 (repo_id,),
114 )
115 return {row[0] for row in cur.fetchall()}
116
117
118# ---------------------------------------------------------------------------
119# Repository discovery
120# ---------------------------------------------------------------------------
121
122def find_repos(gitea_root: str) -> list[tuple[str, str]]:
123 """
124 Walk the Gitea repository root and return all bare Git repositories,
125 excluding wiki repos (*.wiki.git).
126
127 Returns a list of (absolute_path, 'owner/repo') tuples.
128 """
129 repos = []
130 root = Path(gitea_root)
131 for owner_dir in root.iterdir():
132 if not owner_dir.is_dir():
133 continue
134 for repo_dir in owner_dir.iterdir():
135 if not repo_dir.is_dir():
136 continue
137 if repo_dir.name.endswith(".wiki.git"):
138 continue
139 # Bare repo detection: HEAD file + objects/ directory at root
140 if (repo_dir / "HEAD").exists() and (repo_dir / "objects").is_dir():
141 repo_name = repo_dir.name.removesuffix(".git")
142 key = f"{owner_dir.name}/{repo_name}".lower()
143 repos.append((str(repo_dir), key))
144 return repos
145
146
147# ---------------------------------------------------------------------------
148# Git log extraction
149# ---------------------------------------------------------------------------
150
151def extract_commits(repo_path: str) -> list[dict]:
152 """
153 Run `git log --all` on a bare repository and return a list of commits.
154
155 Each commit is a dict with keys: sha, email, iso.
156 """
157 cmd = ["git", "-C", repo_path, "log", "--all", "--format=%H|%ae|%aI"]
158 try:
159 result = subprocess.run(
160 cmd, capture_output=True, text=True, check=True, timeout=120
161 )
162 except subprocess.CalledProcessError as exc:
163 print(f" [✗] git log failed: {exc.stderr.strip()}")
164 return []
165 except subprocess.TimeoutExpired:
166 print(f" [✗] git log timed out")
167 return []
168
169 commits = []
170 for line in result.stdout.splitlines():
171 parts = line.split("|")
172 if len(parts) < 3:
173 continue
174 sha, email, iso = parts[0], parts[1].lower().strip(), parts[2].strip()
175 if sha and email and iso:
176 commits.append({"sha": sha, "email": email, "iso": iso})
177 return commits
178
179
180# ---------------------------------------------------------------------------
181# Date parsing
182# ---------------------------------------------------------------------------
183
184def iso_to_unix(iso_str: str) -> int | None:
185 """
186 Parse an ISO 8601 timestamp with any UTC offset and return a Unix
187 timestamp in UTC. Returns None if parsing fails.
188 """
189 try:
190 dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
191 return int(dt.astimezone(timezone.utc).timestamp())
192 except ValueError as exc:
193 print(f" [⚠] Cannot parse date '{iso_str}': {exc}")
194 return None
195
196
197# ---------------------------------------------------------------------------
198# Deduplication key
199# ---------------------------------------------------------------------------
200
201def make_content_hash(sha: str, repo_id: int, user_id: int) -> str:
202 """
203 Produce a stable, unique identifier for a (commit, repo, user) triple.
204
205 Stored in the action.content column and used to prevent duplicate rows
206 on repeated runs.
207 """
208 return hashlib.sha256(f"{sha}:{repo_id}:{user_id}".encode()).hexdigest()
209
210
211# ---------------------------------------------------------------------------
212# Core backfill logic
213# ---------------------------------------------------------------------------
214
215def backfill(db_path: str, user_map: dict, repo_map: dict, repos: list) -> None:
216 """Insert PUSH_EVENT actions for every matched commit in every repo."""
217 conn = sqlite3.connect(db_path)
218 conn.execute("PRAGMA journal_mode=WAL")
219 conn.execute("PRAGMA synchronous=NORMAL")
220 conn.execute("PRAGMA cache_size=-32000") # 32 MB page cache
221 cur = conn.cursor()
222
223 cur.execute("SELECT COUNT(*) FROM action WHERE op_type = 5")
224 print(f"[2] Existing PUSH_EVENT rows: {cur.fetchone()[0]}\n")
225
226 inserted = 0
227 skipped = 0
228 no_user = 0
229 no_repo = 0
230
231 for repo_path, repo_key in repos:
232 repo_id = repo_map.get(repo_key)
233 if not repo_id:
234 print(f"[⚠] No database entry for repo '{repo_key}' — skipping")
235 no_repo += 1
236 continue
237
238 print(f"[→] {repo_key} (id={repo_id})")
239 commits = extract_commits(repo_path)
240 if not commits:
241 print(" No commits found.")
242 continue
243 print(f" {len(commits)} commits found.")
244
245 existing = load_existing_hashes(cur, repo_id)
246
247 for commit in commits:
248 user_id = user_map.get(commit["email"])
249 if not user_id:
250 no_user += 1
251 continue
252
253 content_hash = make_content_hash(commit["sha"], repo_id, user_id)
254 if content_hash in existing:
255 skipped += 1
256 continue
257
258 created_unix = iso_to_unix(commit["iso"])
259 if created_unix is None:
260 skipped += 1
261 continue
262
263 cur.execute(
264 """
265 INSERT INTO action
266 (user_id, op_type, act_user_id, repo_id,
267 comment_id, ref_name, is_private,
268 content, created_unix)
269 VALUES (?, 5, ?, ?, NULL, NULL, 0, ?, ?)
270 """,
271 (user_id, user_id, repo_id, content_hash, created_unix),
272 )
273 existing.add(content_hash)
274 inserted += 1
275
276 if inserted % BATCH_SIZE == 0:
277 conn.commit()
278 print(f" [{inserted} inserted so far]")
279
280 conn.commit()
281 conn.close()
282
283 width = 44
284 print(f"\n{'─' * width}")
285 print(f" Backfill complete")
286 print(f"{'─' * width}")
287 print(f" Inserted : {inserted}")
288 print(f" Skipped : {skipped:<6} (already present or bad date)")
289 print(f" No user : {no_user:<6} (email not registered in Gitea)")
290 print(f" No repo : {no_repo:<6} (repo not found in database)")
291 print(f"{'─' * width}")
292 print(f"\nRestart Gitea to refresh the heatmap:")
293 print(f" docker restart gitea\n")
294
295
296# ---------------------------------------------------------------------------
297# Entry point
298# ---------------------------------------------------------------------------
299
300def main() -> None:
301 print("Gitea Heatmap Backfill")
302 print("=" * 44)
303
304 if not os.path.exists(DB_PATH):
305 print(f"[✗] Database not found: {DB_PATH}")
306 sys.exit(1)
307
308 create_backup(DB_PATH)
309 print()
310
311 user_map = load_user_map(DB_PATH)
312 print(f"[✓] {len(user_map)} email(s) loaded from email_address table")
313
314 repo_map = load_repo_map(DB_PATH)
315 print(f"[✓] {len(repo_map)} repo(s) loaded from repository table")
316
317 repos = find_repos(GITEA_ROOT)
318 print(f"[✓] {len(repos)} bare repo(s) found on disk")
319
320 if not repos:
321 print("\n[✗] No repositories found. Check GITEA_ROOT.")
322 sys.exit(1)
323
324 print()
325 backfill(DB_PATH, user_map, repo_map, repos)
326
327
328if __name__ == "__main__":
329 main()