#!/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>.+)$"
)

HELP_EPILOG = """\
三种模式（互斥；未给出 --subject / --message-file 时进入 CI 输入校验）:

  --subject "<候选>"     校验单个候选主题；交互式起草提交信息时用这一种
  --message-file <path>  校验提交信息文件的首行；body/footer 不做机械校验
  (无参数)               CI 输入校验：从事件 payload 取回全部待校验主题

策略来源固定优先级：--policy > $COMMIT_POLICY_PATH > 本脚本旁的
references/commit_policy.json。策略缺失、损坏或 schema_version 不支持时报部署
错误，不退回任何内置约定。本脚本是提交信息规则的唯一实现；说明性文档不是机器
策略来源。

CI 输入校验按事件展开：

  pull_request*  pull_request.title 加 base.sha..head.sha 的完整提交范围
  push           before..after 完整范围；before 为零对象时按分支图计算新增提交
  workflow_run   上游为 pull_request* 时，取 <集成分支>..workflow_run.head_sha
                 的完整范围；其它上游事件只校验 workflow_run.head_sha 一条
  其它 / 本地     回退 HEAD 一条

集成分支名取 $COMMIT_LINT_MAIN_BRANCH，默认 main；依次尝试
refs/remotes/origin/<名字> 与 refs/heads/<名字>。无法解析、或上游 head_branch
就是集成分支时，退化为只校验上游 head 一条并打印 WARN，不猜测 base。

workflow_run payload 不含 PR 标题，该架构下标题不参与校验；完整提交范围校验取代
它，且强度更高。

已识别事件缺失或损坏 payload 时失败；shallow repository 一律失败。

退出码：0 全部通过；1 存在不合规主题；2 策略、参数或输入错误。
"""


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",
        epilog=HELP_EPILOG,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    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 == "workflow_run"
        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 _other_branch_refs(branch_name: str, context: str) -> list[str]:
    """Non-symbolic branch refs that are not `branch_name` on any remote."""
    try:
        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 {context}: {exc}") from exc
    if refs_result.returncode != 0:
        detail = refs_result.stderr.strip()
        suffix = f": {detail}" if detail else ""
        raise InputError(f"cannot inspect refs for {context}{suffix}")

    own_ref = f"refs/heads/{branch_name}"
    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(f"invalid git ref output for {context}")
        if symbolic_target:
            continue
        if ref_name == own_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)
    return exclusions


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,
        )
    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}")

    branch_name = target_ref.removeprefix("refs/heads/")
    exclusions = _other_branch_refs(branch_name, "zero-before push")

    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 _integration_branch() -> tuple[str, str | None]:
    """The configured integration branch name and its first resolvable ref."""
    name = (os.getenv("COMMIT_LINT_MAIN_BRANCH") or "").strip() or "main"
    for ref in (f"refs/remotes/origin/{name}", f"refs/heads/{name}"):
        try:
            result = subprocess.run(
                ["git", "rev-parse", "--verify", "--quiet", ref],
                capture_output=True,
                text=True,
                check=False,
            )
        except OSError as exc:
            raise InputError(f"cannot resolve {ref}: {exc}") from exc
        if result.returncode == 0 and result.stdout.strip():
            return name, ref
    return name, None


def _workflow_run_subjects(payload: dict[str, object]) -> list[tuple[str, str]]:
    """Subjects for a run triggered by another run's completion.

    A workflow_run payload carries no PR title and no before/after range, so the
    upstream head SHA is the only anchor. For an upstream pull_request the range
    is taken against the integration branch; every other case validates the
    upstream head commit alone rather than guessing a base.
    """
    run = payload.get("workflow_run")
    if not isinstance(run, dict):
        raise InputError("workflow_run event payload.workflow_run must be an object")
    head_sha = run.get("head_sha")
    if not isinstance(head_sha, str) or GIT_OBJECT_ID_RE.fullmatch(head_sha) is None:
        raise InputError("workflow_run.head_sha must be a full Git object ID")
    upstream_event = run.get("event")
    if not isinstance(upstream_event, str) or not upstream_event:
        raise InputError("workflow_run.event must be a non-empty string")
    _require_complete_git_history("workflow_run event")

    label_prefix = f"workflow_run({upstream_event}).commit"
    head_only = f"validating only upstream head commit {head_sha[:7]}"
    if upstream_event.startswith("pull_request"):
        head_branch = run.get("head_branch")
        if not isinstance(head_branch, str) or not head_branch:
            raise InputError("workflow_run.head_branch must be a non-empty string")
        main_name, main_ref = _integration_branch()
        if main_ref is None:
            _eprint(f"WARN: no {main_name} ref to compare against; {head_only}")
        elif head_branch == main_name:
            _eprint(f"WARN: upstream head_branch is {main_name}; {head_only}")
        else:
            revision_range = f"{main_ref}..{head_sha}"
            subjects = _git_log_subjects(
                [head_sha, "--not", main_ref],
                f"workflow_run pull-request range {revision_range}",
                allow_empty=True,
                label_prefix=label_prefix,
            )
            if subjects:
                return subjects
            _eprint(f"WARN: {revision_range} is empty; {head_only}")

    return _git_log_subjects(
        ["-1", head_sha],
        f"workflow_run head commit {head_sha[:7]}",
        label_prefix=label_prefix,
    )


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)
        # The loop above only asserts payload shape; the Git range is authority.
        return event_name, range_subjects

    if event_name == "workflow_run":
        if not isinstance(payload, dict):
            raise InputError("workflow_run event payload is required")
        return event_name, _workflow_run_subjects(payload)

    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())