git_archaeology.py

# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "marimo",
#     "polars==1.35.2",
#     "altair==6.0.0",
#     "httpx==0.28.1",
#     "pydantic>=2.0.0",
#     "diskcache==5.6.3",
# ]
# ///

import marimo

__generated_with = "0.18.4"
app = marimo.App(width="medium")


@app.cell
def _():
    import marimo as mo
    return (mo,)


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    # Git Code Archaeology

    This notebook analyzes a git repository to visualize how code ages over time.
    It creates a stacked area chart showing lines of code broken down by the year
    each line was originally added, revealing how quickly code gets replaced.
    """)
    return


@app.cell
def _():
    import subprocess
    from datetime import datetime
    from collections import defaultdict
    import polars as pl
    import altair as alt
    from diskcache import Cache

    cache = Cache("git-research")
    return alt, cache, datetime, pl, subprocess


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## Configuration
    """)
    return


@app.cell
def _():
    return


@app.cell
def _(mo):
    repo_url_input = mo.ui.text(
        value="https://github.com/marimo-team/marimo",
        label="Repository URL (HTTPS)",
        full_width=True,
    )
    repo_url_input
    return (repo_url_input,)


@app.cell
def _(mo):
    sample_count_slider = mo.ui.slider(
        start=10,
        stop=200,
        value=100,
        step=5,
        label="Number of commits to sample",
    )
    sample_count_slider
    return (sample_count_slider,)


@app.cell
def _(mo):
    file_extensions_input = mo.ui.text(
        value=".py,.js,.ts,.java,.c,.cpp,.h,.go,.rs,.rb,.md",
        label="File extensions to analyze (comma-separated, leave empty for all)",
        full_width=True,
    )
    file_extensions_input
    return (file_extensions_input,)


@app.cell
def _(mo):
    granularity_select = mo.ui.dropdown(
        options=["Year", "Quarter"],
        value="Quarter",
        label="Time granularity",
    )
    granularity_select
    return (granularity_select,)


@app.cell
def _(mo):
    show_versions = mo.ui.checkbox(label="show versions")
    show_versions
    return (show_versions,)


@app.cell
def _():
    from pydantic import BaseModel, Field
    from pydantic_core import PydanticUndefined

    class RepoParams(BaseModel):
        repo: str = Field(description="Repository URL (HTTPS)")
        samples: int = Field(default=100, description="Number of commits to sample")

    return (RepoParams,)


@app.cell
def _(RepoParams, mo):
    cli_args = mo.cli_args()

    if mo.app_meta().mode == "script":
        if "help" in cli_args or len(cli_args) == 0:
            print("Usage: uv run git_archaeology.py --repo <url> [--samples <n>]")
            print()
            for name, field in RepoParams.model_fields.items():
                default = " (required)" if field.default is PydanticUndefined else f" (default: {field.default})"
                print(f"  --{name:12s} {field.description}{default}")
            exit()
        repo_params = RepoParams(
            **{k.replace("-", "_"): v for k, v in cli_args.items()}
        )
    return cli_args, repo_params


@app.cell(hide_code=True)
def _(subprocess):
    from pathlib import Path
    import hashlib

    DOWNLOADS_DIR = Path(".downloads")


    def get_cached_repo_path(repo_url: str) -> Path:
        """Get the cached path for a repo URL, using a hash for uniqueness."""
        repo_name = repo_url.rstrip("/").split("/")[-1].replace(".git", "")
        url_hash = hashlib.md5(repo_url.encode()).hexdigest()[:8]
        return DOWNLOADS_DIR / f"{repo_name}-{url_hash}"


    def clone_or_update_repo(repo_url: str) -> Path:
        """Clone repo if not cached, otherwise return cached path."""
        DOWNLOADS_DIR.mkdir(exist_ok=True)
        repo_path = get_cached_repo_path(repo_url)

        if repo_path.exists():
            # Repo already cached, fetch latest
            subprocess.run(
                ["git", "fetch", "--all"],
                cwd=repo_path,
                capture_output=True,
            )
        else:
            # Clone fresh
            subprocess.run(
                ["git", "clone", repo_url, str(repo_path)],
                capture_output=True,
                check=True,
            )
        return repo_path
    return Path, clone_or_update_repo


@app.cell(hide_code=True)
def _(cache, datetime, subprocess):
    from concurrent.futures import ThreadPoolExecutor, as_completed
    import re

    # Pre-compile regex for timestamp extraction (used in get_blame_info)
    # Format: hash (author timestamp tz line_num) content
    TIMESTAMP_PATTERN = re.compile(r"\(.*?\s+(\d{10})\s+[+-]\d{4}\s+\d+\)")


    def run_git_command(cmd: list[str], repo_path: str) -> str:
        """Run a git command and return stdout."""
        result = subprocess.run(
            cmd,
            cwd=repo_path,
            capture_output=True,
            text=True,
        )
        if result.returncode != 0:
            raise RuntimeError(f"Git command failed: {result.stderr}")
        return result.stdout


    @cache.memoize()
    def get_commit_list(repo_path: str) -> list[tuple[str, datetime]]:
        """Get list of all commits with their dates."""
        output = run_git_command(
            ["git", "log", "--format=%H %at", "--reverse"],
            repo_path,
        )
        commits = []
        for line in output.strip().split("\n"):
            if line:
                parts = line.split()
                commit_hash = parts[0]
                timestamp = int(parts[1])
                commit_date = datetime.fromtimestamp(timestamp)
                commits.append((commit_hash, commit_date))
        return commits


    def get_tracked_files(
        repo_path: str, commit_hash: str, extensions: list[str] | None = None
    ) -> list[str]:
        """Get list of tracked files at a specific commit."""
        output = run_git_command(
            ["git", "ls-tree", "-r", "--name-only", commit_hash],
            repo_path,
        )
        files = output.strip().split("\n")
        if extensions:
            files = [f for f in files if any(f.endswith(ext) for ext in extensions)]
        return [f for f in files if f]


    def get_blame_info(repo_path: str, commit_hash: str, file_path: str) -> list[int]:
        """
        Get blame info for a file at a specific commit.
        Returns list of timestamps for each line.

        Uses -t flag for raw timestamp output which is much faster than --line-porcelain.
        Format: <hash> <orig_line> <final_line> <num_lines> (<author> <timestamp> <tz>) <content>
        """
        try:
            output = run_git_command(
                ["git", "blame", "-t", commit_hash, "--", file_path],
                repo_path,
            )
        except (RuntimeError, UnicodeDecodeError):
            # File might be binary or have other issues
            return []

        timestamps = []
        for line in output.split("\n"):
            if not line:
                continue
            match = TIMESTAMP_PATTERN.search(line)
            if match:
                timestamps.append(int(match.group(1)))

        return timestamps


    @cache.memoize()
    def sample_commits(
        commits: list[tuple[str, datetime]], n_samples: int
    ) -> list[tuple[str, datetime]]:
        """Sample n commits evenly distributed across history."""
        if len(commits) <= n_samples:
            return commits
        step = len(commits) / n_samples
        indices = [int(i * step) for i in range(n_samples)]
        # Always include the last commit
        if indices[-1] != len(commits) - 1:
            indices[-1] = len(commits) - 1
        return [commits[i] for i in indices]


    @cache.memoize()
    def analyze_single_commit(
        repo_path: str,
        commit_hash: str,
        commit_date: datetime,
        extensions: list[str] | None,
        file_workers: int = 4,
    ) -> list[tuple[datetime, int]]:
        """Analyze a single commit - designed for parallel execution.

        Uses nested parallelization: commits in parallel, files within each commit also in parallel.
        """
        files = get_tracked_files(repo_path, commit_hash, extensions)

        def blame_file(file_path: str) -> list[int]:
            return get_blame_info(repo_path, commit_hash, file_path)

        results = []
        # Parallelize file processing within each commit
        with ThreadPoolExecutor(max_workers=file_workers) as file_executor:
            file_futures = {file_executor.submit(blame_file, f): f for f in files}
            for future in as_completed(file_futures):
                timestamps = future.result()
                for ts in timestamps:
                    results.append((commit_date, ts))
        return results


    @cache.memoize(ignore=["progress_bar", "is_script"])
    def collect_blame_data(
        repo_path: str,
        sampled_commits: list[tuple[str, datetime]],
        extensions: list[str] | None,
        progress_bar=None,
        is_script: bool = False,
        max_workers: int = 12,
    ) -> list[tuple[datetime, int]]:
        """Collect raw blame data from sampled commits in parallel."""
        import time

        raw_data = []
        total = len(sampled_commits)
        done = 0
        last_time = time.time()

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(analyze_single_commit, str(repo_path), h, d, extensions): (h, d)
                for h, d in sampled_commits
            }
            for future in as_completed(futures):
                commit_hash, _ = futures[future]
                done += 1
                now = time.time()
                step_duration = now - last_time
                last_time = now
                if progress_bar:
                    progress_bar.update(title=f"Analyzed {commit_hash[:8]}...")
                if is_script:
                    print(f"  [{done}/{total}] Analyzed {commit_hash[:8]} ({step_duration:.1f}s)")
                raw_data.extend(future.result())

        return raw_data
    return collect_blame_data, get_commit_list, sample_commits


@app.cell
def _(
    clone_or_update_repo,
    file_extensions_input,
    get_commit_list,
    mo,
    repo_params,
    repo_url_input,
    sample_commits,
    sample_count_slider,
):
    # Clone or use cached repo
    repo_url = repo_params.repo if mo.app_meta().mode == "script" else repo_url_input.value.strip()
    with mo.status.spinner(f"Cloning/updating repository..."):
        repo_path = clone_or_update_repo(repo_url)

    # Parse configuration
    n_samples = repo_params.samples if mo.app_meta().mode == "script" else sample_count_slider.value
    extensions_str = file_extensions_input.value.strip()
    extensions = [ext.strip() for ext in extensions_str.split(",")] if extensions_str else None

    # Get commits
    with mo.status.spinner("Getting commit history..."):
        all_commits = get_commit_list(str(repo_path))
        sampled = sample_commits(all_commits, n_samples)

    mo.md(f"Found **{len(all_commits)}** commits, sampling **{len(sampled)}** for analysis")
    return extensions, repo_path, sampled


@app.cell
def _(collect_blame_data, extensions, mo, pl, repo_path, sampled):
    with mo.status.progress_bar(
        total=len(sampled),
        title="Analyzing commits",
        show_rate=True,
        show_eta=True,
    ) as bar:
        raw_data = collect_blame_data(repo_path, sampled, extensions, progress_bar=bar, is_script=mo.app_meta().mode == "script")

    # Store raw data as DataFrame with timestamps
    raw_df = pl.DataFrame(raw_data, schema=["commit_date", "line_timestamp"], orient="row")
    return (raw_df,)


@app.cell(hide_code=True)
def _(mo):
    mo.md("""
    ## Visualization
    """)
    return


@app.cell
def _(datetime, granularity_select, pl, raw_df):
    granularity = granularity_select.value


    def get_period(ts: int, granularity: str) -> str:
        dt = datetime.fromtimestamp(ts)
        if granularity == "Year":
            return str(dt.year)
        else:  # Quarter
            q = (dt.month - 1) // 3 + 1
            return f"{dt.year}-Q{q}"


    # Apply granularity and aggregate
    df = (
        raw_df.with_columns(
            pl.col("line_timestamp")
            .map_elements(lambda ts: get_period(ts, granularity), return_dtype=pl.Utf8)
            .alias("period")
        )
        .group_by(["commit_date", "period"])
        .len()
        .rename({"len": "line_count"})
        .sort(["commit_date", "period"])
    )
    return (df,)


@app.cell
def _(mo, repo_params, repo_url_input):
    import httpx

    _repo = repo_params.repo if mo.app_meta().mode == "script" else repo_url_input.value
    parts = _repo.split("/")
    repo_name = parts[-2] if _repo.endswith("/") else parts[-1]

    res = httpx.get(f"https://pypi.org/pypi/{repo_name}/json").json()
    return repo_name, res


@app.cell
def _(alt, pl, res):
    _version_data = [
        {"version": key, "datetime": value[0]["upload_time"]}
        for key, value in res["releases"].items()
        if key.endswith(".0") and key != "0.0.0" and len(value) > 0
    ]
    df_versions = pl.DataFrame(
        _version_data,
        schema={"version": pl.Utf8, "datetime": pl.Utf8},
    ).with_columns(datetime=pl.col("datetime").str.to_datetime())

    base_chart = alt.Chart(df_versions)

    date_lines = base_chart.mark_rule(strokeDash=[5, 5]).encode(
        x=alt.X("datetime:T", title="Date"), tooltip=["version:N", "datetime:T"]
    )

    date_text = base_chart.mark_text(angle=270, align="left", dx=15, dy=0).encode(
        x="datetime:T", y=alt.value(10), text="version:N"
    )
    return date_lines, date_text


@app.cell
def _(alt, date_lines, date_text, df, granularity_select, show_versions):
    color_title = "Year Added" if granularity_select.value == "Year" else "Quarter Added"

    chart = (
        alt.Chart(df)
        .mark_area()
        .encode(
            x=alt.X("commit_date:T", title="Date"),
            y=alt.Y("line_count:Q", title="Lines of Code"),
            color=alt.Color(
                "period:O",
                scale=alt.Scale(scheme="viridis"),
                title=color_title,
            ),
            order=alt.Order("period:O"),
            tooltip=["commit_date:T", "period:O", "line_count:Q"],
        )
    )

    out = chart
    if show_versions.value:
        out += date_lines + date_text

    out = out.properties(
        title="Code Archaeology: Lines of Code by Period Added",
        width=800,
        height=500,
    )

    out
    return chart, out


@app.cell
def _(Path, alt, chart, date_lines, date_text, out, repo_name):
    Path("charts").mkdir(exist_ok=True)

    clean_path = Path("charts") / (repo_name + "-clean.json")
    clean_path.write_text(out.to_json())

    versioned_path = Path("charts") / (repo_name + "-versioned.json")
    versioned_chart = (
        (chart + date_lines + date_text)
        .properties(
            title="Code Archaeology: Lines of Code by Period Added",
            width=800,
            height=500,
        )
        .to_dict()
    )
    versioned_path.write_text(alt.Chart.from_dict(versioned_chart).to_json())
    return


@app.cell
def _():
    return


if __name__ == "__main__":
    app.run()