gitstats.py

#!/usr/bin/env python3
"""
gitstats.py

Compute Diff Delta stats for Git repos.

Supports:
  - whole repo score
  - one commit score
  - date ranges
  - monthly breakdown
  - monthly breakdown by author
  - direct-subfolder repo recursion
  - aggregate stats across repos
  - recursive monthly breakdown by author
  - full-range aggregate by user
  - full-range aggregate total row
  - filter summary
  - excluded filetypes
  - HTML report with charts
  - JSON output

Usage:

  python gitstats.py .\repo

  python gitstats.py .\repo --commit HEAD

  python gitstats.py .\repo --from "2025-06-01 00:00:00" --to "2025-11-30 23:59:59"

  python gitstats.py .\repo --by-month-user

  python gitstats.py --recurse C:\Projects --by-month-user

  python gitstats.py --recurse C:\Projects --by-month-user --html-report .\gitstats_report.html

  python gitstats.py --recurse C:\Projects --exclude-ext .csv --exclude-ext .xml
"""

from __future__ import annotations

import argparse
import base64
import hashlib
import html
import io
import json
import math
import re
import subprocess
import sys

from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable


try:
    import matplotlib.pyplot as plt
except ImportError:
    plt = None


BASE_SCORE = {
    "add": 0.80,
    "delete": 1.00,
    "update": 1.00,
    "move": 0.35,
    "copy_paste": 0.25,
    "find_replace": 0.60,
}

LANGUAGE_WEIGHT = {
    ".py": 1.00,
    ".go": 1.00,
    ".rs": 1.00,
    ".c": 1.05,
    ".h": 1.00,
    ".cpp": 1.05,
    ".hpp": 1.00,
    ".cs": 1.00,
    ".java": 1.00,
    ".js": 0.95,
    ".ts": 0.95,
    ".html": 0.60,
    ".css": 0.55,
    ".json": 0.45,
    ".yaml": 0.45,
    ".yml": 0.45,
    ".md": 0.35,
    ".txt": 0.25,
}

GENERATED_OR_NOISY_PATHS = [
    "dist/",
    "build/",
    "target/",
    "node_modules/",
    "__pycache__/",
    ".min.js",
    ".pb.go",
    "package-lock.json",
    "yarn.lock",
    "pnpm-lock.yaml",
]

COMPILED_EXTENSIONS = {
    ".exe",
    ".dll",
    ".so",
    ".dylib",
    ".class",
    ".pyc",
    ".o",
    ".obj",
    ".bin",
}

DEFAULT_EXCLUDED_EXTENSIONS = {
    ".csv",
}

EXCLUDED_EXTENSIONS: set[str] = set(DEFAULT_EXCLUDED_EXTENSIONS)

FILTER_LABELS = {
    "scored": "Scored lines",
    "compiled_file": "Compiled/binary file lines",
    "generated_or_noisy_path": "Generated/noisy path lines",
    "excluded_filetype": "Excluded filetype lines",
    "blank_line": "Blank lines",
    "comment_line": "Comment lines",
    "delimiter_only_line": "Delimiter-only lines",
}


@dataclass
class LineEvent:
    commit: str
    commit_time: int
    file: str
    operation: str
    line: str


CommitInfo = tuple[str, int, str]
# commit_hash, commit_time_epoch, author_name


def normalise_extension(value: str) -> str:
    value = value.strip().lower()

    if not value:
        return value

    if not value.startswith("."):
        value = "." + value

    return value


def run_git(repo: Path, args: list[str]) -> str:
    cmd = ["git", "-C", str(repo), *args]

    proc = subprocess.run(
        cmd,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        encoding="utf-8",
        errors="replace",
    )

    if proc.returncode != 0:
        raise RuntimeError(proc.stderr.strip())

    return proc.stdout


def validate_repo(repo: Path) -> None:
    if not repo.exists():
        raise ValueError(f"repo does not exist: {repo}")

    if not repo.is_dir():
        raise ValueError(f"repo path is not a directory: {repo}")

    result = run_git(repo, ["rev-parse", "--is-inside-work-tree"]).strip()

    if result != "true":
        raise ValueError(f"not a git repo: {repo}")


def find_direct_git_repos(root: Path) -> list[Path]:
    if not root.exists():
        raise ValueError(f"recurse path does not exist: {root}")

    if not root.is_dir():
        raise ValueError(f"recurse path is not a directory: {root}")

    repos: list[Path] = []

    for child in sorted(root.iterdir()):
        if not child.is_dir():
            continue

        try:
            validate_repo(child)
            repos.append(child.resolve())
        except Exception:
            continue

    return repos


def get_commits(
    repo: Path,
    since: str | None,
    until: str | None,
    max_count: int | None,
) -> list[CommitInfo]:
    args = ["log", "--format=%H%x00%ct%x00%an"]

    if since:
        args.append(f"--since={since}")

    if until:
        args.append(f"--until={until}")

    if max_count:
        args.append(f"--max-count={max_count}")

    output = run_git(repo, args)

    commits: list[CommitInfo] = []

    for line in output.splitlines():
        if not line.strip():
            continue

        commit, timestamp, author = line.split("\x00", 2)
        commits.append((commit, int(timestamp), author))

    return commits


def get_one_commit(repo: Path, commit: str) -> CommitInfo:
    output = run_git(
        repo,
        ["show", "-s", "--format=%H%x00%ct%x00%an", commit],
    ).strip()

    commit_hash, timestamp, author = output.split("\x00", 2)
    return commit_hash, int(timestamp), author


def parse_patch(repo: Path, commit: str, commit_time: int) -> list[LineEvent]:
    patch = run_git(
        repo,
        [
            "show",
            "--format=",
            "--find-renames",
            "--find-copies",
            "--unified=0",
            "--no-ext-diff",
            commit,
        ],
    )

    events: list[LineEvent] = []
    current_file: str | None = None

    for raw in patch.splitlines():
        line = raw.rstrip("\n")

        if line.startswith("+++ b/"):
            current_file = line[6:]
            continue

        if line.startswith("+++ /dev/null"):
            current_file = None
            continue

        if not current_file:
            continue

        if line.startswith("+++ ") or line.startswith("--- "):
            continue

        if line.startswith("@@"):
            continue

        if line.startswith("+") and not line.startswith("+++"):
            events.append(
                LineEvent(
                    commit=commit,
                    commit_time=commit_time,
                    file=current_file,
                    operation="add",
                    line=line[1:],
                )
            )

        elif line.startswith("-") and not line.startswith("---"):
            events.append(
                LineEvent(
                    commit=commit,
                    commit_time=commit_time,
                    file=current_file,
                    operation="delete",
                    line=line[1:],
                )
            )

    return infer_updates(events)


def infer_updates(events: list[LineEvent]) -> list[LineEvent]:
    by_file: dict[str, list[LineEvent]] = {}

    for event in events:
        by_file.setdefault(event.file, []).append(event)

    inferred: list[LineEvent] = []

    for file_events in by_file.values():
        adds = [e for e in file_events if e.operation == "add"]
        deletes = [e for e in file_events if e.operation == "delete"]

        update_pairs = min(len(adds), len(deletes))

        add_ids = {id(e) for e in adds[:update_pairs]}
        del_ids = {id(e) for e in deletes[:update_pairs]}

        for event in file_events:
            if id(event) in add_ids or id(event) in del_ids:
                event.operation = "update"

            inferred.append(event)

    return inferred


def file_branch_filter_reason(event: LineEvent) -> str:
    path = event.file.lower()
    suffix = Path(path).suffix.lower()

    if suffix in COMPILED_EXTENSIONS:
        return "compiled_file"

    if suffix in EXCLUDED_EXTENSIONS:
        return "excluded_filetype"

    if any(marker in path for marker in GENERATED_OR_NOISY_PATHS):
        return "generated_or_noisy_path"

    return "scored"


def file_branch_filter(event: LineEvent) -> float:
    return 0.0 if file_branch_filter_reason(event) != "scored" else 1.0


def context_filter_reason(event: LineEvent) -> str:
    text = event.line.strip()

    if not text:
        return "blank_line"

    comment_prefixes = ("#", "//", "--", "/*", "*", "*/")

    if text.startswith(comment_prefixes):
        return "comment_line"

    delimiter_only = {
        "{",
        "}",
        "};",
        ");",
        "(",
        ")",
        "[",
        "]",
        ",",
        ";",
    }

    if text in delimiter_only:
        return "delimiter_only_line"

    return "scored"


def context_filter(event: LineEvent) -> float:
    return 0.0 if context_filter_reason(event) != "scored" else 1.0


def filter_summary(events: list[LineEvent]) -> dict[str, int]:
    summary = {key: 0 for key in FILTER_LABELS}

    for event in events:
        file_reason = file_branch_filter_reason(event)

        if file_reason != "scored":
            summary[file_reason] += 1
            continue

        context_reason = context_filter_reason(event)

        if context_reason != "scored":
            summary[context_reason] += 1
            continue

        summary["scored"] += 1

    return summary


def duplication_groups(events: Iterable[LineEvent]) -> dict[str, int]:
    counts: dict[str, int] = {}

    for event in events:
        key = duplicate_key(event)
        counts[key] = counts.get(key, 0) + 1

    return counts


def duplicate_key(event: LineEvent) -> str:
    normalised = re.sub(r"\s+", " ", event.line.strip())
    payload = f"{event.operation}:{normalised}"
    return hashlib.sha1(payload.encode("utf-8", errors="replace")).hexdigest()


def duplication_filter(event: LineEvent, groups: dict[str, int]) -> float:
    size = groups.get(duplicate_key(event), 1)

    if size <= 1:
        return 1.0

    return 1.0 / size


def base_score(event: LineEvent) -> float:
    return BASE_SCORE.get(event.operation, 0.50)


def time_scalar(event: LineEvent, newest_commit_time: int) -> float:
    age_seconds = max(0, newest_commit_time - event.commit_time)
    age_days = age_seconds / 86_400

    return 0.5 + 0.7 * (1.0 - math.exp(-age_days / 60.0))


def context_scalar(event: LineEvent) -> float:
    suffix = Path(event.file).suffix.lower()
    language_weight = LANGUAGE_WEIGHT.get(suffix, 0.75)

    invocation_bonus = 1.08 if looks_like_invocation(event.line) else 1.00

    return language_weight * invocation_bonus


def looks_like_invocation(line: str) -> bool:
    text = line.strip()

    if not text:
        return False

    if re.search(r"\b(if|for|while|switch|catch)\s*\(", text):
        return False

    return bool(re.search(r"\w+\s*\(", text))


def score_event(
    event: LineEvent,
    groups: dict[str, int],
    newest_commit_time: int,
) -> dict:
    phi = file_branch_filter(event)
    theta = context_filter(event)
    dup = duplication_filter(event, groups)
    beta = base_score(event)
    tau = time_scalar(event, newest_commit_time)
    sigma = context_scalar(event)

    score = phi * theta * dup * beta * tau * sigma

    return {
        "commit": event.commit[:12],
        "file": event.file,
        "operation": event.operation,
        "line": event.line,
        "phi": round(phi, 4),
        "theta": round(theta, 4),
        "dup": round(dup, 4),
        "beta": round(beta, 4),
        "tau": round(tau, 4),
        "sigma": round(sigma, 4),
        "score": round(score, 6),
    }


def compute(repo: Path, commits: list[CommitInfo]) -> dict:
    all_events: list[LineEvent] = []

    for commit, commit_time, _author in commits:
        all_events.extend(parse_patch(repo, commit, commit_time))

    if not all_events:
        return {
            "diff_delta": 0.0,
            "commits": len(commits),
            "events": 0,
            "scored_events": 0,
            "filter_summary": {key: 0 for key in FILTER_LABELS},
            "by_commit": {},
            "by_file": {},
            "rows": [],
        }

    newest_commit_time = max(event.commit_time for event in all_events)

    filters = filter_summary(all_events)
    groups = duplication_groups(all_events)

    rows = [score_event(event, groups, newest_commit_time) for event in all_events]

    total = sum(row["score"] for row in rows)

    by_commit: dict[str, float] = {}
    by_file: dict[str, float] = {}

    for row in rows:
        by_commit[row["commit"]] = by_commit.get(row["commit"], 0.0) + row["score"]
        by_file[row["file"]] = by_file.get(row["file"], 0.0) + row["score"]

    return {
        "diff_delta": round(total, 6),
        "commits": len(commits),
        "events": len(rows),
        "scored_events": sum(1 for row in rows if row["score"] > 0),
        "filter_summary": filters,
        "by_commit": {
            k: round(v, 6)
            for k, v in sorted(by_commit.items(), key=lambda item: item[1], reverse=True)
        },
        "by_file": {
            k: round(v, 6)
            for k, v in sorted(by_file.items(), key=lambda item: item[1], reverse=True)
        },
        "rows": rows,
    }


def month_key(timestamp: int) -> str:
    dt = datetime.fromtimestamp(timestamp, tz=timezone.utc)
    return f"{dt.year:04d}-{dt.month:02d}"


def compute_by_month(repo: Path, commits: list[CommitInfo]) -> dict:
    groups: dict[str, list[CommitInfo]] = {}

    for commit, commit_time, author in commits:
        month = month_key(commit_time)
        groups.setdefault(month, []).append((commit, commit_time, author))

    result = {}

    for month in sorted(groups):
        stats = compute(repo, groups[month])

        result[month] = {
            "diff_delta": stats["diff_delta"],
            "commits": stats["commits"],
            "events": stats["events"],
            "scored_events": stats["scored_events"],
            "filter_summary": stats["filter_summary"],
            "top_files": dict(list(stats["by_file"].items())[:10]),
            "top_commits": dict(list(stats["by_commit"].items())[:10]),
        }

    return result


def compute_by_month_user(repo: Path, commits: list[CommitInfo]) -> dict:
    groups: dict[str, dict[str, list[CommitInfo]]] = {}

    for commit, commit_time, author in commits:
        month = month_key(commit_time)

        groups.setdefault(month, {})
        groups[month].setdefault(author, [])
        groups[month][author].append((commit, commit_time, author))

    result = {}

    for month in sorted(groups):
        result[month] = {}

        for author in sorted(groups[month]):
            stats = compute(repo, groups[month][author])

            result[month][author] = {
                "diff_delta": stats["diff_delta"],
                "commits": stats["commits"],
                "events": stats["events"],
                "scored_events": stats["scored_events"],
                "filter_summary": stats["filter_summary"],
                "top_files": dict(list(stats["by_file"].items())[:10]),
                "top_commits": dict(list(stats["by_commit"].items())[:10]),
            }

    return result


def merge_filter_summaries(summaries: list[dict[str, int]]) -> dict[str, int]:
    merged = {key: 0 for key in FILTER_LABELS}

    for summary in summaries:
        for key in FILTER_LABELS:
            merged[key] += summary.get(key, 0)

    return merged


def merge_score_maps(maps: list[dict[str, float]]) -> dict[str, float]:
    merged: dict[str, float] = {}

    for score_map in maps:
        for key, value in score_map.items():
            merged[key] = merged.get(key, 0.0) + value

    return {
        k: round(v, 6)
        for k, v in sorted(merged.items(), key=lambda item: item[1], reverse=True)
    }


def namespace_result(repo_name: str, result: dict) -> dict:
    namespaced = dict(result)

    namespaced["by_file"] = {
        f"{repo_name}/{path}": score
        for path, score in result["by_file"].items()
    }

    namespaced["by_commit"] = {
        f"{repo_name}:{commit}": score
        for commit, score in result["by_commit"].items()
    }

    return namespaced


def aggregate_results(results: list[dict]) -> dict:
    return {
        "diff_delta": round(sum(r["diff_delta"] for r in results), 6),
        "repos": len(results),
        "commits": sum(r["commits"] for r in results),
        "events": sum(r["events"] for r in results),
        "scored_events": sum(r["scored_events"] for r in results),
        "filter_summary": merge_filter_summaries(
            [r["filter_summary"] for r in results]
        ),
        "by_commit": merge_score_maps([r["by_commit"] for r in results]),
        "by_file": merge_score_maps([r["by_file"] for r in results]),
        "rows": [],
    }


def merge_month_user_results(repo_month_user_results: dict[str, dict]) -> dict:
    aggregate: dict[str, dict[str, dict]] = {}

    for repo_name, monthly_user in repo_month_user_results.items():
        for month, authors in monthly_user.items():
            aggregate.setdefault(month, {})

            for author, stats in authors.items():
                aggregate[month].setdefault(
                    author,
                    {
                        "diff_delta": 0.0,
                        "commits": 0,
                        "events": 0,
                        "scored_events": 0,
                        "filter_summary": {key: 0 for key in FILTER_LABELS},
                        "top_files": {},
                        "top_commits": {},
                    },
                )

                target = aggregate[month][author]

                target["diff_delta"] += stats["diff_delta"]
                target["commits"] += stats["commits"]
                target["events"] += stats["events"]
                target["scored_events"] += stats["scored_events"]

                target["filter_summary"] = merge_filter_summaries(
                    [
                        target["filter_summary"],
                        stats["filter_summary"],
                    ]
                )

                namespaced_files = {
                    f"{repo_name}/{path}": score
                    for path, score in stats["top_files"].items()
                }

                namespaced_commits = {
                    f"{repo_name}:{commit}": score
                    for commit, score in stats["top_commits"].items()
                }

                target["top_files"] = merge_score_maps(
                    [
                        target["top_files"],
                        namespaced_files,
                    ]
                )

                target["top_commits"] = merge_score_maps(
                    [
                        target["top_commits"],
                        namespaced_commits,
                    ]
                )

    for month in aggregate:
        for author in aggregate[month]:
            aggregate[month][author]["diff_delta"] = round(
                aggregate[month][author]["diff_delta"],
                6,
            )

            aggregate[month][author]["top_files"] = dict(
                list(aggregate[month][author]["top_files"].items())[:10]
            )

            aggregate[month][author]["top_commits"] = dict(
                list(aggregate[month][author]["top_commits"].items())[:10]
            )

    return {
        month: aggregate[month]
        for month in sorted(aggregate)
    }


def collapse_month_user_to_user_totals(monthly_user: dict) -> dict:
    totals: dict[str, dict] = {}

    for _month, authors in monthly_user.items():
        for author, stats in authors.items():
            totals.setdefault(
                author,
                {
                    "commits": 0,
                    "events": 0,
                    "scored_events": 0,
                    "diff_delta": 0.0,
                    "filter_summary": {key: 0 for key in FILTER_LABELS},
                },
            )

            target = totals[author]

            target["commits"] += stats["commits"]
            target["events"] += stats["events"]
            target["scored_events"] += stats["scored_events"]
            target["diff_delta"] += stats["diff_delta"]

            target["filter_summary"] = merge_filter_summaries(
                [
                    target["filter_summary"],
                    stats["filter_summary"],
                ]
            )

    for author in totals:
        totals[author]["diff_delta"] = round(totals[author]["diff_delta"], 6)

    return {
        author: totals[author]
        for author in sorted(
            totals,
            key=lambda name: totals[name]["diff_delta"],
            reverse=True,
        )
    }


def collapse_month_user_to_total(monthly_user: dict) -> dict:
    total = {
        "commits": 0,
        "events": 0,
        "scored_events": 0,
        "diff_delta": 0.0,
        "filter_summary": {key: 0 for key in FILTER_LABELS},
    }

    for _month, authors in monthly_user.items():
        for _author, stats in authors.items():
            total["commits"] += stats["commits"]
            total["events"] += stats["events"]
            total["scored_events"] += stats["scored_events"]
            total["diff_delta"] += stats["diff_delta"]

            total["filter_summary"] = merge_filter_summaries(
                [
                    total["filter_summary"],
                    stats["filter_summary"],
                ]
            )

    total["diff_delta"] = round(total["diff_delta"], 6)

    return total


def print_filter_summary(summary: dict[str, int]) -> None:
    print()
    print("Filter summary:")
    print("-" * 44)

    total = sum(summary.values())

    for key, label in FILTER_LABELS.items():
        count = summary.get(key, 0)
        percent = (count / total * 100) if total else 0.0
        print(f"  {label:<30} {count:>8}  {percent:>6.2f}%")


def print_active_exclusions() -> None:
    if not EXCLUDED_EXTENSIONS:
        return

    print()
    print("Excluded filetypes:")
    print("-" * 44)

    for ext in sorted(EXCLUDED_EXTENSIONS):
        print(f"  {ext}")


def print_total_user_report(user_totals: dict) -> None:
    print()
    print("Aggregate by user for full range:")
    print(
        f"{'Author':<28} "
        f"{'Commits':>8} "
        f"{'Events':>8} "
        f"{'Scored':>8} "
        f"{'Diff Delta':>12}"
    )
    print("-" * 74)

    for author, stats in user_totals.items():
        print(
            f"{author[:28]:<28} "
            f"{stats['commits']:>8} "
            f"{stats['events']:>8} "
            f"{stats['scored_events']:>8} "
            f"{stats['diff_delta']:>12.6f}"
        )


def print_total_range_report(total: dict) -> None:
    print()
    print("Aggregate for full range:")
    print(
        f"{'Commits':>8} "
        f"{'Events':>8} "
        f"{'Scored':>8} "
        f"{'Diff Delta':>12}"
    )
    print("-" * 44)

    print(
        f"{total['commits']:>8} "
        f"{total['events']:>8} "
        f"{total['scored_events']:>8} "
        f"{total['diff_delta']:>12.6f}"
    )


def print_text_report(result: dict) -> None:
    if "repos" in result:
        print(f"Repos:        {result['repos']}")

    print(f"Diff Delta:   {result['diff_delta']}")
    print(f"Commits:      {result['commits']}")
    print(f"Events:       {result['events']}")
    print(f"Scored:       {result['scored_events']}")
    print()

    print("Top commits:")
    for commit, score in list(result["by_commit"].items())[:10]:
        print(f"  {commit:<32} {score}")

    print()

    print("Top files:")
    for file, score in list(result["by_file"].items())[:15]:
        print(f"  {score:<10} {file}")

    print_filter_summary(result["filter_summary"])
    print_active_exclusions()


def print_monthly_report(monthly: dict) -> None:
    print(f"{'Month':<10} {'Commits':>8} {'Events':>8} {'Scored':>8} {'Diff Delta':>12}")
    print("-" * 52)

    for month, stats in monthly.items():
        print(
            f"{month:<10} "
            f"{stats['commits']:>8} "
            f"{stats['events']:>8} "
            f"{stats['scored_events']:>8} "
            f"{stats['diff_delta']:>12.6f}"
        )

    merged = merge_filter_summaries(
        [stats["filter_summary"] for stats in monthly.values()]
    )

    print_filter_summary(merged)
    print_active_exclusions()


def print_month_user_report(monthly_user: dict) -> None:
    print(
        f"{'Month':<10} "
        f"{'Author':<28} "
        f"{'Commits':>8} "
        f"{'Events':>8} "
        f"{'Scored':>8} "
        f"{'Diff Delta':>12}"
    )
    print("-" * 82)

    for month, authors in monthly_user.items():
        for author, stats in sorted(
            authors.items(),
            key=lambda item: item[1]["diff_delta"],
            reverse=True,
        ):
            print(
                f"{month:<10} "
                f"{author[:28]:<28} "
                f"{stats['commits']:>8} "
                f"{stats['events']:>8} "
                f"{stats['scored_events']:>8} "
                f"{stats['diff_delta']:>12.6f}"
            )

    summaries = []

    for authors in monthly_user.values():
        for stats in authors.values():
            summaries.append(stats["filter_summary"])

    merged = merge_filter_summaries(summaries)
    print_filter_summary(merged)
    print_active_exclusions()


def print_recurse_report(repo_results: dict[str, dict], aggregate: dict) -> None:
    print("Repos:")
    print(f"{'Repo':<32} {'Commits':>8} {'Events':>8} {'Scored':>8} {'Diff Delta':>12}")
    print("-" * 76)

    for repo_name, result in sorted(
        repo_results.items(),
        key=lambda item: item[1]["diff_delta"],
        reverse=True,
    ):
        print(
            f"{repo_name[:32]:<32} "
            f"{result['commits']:>8} "
            f"{result['events']:>8} "
            f"{result['scored_events']:>8} "
            f"{result['diff_delta']:>12.6f}"
        )

    print()
    print("Aggregate:")
    print_text_report(aggregate)


def print_recurse_month_user_report(
    repo_month_user_results: dict[str, dict],
    aggregate_month_user: dict,
) -> None:
    print("Repos:")
    print(
        f"{'Repo':<32} "
        f"{'Month':<10} "
        f"{'Author':<28} "
        f"{'Commits':>8} "
        f"{'Events':>8} "
        f"{'Scored':>8} "
        f"{'Diff Delta':>12}"
    )
    print("-" * 122)

    for repo_name, monthly_user in sorted(repo_month_user_results.items()):
        for month, authors in monthly_user.items():
            for author, stats in sorted(
                authors.items(),
                key=lambda item: item[1]["diff_delta"],
                reverse=True,
            ):
                print(
                    f"{repo_name[:32]:<32} "
                    f"{month:<10} "
                    f"{author[:28]:<28} "
                    f"{stats['commits']:>8} "
                    f"{stats['events']:>8} "
                    f"{stats['scored_events']:>8} "
                    f"{stats['diff_delta']:>12.6f}"
                )

    print()
    print("Aggregate by month/user:")
    print_month_user_report(aggregate_month_user)

    user_totals = collapse_month_user_to_user_totals(aggregate_month_user)
    print_total_user_report(user_totals)

    total_range = collapse_month_user_to_total(aggregate_month_user)
    print_total_range_report(total_range)


# -----------------------------
# HTML reporting
# -----------------------------


def require_matplotlib() -> None:
    if plt is None:
        raise RuntimeError(
            "matplotlib is required for --html-report. Install it with: pip install matplotlib"
        )


def fig_to_base64() -> str:
    buffer = io.BytesIO()
    plt.tight_layout()
    plt.savefig(buffer, format="png", dpi=140, bbox_inches="tight")
    plt.close()
    buffer.seek(0)
    return base64.b64encode(buffer.read()).decode("ascii")


def make_bar_chart(title: str, labels: list[str], values: list[float], ylabel: str) -> str:
    require_matplotlib()

    plt.figure(figsize=(11, 5))
    plt.bar(labels, values)
    plt.title(title)
    plt.ylabel(ylabel)
    plt.xticks(rotation=45, ha="right")
    return fig_to_base64()


def make_month_user_line_chart(monthly_user: dict, title: str) -> str:
    """
    Time chart: Diff Delta per month, one line per user.

    Used for:
      - aggregate user/month across all repos
      - per-repo user/month charts
    """

    require_matplotlib()

    months = sorted(monthly_user.keys())
    authors = sorted(
        {
            author
            for authors in monthly_user.values()
            for author in authors.keys()
        }
    )

    if not months or not authors:
        return ""

    plt.figure(figsize=(13, 6))

    for author in authors:
        values = [
            monthly_user.get(month, {}).get(author, {}).get("diff_delta", 0.0)
            for month in months
        ]

        plt.plot(
            months,
            values,
            marker="o",
            linewidth=2,
            label=author,
        )

    plt.title(title)
    plt.ylabel("Diff Delta")
    plt.xlabel("Month")
    plt.xticks(rotation=45, ha="right")
    plt.grid(True, axis="y", alpha=0.3)
    plt.legend()

    return fig_to_base64()


def make_repo_total_chart(repo_results: dict[str, dict]) -> str:
    labels = []
    values = []

    for repo_name, result in sorted(
        repo_results.items(),
        key=lambda item: item[1]["diff_delta"],
        reverse=True,
    ):
        labels.append(repo_name)
        values.append(result["diff_delta"])

    return make_bar_chart(
        title="Diff Delta by Repo",
        labels=labels,
        values=values,
        ylabel="Diff Delta",
    )


def make_user_total_chart(user_totals: dict) -> str:
    labels = list(user_totals.keys())
    values = [stats["diff_delta"] for stats in user_totals.values()]

    return make_bar_chart(
        title="Diff Delta by User Across Full Range",
        labels=labels,
        values=values,
        ylabel="Diff Delta",
    )


def make_filter_chart(filter_summary: dict[str, int]) -> str:
    labels = []
    values = []

    for key, label in FILTER_LABELS.items():
        labels.append(label)
        values.append(filter_summary.get(key, 0))

    return make_bar_chart(
        title="Filter Summary",
        labels=labels,
        values=values,
        ylabel="Line Events",
    )


def html_escape(value: object) -> str:
    return html.escape(str(value))


def html_table(headers: list[str], rows: list[list[object]]) -> str:
    head = "".join(f"<th>{html_escape(header)}</th>" for header in headers)

    body_rows = []

    for row in rows:
        cells = "".join(f"<td>{html_escape(cell)}</td>" for cell in row)
        body_rows.append(f"<tr>{cells}</tr>")

    body = "\n".join(body_rows)

    return f"""
<table>
  <thead>
    <tr>{head}</tr>
  </thead>
  <tbody>
    {body}
  </tbody>
</table>
"""


def write_recurse_month_user_html_report(
    output_path: Path,
    repo_month_user_results: dict[str, dict],
    aggregate_month_user: dict,
    aggregate_user_totals: dict,
    aggregate_total_range: dict,
) -> None:
    require_matplotlib()

    repo_rows = []

    for repo_name, monthly_user in sorted(repo_month_user_results.items()):
        for month, authors in monthly_user.items():
            for author, stats in sorted(
                authors.items(),
                key=lambda item: item[1]["diff_delta"],
                reverse=True,
            ):
                repo_rows.append(
                    [
                        repo_name,
                        month,
                        author,
                        stats["commits"],
                        stats["events"],
                        stats["scored_events"],
                        f"{stats['diff_delta']:.6f}",
                    ]
                )

    aggregate_month_rows = []

    for month, authors in aggregate_month_user.items():
        for author, stats in sorted(
            authors.items(),
            key=lambda item: item[1]["diff_delta"],
            reverse=True,
        ):
            aggregate_month_rows.append(
                [
                    month,
                    author,
                    stats["commits"],
                    stats["events"],
                    stats["scored_events"],
                    f"{stats['diff_delta']:.6f}",
                ]
            )

    user_total_rows = []

    for author, stats in aggregate_user_totals.items():
        user_total_rows.append(
            [
                author,
                stats["commits"],
                stats["events"],
                stats["scored_events"],
                f"{stats['diff_delta']:.6f}",
            ]
        )

    repo_totals: dict[str, dict] = {}

    for repo_name, monthly_user in repo_month_user_results.items():
        repo_total = collapse_month_user_to_total(monthly_user)
        repo_totals[repo_name] = {
            "diff_delta": repo_total["diff_delta"],
            "commits": repo_total["commits"],
            "events": repo_total["events"],
            "scored_events": repo_total["scored_events"],
            "filter_summary": repo_total["filter_summary"],
        }

    repo_total_rows = []

    for repo_name, stats in sorted(
        repo_totals.items(),
        key=lambda item: item[1]["diff_delta"],
        reverse=True,
    ):
        repo_total_rows.append(
            [
                repo_name,
                stats["commits"],
                stats["events"],
                stats["scored_events"],
                f"{stats['diff_delta']:.6f}",
            ]
        )

    charts = {
        "aggregate_month_user": make_month_user_line_chart(
            aggregate_month_user,
            "Diff Delta per User per Month Across All Repos",
        ),
        "aggregate_user_total": make_user_total_chart(aggregate_user_totals),
        "repo_total": make_repo_total_chart(repo_totals),
        "filter_summary": make_filter_chart(aggregate_total_range["filter_summary"]),
    }

    repo_charts = []

    for repo_name, monthly_user in sorted(repo_month_user_results.items()):
        chart = make_month_user_line_chart(
            monthly_user,
            f"{repo_name}: Diff Delta per User per Month",
        )

        if chart:
            repo_charts.append(
                f"""
<section>
  <h3>{html_escape(repo_name)}</h3>
  <img src="data:image/png;base64,{chart}" />
</section>
"""
            )

    html_doc = f"""<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<title>Git Diff Delta Report</title>
<style>
  body {{
    font-family: Arial, sans-serif;
    margin: 32px;
    background: #111;
    color: #eee;
  }}

  h1, h2, h3 {{
    color: #fff;
  }}

  .summary {{
    display: grid;
    grid-template-columns: repeat(4, minmax(120px, 1fr));
    gap: 16px;
    margin: 24px 0;
  }}

  .card {{
    background: #1b1b1b;
    border: 1px solid #333;
    border-radius: 8px;
    padding: 16px;
  }}

  .card .label {{
    color: #aaa;
    font-size: 0.9rem;
  }}

  .card .value {{
    font-size: 1.5rem;
    font-weight: bold;
    margin-top: 8px;
  }}

  section {{
    margin: 36px 0;
  }}

  img {{
    max-width: 100%;
    background: #fff;
    padding: 12px;
    border-radius: 8px;
  }}

  table {{
    border-collapse: collapse;
    width: 100%;
    margin-top: 12px;
    font-size: 0.92rem;
  }}

  th, td {{
    border: 1px solid #333;
    padding: 8px 10px;
    text-align: left;
  }}

  th {{
    background: #222;
  }}

  tr:nth-child(even) {{
    background: #181818;
  }}

  code {{
    background: #222;
    padding: 2px 5px;
    border-radius: 4px;
  }}
</style>
</head>
<body>

<h1>Git Diff Delta Report</h1>

<section class="summary">
  <div class="card">
    <div class="label">Commits</div>
    <div class="value">{aggregate_total_range["commits"]}</div>
  </div>
  <div class="card">
    <div class="label">Events</div>
    <div class="value">{aggregate_total_range["events"]}</div>
  </div>
  <div class="card">
    <div class="label">Scored Events</div>
    <div class="value">{aggregate_total_range["scored_events"]}</div>
  </div>
  <div class="card">
    <div class="label">Diff Delta</div>
    <div class="value">{aggregate_total_range["diff_delta"]:.6f}</div>
  </div>
</section>

<section>
  <h2>Diff Delta per User per Month Across All Repos</h2>
  <img src="data:image/png;base64,{charts["aggregate_month_user"]}" />
</section>

<section>
  <h2>Diff Delta by User Across Full Range</h2>
  <img src="data:image/png;base64,{charts["aggregate_user_total"]}" />
  {html_table(
      ["Author", "Commits", "Events", "Scored", "Diff Delta"],
      user_total_rows,
  )}
</section>

<section>
  <h2>Diff Delta by Repo</h2>
  <img src="data:image/png;base64,{charts["repo_total"]}" />
  {html_table(
      ["Repo", "Commits", "Events", "Scored", "Diff Delta"],
      repo_total_rows,
  )}
</section>

<section>
  <h2>Per-Repo Monthly/User Charts</h2>
  {''.join(repo_charts)}
</section>

<section>
  <h2>Aggregate by Month/User</h2>
  {html_table(
      ["Month", "Author", "Commits", "Events", "Scored", "Diff Delta"],
      aggregate_month_rows,
  )}
</section>

<section>
  <h2>Per-Repo Month/User Rows</h2>
  {html_table(
      ["Repo", "Month", "Author", "Commits", "Events", "Scored", "Diff Delta"],
      repo_rows,
  )}
</section>

<section>
  <h2>Filter Summary</h2>
  <img src="data:image/png;base64,{charts["filter_summary"]}" />
  {html_table(
      ["Filter", "Lines"],
      [
          [FILTER_LABELS[key], aggregate_total_range["filter_summary"].get(key, 0)]
          for key in FILTER_LABELS
      ],
  )}
</section>

</body>
</html>
"""

    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(html_doc, encoding="utf-8")


def run_recurse_mode(args: argparse.Namespace) -> int:
    root = args.recurse.resolve()
    repos = find_direct_git_repos(root)

    if not repos:
        print(f"No Git repos found under direct subfolders of: {root}")
        return 0

    since = args.date_from or args.since
    until = args.date_to

    if args.by_month_user:
        repo_month_user_results: dict[str, dict] = {}

        for repo in repos:
            repo_name = repo.name

            try:
                if args.commit:
                    commits = [get_one_commit(repo, args.commit)]
                else:
                    commits = get_commits(repo, since, until, args.max_count)

                if not commits:
                    continue

                repo_month_user_results[repo_name] = compute_by_month_user(
                    repo,
                    commits,
                )

            except Exception as exc:
                print(f"warning: skipped {repo_name}: {exc}", file=sys.stderr)

        if not repo_month_user_results:
            print("No commits found in any repo for the selected range.")
            print(f"Root:  {root}")
            print(f"From:  {since or '(none)'}")
            print(f"To:    {until or '(none)'}")
            return 0

        aggregate_month_user = merge_month_user_results(repo_month_user_results)
        aggregate_user_totals = collapse_month_user_to_user_totals(aggregate_month_user)
        aggregate_total_range = collapse_month_user_to_total(aggregate_month_user)

        if args.html_report:
            write_recurse_month_user_html_report(
                args.html_report,
                repo_month_user_results,
                aggregate_month_user,
                aggregate_user_totals,
                aggregate_total_range,
            )

        if args.json:
            print(
                json.dumps(
                    {
                        "repos": repo_month_user_results,
                        "aggregate_by_month_user": aggregate_month_user,
                        "aggregate_by_user_total_range": aggregate_user_totals,
                        "aggregate_total_range": aggregate_total_range,
                    },
                    indent=2,
                )
            )
        else:
            print_recurse_month_user_report(
                repo_month_user_results,
                aggregate_month_user,
            )

            if args.html_report:
                print()
                print(f"HTML report written to: {args.html_report.resolve()}")

        return 0

    if args.by_month:
        print(
            "error: --recurse --by-month is not implemented. "
            "Use --recurse --by-month-user or plain --recurse.",
            file=sys.stderr,
        )
        return 1

    repo_results: dict[str, dict] = {}
    aggregate_inputs: list[dict] = []

    for repo in repos:
        repo_name = repo.name

        try:
            if args.commit:
                commits = [get_one_commit(repo, args.commit)]
            else:
                commits = get_commits(repo, since, until, args.max_count)

            if not commits:
                continue

            result = compute(repo, commits)

            repo_results[repo_name] = result
            aggregate_inputs.append(namespace_result(repo_name, result))

        except Exception as exc:
            print(f"warning: skipped {repo_name}: {exc}", file=sys.stderr)

    if not repo_results:
        print("No commits found in any repo for the selected range.")
        print(f"Root:  {root}")
        print(f"From:  {since or '(none)'}")
        print(f"To:    {until or '(none)'}")
        return 0

    aggregate = aggregate_results(aggregate_inputs)

    if args.json:
        print(
            json.dumps(
                {
                    "repos": repo_results,
                    "aggregate": aggregate,
                },
                indent=2,
            )
        )
    else:
        print_recurse_report(repo_results, aggregate)

    return 0


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Compute Diff Delta stats for Git repos."
    )

    parser.add_argument(
        "repo",
        type=Path,
        nargs="?",
        help="Path to Git repo.",
    )

    parser.add_argument(
        "--recurse",
        type=Path,
        help="Look at every direct subfolder of this path and analyse each Git repo found.",
    )

    parser.add_argument(
        "--since",
        help='Git date string. Example: "30 days ago", "2025-06-01".',
    )

    parser.add_argument(
        "--from",
        dest="date_from",
        help="Start date. Example: 2025-06-01.",
    )

    parser.add_argument(
        "--to",
        dest="date_to",
        help="End date. Example: 2025-11-30.",
    )

    parser.add_argument(
        "--max-count",
        type=int,
        help="Maximum number of commits to analyse.",
    )

    parser.add_argument(
        "--commit",
        help="Score one commit. Example: HEAD or a commit hash.",
    )

    parser.add_argument(
        "--by-month",
        action="store_true",
        help="Break stats down by month.",
    )

    parser.add_argument(
        "--by-month-user",
        action="store_true",
        help="Break stats down by month and author.",
    )

    parser.add_argument(
        "--exclude-ext",
        action="append",
        default=[],
        help=(
            "Exclude a file extension from scoring. "
            "Can be used more than once. Example: --exclude-ext .csv --exclude-ext .xml"
        ),
    )

    parser.add_argument(
        "--html-report",
        type=Path,
        help="Write an HTML report with charts to this file.",
    )

    parser.add_argument(
        "--json",
        action="store_true",
        help="Print JSON output.",
    )

    return parser.parse_args()


def main() -> int:
    args = parse_args()

    global EXCLUDED_EXTENSIONS

    EXCLUDED_EXTENSIONS = set(DEFAULT_EXCLUDED_EXTENSIONS)

    for ext in args.exclude_ext:
        normalised = normalise_extension(ext)
        if normalised:
            EXCLUDED_EXTENSIONS.add(normalised)

    try:
        if args.html_report and not (args.recurse and args.by_month_user):
            print(
                "error: --html-report currently requires --recurse --by-month-user",
                file=sys.stderr,
            )
            return 1

        if args.recurse:
            return run_recurse_mode(args)

        if not args.repo:
            print("error: repo path is required unless --recurse is used", file=sys.stderr)
            return 1

        repo = args.repo.resolve()
        validate_repo(repo)

        if args.commit:
            commits = [get_one_commit(repo, args.commit)]
        else:
            since = args.date_from or args.since
            until = args.date_to
            commits = get_commits(repo, since, until, args.max_count)

            if not commits:
                print("No commits found for the selected range.")
                print(f"Repo:  {repo}")
                print(f"From:  {since or '(none)'}")
                print(f"To:    {until or '(none)'}")
                return 0

        if args.by_month_user:
            monthly_user = compute_by_month_user(repo, commits)

            if args.json:
                print(json.dumps(monthly_user, indent=2))
            else:
                print_month_user_report(monthly_user)

        elif args.by_month:
            monthly = compute_by_month(repo, commits)

            if args.json:
                print(json.dumps(monthly, indent=2))
            else:
                print_monthly_report(monthly)

        else:
            result = compute(repo, commits)

            if args.json:
                print(json.dumps(result, indent=2))
            else:
                print_text_report(result)

        return 0

    except Exception as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())