297 lines
9.5 KiB
Python
297 lines
9.5 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:
|
||
langs = session.get(
|
||
f"{BASE_URL}/api/v1/repos/{USERNAME}/{repo['name']}/languages",
|
||
timeout=30,
|
||
).json()
|
||
for lang, bytes_ in langs.items():
|
||
totals[lang] += int(bytes_)
|
||
|
||
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())
|