"""cross-project errors-*.jsonl 수집·집계·익명화·zip 생성 (read-only).

글로벌 run-index(recent/active)를 순회해 모든 타겟의 에러 로그를 모은다.
errorType 값에 분기하지 않는 data-driven 처리.
"""
from __future__ import annotations

import argparse
import datetime as dt
import json
import re
import sys
import zipfile
from pathlib import Path

from .fixed_text import line

from okstra_ctl.error_log_core import aggregate, parse_records
from okstra_ctl.jsonl import read_jsonl
from okstra_ctl.locks import central_lock
from okstra_ctl.paths import okstra_home, resolve_under_root
from okstra_ctl.json_boundary import load_owned_object, write_owned_object_atomic


def run_dirs(home: Path) -> list[tuple[str, Path, dict]]:
    """글로벌 run-index 순회의 단일 참조점 — `(project_root, run_dir, row)`.

    인덱스 행을 함께 돌려주는 이유: 행에는 `status` 처럼 디스크만 봐서는 알 수
    없는 사실이 실려 있다(`run_index_row.build_run_index_row`). 행을 버리면 그게
    필요한 소비자가 recent/active 순회를 통째로 복제하게 된다.
    """
    rows = read_jsonl(home / "recent.jsonl") + read_jsonl(home / "active.jsonl")
    seen: set[tuple[str, str]] = set()
    out: list[tuple[str, Path, dict]] = []
    for row in rows:
        project_root = str(row.get("projectRoot", ""))
        run_dir_rel = str(row.get("runDirRel", ""))
        key = (project_root, run_dir_rel)
        if key in seen:
            continue
        seen.add(key)
        run_dir = resolve_under_root(project_root, run_dir_rel)
        if run_dir is not None:
            out.append((project_root, run_dir, row))
    return out


def _logs_in_run_dir(run_dir: Path) -> list[Path]:
    flat = run_dir.glob("logs/errors-*.jsonl")
    # stage-*/logs: 정상 index 에선 비활성, 비정형 트리(과거 stage-isolated 잔재) 방어용.
    staged = run_dir.glob("stage-*/logs/errors-*.jsonl")
    return sorted(set(flat) | set(staged))


def collect_records(home: Path) -> tuple[list[dict], dict]:
    records: list[dict] = []
    project_roots: set[str] = set()
    reachable = 0
    unreachable = 0
    for project_root, run_dir, _row in run_dirs(home):
        if not run_dir.exists():
            unreachable += 1
            continue
        logs = _logs_in_run_dir(run_dir)
        if not logs:
            continue
        reachable += 1
        recs, _ = parse_records(logs)
        for rec in recs:
            rec["_projectRoot"] = project_root
            records.append(rec)
        project_roots.add(project_root)
    stats = {
        "runCount": reachable,
        "unreachableRuns": unreachable,
        "projectRoots": sorted(project_roots),
    }
    return records, stats


KEEP_FIELDS = (
    "ts", "taskKey", "phase", "agent", "agentRole", "source",
    "errorType", "command", "exitCode", "message", "stderrExcerpt",
)


def _token_map(records: list[dict]) -> dict:
    roots = sorted({str(r.get("_projectRoot", "")) for r in records if r.get("_projectRoot")})
    return {root: f"proj-{i}" for i, root in enumerate(roots, start=1)}


_HOME_USER = re.compile(r"/Users/[^/\s]+")
_LINUX_USER = re.compile(r"/home/[^/\s]+")
# zip 은 프로젝트 간 공유되므로 자유 텍스트의 잔여 절대경로·이메일·IP·토큰을 일괄 마스킹한다.
# token_map 치환을 먼저 돌려 알려진 프로젝트 루트는 proj-N 신호로 보존한 뒤 나머지 경로만 붕괴시킨다.
_PATHISH = re.compile(r"[/\\][^\s]+")
_EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+")
_IPV4 = re.compile(r"\b\d{1,3}(?:\.\d{1,3}){3}\b")
# sk-... / 20자 이상 연속 hex·base64 런 → 토큰/시크릿 추정.
_SECRET = re.compile(r"\b(?:sk-[A-Za-z0-9_-]{8,}|[A-Za-z0-9+/=_-]{20,})\b")
# 알려진 자격증명 패턴 — 20자 미만이거나 :@ 인접으로 위 일반 규칙의 \b 런이
# 끊겨 놓치는 짧은 비밀을 명시적으로 잡는다.
_KNOWN_SECRETS = re.compile(
    r"(?:AKIA|ASIA)[0-9A-Z]{16}"            # AWS access key id
    r"|gh[pousr]_[A-Za-z0-9_]{8,}"          # GitHub token (PAT/OAuth/server/refresh)
    r"|github_pat_[A-Za-z0-9_]{8,}"
    r"|xox[abprs]-[A-Za-z0-9-]{8,}"         # Slack token
)
# URL basic-auth(scheme://user:pass@host) 의 비밀번호. _PATHISH 보다 먼저 돌린다.
_URL_CRED = re.compile(r"([\w.+-]+://[^\s:/@]+:)[^\s@/]+(@)")
# 자격증명을 가리키는 키워드(부분일치). 긴 플래그는 철자 열거(denylist) 대신
# 이 키워드를 품은 모든 플래그를 잡아 --api-key/--access-token/--db-password 등
# 새 도구의 플래그도 자동 포함한다. 공유 zip 이라 과마스킹이 누출보다 안전.
_CRED_KEYWORD = (
    r"user(?:name)?|pass(?:wd|word)?|pwd|token|secret"
    r"|api[-_]?key|key|auth|cred(?:ential)?|access[-_]?key"
)
# 자격증명 플래그 뒤 값. 긴 플래그는 키워드 부분일치, 짧은 플래그는 -u/-p/-a
# (mysql/redis). 따옴표로 감싼 값은 공백을 포함할 수 있으므로 닫는 따옴표까지
# 먹는다 — `--password 'my pass'` 의 공백 뒤 잔여(pass')가 새지 않도록. 비-따옴표
# 값은 다른 플래그(-) 로 시작하면 마스킹하지 않는다 — `-a -m msg` 같은
# 비-자격증명 조합(예: git commit -a)을 잘못 먹지 않도록.
_FLAG_CRED = re.compile(
    rf"((?:--[\w-]*(?:{_CRED_KEYWORD})[\w-]*|(?<![\w-])-[upa])[ =])"
    r"(?:'[^']*'|\"[^\"]*\"|(?!-)\S+)",
    re.IGNORECASE,
)
# 단문자 플래그 붙여쓰기형(-uroot -ppassw0rd 같은 MySQL/psql 스타일) — 위 규칙은
# 구분자를 요구해 놓치므로, 토큰 경계의 -u/-p 에 붙은 값도 잡는다.
_FLAG_CRED_ATTACHED = re.compile(r"(?<![\w-])(-[up])\S+")
# HTTP Authorization 스킴 토큰(Authorization: Bearer <jwt> / Basic <base64>).
_AUTH_SCHEME = re.compile(r"\b(Bearer|Basic) +\S+", re.IGNORECASE)

# 자유 텍스트(메시지·stderr·command)는 경로 붕괴까지, 구조 필드(taskKey 등)는 token_map 만.
_FREETEXT_FIELDS = ("message", "stderrExcerpt", "command")


def _scrub(value: object, token_map: dict) -> object:
    if not isinstance(value, str):
        return value
    out = value
    # 긴 루트부터 치환 — 한 루트가 다른 루트의 prefix 일 때(예: /w/app vs
    # /w/app-v2) 짧은 쪽이 먼저 박혀 긴 쪽이 부분치환되는 오귀속을 막는다.
    for root, token in sorted(token_map.items(), key=lambda kv: -len(kv[0])):
        if root:
            out = out.replace(root, token)
    out = _HOME_USER.sub("/Users/<user>", out)
    out = _LINUX_USER.sub("/home/<user>", out)
    return out


def _scrub_freetext(value: object, token_map: dict) -> object:
    if not isinstance(value, str):
        return value
    out = _scrub(value, token_map)
    out = _EMAIL.sub("<email>", out)
    out = _IPV4.sub("<ip>", out)
    out = _KNOWN_SECRETS.sub("<token>", out)
    out = _URL_CRED.sub(r"\1<secret>\2", out)
    out = _AUTH_SCHEME.sub(r"\1 <secret>", out)
    out = _FLAG_CRED.sub(r"\1<secret>", out)
    out = _FLAG_CRED_ATTACHED.sub(r"\1<secret>", out)
    out = _SECRET.sub("<token>", out)
    # 경로 붕괴는 마지막에: 위 마스킹 토큰(<email> 등)은 경로 모양이 아니므로 안전.
    out = _PATHISH.sub("<path>", out)
    return out


def anonymize(records: list[dict]) -> tuple[list[dict], dict]:
    token_map = _token_map(records)
    clean: list[dict] = []
    for rec in records:
        root = str(rec.get("_projectRoot", ""))
        token = token_map.get(root, "proj-?")
        item = {}
        for k in KEEP_FIELDS:
            if k not in rec:
                continue
            scrub = _scrub_freetext if k in _FREETEXT_FIELDS else _scrub
            item[k] = scrub(rec.get(k), token_map)
        if "taskKey" in item:
            # 프로젝트 접두만 토큰화하고 나머지를 남기면 타겟의 티켓 번호가
            # 그대로 나간다 — `proj-2:DEV-9388:DEV-9428` 은 익명이 아니다.
            # 클러스터를 식별하는 것은 이미 지문이므로, 태스크 아이디는 공개
            # 이슈나 공유 zip 을 읽는 사람에게 아무것도 더 주지 않는다.
            item["taskKey"] = token
        item["sourceProject"] = token
        clean.append(item)
    return clean, token_map


_NUM = re.compile(r"\d+")


def cluster_key(rec: dict) -> str:
    msg = str(rec.get("message", ""))
    msg = _PATHISH.sub("<path>", msg)
    msg = _NUM.sub("<n>", msg)
    msg = msg.strip()[:80]
    return "|".join([
        str(rec.get("errorType", "")),
        str(rec.get("phase", "")),
        str(rec.get("agent", "")),
        msg,
    ])


def build_clusters(records: list[dict], *,
                   project_field: str = "sourceProject") -> tuple[list[dict], list[str]]:
    """클러스터 목록과, records 와 같은 순서의 레코드별 cluster_key 를 함께 반환.
    호출자가 직렬화 시 키를 재계산(정규식 중복)하지 않도록 키를 노출한다.

    project_field 는 확산을 세는 소스다. 익명화 레코드는 anonymize() 가 붙인
    `sourceProject`, 원본 레코드는 collect_records() 가 붙인 `_projectRoot` 를 쓴다."""
    buckets: dict[str, dict] = {}
    keys: list[str] = []
    for rec in records:
        key = cluster_key(rec)
        keys.append(key)
        b = buckets.setdefault(key, {
            "key": key, "errorType": str(rec.get("errorType", "")),
            "phase": str(rec.get("phase", "")), "agent": str(rec.get("agent", "")),
            "count": 0, "projects": set(), "sample": str(rec.get("message", "")),
        })
        b["count"] += 1
        b["projects"].add(str(rec.get(project_field, "")))
    clusters = []
    for b in buckets.values():
        b["projects"] = sorted(p for p in b["projects"] if p)
        clusters.append(b)
    clusters.sort(key=lambda c: (-c["count"], c["key"]))
    return clusters, keys


def cluster_id_map(clusters: list[dict]) -> dict:
    """cluster_key → report 표의 # (1-based, build_clusters 정렬 순서)."""
    return {c["key"]: i for i, c in enumerate(clusters, start=1)}


def render_report(*, agg, clusters, stats, generated_at) -> str:
    lines = [
        "# okstra cross-project 에러 환류 리포트",
        "",
        f"- 생성 시각: {generated_at}",
        f"- 총 에러: {agg['errorCount']} (에러 로그 보유 run {stats['runCount']}개)",
        f"- 도달 불가 run: {stats['unreachableRuns']}",
        f"- errorType 분포: {agg['byErrorType']}",
        "",
        "## 빈발 클러스터 (brief 분할 단위)",
        "",
        "| # | errorType | phase | agent | count | projects | 대표 메시지 |",
        "|---|---|---|---|---:|---|---|",
    ]
    for i, c in enumerate(clusters, start=1):
        sample = c["sample"].replace("|", "\\|").replace("\n", " ")[:60]
        lines.append(
            f"| {i} | {c['errorType']} | {c['phase']} | {c['agent']} "
            f"| {c['count']} | {len(c['projects'])} | {sample} |"
        )
    return "\n".join(lines) + "\n"


_CONFIG_NAME = "error-zip.json"


def remember_output_path(home: Path, out_path: Path) -> None:
    cfg = home / _CONFIG_NAME
    write_owned_object_atomic(
        cfg,
        {"lastOutputPath": str(out_path)},
        artifact="error zip configuration",
    )


def last_output_path(home: Path) -> str:
    cfg = home / _CONFIG_NAME
    if not cfg.is_file():
        return ""
    try:
        return str(
            load_owned_object(cfg, artifact="error zip configuration").get(
                "lastOutputPath", ""
            )
        )
    except Exception:
        return ""


def _write_zip(out_path: Path, report: str, serialized: list[dict]) -> None:
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with zipfile.ZipFile(out_path, "w", zipfile.ZIP_DEFLATED) as zf:
        zf.writestr("report.md", report)
        payload = "\n".join(json.dumps(r, ensure_ascii=False) for r in serialized)
        zf.writestr("errors/anonymized.jsonl", payload + ("\n" if payload else ""))


def build_zip(home: Path, out_path: Path, now: dt.datetime) -> dict:
    records, stats = collect_records(home)
    # 집계는 _sourceLog 가 살아있는 raw 레코드에서 한다 — 익명화 레코드는
    # _sourceLog 가 KEEP_FIELDS 밖이라 runCount 가 무너진다.
    agg = aggregate(records)
    clean, _ = anonymize(records)
    clusters, keys = build_clusters(clean)
    report = render_report(
        agg=agg, clusters=clusters, stats=stats,
        generated_at=now.isoformat(),
    )
    id_map = cluster_id_map(clusters)
    serialized = [{**r, "clusterId": id_map.get(k)} for r, k in zip(clean, keys)]
    # 출력 zip 과 공유 error-zip.json 쓰기를 중앙 락으로 직렬화한다 — 동시 error-zip
    # 두 개가 같은 경로에 쓰면 한쪽이 다른 쪽의 부분 기록 zip 을 절단하고
    # lastOutputPath 가 last-writer-wins 로 덮인다.
    with central_lock(home):
        _write_zip(out_path, report, serialized)
        remember_output_path(home, out_path)
    return {
        "outPath": str(out_path),
        "errorCount": len(clean),
        "runCount": stats["runCount"],
        "unreachableRuns": stats["unreachableRuns"],
        "clusterCount": len(clusters),
        "projectCount": len(stats["projectRoots"]),
    }


def render_result_text(result: dict) -> str:
    labels = (
        ("outPath", "Output zip"), ("errorCount", "Total errors"),
        ("runCount", "Run count"), ("unreachableRuns", "Unreachable runs"),
        ("clusterCount", "Cluster count"), ("projectCount", "Project count"),
    )
    lines = ["# Okstra Error Zip Result\n\n"]
    for key, label in labels:
        lines.append(line(label, result.get(key)))
    return "".join(lines)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="okstra error-zip",
        description="cross-project okstra 에러를 수집·익명화해 zip 으로 묶는다.",
    )
    parser.add_argument("--out", required=True, help="출력 .zip 절대/상대 경로")
    parser.add_argument("--text", action="store_true", help="emit fixed text fields")
    args = parser.parse_args(argv)
    home = okstra_home()
    result = build_zip(home, Path(args.out), dt.datetime.now(dt.timezone.utc))
    output = render_result_text(result) if args.text else json.dumps(result, ensure_ascii=False, indent=2)
    print(output, end="" if args.text else "\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))