#!/usr/bin/env python3 """ Gitea Language Stats Calculates the percentage of every programming language used in all repositories of a given account and updates the README.md in the .profile repository between the markers and . 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 (needs write access) 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 base64 import json import os import re import sys from collections import defaultdict from pathlib import Path import requests from dotenv import load_dotenv load_dotenv() BASE_URL = os.getenv("GITEA_BASE_URL", "").rstrip("/") TOKEN = os.getenv("GITEA_TOKEN") USERNAME = os.getenv("GITEA_USERNAME") DEFAULT_STATE_DIR = Path.home() / ".local" / "state" STATE_FILE = Path(os.getenv("STATE_FILE", DEFAULT_STATE_DIR / "language_stats.json")) # Profile repository where README will be updated PROFILE_REPO = ".profile" README_PATH = "README.md" # Markers for content replacement MARKER_START = "" MARKER_END = "" # List of files whose byte-counts should be excluded from the totals. # Each tuple: (repository_name, file_path, language_name) EXCLUDE_FILES: list[tuple[str, str, str]] = [ ("ar-scavenger-hunt", "static/aframe.js", "JavaScript"), ("Kingsborough-LibGuide", "groups/home/bootstrap3/Vue3-v.4.html", "HTML"), ("Kingsborough-LibGuide", "groups/home/bootstrap3/Vue3-v.4-css.html", "HTML"), ("Kingsborough-LibGuide", "groups/home/bootstrap3/Vue3-v.5.html", "HTML"), ("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"), ] def get_session() -> requests.Session: """Create an authenticated requests session.""" 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}, timeout=30, ) resp.raise_for_status() batch = resp.json() if not batch: break repos.extend(batch) page += 1 return repos def get_repo_timestamps(repos: list[dict]) -> dict[str, str]: """Extract repository names and their updated_at timestamps. Excludes the profile repo to avoid infinite loops (updating it would trigger another update). """ return { repo["name"]: repo.get("updated_at", "") for repo in repos if repo["name"] != PROFILE_REPO } 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: resp = session.get( f"{BASE_URL}/api/v1/repos/{USERNAME}/{repo['name']}/languages", timeout=30, ) if not resp.ok: # Skip repos we can't access (e.g., token scope limitations) print(f"Warning: Could not fetch languages for {repo['name']}: {resp.status_code}", file=sys.stderr) continue langs = resp.json() # Skip error responses (e.g., {"message": "..."}) if not isinstance(langs, dict) or "message" in langs: print(f"Warning: Unexpected response for {repo['name']}: {langs}", file=sys.stderr) continue for lang, bytes_ in langs.items(): # Handle bytes as either int or string if isinstance(bytes_, int): totals[lang] += bytes_ elif isinstance(bytes_, str) and bytes_.isdigit(): totals[lang] += int(bytes_) # Skip non-numeric values return dict(totals) 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}", timeout=30, ) if meta.ok: size = meta.json().get("size", 0) totals[language] = max(0, totals.get(language, 0) - size) except requests.RequestException: # Ignore failures – continue with best-effort stats pass return totals def generate_markdown(totals: dict[str, int]) -> str: """Generate markdown output for language statistics.""" total_bytes = sum(totals.values()) if total_bytes == 0: return "" pairs = sorted(totals.items(), key=lambda x: x[1], reverse=True) lang_lines = [] for lang, nbytes in pairs: pct = nbytes * 100 / total_bytes bar = "█" * int(pct / 2) # up to 50 chars lang_lines.append(f"{bar} {lang}{pct:5.1f}%") # Join language lines with backslash + newline for markdown line breaks return "## 📊 Languages\n" + "\\\n".join(lang_lines) def fetch_readme(session: requests.Session) -> tuple[str, str]: """Fetch the README from the profile repository. Returns: Tuple of (content, sha) where sha is needed for updating. """ resp = session.get( f"{BASE_URL}/api/v1/repos/{USERNAME}/{PROFILE_REPO}/contents/{README_PATH}", timeout=30, ) resp.raise_for_status() data = resp.json() content = base64.b64decode(data["content"]).decode("utf-8") return content, data["sha"] def replace_between_markers(content: str, new_content: str) -> str: """Replace content between LANG_STATS markers.""" pattern = re.compile( rf"({re.escape(MARKER_START)})(.*?)({re.escape(MARKER_END)})", re.DOTALL, ) replacement = rf"\1\n{new_content}\n\3" new_readme, count = pattern.subn(replacement, content) if count == 0: raise ValueError(f"Markers {MARKER_START} and {MARKER_END} not found in README") return new_readme def update_readme(session: requests.Session, content: str, sha: str) -> None: """Update the README in the profile repository.""" encoded_content = base64.b64encode(content.encode("utf-8")).decode("ascii") resp = session.put( f"{BASE_URL}/api/v1/repos/{USERNAME}/{PROFILE_REPO}/contents/{README_PATH}", json={ "content": encoded_content, "sha": sha, "message": "Update language statistics", }, timeout=30, ) resp.raise_for_status() 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_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.") markdown = generate_markdown(totals) # Fetch current README, update it, and push readme_content, readme_sha = fetch_readme(session) updated_readme = replace_between_markers(readme_content, markdown) update_readme(session, updated_readme, readme_sha) print("README.md updated successfully.", file=sys.stderr) # Save current state for next run new_state = {"repo_timestamps": current_timestamps} save_state(new_state) return 0 if __name__ == "__main__": sys.exit(main())