commitose.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import math
import os
import random
import secrets
import subprocess
import sys
from dataclasses import dataclass, field
from datetime import date, timedelta
from enum import Enum
from pathlib import Path
from types import UnionType
from typing import Any, Optional, get_args, get_origin, get_type_hints

import tomllib

TODAY = date.today()

Week = tuple[float, float, float, float, float, float, float]


@dataclass
class Commit:
    date: date
    message: str


@dataclass
class Break:
    """A time period with scaled commit activity."""

    start_date: date
    end_date: date
    factor: float = 0.0


class MarkovChain:
    Vector = tuple[float, float, float, float]
    Matrix = tuple[Vector, Vector, Vector, Vector]

    class State(Enum):
        OFF = 0
        QUIET = 1
        NORMAL = 2
        BUSY = 3

    transition_matrix: Matrix
    rng: random.Random
    current_state: State

    def __init__(self, transition_matrix: Matrix, seed: int | None = None):
        self.transition_matrix = transition_matrix
        self.rng = random.Random(seed)
        self.current_state = MarkovChain.State.OFF

    def next(self) -> MarkovChain.State:
        probabilities = self.transition_matrix[self.current_state.value]
        self.current_state = self.rng.choices(
            list(MarkovChain.State), weights=probabilities
        )[0]
        return self.current_state


@dataclass
class Sampler:
    seed: int | None = None
    rng: random.Random = field(init=False)

    def __post_init__(self):
        self.rng = random.Random(self.seed)

    def zinb(self, mean: float, dispersion: float, zero_inflation: float) -> int:
        # Zero inflation
        if self.rng.random() < zero_inflation:
            return 0

        # Negative binomial via gamma-Poisson mixture
        # See: https://en.wikipedia.org/wiki/Negative_binomial_distribution
        if mean > 0:
            lam = self.rng.gammavariate(dispersion, mean / dispersion)
            return self.poisson(lam)

        return 0

    def poisson(self, lam: float) -> int:
        if lam <= 0:
            return 0

        # Knuth algorithm
        # See: https://en.wikipedia.org/wiki/Poisson_distribution#Random_variate_generation
        L = math.exp(-lam)
        k = 0
        p = 1.0

        while p > L:
            k += 1
            p *= self.rng.random()

        return k - 1


@dataclass
class Simulator:
    start_date: date
    end_date: date

    # Weights for each weekday. Allows commit activity to be scaled by a factor
    # depending on the day of the week.
    weekday_weights: Week

    breaks: list[Break]

    # Mean number of commits per state
    commit_means: MarkovChain.Vector
    # Dispersion parameter for commit count distribution
    commit_dispersion: float
    # Zero inflation per state
    commit_zero_inflation: MarkovChain.Vector

    transition_matrix: MarkovChain.Matrix

    seed: int | None = None
    sampler: Sampler = field(init=False)
    markov_chain: MarkovChain = field(init=False)

    def __post_init__(self):
        self.sampler = Sampler(self.seed)
        self.markov_chain = MarkovChain(self.transition_matrix, self.seed)

    def simulate(self) -> list[Commit]:
        commits: list[Commit] = []
        current_date = self.start_date

        while current_date <= self.end_date:
            commit_count = self.sample_daily_commits(current_date)

            for _ in range(commit_count):
                commits.append(
                    Commit(date=current_date, message=self.generate_commit_msg())
                )

            current_date += timedelta(days=1)

        return commits

    def sample_daily_commits(self, date: date) -> int:
        state = self.markov_chain.next()

        if state == MarkovChain.State.OFF:
            return 0

        weekday_weight = self.weekday_weights[date.weekday()]
        break_factor = self.get_break_factor(date)
        commit_mean = self.commit_means[state.value] * weekday_weight * break_factor
        commit_zero_inflation = self.commit_zero_inflation[state.value]

        commit_count = self.sampler.zinb(
            commit_mean, self.commit_dispersion, commit_zero_inflation
        )

        return commit_count

    def get_break_factor(self, date: date) -> float:
        """Return activity scaling factor for a given date."""
        for b in self.breaks:
            if b.start_date <= date <= b.end_date:
                return b.factor
        return 1.0

    def generate_commit_msg(self) -> str:
        """Generate a random commit message."""
        return secrets.token_hex(8)


@dataclass
class GitClient:
    user_name: str
    user_email: str
    repo_path: Path
    branch: str

    class WouldOverwriteRepoError(Exception):
        pass

    def exec_git(
        self, cmd: list[str], env: dict[str, str] | None = None, check: bool = False
    ):
        """Execute a git command in the repository."""
        subprocess.run(
            ["git", *cmd],
            cwd=self.repo_path,
            env=env,
            check=check,
            capture_output=True,
        )

    def commit_all(self, commits: list[Commit]) -> None:
        """Commit all commits in the list."""
        for i, commit in enumerate(commits, 1):
            self.commit(commit)
            if i == len(commits):
                print(f"\033[KCommitted {i}/{len(commits)} commits!")
            elif i % 10 == 0:
                print(f"\033[KCommitted {i}/{len(commits)} commits...", end="\r")

    def commit(self, commit: Commit) -> None:
        env = os.environ.copy()
        date_str = commit.date.strftime("%Y-%m-%dT%H:%M:%S%z")
        env.update(
            {
                "GIT_AUTHOR_NAME": self.user_name,
                "GIT_AUTHOR_EMAIL": self.user_email,
                "GIT_AUTHOR_DATE": date_str,
                "GIT_COMMITTER_NAME": self.user_name,
                "GIT_COMMITTER_EMAIL": self.user_email,
                "GIT_COMMITTER_DATE": date_str,
            }
        )
        self.exec_git(["commit", "--allow-empty", "-m", commit.message], env)

    def wipe_repo(self) -> None:
        self.exec_git(["checkout", "--orphan", "temp-branch"])
        self.exec_git(["rm", "-rf", "."], check=False)
        # If user has switched branch, the new branch may not yet exist
        self.exec_git(["branch", "-D", self.branch], check=False)
        self.exec_git(["branch", "-m", self.branch])

    @staticmethod
    def check_repo_path(repo_path: Path) -> None:
        repo_path.mkdir(parents=True, exist_ok=True)

        # Ensure that directory is empty, or only contains .git
        for entry in repo_path.iterdir():
            if not (entry.name == ".git" and entry.is_dir()):
                raise ValueError(
                    f"Directory '{repo_path}' is not empty (found '{entry.name}')"
                )

        if (repo_path / ".git").exists():
            # Verify it's a valid git repository
            try:
                subprocess.run(
                    ["git", "rev-parse", "--git-dir"],
                    cwd=repo_path,
                    check=True,
                    capture_output=True,
                )
            except subprocess.CalledProcessError:
                raise ValueError(
                    f"Directory {repo_path} contains invalid .git directory"
                )
            raise GitClient.WouldOverwriteRepoError()
        else:
            # Create new repo
            subprocess.run(
                ["git", "init"],
                cwd=repo_path,
                check=True,
                capture_output=True,
            )

    @staticmethod
    def get_git_config(key: str) -> str | None:
        """Get a value from git config."""
        try:
            result = subprocess.run(
                ["git", "config", "--get", key],
                capture_output=True,
                text=True,
                check=True,
            )
            return result.stdout.strip()
        except subprocess.CalledProcessError:
            return None


@dataclass
class Config:
    start_date: date = TODAY - timedelta(days=365)
    end_date: date = TODAY
    breaks: list[Break] = field(default_factory=list)

    weekday_weights: Week = (1, 1, 1, 1, 1, 1, 1)
    commit_means: MarkovChain.Vector = (0, 1, 3, 7)
    commit_dispersion: float = 1
    commit_zero_inflation: MarkovChain.Vector = (1, 0.3, 0.1, 0.05)

    transition_matrix: MarkovChain.Matrix = (
        (0.65, 0.25, 0.10, 0.00),  # off
        (0.15, 0.60, 0.20, 0.05),  # quiet
        (0.05, 0.15, 0.65, 0.15),  # normal
        (0.00, 0.05, 0.35, 0.60),  # busy
    )

    seed: int | None = None

    repo_path: Path = Path("commits_repo")
    user_name: str | None = None
    user_email: str | None = None
    branch: str = "main"
    dry_run: bool = False

    def load_toml(self, path: str) -> None:
        """Load configuration from TOML file."""
        with open(path, "rb") as f:
            raw_data = tomllib.load(f)
            coerced_data = Config.coerce_config(raw_data)
            for field_name in coerced_data:
                setattr(self, field_name, coerced_data[field_name])
        self._validate()

    def load_args(self, args: dict[str, Any]):
        """Load configuration from parsed CLI arguments."""
        coerced_data = Config.coerce_config(args)
        for field_name in coerced_data:
            setattr(self, field_name, coerced_data[field_name])
        self._validate()

    @staticmethod
    def coerce_config(raw_data: dict[str, Any]) -> dict[str, Any]:
        coerced_data = {}
        field_types = get_type_hints(Config)

        for field_name in raw_data:
            if field_name not in field_types:
                continue
            value = raw_data[field_name]
            if value is None:
                # Value not passed
                continue
            expected_type = field_types[field_name]
            coerced_data[field_name] = Config.coerce_field(
                field_name, value, expected_type
            )

        return coerced_data

    @staticmethod
    def coerce_field(field_name: str, value: Any, expected_type: Any) -> Any:
        origin = get_origin(expected_type)
        args = get_args(expected_type)

        # Date
        if expected_type is date:
            # If passed as date (from TOML conversion), no need to coerce
            if isinstance(value, date):
                return value
            # If passed as string, should be ISO format
            if isinstance(value, str):
                try:
                    return date.fromisoformat(value)
                except ValueError:
                    raise ValueError(
                        f"Invalid {field_name} '{value}', expected ISO format (YYYY-MM-DD)"
                    )
            raise TypeError(
                f"Expected date or ISO string for {field_name}, got {type(value).__name__}"
            )

        # Break
        if expected_type is Break:
            return Config.coerce_break(value)

        # Path
        if expected_type is Path:
            # Should be passed as string or Path
            if isinstance(value, str):
                return Path(value)
            if isinstance(value, Path):
                return value
            raise TypeError(
                f"Expected path for {field_name}, got {type(value).__name__}"
            )

        # String
        if expected_type is str:
            # Should be passed as string
            if isinstance(value, str):
                return value
            raise TypeError(
                f"Expected string for {field_name}, got {type(value).__name__}"
            )

        # Int
        if expected_type is int:
            # Should be passed as int
            if isinstance(value, int):
                return value
            raise TypeError(
                f"Expected int for {field_name}, got {type(value).__name__}"
            )

        # Float
        if expected_type is float:
            # Should be passed as float
            if isinstance(value, float):
                return value
            # But we can also accept ints
            if isinstance(value, int):
                return float(value)
            raise TypeError(
                f"Expected float for {field_name}, got {type(value).__name__}"
            )

        # Boolean
        if expected_type is bool:
            # Should be passed as bool
            if isinstance(value, bool):
                return value
            raise TypeError(
                f"Expected bool for {field_name}, got {type(value).__name__}"
            )

        # Union
        if origin is UnionType:
            # Union types for this class consist of None and one other type
            # If the config option has been passed, we expect the other type
            some_type = next(arg for arg in args if arg is not type(None))
            return Config.coerce_field(field_name, value, some_type)

        # List
        if origin is list:
            # Should be passed as list
            if isinstance(value, list):
                return [Config.coerce_field(field_name, v, args[0]) for v in value]
            raise TypeError(
                f"Expected list for {field_name}, got {type(value).__name__}"
            )

        # Tuple
        if origin is tuple:
            # Should be passed as list
            if isinstance(value, list):
                if len(value) != len(args):
                    raise ValueError(
                        f"Expected tuple of length {len(args)} for {field_name}, got {len(value)}"
                    )
                return tuple(
                    Config.coerce_field(field_name, v, args[i])
                    for i, v in enumerate(value)
                )
            raise TypeError(
                f"Expected list (to convert to tuple), got {type(value).__name__}"
            )

    # TODO: Also support dict input
    @staticmethod
    def coerce_break(value: Any) -> Break:
        """Parse break from string format START:END[:FACTOR]."""
        BREAK_FORMAT = "START_DATE:END_DATE[:FLOAT]"
        if isinstance(value, str):
            try:
                start_date, end_date, *rest = value.split(":")
            except ValueError:
                raise ValueError(
                    f"Invalid break '{value}', expected format: {BREAK_FORMAT}"
                )
            if len(rest) == 0:
                factor = None
            elif len(rest) == 1:
                try:
                    factor = float(rest[0])
                except ValueError:
                    raise ValueError(
                        f"Invalid break factor '{rest[0]}', expected a float"
                    )
            else:
                raise ValueError(f"Invalid break '{value}', expected {BREAK_FORMAT}")
            try:
                start_date = date.fromisoformat(start_date)
            except ValueError:
                raise ValueError(
                    f"Invalid break start date '{start_date}', expected ISO format (YYYY-MM-DD)"
                )
            try:
                end_date = date.fromisoformat(end_date)
            except ValueError:
                raise ValueError(
                    f"Invalid break end date '{end_date}', expected ISO format (YYYY-MM-DD)"
                )
            if factor is not None:
                return Break(start_date, end_date, factor)
            else:
                return Break(start_date, end_date)

        if isinstance(value, dict):
            start_date = value.get("start_date")
            end_date = value.get("end_date")
            factor = value.get("factor")

            if start_date is None:
                raise ValueError("Break dict must contain 'start_date'")
            if end_date is None:
                raise ValueError("Break dict must contain 'end_date'")

            # Convert dates if they're strings
            if isinstance(start_date, str):
                try:
                    start_date = date.fromisoformat(start_date)
                except ValueError:
                    raise ValueError(
                        f"Invalid break start date '{start_date}', expected ISO format (YYYY-MM-DD)"
                    )
            if isinstance(end_date, str):
                try:
                    end_date = date.fromisoformat(end_date)
                except ValueError:
                    raise ValueError(
                        f"Invalid break end date '{end_date}', expected ISO format (YYYY-MM-DD)"
                    )

            if factor is None:
                return Break(start_date, end_date)
            else:
                return Break(start_date, end_date, factor)

        raise TypeError(f"Expected break string or dict, got {type(value).__name__}")

    def _validate(self) -> None:
        if self.start_date > self.end_date:
            raise ValueError(
                f"Simulation start date ({self.start_date}) must be before end date ({self.end_date})"
            )
        for weight in self.weekday_weights:
            if weight < 0:
                raise ValueError(
                    f"All weekday weights must be non-negative (got {self.weekday_weights})"
                )
        for mean in self.commit_means:
            if mean < 0:
                raise ValueError(
                    f"All commit means must be non-negative (got {self.commit_means})"
                )
        if self.commit_dispersion <= 0:
            raise ValueError(
                f"Commit dispersion must be positive (got {self.commit_dispersion})"
            )
        for zero_inflation in self.commit_zero_inflation:
            if not 0 <= zero_inflation <= 1:
                raise ValueError(
                    f"Zero inflation probabilities must be between 0 and 1 (got {self.commit_zero_inflation})"
                )
        for i, row in enumerate(self.transition_matrix):
            if not math.isclose(sum(row), 1.0, rel_tol=1e-9):
                raise ValueError(
                    f"Transition matrix rows must sum to 1.0 (got {sum(row):.6f} in row {i})"
                )
        self._validate_breaks()

    def _validate_breaks(self) -> None:
        """Ensure breaks are valid and don't overlap."""
        # Validate
        for b in self.breaks:
            if b.start_date > b.end_date:
                raise ValueError(
                    f"Break start date must be before end date (got {b.start_date} to {b.end_date})"
                )
            if b.factor < 0:
                raise ValueError(f"Break factor must be non-negative (got {b.factor})")

        # Check for overlaps
        sorted_breaks = sorted(self.breaks, key=lambda b: b.start_date)
        for i in range(len(sorted_breaks) - 1):
            current_break = sorted_breaks[i]
            next_break = sorted_breaks[i + 1]
            if current_break.end_date >= next_break.start_date:
                raise ValueError(
                    f"Breaks must not overlap: {current_break.start_date} to "
                    f"{current_break.end_date} overlaps with {next_break.start_date} to "
                    f"{next_break.end_date}"
                )


class Visualizer:
    """Renders commit schedule as GitHub-style contribution graph."""

    # Colour scheme (GitHub-style green)
    COLOUR_SCHEME = [
        (239, 242, 245),
        (172, 238, 187),
        (74, 194, 107),
        (45, 164, 78),
        (17, 99, 41),
    ]
    BLOCK = "█"

    def __init__(self, schedule: list[Commit], start_date: date, end_date: date):
        self.schedule = schedule
        self.start_date = start_date
        self.end_date = end_date
        self.commit_map = self._build_commit_map()
        self.colour_cutoffs = self._calculate_colour_cutoffs()

    def _build_commit_map(self) -> dict[date, int]:
        """Create mapping of date to commit count."""
        commit_map = {}

        # Initialize all dates with zero (including days with no commits)
        current = self.start_date
        while current <= self.end_date:
            commit_map[current] = 0
            current += timedelta(days=1)

        # Count commits per day
        for commit in self.schedule:
            commit_map[commit.date] += 1

        return commit_map

    def render(self) -> str:
        """Render the contribution graph adaptively."""
        try:
            terminal_width = os.get_terminal_size().columns
        except OSError:
            terminal_width = 80

        total_days = (self.end_date - self.start_date).days + 1
        total_weeks = (total_days + 6) // 7
        horizontal_width = 8 + total_weeks * 2

        lines = []

        if horizontal_width <= terminal_width:
            lines += self._render_horizontal()
        elif terminal_width >= 70:
            lines += self._render_quarterly()
        else:
            lines += self._render_vertical()

        lines += self._render_statistics()

        return "\n".join(lines)

    def _render_horizontal(self) -> list[str]:
        """Render full horizontal GitHub-style graph."""
        lines = [self._bold("Commit Graph Preview\n")]

        weeks = self._build_weeks(self.start_date, self.end_date)

        # Month labels
        lines.append(self._muted(self._build_month_labels(weeks)))

        # Day rows
        day_labels = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
        for day_idx in range(7):
            line = self._muted(day_labels[day_idx]) + " "
            for week in weeks:
                count = week[day_idx]
                line += "  " if count is None else self._coloured_block(count) * 2
            lines.append(line)

        lines.append("")
        lines.append(self._render_legend())

        return lines

    def _render_quarterly(self) -> list[str]:
        """Render in quarterly sections."""
        lines = [
            self._bold("Commit Graph Preview") + " " + self._muted("(quarterly view)\n")
        ]

        current_date = self.start_date

        while current_date <= self.end_date:
            year = current_date.year
            quarter = (current_date.month - 1) // 3 + 1
            quarter_end = self._get_quarter_end(year, quarter)
            quarter_end = min(quarter_end, self.end_date)

            lines.append(
                self._bold(f"Q{quarter} {year}")
                + self._muted(
                    f" {current_date.strftime('%b %d')} - {quarter_end.strftime('%b %d, %Y')}"
                )
            )

            weeks = self._build_weeks(current_date, quarter_end)
            day_labels = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]

            for day_idx in range(7):
                line = self._muted(day_labels[day_idx]) + " "
                for week in weeks:
                    count = week[day_idx]
                    line += "  " if count is None else self._coloured_block(count) * 2
                lines.append(line)

            lines.append("")
            current_date = quarter_end + timedelta(days=1)

        return lines

    def _render_vertical(self) -> list[str]:
        """Render in vertical/transposed layout."""
        lines = [
            self._bold("Commit Graph Preview") + " " + self._muted("(vertical view)\n")
        ]

        lines.append(self._bold("Date") + "   " + self._muted("M T W T F S S"))

        weeks = self._build_weeks(self.start_date, self.end_date)
        current_date = self.start_date

        # Find first Monday
        days_to_monday = (7 - current_date.weekday()) % 7  # 0 is Monday
        current_date -= timedelta(days=days_to_monday)

        for week in weeks:
            date_str = self._muted(current_date.strftime("%b %d"))
            line = f"{date_str} "

            for day_idx in range(7):
                count = week[day_idx]
                line += "  " if count is None else self._coloured_block(count) * 2

            lines.append(line)
            current_date += timedelta(days=7)

        lines.append("")

        return lines

    def _render_statistics(self) -> list[str]:
        """Render commit statistics."""
        active_days = sum(1 for count in self.commit_map.values() if count > 0)
        avg_commits = len(self.schedule) / max(active_days, 1)

        return [
            self._bold("Statistics"),
            f"  Total commits: {len(self.schedule)}",
            f"  Days with commits: {active_days}",
            f"  Average per active day: {avg_commits:.1f}",
            f"  Max commits in a day: {max(self.commit_map.values())}",
            "",
        ]

    def _render_legend(self) -> str:
        """Render colour gradient legend."""
        legend = "Less "

        for colour in self.COLOUR_SCHEME:
            legend += (
                f"\033[38;2;{colour[0]};{colour[1]};{colour[2]}m{self.BLOCK}\033[0m"
            ) * 2

        legend += " More\n"
        return legend

    def _build_weeks(self, start: date, end: date) -> list[list[Optional[int]]]:
        """Build week data structure for date range."""
        current = start
        while current.weekday() != 0:
            current -= timedelta(days=1)

        weeks = []
        week: list[Optional[int]] = [None] * current.weekday()

        while current <= end:
            if current >= start:
                week.append(self.commit_map.get(current, 0))
            else:
                week.append(None)

            if len(week) == 7:
                weeks.append(week)
                week = []

            current += timedelta(days=1)

        if week:
            week.extend([None] * (7 - len(week)))
            weeks.append(week)

        return weeks

    def _build_month_labels(self, weeks: list[list[Optional[int]]]) -> str:
        """Build month label row for horizontal view."""
        # Main idea: Build up a list of months associated with the week they
        # start in
        months = [(self.start_date.strftime("%b"), 0)]
        current_week = 0

        for d in sorted(self.commit_map.keys()):
            month_abbr = d.strftime("%b")
            if month_abbr != months[-1][0]:
                # Encountered a new month
                if current_week - months[-1][1] < 2:
                    # Previous month would be cut off, so blank it
                    months[-1] = ("", months[-1][1])
                months.append((month_abbr, current_week))
            if d.weekday() == 6:
                # New week is beginning, so update column
                current_week += 1

        labels = " " * 4
        for i in range(len(months) - 1):
            labels += months[i][0]
            # Add spacing until next month's first week
            week_delta = months[i + 1][1] - months[i][1]
            spacing = week_delta * 2 - len(months[i][0])
            labels += " " * spacing
        labels += months[-1][0]

        return labels

    def _get_quarter_end(self, year: int, quarter: int) -> date:
        """Get last day of quarter."""
        end_month = quarter * 3
        if end_month == 12:
            return date(year, 12, 31)
        return date(year, end_month + 1, 1) - timedelta(days=1)

    def _calculate_colour_cutoffs(self) -> list[float]:
        """Calculate colour cutoffs based on quartiles after removing outliers."""
        # Get all non-zero commit counts
        counts = [c for c in self.commit_map.values() if c > 0]

        if len(counts) == 0:
            # No commits, all will be gray
            return [0, 0, 0]

        max_val = max(counts)
        fallback_cutoffs = [max_val * 0.25, max_val * 0.5, max_val * 0.75]

        if len(counts) < 4:
            # Too few data points for meaningful quartiles, use simple division
            return fallback_cutoffs

        # Remove outliers using IQR method
        filtered_counts = self._remove_outliers(counts)

        if len(filtered_counts) < 4:
            # After outlier removal, too few points remain
            return fallback_cutoffs

        # Calculate quartiles on filtered data
        q1, q2, q3 = self._calculate_quartiles(filtered_counts)

        return [q1, q2, q3]

    def _remove_outliers(self, values: list[int]) -> list[int]:
        """Remove outliers using IQR (Interquartile Range) method.

        Outliers are values outside the range [Q1 - 1.5*IQR, Q3 + 1.5*IQR].
        """
        if len(values) < 4:
            return values.copy()

        sorted_values = sorted(values)
        n = len(sorted_values)

        # Calculate Q1 and Q3
        q1_idx = n // 4
        q3_idx = (3 * n) // 4

        q1 = sorted_values[q1_idx]
        q3 = sorted_values[q3_idx]

        iqr = q3 - q1

        # Define outlier bounds
        lower_bound = q1 - 1.5 * iqr
        upper_bound = q3 + 1.5 * iqr

        # Filter out outliers
        filtered = [v for v in values if lower_bound <= v <= upper_bound]

        return filtered if filtered else values.copy()

    def _calculate_quartiles(self, values: list[int]) -> tuple[float, float, float]:
        """Calculate Q1, Q2 (median), and Q3 from sorted values.

        Uses the next distinct value after each quartile index to ensure
        proper distribution when there are duplicate values.
        """
        if not values:
            return (0.0, 0.0, 0.0)

        sorted_values = sorted(values)
        n = len(sorted_values)

        # If all values are the same, use the value itself as all cutoffs
        if sorted_values[0] == sorted_values[-1]:
            return (
                float(sorted_values[0]),
                float(sorted_values[0]),
                float(sorted_values[0]),
            )

        # Find the next distinct value after each quartile index
        def get_cutoff(idx: int) -> float:
            """Get the smallest value greater than the value at idx."""
            val = sorted_values[idx]
            for i in range(idx + 1, n):
                if sorted_values[i] > val:
                    return float(sorted_values[i])
            # No greater value found
            return float(val)

        # Calculate quartile indices
        q1_idx = n // 4
        q2_idx = n // 2
        q3_idx = (3 * n) // 4

        q1 = get_cutoff(q1_idx)
        q2 = get_cutoff(q2_idx)
        q3 = get_cutoff(q3_idx)

        return (q1, q2, q3)

    def _coloured_block(self, count: int) -> str:
        """Get coloured block for commit count.

        Colour mapping:
        - 0 commits: gray (colour 0)
        - 0 < count < Q1: lightest green (colour 1)
        - Q1 <= count < Q2: medium-light green (colour 2)
        - Q2 <= count < Q3: medium-dark green (colour 3)
        - count >= Q3: darkest green (colour 4)
        """
        if count == 0:
            colour_idx = 0
        elif count < self.colour_cutoffs[0]:
            colour_idx = 1
        elif count < self.colour_cutoffs[1]:
            colour_idx = 2
        elif count < self.colour_cutoffs[2]:
            colour_idx = 3
        else:
            colour_idx = 4

        r, g, b = self.COLOUR_SCHEME[colour_idx]

        return f"\033[38;2;{r};{g};{b}m{self.BLOCK}\033[0m"

    def _bold(self, text: str) -> str:
        return f"\033[1m{text}\033[0m"

    def _muted(self, text: str) -> str:
        return f"\033[90m{text}\033[0m"


def main() -> None:
    config = Config()

    parser = argparse.ArgumentParser(
        description="Commitose: Markov chain-based commit history simulator"
    )

    # Date arguments
    parser.add_argument(
        "--start-date",
        help="simulation start date (format: YYYY-MM-DD, default: 365 days ago)",
    )
    parser.add_argument(
        "--end-date",
        help="simulation end date (format: YYYY-MM-DD, default: today)",
    )

    # Git arguments
    parser.add_argument(
        "--repo-path",
        type=Path,
        help=f"path to Git repository (default: {config.repo_path})",
    )
    parser.add_argument("--user-name", help="Git author name")
    parser.add_argument("--user-email", help="Git author email")
    parser.add_argument("--branch", help=f"Git branch name (default: {config.branch})")

    # Generation arguments
    parser.add_argument("--seed", type=int, help="random seed for reproducibility")
    parser.add_argument(
        "--break",
        nargs="*",
        action="extend",
        dest="breaks",
        help="scale average commits over a period by a factor (START-DATE:END-DATE[:FACTOR], factor default: 0)",
    )

    # Configuration file
    parser.add_argument("--config", type=Path, help="load configuration from TOML")

    # Runtime options
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Generate schedule without making commits",
        default=None,
    )

    args = parser.parse_args()

    try:
        if args.config:
            config.load_toml(args.config)

        # CLI arguments overwrite config file
        config.load_args(vars(args))
    except (ValueError, TypeError) as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

    if not config.dry_run and (not config.user_name or not config.user_email):
        # Try to get values from git config
        if config.user_name is None:
            git_name = GitClient.get_git_config("user.name")
            if git_name:
                response = (
                    input(f"Found Git username '{git_name}'. Use this? [Y/n]: ")
                    .lower()
                    .strip()
                )
                if response == "" or response.startswith("y"):
                    config.user_name = git_name

        if config.user_email is None:
            git_email = GitClient.get_git_config("user.email")
            if git_email:
                response = (
                    input(f"Found Git email '{git_email}'. Use this? [Y/n]: ")
                    .lower()
                    .strip()
                )
                if response == "" or response.startswith("y"):
                    config.user_email = git_email

        if not config.user_name or not config.user_email:
            print(
                "Error: Git username and email must be configured for non-dry-run mode",
                file=sys.stderr,
            )
            sys.exit(1)

    simulator = Simulator(
        start_date=config.start_date,
        end_date=config.end_date,
        weekday_weights=config.weekday_weights,
        breaks=config.breaks,
        commit_means=config.commit_means,
        commit_dispersion=config.commit_dispersion,
        commit_zero_inflation=config.commit_zero_inflation,
        transition_matrix=config.transition_matrix,
        seed=config.seed,
    )

    commits = simulator.simulate()

    visualizer = Visualizer(commits, config.start_date, config.end_date)

    print(visualizer.render())

    while not config.dry_run:
        # Ask for user confirmation
        if config.seed is not None:
            prompt = "Do you want to proceed with these commits? [Y/n]: "
        else:
            prompt = "Do you want to proceed with these commits? [Y/n/r(egenerate)]: "

        response = input(prompt).lower().strip()

        if response == "" or response.startswith("y"):
            # Proceed with commits (default)
            break
        elif response.startswith("n"):
            print("Aborted.", file=sys.stderr)
            sys.exit(0)
        elif response.startswith("r"):
            if config.seed is not None:
                print(
                    "Cannot regenerate new results with a fixed seed. Please remove the seed option or choose [Y/n].",
                    file=sys.stderr,
                )
                continue
            # Regenerate commits by running simulator again
            commits = simulator.simulate()
            visualizer = Visualizer(commits, config.start_date, config.end_date)
            print(visualizer.render())
            continue
        else:
            print("Invalid response. Please choose Y (yes), N (no)", end="")
            if config.seed is None:
                print(", or R (regenerate)", end="")
            print(".")

    if not config.dry_run and config.user_name and config.user_email:
        git_client = GitClient(
            user_name=config.user_name,
            user_email=config.user_email,
            repo_path=config.repo_path,
            branch=config.branch,
        )

        try:
            git_client.check_repo_path(config.repo_path)
        except GitClient.WouldOverwriteRepoError:
            if (
                not input(
                    f"Repository at '{config.repo_path}' already exists. Wipe and regenerate? [y/N]: "
                )
                .lower()
                .startswith("y")
            ):
                print("Aborted.", file=sys.stderr)
                sys.exit(1)
            git_client.wipe_repo()

        git_client.exec_git(["checkout", "-b", config.branch])
        git_client.commit_all(commits)


if __name__ == "__main__":
    main()