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

import argparse
import json
import os
import re
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Pattern

TYPE_RE = re.compile(r"^[a-z][a-z0-9-]*$")
EMOJI_RE = re.compile(r"^:[a-z0-9_+-]+:$")
GIT_OBJECT_ID_RE = re.compile(r"^[0-9a-fA-F]{40}(?:[0-9a-fA-F]{24})?$")
HEADER_RE = re.compile(
    r"^(?:(?P<emoji>:[a-z0-9_+-]+:)\s+)?"
    r"(?P<type>[a-z][a-z0-9-]*)"
    r"(?:\((?P<scope>[^()]*)\))?"
    r":\s+(?P<text>.+)$"
)


class PolicyError(ValueError):
    pass


class InputError(ValueError):
    pass


@dataclass(frozen=True)
class CommitPolicy:
    source: Path
    types: dict[str, str]
    scope_optional: bool
    scope_pattern: Pattern[str]
    subject_max_length: int
    subject_forbidden_suffixes: tuple[str, ...]
    emoji_required_by_default: bool
    emoji_requirement_env: str
    emoji_false_values: frozenset[str]


def _eprint(*args: object) -> None:
    print(*args, file=sys.stderr)


def resolve_policy_path(cli_path: str | None) -> Path:
    if cli_path is not None:
        return Path(cli_path).expanduser()
    environment_path = os.getenv("COMMIT_POLICY_PATH")
    if environment_path:
        return Path(environment_path).expanduser()
    return Path(__file__).resolve().parent.parent / "references" / "commit_policy.json"


def _load_json(path: Path) -> object:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
        raise PolicyError(f"cannot read policy {path}: {exc}") from exc


def load_policy(path: Path) -> CommitPolicy:
    source = path.resolve()
    raw = _load_json(source)
    if not isinstance(raw, dict):
        raise PolicyError(f"policy root must be an object: {source}")

    schema_version = raw.get("schema_version")
    if type(schema_version) is not int:
        raise PolicyError(f"policy schema_version must be the integer 1: {source}")
    if schema_version != 1:
        raise PolicyError(
            f"unsupported policy schema_version {schema_version}: {source}"
        )

    types = raw.get("types")
    if not isinstance(types, dict) or not types:
        raise PolicyError("policy.types must be a non-empty object")
    if not all(
        isinstance(name, str)
        and TYPE_RE.fullmatch(name)
        and isinstance(emoji, str)
        and EMOJI_RE.fullmatch(emoji)
        for name, emoji in types.items()
    ):
        raise PolicyError(
            "policy.types must map lowercase type names to colon-delimited emoji codes"
        )

    scope = raw.get("scope")
    if not isinstance(scope, dict):
        raise PolicyError("policy.scope must be an object")
    scope_optional = scope.get("optional")
    scope_expression = scope.get("pattern")
    if not isinstance(scope_optional, bool):
        raise PolicyError("policy.scope.optional must be boolean")
    if not isinstance(scope_expression, str) or not scope_expression:
        raise PolicyError("policy.scope.pattern must be a non-empty string")
    try:
        scope_pattern = re.compile(scope_expression)
    except re.error as exc:
        raise PolicyError(f"invalid policy.scope.pattern: {exc}") from exc

    subject = raw.get("subject")
    if not isinstance(subject, dict):
        raise PolicyError("policy.subject must be an object")
    max_length = subject.get("max_length")
    forbidden_suffixes = subject.get("forbidden_suffixes")
    if type(max_length) is not int or max_length <= 0:
        raise PolicyError("policy.subject.max_length must be a positive integer")
    if not isinstance(forbidden_suffixes, list) or not all(
        isinstance(suffix, str) and suffix for suffix in forbidden_suffixes
    ):
        raise PolicyError(
            "policy.subject.forbidden_suffixes must contain non-empty strings"
        )

    emoji = raw.get("emoji")
    if not isinstance(emoji, dict):
        raise PolicyError("policy.emoji must be an object")
    required_by_default = emoji.get("required_by_default")
    requirement_env = emoji.get("requirement_env")
    false_values = emoji.get("false_values")
    if not isinstance(required_by_default, bool):
        raise PolicyError("policy.emoji.required_by_default must be boolean")
    if not isinstance(requirement_env, str) or not requirement_env.strip():
        raise PolicyError("policy.emoji.requirement_env must be a non-empty string")
    if not isinstance(false_values, list) or not false_values or not all(
        isinstance(value, str) and value.strip() for value in false_values
    ):
        raise PolicyError("policy.emoji.false_values must contain non-empty strings")

    return CommitPolicy(
        source=source,
        types=dict(types),
        scope_optional=scope_optional,
        scope_pattern=scope_pattern,
        subject_max_length=max_length,
        subject_forbidden_suffixes=tuple(forbidden_suffixes),
        emoji_required_by_default=required_by_default,
        emoji_requirement_env=requirement_env.strip(),
        emoji_false_values=frozenset(
            value.strip().casefold() for value in false_values
        ),
    )


def emoji_is_required(policy: CommitPolicy) -> bool:
    override = os.getenv(policy.emoji_requirement_env)
    if override is None:
        return policy.emoji_required_by_default
    return override.strip().casefold() not in policy.emoji_false_values


def validate_subject(
    line: str,
    policy: CommitPolicy,
    *,
    require_emoji: bool,
) -> str | None:
    if "\n" in line or "\r" in line:
        return "subject must be a single line"

    if not line.strip():
        return "empty subject"
    if line != line.strip():
        return "subject must not have leading or trailing whitespace"
    subject = line

    match = HEADER_RE.fullmatch(subject)
    if not match:
        return "does not match ':emoji: type(scope): subject' or 'type(scope): subject'"

    type_name = match.group("type")
    scope = match.group("scope")
    text = match.group("text").rstrip()
    if type_name not in policy.types:
        return f"unknown type: {type_name}"
    if scope is None and not policy.scope_optional:
        return "missing required scope"
    if scope is not None and policy.scope_pattern.fullmatch(scope) is None:
        return f"invalid scope: {scope}"

    supplied_emoji = match.group("emoji")
    if supplied_emoji is None and require_emoji:
        false_values = ", ".join(
            repr(value) for value in sorted(policy.emoji_false_values)
        )
        return (
            "missing emoji "
            f"(set {policy.emoji_requirement_env} to one of {false_values} to allow)"
        )
    expected_emoji = policy.types[type_name]
    if supplied_emoji is not None and supplied_emoji != expected_emoji:
        return (
            "emoji/type mismatch: "
            f"got {supplied_emoji} {type_name}, expected {expected_emoji}"
        )

    if not text:
        return "empty subject"
    if len(text) > policy.subject_max_length:
        return f"subject exceeds {policy.subject_max_length} characters"
    forbidden_suffix = next(
        (
            suffix
            for suffix in policy.subject_forbidden_suffixes
            if text.endswith(suffix)
        ),
        None,
    )
    if forbidden_suffix is not None:
        if forbidden_suffix in {".", "。"}:
            return (
                "subject should not end with a period "
                f"(forbidden suffix {forbidden_suffix!r})"
            )
        return f"subject must not end with forbidden suffix {forbidden_suffix!r}"
    return None


def _read_message_file(path: Path) -> str:
    try:
        content = path.read_text(encoding="utf-8")
    except (OSError, UnicodeError) as exc:
        raise InputError(f"cannot read message file {path}: {exc}") from exc
    lines = content.splitlines()
    return lines[0] if lines else ""


def _parse_args(argv: list[str] | None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Validate commit-message subjects")
    parser.add_argument("--policy", help="path to a versioned commit policy JSON")
    candidates = parser.add_mutually_exclusive_group()
    candidates.add_argument("--subject", help="single commit subject to validate")
    candidates.add_argument(
        "--message-file",
        type=Path,
        help="UTF-8 commit message file whose first line is validated",
    )
    return parser.parse_args(argv)


def _requires_event_payload(event_name: str) -> bool:
    return event_name == "push" or event_name.startswith("pull_request")


def _load_event_payload() -> tuple[str, dict[str, object] | None]:
    event_name = os.getenv("GITHUB_EVENT_NAME") or os.getenv("GITEA_EVENT_NAME") or ""
    event_path = os.getenv("GITHUB_EVENT_PATH") or os.getenv("GITEA_EVENT_PATH") or ""
    if not event_path:
        if _requires_event_payload(event_name):
            raise InputError(f"{event_name} event requires an event payload path")
        return event_name, None

    path = Path(event_path)
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
        if _requires_event_payload(event_name):
            raise InputError(f"cannot parse {event_name} event payload {path}: {exc}") from exc
        _eprint(f"WARN: failed to parse event payload: {path} ({exc})")
        return event_name, None
    if not isinstance(payload, dict):
        if _requires_event_payload(event_name):
            raise InputError(f"{event_name} event payload must be an object: {path}")
        _eprint(f"WARN: event payload is not an object: {path}")
        return event_name, None
    return event_name, payload


def _message_subject(message: str, label: str) -> str:
    if not message:
        raise InputError(f"{label}.message must be a non-empty string")
    lines = message.splitlines()
    if not lines:
        raise InputError(f"{label}.message must contain a subject line")
    return lines[0]


def _git_log_subjects(
    revisions: list[str],
    description: str,
    *,
    allow_empty: bool = False,
    label_prefix: str = "push.commit",
) -> list[tuple[str, str]]:
    try:
        result = subprocess.run(
            ["git", "log", "--reverse", "--format=%H%x00%s", *revisions],
            capture_output=True,
            text=True,
            check=False,
        )
    except OSError as exc:
        raise InputError(f"cannot read {description}: {exc}") from exc
    if result.returncode != 0:
        detail = result.stderr.strip()
        suffix = f": {detail}" if detail else ""
        raise InputError(f"cannot read {description}{suffix}")

    subjects: list[tuple[str, str]] = []
    for line in result.stdout.splitlines():
        sha, separator, subject = line.partition("\0")
        if not separator or not sha or not subject:
            raise InputError(f"invalid git log output for {description}")
        subjects.append((f"{label_prefix} {sha[:7]}", subject))
    if not subjects and not allow_empty:
        raise InputError(f"{description} is empty")
    return subjects


def _require_complete_git_history(context: str) -> None:
    try:
        result = subprocess.run(
            ["git", "rev-parse", "--is-shallow-repository"],
            capture_output=True,
            text=True,
            check=False,
        )
    except OSError as exc:
        raise InputError(f"cannot inspect repository depth for {context}: {exc}") from exc
    if result.returncode != 0:
        detail = result.stderr.strip()
        suffix = f": {detail}" if detail else ""
        raise InputError(f"cannot inspect repository depth for {context}{suffix}")
    shallow_state = result.stdout.strip()
    if shallow_state == "true":
        raise InputError(f"{context} requires complete, non-shallow Git history")
    if shallow_state != "false":
        raise InputError(f"cannot determine repository depth for {context}")


def _zero_before_subjects(
    payload: dict[str, object], after: str
) -> list[tuple[str, str]]:
    target_ref = payload.get("ref")
    if not isinstance(target_ref, str) or not target_ref.startswith("refs/heads/"):
        raise InputError("zero-before push event requires a refs/heads/* ref")
    try:
        check_ref = subprocess.run(
            ["git", "check-ref-format", target_ref],
            capture_output=True,
            text=True,
            check=False,
        )
        refs_result = subprocess.run(
            [
                "git",
                "for-each-ref",
                "--format=%(refname)%00%(symref)",
                "refs/heads",
                "refs/remotes",
            ],
            capture_output=True,
            text=True,
            check=False,
        )
    except OSError as exc:
        raise InputError(f"cannot inspect refs for zero-before push: {exc}") from exc
    if check_ref.returncode != 0:
        raise InputError(f"invalid zero-before push ref: {target_ref}")
    if refs_result.returncode != 0:
        detail = refs_result.stderr.strip()
        suffix = f": {detail}" if detail else ""
        raise InputError(f"cannot inspect refs for zero-before push{suffix}")

    branch_name = target_ref.removeprefix("refs/heads/")
    exclusions: list[str] = []
    for line in refs_result.stdout.splitlines():
        ref_name, separator, symbolic_target = line.partition("\0")
        if not separator or not ref_name:
            raise InputError("invalid git ref output for zero-before push")
        if symbolic_target:
            continue
        if ref_name == target_ref:
            continue
        if ref_name.startswith("refs/remotes/"):
            remote_parts = ref_name.split("/", 3)
            if len(remote_parts) == 4 and remote_parts[3] == branch_name:
                continue
        exclusions.append(ref_name)

    revisions = [after]
    if exclusions:
        revisions.extend(["--not", *exclusions])
    return _git_log_subjects(
        revisions,
        f"zero-before push history for {target_ref}",
        allow_empty=True,
    )


def _push_range_subjects(payload: dict[str, object]) -> list[tuple[str, str]]:
    before = payload.get("before")
    after = payload.get("after")
    if (
        not isinstance(before, str)
        or not isinstance(after, str)
        or GIT_OBJECT_ID_RE.fullmatch(before) is None
        or GIT_OBJECT_ID_RE.fullmatch(after) is None
    ):
        raise InputError("push event before/after must be full Git object IDs")
    if set(after) == {"0"}:
        raise InputError("push event after cannot be the zero object ID")
    _require_complete_git_history("push event")
    if set(before) == {"0"}:
        return _zero_before_subjects(payload, after)

    revision_range = f"{before}..{after}"
    return _git_log_subjects(
        [revision_range], f"push commit range {revision_range}"
    )


def _gather_ci_subjects() -> tuple[str, list[tuple[str, str]]]:
    event_name, payload = _load_event_payload()
    subjects: list[tuple[str, str]] = []

    if event_name.startswith("pull_request"):
        if not isinstance(payload, dict):
            raise InputError(f"{event_name} event payload is required")
        pull_request = payload.get("pull_request")
        if not isinstance(pull_request, dict):
            raise InputError("pull_request event payload.pull_request must be an object")
        title = pull_request.get("title")
        if not isinstance(title, str) or not title:
            raise InputError("pull_request.title must be a non-empty string")
        subjects.append(("pull_request.title", _message_subject(title, "pull_request.title")))
        base = pull_request.get("base")
        head = pull_request.get("head")
        if not isinstance(base, dict) or not isinstance(head, dict):
            raise InputError("pull_request.base/head must be objects")
        base_sha = base.get("sha")
        head_sha = head.get("sha")
        if (
            not isinstance(base_sha, str)
            or not isinstance(head_sha, str)
            or GIT_OBJECT_ID_RE.fullmatch(base_sha) is None
            or GIT_OBJECT_ID_RE.fullmatch(head_sha) is None
        ):
            raise InputError("pull_request.base/head.sha must be full Git object IDs")
        _require_complete_git_history("pull request event")
        revision_range = f"{base_sha}..{head_sha}"
        subjects.extend(
            _git_log_subjects(
                [revision_range],
                f"pull request commit range {revision_range}",
                allow_empty=True,
                label_prefix="pull_request.commit",
            )
        )
        return event_name, subjects

    if event_name == "push":
        if not isinstance(payload, dict):
            raise InputError("push event payload is required")
        commits = payload.get("commits")
        if not isinstance(commits, list):
            raise InputError("push event payload.commits must be a list")
        for index, commit in enumerate(commits):
            item_label = f"push.commits[{index}]"
            if not isinstance(commit, dict):
                raise InputError(f"{item_label} must be an object")
            message = commit.get("message")
            if not isinstance(message, str):
                raise InputError(f"{item_label}.message must be a non-empty string")
            subject = _message_subject(message, item_label)
            sha = commit.get("id") or commit.get("sha") or ""
            sha_text = str(sha)
            label = f"push.commit {sha_text[:7]}" if sha_text else "push.commit"
            subjects.append((label, subject))
        range_subjects = _push_range_subjects(payload)
        return event_name, range_subjects

    try:
        result = subprocess.run(
            ["git", "log", "-1", "--format=%s", "HEAD"],
            capture_output=True,
            text=True,
            check=False,
        )
    except OSError as exc:
        raise InputError(f"cannot read HEAD subject: {exc}") from exc
    lines = result.stdout.splitlines()
    if result.returncode == 0 and lines:
        return event_name, [("HEAD", lines[0])]

    detail = result.stderr.strip()
    suffix = f": {detail}" if detail else ""
    raise InputError(f"no commit subject found in event payload or HEAD{suffix}")


def main(argv: list[str] | None = None) -> int:
    args = _parse_args(argv)
    try:
        policy = load_policy(resolve_policy_path(args.policy))
        if args.subject is not None:
            event_name = ""
            subjects = [("--subject", args.subject)]
        elif args.message_file is not None:
            event_name = ""
            subjects = [(str(args.message_file), _read_message_file(args.message_file))]
        else:
            event_name, subjects = _gather_ci_subjects()
    except (InputError, PolicyError) as exc:
        _eprint(f"ERROR: {exc}")
        return 2

    require_emoji = emoji_is_required(policy)
    print(f"commit policy: {policy.source}")
    if event_name:
        print(f"event: {event_name}")
    print(f"require emoji: {require_emoji}")
    print(f"checks: {len(subjects)} subject(s)")
    for label, _subject in subjects:
        print(f"input: {label}")

    errors: list[str] = []
    for label, subject in subjects:
        error = validate_subject(subject, policy, require_emoji=require_emoji)
        if error:
            errors.append(f"- {label}: {error}\n  subject: {subject}")

    if errors:
        _eprint("ERROR: commit message lint failed:")
        for error in errors:
            _eprint(error)
        return 1

    print("OK")
    return 0


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