#!/usr/bin/env python3 """Refresh README stats badges from the production Inspector bucket.""" from __future__ import annotations import argparse import json import os import sqlite3 import tempfile from collections.abc import Callable from dataclasses import dataclass from pathlib import Path from urllib.parse import quote PROD_BUCKET_ID = "hetchyy/quranic-inspector-bucket" DB_BUCKET_PATH = "db/inspector.db" DEFAULT_README = Path(__file__).resolve().parents[2] / "README.md" START_MARKER = " " END_MARKER = " " @dataclass(frozen=True) class BadgeStats: reciters: int mushafs: int riwayat: int seconds: int def format_hours(seconds: int, *, step: int = 50) -> str: hours = max(0, int(seconds // 3600)) floored = (hours // step) * step return f"{floored:,}h+" def badge_url(label: str, value: str, color: str) -> str: return f"https://img.shields.io/badge/{quote(label, safe='')}-{quote(value, safe='')}-{color}" def _bucket_uri(bucket_id: str, path: str) -> str: return f"buckets/{bucket_id}/{path}" def read_bucket_bytes(bucket_id: str, path: str, *, token: str | None = None) -> bytes: if token: os.environ.setdefault("HF_TOKEN", token) from huggingface_hub import hffs # type: ignore[import-not-found] result = hffs.cat_file(_bucket_uri(bucket_id, path)) return result.encode("utf-8") if isinstance(result, str) else result def download_db(bucket_id: str, *, token: str | None = None) -> Path: data = read_bucket_bytes(bucket_id, DB_BUCKET_PATH, token=token) fd, tmp = tempfile.mkstemp(prefix="inspector-badges-", suffix=".db") os.close(fd) path = Path(tmp) path.write_bytes(data) return path def _duration_from_manifest(doc: dict, slug: str) -> int: chapters = doc.get("chapters") if not isinstance(chapters, dict) or not chapters: raise RuntimeError(f"{slug}: audio manifest has no chapters") total = 0.0 missing: list[str] = [] for chapter, entry in chapters.items(): if not isinstance(entry, dict): missing.append(str(chapter)) continue value = entry.get("duration_sec") if value is None: missing.append(str(chapter)) continue total += float(value) if missing: sample = ", ".join(missing[:8]) suffix = "..." if len(missing) > 8 else "" raise RuntimeError( f"{slug}: missing duration_sec for {len(missing)} chapter(s): {sample}{suffix}" ) return int(total) def _rows_to_stats( rows: list[sqlite3.Row], *, manifest_reader: Callable[[str], dict] | None, label: str, ) -> BadgeStats: reciters = len({row["reciter_id"] for row in rows}) mushafs = len(rows) riwayat = len({row["riwayah"] for row in rows}) seconds = 0 missing_duration: list[str] = [] for row in rows: duration = row["total_duration_sec"] if duration is not None: seconds += int(duration) continue if manifest_reader is None: missing_duration.append(row["slug"]) continue seconds += _duration_from_manifest(manifest_reader(row["slug"]), row["slug"]) if missing_duration: raise RuntimeError( f"missing total_duration_sec for {label} delivery(s): " + ", ".join(missing_duration) ) return BadgeStats(reciters=reciters, mushafs=mushafs, riwayat=riwayat, seconds=seconds) def collect_stats( db_path: Path, *, manifest_reader: Callable[[str], dict] | None = None, ) -> tuple[BadgeStats, BadgeStats]: """Return (catalog_stats, aligned_stats): all sourced audio vs. released+public only.""" conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row try: catalog_rows = conn.execute( """ SELECT d.slug, d.reciter_id, d.total_duration_sec, d.riwayah FROM deliveries d ORDER BY d.slug """ ).fetchall() aligned_rows = conn.execute( """ SELECT d.slug, d.reciter_id, d.total_duration_sec, d.riwayah FROM deliveries d JOIN delivery_states s ON s.slug = d.slug WHERE s.state = 'released' AND s.visibility = 'public' ORDER BY d.slug """ ).fetchall() finally: conn.close() catalog_stats = _rows_to_stats(catalog_rows, manifest_reader=manifest_reader, label="catalog") aligned_stats = _rows_to_stats(aligned_rows, manifest_reader=manifest_reader, label="aligned") return catalog_stats, aligned_stats LABEL_COLOR = "64748b" STATS_COLOR = "d4842a" RIWAYAT_COLOR = "f0ad4e" def render_badges(catalog: BadgeStats, aligned: BadgeStats) -> str: catalog_hours = format_hours(catalog.seconds, step=500) aligned_hours = format_hours(aligned.seconds, step=50) def label_badge(text: str) -> str: return f'{text}' def badge(label: str, value: str, color: str) -> str: return f'{label}: {value}' lines = [ START_MARKER, "
", " " + label_badge("\U0001f509 Catalog") + " " + badge("Reciters", f"{catalog.reciters:,}", STATS_COLOR) + " " + badge("Riwayat", f"{catalog.riwayat:,}", RIWAYAT_COLOR) + " " + badge("Hours", catalog_hours, STATS_COLOR), "
", " " + label_badge("\U0001f3af Aligned") + " " + badge("Mushafs", f"{aligned.mushafs:,}", STATS_COLOR) + " " + badge("Riwayat", f"{aligned.riwayat:,}", RIWAYAT_COLOR) + " " + badge("Hours", aligned_hours, STATS_COLOR), END_MARKER, ] return "\n".join(lines) def replace_badges(readme: str, badges: str) -> str: if START_MARKER in readme and END_MARKER in readme: before, rest = readme.split(START_MARKER, 1) _, after = rest.split(END_MARKER, 1) return before + badges + after old_block = """ """ if old_block in readme: return readme.replace(old_block, badges) raise RuntimeError("README badge block not found") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--readme", type=Path, default=DEFAULT_README) parser.add_argument("--db-path", type=Path) parser.add_argument("--bucket-id", default=PROD_BUCKET_ID) parser.add_argument("--dry-run", action="store_true") return parser.parse_args() def main() -> int: args = parse_args() token = os.environ.get("HF_TOKEN") temp_db: Path | None = None try: db_path = args.db_path if db_path is None: temp_db = download_db(args.bucket_id, token=token) db_path = temp_db def manifest_reader(slug: str) -> dict: raw = read_bucket_bytes( args.bucket_id, f"catalog/audio_manifest/{slug}.json", token=token ) return json.loads(raw) catalog_stats, aligned_stats = collect_stats(db_path, manifest_reader=manifest_reader) badges = render_badges(catalog_stats, aligned_stats) print( f"catalog: reciters={catalog_stats.reciters} riwayat={catalog_stats.riwayat} " f"hours={format_hours(catalog_stats.seconds, step=500)}" ) print( f"aligned: mushafs={aligned_stats.mushafs} riwayat={aligned_stats.riwayat} " f"hours={format_hours(aligned_stats.seconds, step=50)}" ) print(badges) if args.dry_run: return 0 readme = args.readme.read_text(encoding="utf-8") updated = replace_badges(readme, badges) args.readme.write_text(updated, encoding="utf-8") return 0 finally: if temp_db is not None: try: temp_db.unlink() except OSError: pass if __name__ == "__main__": raise SystemExit(main())