expand language_stats.py to include check of gitea repos

scripts/language_stats.py

@@ -6,16 +6,27 @@ in all repositories of a given account and prints a markdown
 table to stdout, delimited by the
 markers  <!--LANG_STATS_START--> … <!--LANG_STATS_END-->.
 
+When run as a systemd job, it checks for new commits since the last run
+and only regenerates if there are changes. Use --force to always regenerate.
+
 Required environment variables
 --------------------------------
 GITEA_BASE_URL   e.g. https://gitea.example.com
 GITEA_TOKEN      personal access-token with `repo` scope
 GITEA_USERNAME   the account to analyse
+
+Optional environment variables
+--------------------------------
+STATE_FILE       path to store last-run state (default: ~/.local/state/language_stats.json)
 """
+import argparse
+import json
 import os
 import sys
-import requests
 from collections import defaultdict
+from pathlib import Path
+
+import requests
 from dotenv import load_dotenv
 
 load_dotenv()
@@ -24,11 +35,8 @@ BASE_URL   = os.getenv("GITEA_BASE_URL", "").rstrip("/")
 TOKEN = os.getenv("GITEA_TOKEN")
 USERNAME = os.getenv("GITEA_USERNAME")
 
-if not BASE_URL or not TOKEN or not USERNAME:
+DEFAULT_STATE_DIR = Path.home() / ".local" / "state"
-    sys.exit("GITEA_BASE_URL, GITEA_TOKEN and GITEA_USERNAME must be set")
+STATE_FILE = Path(os.getenv("STATE_FILE", DEFAULT_STATE_DIR / "language_stats.json"))
-
-session = requests.Session()
-session.headers["Authorization"] = f"token {TOKEN}"
 
 # List of files whose byte-counts should be excluded from the totals.
 # Each tuple: (repository_name, file_path, language_name)
@@ -40,12 +48,23 @@ EXCLUDE_FILES: list[tuple[str, str, str]] = [
     ("Kingsborough-LibGuide", "groups/home/bootstrap3/Vue3-v.5-css.html", "HTML"),
     ("Kingsborough-LibGuide", "groups/home/bootstrap3/Vue3-v.6.html", "HTML"),
     ("Kingsborough-LibGuide", "groups/home/bootstrap3/Vue3-v.6-css.html", "HTML"),
+    ("Kingsborough-LibGuide", "groups/home/bootstrap3/Vue3-v.7.html", "HTML"),
+    ("Kingsborough-LibGuide", "groups/home/bootstrap3/Vue3-v.7-css.html", "HTML"),
 ]
 
-# ---------- collect repositories ----------
+
-page = 1
+def get_session() -> requests.Session:
-repos = []
+    """Create an authenticated requests session."""
-while True:
+    session = requests.Session()
+    session.headers["Authorization"] = f"token {TOKEN}"
+    return session
+
+
+def fetch_repos(session: requests.Session) -> list[dict]:
+    """Fetch all repositories for the configured user."""
+    page = 1
+    repos = []
+    while True:
         resp = session.get(
             f"{BASE_URL}/api/v1/users/{USERNAME}/repos",
             params={"page": page, "limit": 50},
@@ -57,11 +76,54 @@ while True:
             break
         repos.extend(batch)
         page += 1
+    return repos
 
-# ---------- aggregate language bytes ----------
-totals: dict[str, int] = defaultdict(int)
 
-for repo in repos:
+def get_repo_push_timestamps(repos: list[dict]) -> dict[str, str]:
+    """Extract repository names and their pushed_at timestamps."""
+    return {repo["name"]: repo.get("pushed_at", "") for repo in repos}
+
+
+def load_state() -> dict:
+    """Load the previous state from the state file."""
+    if STATE_FILE.exists():
+        try:
+            return json.loads(STATE_FILE.read_text())
+        except (json.JSONDecodeError, OSError):
+            return {}
+    return {}
+
+
+def save_state(state: dict) -> None:
+    """Save the current state to the state file."""
+    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
+    STATE_FILE.write_text(json.dumps(state, indent=2))
+
+
+def has_new_commits(current_timestamps: dict[str, str], saved_state: dict) -> bool:
+    """Check if any repository has new commits since last run."""
+    saved_timestamps = saved_state.get("repo_timestamps", {})
+
+    # Check for new or updated repos
+    for repo_name, pushed_at in current_timestamps.items():
+        if repo_name not in saved_timestamps:
+            return True
+        if pushed_at != saved_timestamps[repo_name]:
+            return True
+
+    # Check for deleted repos (also counts as a change)
+    for repo_name in saved_timestamps:
+        if repo_name not in current_timestamps:
+            return True
+
+    return False
+
+
+def aggregate_language_bytes(session: requests.Session, repos: list[dict]) -> dict[str, int]:
+    """Aggregate language byte counts across all repositories."""
+    totals: dict[str, int] = defaultdict(int)
+
+    for repo in repos:
         langs = session.get(
             f"{BASE_URL}/api/v1/repos/{USERNAME}/{repo['name']}/languages",
             timeout=30,
@@ -69,8 +131,12 @@ for repo in repos:
         for lang, bytes_ in langs.items():
             totals[lang] += bytes_
 
-# ---------- subtract excluded files ----------
+    return dict(totals)
-for repo_name, file_path, language in EXCLUDE_FILES:
+
+
+def subtract_excluded_files(session: requests.Session, totals: dict[str, int]) -> dict[str, int]:
+    """Subtract byte counts for explicitly excluded files."""
+    for repo_name, file_path, language in EXCLUDE_FILES:
         try:
             meta = session.get(
                 f"{BASE_URL}/api/v1/repos/{USERNAME}/{repo_name}/contents/{file_path}",
@@ -78,25 +144,85 @@ for repo_name, file_path, language in EXCLUDE_FILES:
             )
             if meta.ok:
                 size = meta.json().get("size", 0)
-            totals[language] = max(0, totals[language] - size)
+                totals[language] = max(0, totals.get(language, 0) - size)
         except requests.RequestException:
             # Ignore failures – continue with best-effort stats
             pass
+    return totals
 
-total_bytes = sum(totals.values())
-if total_bytes == 0:
-    sys.exit("No language data returned from API.")
 
-# ---------- build markdown table ----------
+def generate_markdown(totals: dict[str, int]) -> list[str]:
-pairs = sorted(totals.items(), key=lambda x: x[1], reverse=True)
+    """Generate markdown output for language statistics."""
+    total_bytes = sum(totals.values())
+    if total_bytes == 0:
+        return []
 
-lines = [
+    pairs = sorted(totals.items(), key=lambda x: x[1], reverse=True)
-    "## 📊 Languages\n",
+
-]
+    lines = ["## 📊 Languages\n"]
-for lang, nbytes in pairs:
+    for lang, nbytes in pairs:
         pct = nbytes * 100 / total_bytes
         bar = "█" * int(pct / 2)  # up to 50 chars
         lines.append(f"{bar} {lang}{pct:5.1f}%\\")
 
-for line in lines:
+    return lines
+
+
+def main() -> int:
+    """Main entry point for the script."""
+    parser = argparse.ArgumentParser(
+        description="Generate language statistics from Gitea repositories"
+    )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="Force regeneration even if no new commits detected",
+    )
+    parser.add_argument(
+        "--check-only",
+        action="store_true",
+        help="Only check for new commits, don't generate output",
+    )
+    args = parser.parse_args()
+
+    # Validate environment
+    if not BASE_URL or not TOKEN or not USERNAME:
+        sys.exit("GITEA_BASE_URL, GITEA_TOKEN and GITEA_USERNAME must be set")
+
+    session = get_session()
+
+    # Fetch current repository state
+    repos = fetch_repos(session)
+    current_timestamps = get_repo_push_timestamps(repos)
+
+    # Load previous state and check for changes
+    saved_state = load_state()
+
+    if not args.force and not has_new_commits(current_timestamps, saved_state):
+        print("No new commits detected. Skipping regeneration.", file=sys.stderr)
+        return 0
+
+    if args.check_only:
+        print("New commits detected.", file=sys.stderr)
+        return 0
+
+    # Generate the stats
+    totals = aggregate_language_bytes(session, repos)
+    totals = subtract_excluded_files(session, totals)
+
+    if sum(totals.values()) == 0:
+        sys.exit("No language data returned from API.")
+
+    lines = generate_markdown(totals)
+    for line in lines:
         print(line)
+
+    # Save current state for next run
+    new_state = {"repo_timestamps": current_timestamps}
+    save_state(new_state)
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())