Mark EatonandClaude Opus 4.5
2b4ef71458
Handle cases where the Gitea API returns an error message (e.g., "token scope is limited to public repos") instead of language data. Now skips repos with non-OK responses or error messages gracefully. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
311 lines
10 KiB
Python
311 lines
10 KiB
Python
#!/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
|
||
<!--LANG_STATS_START--> and <!--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 (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 = "<!--LANG_STATS_START-->"
|
||
MARKER_END = "<!--LANG_STATS_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())
|