#!/usr/bin/env python3
"""PreToolUse(Bash) 가드 — 운영 안전장치를 사람이 아니라 훅이 강제한다.

deny : 로컬 스택 기동, 운영 DB 대상 alembic, _test 아닌 DB 로 pytest, --allow-network,
       main 강제 푸시, 레포 핵심 경로 rm -rf, 유닛 스위트 실패 상태의 backend 커밋
ask  : main 갱신 푸시 / deploy.sh 실행 (KST 시각·장중 경고 포함)
파싱 실패는 조용히 통과(exit 0). 인용부호 안의 문자열은 명령으로 보지 않는다.
"""
import json
import os
import re
import shlex
import subprocess
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from _lib import git, kst_now, market_state, project_dir, read_stdin_json  # noqa: E402

WRAPPERS = {"sudo", "env", "time", "nohup", "command", "exec", "nice", "builtin"}
SHELLS = {"bash", "sh", "zsh", "dash"}
COMPOSE_BAD = {"up", "start", "restart", "run", "exec"}
# Makefile(레포 루트) 기준: dev/deploy/restart/rebuild/install 이 scripts/*.sh 또는 docker-compose up 을 호출.
# `up` 은 Makefile 에 없지만 관용적 별칭이라 함께 막는다. help/logs*/status/stop/backup 은 허용.
MAKE_BAD_TARGETS = {"up", "start", "dev", "deploy", "restart", "rebuild", "install"}
SCRIPT_BAD_RE = re.compile(r"(^|/)scripts/(dev|deploy|restart|start[^/]*)\.sh$")
PY_CODE_BAD_RE = re.compile(r"uvicorn\.(run|Server)|create_app\s*\(|gunicorn|app\.main[\s\S]*(uvicorn|\.run\()|(uvicorn|\.run\()[\s\S]*app\.main")
PYTEST_CMD_DEFAULT = (
    'cd backend && DATABASE_URL="postgresql+asyncpg://x:x@localhost/autostock_test" '
    "../.venv/bin/python -m pytest tests/ --ignore=tests/integration -q -x -p no:cacheprovider"
)


# ───────────────────────── 파싱 ─────────────────────────
def _strip_heredocs(cmd: str) -> str:
    out, lines, delim = [], cmd.split("\n"), None
    for ln in lines:
        if delim is not None:
            if ln.strip() == delim:
                delim = None
            continue
        out.append(ln)
        m = re.search(r"<<-?\s*['\"]?(\w+)['\"]?", ln)
        if m:
            delim = m.group(1)
    return "\n".join(out)


def split_segments(cmd: str) -> list[str]:
    """따옴표를 존중하며 ; && || | & 줄바꿈 ( ) ` 로 분리."""
    segs, cur, q, i, n = [], [], None, 0, len(cmd)
    while i < n:
        c = cmd[i]
        if q:
            cur.append(c)
            if c == "\\" and q == '"' and i + 1 < n:
                cur.append(cmd[i + 1])
                i += 1
            elif c == q:
                q = None
        elif c == "\\" and i + 1 < n:
            cur.append(c + cmd[i + 1])
            i += 1
        elif c in "'\"":
            q = c
            cur.append(c)
        elif c in ";|&\n()`":
            segs.append("".join(cur))
            cur = []
        else:
            cur.append(c)
        i += 1
    segs.append("".join(cur))
    return [s.strip() for s in segs if s.strip()]


def tokenize(seg: str) -> list[str]:
    try:
        return shlex.split(seg, posix=True)
    except ValueError:
        return seg.split()


def split_assignments(argv: list[str]) -> tuple[dict, list[str]]:
    env, i = {}, 0
    while i < len(argv):
        t = argv[i]
        m = re.match(r"^([A-Za-z_][A-Za-z0-9_]*)=(.*)$", t)
        if m:
            env[m.group(1)] = m.group(2)
            i += 1
        elif t in WRAPPERS:
            i += 1
        elif t == "export" and i + 1 < len(argv):
            i += 1
        else:
            break
    return env, argv[i:]


def base(tok: str) -> str:
    return os.path.basename(tok)


# ───────────────────────── 판정 ─────────────────────────
def deny(reason: str) -> dict:
    return {"decision": "deny", "reason": reason}


def ask(reason: str) -> dict:
    return {"decision": "ask", "reason": reason}


def is_pytest(argv: list[str]) -> bool:
    return any(base(t) in ("pytest", "py.test") for t in argv[:4]) or (
        "-m" in argv and "pytest" in argv
    )


def is_prod_url(url: str) -> bool:
    low = url.lower()
    return any(k in low for k in ("nas", "15433", "autostock")) and not url.split("?")[0].endswith("_test")


def check_local_stack(argv: list[str]) -> dict | None:
    b0 = base(argv[0]) if argv else ""
    names = [base(t) for t in argv]
    uvicorn_main = ("uvicorn" in names or b0 == "gunicorn") and any(
        t.startswith("app.main") or t.startswith("app:") for t in argv
    )
    py_dash_m = "-m" in argv and argv.index("-m") + 1 < len(argv) and argv[argv.index("-m") + 1].startswith(
        ("app.main", "uvicorn", "gunicorn")
    )
    if uvicorn_main or py_dash_m:
        return deny(
            "로컬 백엔드 기동 금지: NAS 와 KIS 앱키를 공유하므로 토큰이 무효화된다"
            "(local-docker-removed-kis-token-conflict 메모리). 검증은 격리 pytest(_test DB)로만 하라."
        )
    if b0 == "fastapi" and any(t in ("run", "dev") for t in argv[1:3]):
        return deny("로컬 fastapi 기동 금지(KIS 토큰 충돌).")
    # 실행 파일명과 무관: docker/podman/미해석 $VAR 가 compose 서브커맨드를 실행하는 경우.
    # (grep/echo 등 비실행 명령의 인자는 제외해 오탐을 막는다)
    if b0 in ("docker", "docker-compose", "podman", "podman-compose") or b0.startswith("$"):
        start = argv.index("compose") if "compose" in argv else (0 if b0.endswith("-compose") else -1)
        if start >= 0 and any(t in COMPOSE_BAD for t in argv[start + 1:]):
            return deny(
                "로컬 docker compose up/start/restart/run/exec 금지: 컨테이너·의존 서비스가 기동되어 KIS 앱키 공유로 "
                "NAS 토큰이 무효화된다. 배포는 /deploy 스킬(Jenkins), 검증은 격리 pytest."
            )
    if b0 == "make" and any(t in MAKE_BAD_TARGETS for t in argv[1:]):
        return deny(
            "make up/dev/deploy/restart/rebuild/install 금지: Makefile 이 scripts/*.sh 로 로컬 스택을 기동한다(KIS 토큰 충돌)."
        )
    for t in argv[:2]:
        if SCRIPT_BAD_RE.search(t) and ".claude/" not in t:
            return deny("scripts/dev|deploy|restart|start*.sh 는 로컬 스택을 기동한다(KIS 토큰 충돌). 배포는 /deploy 스킬.")
    if re.match(r"^python[\d.]*$", b0) or b0 == "python":
        for i, t in enumerate(argv[:-1]):
            if t == "-c" and PY_CODE_BAD_RE.search(argv[i + 1]):
                return deny("python -c 로 uvicorn/앱 기동 금지(KIS 토큰 충돌).")
    return None


def check_alembic(argv: list[str], env: dict) -> dict | None:
    if not (any(base(t) == "alembic" for t in argv) or ("-m" in argv and "alembic" in argv)):
        return None
    if not any(t in ("upgrade", "downgrade") for t in argv):
        return None
    if base(argv[0]) == "ssh":
        return None
    url = env.get("DATABASE_URL", os.environ.get("DATABASE_URL", ""))
    if url and is_prod_url(url):
        return deny("운영 DB(nas/15433/autostock, _test 아님)를 대상으로 로컬 alembic upgrade/downgrade 금지. 마이그레이션은 배포 파이프라인(Jenkins)이 앱 기동 전에 실행한다.")
    if not url:
        return ask("DATABASE_URL 미지정 alembic upgrade/downgrade — .env 가 운영 DB 를 가리킬 수 있다. 대상 DB 가 *_test 인지 확인했는가?")
    return None


def check_pytest(argv: list[str], env: dict) -> dict | None:
    if not is_pytest(argv):
        return None
    if "--allow-network" in argv:
        return deny("--allow-network 금지: KIS 호출 차단을 끄면 실주문이 나갈 수 있다(CLAUDE.md §4.1, 2026-09-27 실주문 2회 사고).")
    if any(".claude/hooks" in t for t in argv):
        return None  # 하네스 훅 자체 테스트는 DB 무관
    url = env.get("DATABASE_URL", os.environ.get("DATABASE_URL", ""))
    if not url.split("?")[0].endswith("_test"):
        return deny(
            "pytest 는 DATABASE_URL 이 *_test 로 끝나야 한다(CLAUDE.md §4.1). 예: "
            'cd backend && DATABASE_URL="postgresql+asyncpg://x:x@localhost/autostock_test" '
            "../.venv/bin/python -m pytest tests/ --ignore=tests/integration"
        )
    return None


def check_git_push(argv: list[str], cwd: str) -> dict | None:
    if base(argv[0]) != "git" or "push" not in argv:
        return None
    args = argv[argv.index("push") + 1:]
    force = any(a in ("-f", "--force", "--force-with-lease") or a.startswith("--force-with-lease=") for a in args) \
        or any(re.match(r"^-[a-zA-Z]*f[a-zA-Z]*$", a) for a in args)
    pos = [a for a in args if not a.startswith("-")]
    refspecs = pos[1:] if len(pos) > 1 else []
    to_main = False
    for r in refspecs:
        r2 = r.lstrip("+")
        dst = r2.split(":")[-1]
        if dst in ("main", "refs/heads/main"):
            to_main = True
        if r.startswith("+") and dst in ("main", "refs/heads/main"):
            force = True
        if r.startswith(":") and dst in ("main", "refs/heads/main"):
            return deny("main 삭제 푸시(`:main`) 금지.")
    if "--delete" in args or "-d" in args:
        if any(r in ("main", "refs/heads/main") for r in refspecs):
            return deny("main 브랜치 삭제 푸시 금지.")
    if not refspecs:  # git push / git push origin → 현재 브랜치
        _, br = git(["rev-parse", "--abbrev-ref", "HEAD"], cwd)
        if br.strip() == "main":
            to_main = True
    if force and to_main:
        return deny("main 강제 푸시(--force/-f/+ref) 금지.")
    if to_main:
        return ask(deploy_reason("main 푸시 = 운영 배포(Jenkins 재기동)"))
    return None


def deploy_reason(prefix: str) -> str:
    now = kst_now()
    st = market_state(now)
    msg = f"{prefix}. 현재 KST {now:%Y-%m-%d %H:%M} ({'월화수목금토일'[now.weekday()]}) {st}."
    if st == "장중":
        msg += " ⚠ 장중 배포 — 재기동이 매매에 영향(DT 포지션·스케줄러 상태 유실, 토큰 재발급)."
    return msg


def check_deploy_script(argv: list[str]) -> dict | None:
    if any(t.endswith("deploy.sh") for t in argv[:3]) and "--watch-only" not in argv:
        return ask(deploy_reason("deploy.sh 실행(원격 main 푸시 포함)"))
    return None


def _resolve(path: str, cwd: str) -> str:
    p = os.path.expandvars(os.path.expanduser(path))
    if p.startswith("$"):
        p = cwd
    return os.path.realpath(p if os.path.isabs(p) else os.path.join(cwd, p))


def check_find_delete(argv: list[str], cwd: str, root: str) -> dict | None:
    if base(argv[0]) != "find":
        return None
    destructive = "-delete" in argv or any(
        t in ("-exec", "-execdir", "-ok") and i + 1 < len(argv) and base(argv[i + 1]) == "rm"
        for i, t in enumerate(argv)
    )
    if not destructive:
        return None
    starts = []
    for t in argv[1:]:
        if t.startswith(("-", "!", "(")):
            break
        starts.append(t)
    protected = {os.path.realpath(root), os.path.realpath(os.path.join(root, "backend")),
                 os.path.realpath(os.path.join(root, ".git"))}
    for t in starts or ["."]:
        target = _resolve(t, cwd)
        if target in protected or os.path.realpath(root).startswith(target.rstrip("/") + "/") or target == "/":
            return deny(f"find '{t}' -delete/-exec rm 은 레포 루트/backend/.git(또는 상위)를 대량 삭제한다. 하위 캐시 경로로 한정하라.")
    return None


def check_rm(argv: list[str], cwd: str, root: str) -> dict | None:
    if base(argv[0]) != "rm":
        return None
    flags = [a for a in argv[1:] if a.startswith("-")]
    recursive = any(a in ("--recursive",) or re.match(r"^-[a-zA-Z]*[rR][a-zA-Z]*$", a) for a in flags)
    if not recursive:
        return None
    protected = {os.path.realpath(root), os.path.realpath(os.path.join(root, "backend")),
                 os.path.realpath(os.path.join(root, ".git"))}
    for t in argv[1:]:
        if t.startswith("-"):
            continue
        has_glob = any(ch in os.path.basename(t) for ch in "*?[")
        target = _resolve(os.path.dirname(t) or ".", cwd) if has_glob else _resolve(t, cwd)
        anc = os.path.realpath(root).startswith(target.rstrip("/") + "/") or target == "/"
        if target in protected or anc:
            return deny(f"rm -rf 대상 '{t}' 가 레포 루트/backend/.git(또는 그 상위)이다. 삭제 금지.")
    return None


def commit_gate(cmd_segments: list[tuple[list[str], int]], cwd: str) -> dict | None:
    commit_idx = None
    for idx, (argv, _) in enumerate(cmd_segments):
        if base(argv[0]) == "git" and "commit" in argv[1:]:
            commit_idx = idx
            break
    if commit_idx is None:
        return None
    argv = cmd_segments[commit_idx][0]
    rc, top = git(["rev-parse", "--show-toplevel"], cwd)
    if rc != 0:
        return None
    root = top.strip()
    preceded_by_add = any(base(a[0]) == "git" and "add" in a[1:] for a, _ in cmd_segments[:commit_idx])
    commit_all = any(
        a == "--all" or (a.startswith("-") and not a.startswith("--") and "a" in a[1:])
        for a in argv[argv.index("commit") + 1:]
    )
    if preceded_by_add or commit_all:
        _, out = git(["status", "--porcelain", "--", "backend"], root)
        touched = bool(out.strip())
    else:
        _, out = git(["diff", "--cached", "--name-only"], root)
        touched = any(l.startswith("backend/") for l in out.splitlines())
    if not touched:
        return None
    cmd = os.environ.get("HARNESS_PYTEST_CMD")
    if not cmd:
        if not os.path.exists(os.path.join(root, ".venv", "bin", "python")):
            sys.stderr.write("[bash_guard] .venv 없음 — 커밋 게이트(유닛 스위트) 생략\n")
            return {"decision": "warn", "reason": ".venv 가 없어 커밋 전 유닛 스위트를 건너뛰었다. 직접 검증하라."}
        cmd = PYTEST_CMD_DEFAULT
    try:
        p = subprocess.run(["bash", "-c", cmd], cwd=root, capture_output=True, text=True, timeout=120)
    except subprocess.TimeoutExpired:
        return deny("커밋 게이트: 유닛 스위트가 120초 내 끝나지 않았다(실 KIS 호출 의심, CLAUDE.md §4.1). 커밋 보류.")
    except Exception as e:  # noqa: BLE001
        sys.stderr.write(f"[bash_guard] 게이트 실행 오류 {e!r} — 통과\n")
        return None
    if p.returncode != 0:
        all_lines = (p.stdout + "\n" + p.stderr).strip().splitlines()
        key = [l for l in all_lines if re.match(r"^(FAILED|ERROR)\b", l.strip())
               or re.search(r"\d+ (failed|passed|error)", l)]
        tail = "\n".join((key or all_lines)[-15:])
        return deny(f"커밋 게이트: backend 변경이 스테이징돼 있는데 유닛 스위트 실패(rc={p.returncode}). 고친 뒤 다시 커밋.\n--- tail ---\n{tail}")
    return None


# ───────────────────────── 메인 ─────────────────────────
def evaluate(cmd: str, cwd: str, root: str) -> dict | None:
    cmd = _strip_heredocs(cmd)
    results: list[dict] = []
    env_acc: dict = {}
    parsed: list[tuple[list[str], int]] = []

    def handle(argv_all: list[str], depth: int = 0) -> None:
        env, argv = split_assignments(argv_all)
        if not argv:
            env_acc.update(env)  # 'A=b' 단독 또는 export A=b
            return
        eff = {**env_acc, **env}
        m = re.match(r"^\$(\w+)$|^\$\{(\w+)\}$", argv[0])
        if m and eff.get(m.group(1) or m.group(2)):
            argv = tokenize(eff[m.group(1) or m.group(2)]) + argv[1:]
        if base(argv[0]) in SHELLS and depth < 3:
            for i, t in enumerate(argv[1:], 1):
                if re.match(r"^-[a-zA-Z]*c[a-zA-Z]*$", t) and i + 1 < len(argv):
                    for s in split_segments(argv[i + 1]):
                        handle(tokenize(s), depth + 1)
                    return
        if base(argv[0]) == "eval" and depth < 3:
            for s in split_segments(" ".join(argv[1:])):
                handle(tokenize(s), depth + 1)
            return
        if base(argv[0]) == "ssh":
            return  # 원격 실행은 로컬 가드 대상 아님
        parsed.append((argv, depth))
        for fn in (check_local_stack, ):
            r = fn(argv)
            if r:
                results.append(r)
        for r in (check_alembic(argv, eff), check_pytest(argv, eff), check_git_push(argv, cwd),
                  check_rm(argv, cwd, root), check_find_delete(argv, cwd, root), check_deploy_script(argv)):
            if r:
                results.append(r)

    for seg in split_segments(cmd):
        handle(tokenize(seg))

    g = commit_gate(parsed, cwd)
    if g:
        results.append(g)
    for kind in ("deny", "ask", "warn"):
        hit = [r for r in results if r["decision"] == kind]
        if hit:
            return {"decision": kind, "reason": " | ".join(r["reason"] for r in hit)}
    return None


def main() -> int:
    data = read_stdin_json()
    if data.get("tool_name") not in (None, "Bash"):
        return 0
    cmd = (data.get("tool_input") or {}).get("command")
    if not isinstance(cmd, str) or not cmd.strip():
        return 0
    cwd = data.get("cwd") or os.getcwd()
    try:
        res = evaluate(cmd, cwd, project_dir(data))
    except Exception as e:  # noqa: BLE001  훅 자체 오류는 통과
        sys.stderr.write(f"[bash_guard] internal error {e!r}\n")
        return 0
    if not res:
        return 0
    if res["decision"] == "warn":
        json.dump({"hookSpecificOutput": {"hookEventName": "PreToolUse", "additionalContext": res["reason"]}}, sys.stdout, ensure_ascii=False)
        return 0
    json.dump(
        {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": res["decision"],
                                "permissionDecisionReason": f"[harness] {res['reason']}"}},
        sys.stdout, ensure_ascii=False,
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())