auto-stock/.claude/hooks/rules_check.py

#!/usr/bin/env python3
"""PostToolUse(Edit|Write|MultiEdit) — 편집된 backend/**/*.py 만 검사.

1) py_compile 실패 → exit 2
2) CLAUDE.md 정적 규칙: git diff(HEAD 대비) 추가 라인만 대상 → 기존 위반은 보고하지 않는다.
   (추적되지 않는 새 파일은 전체 라인이 추가로 간주됨)
위반이 있으면 exit 2 + stderr 에 '파일:라인 규칙 — 근거(CLAUDE.md §)'. 경고성 항목은 stderr 로만 알리고 exit 0.
"""
import os
import py_compile
import re
import sys
import tempfile

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

REALTIME_MARKERS = (
    "/app/realtime/",
    "/services/daytrading/",
    "/services/swing/scheduler.py",
    "/kis_realtime_stream.py",
)
DB_READ_FUNCS = {"close_stale_positions", "populate_daytrading_watchlist", "restore_from_db"}
ALLOW_RE = re.compile(r"harness:\s*allow-(json-dumps|except-pass)\b")
EXCEPT_RE = re.compile(r"^\s*except\b[^:]*:\s*(.*)$")


def added_lines(root: str, rel: str, full_text: str) -> set[int]:
    rc, _ = git(["ls-files", "--error-unmatch", rel], root)
    if rc != 0:  # 새 파일(미추적) → 전부 추가
        return set(range(1, full_text.count("\n") + 2))
    rc, out = git(["diff", "-U0", "HEAD", "--", rel], root)
    if rc != 0:
        return set()
    res: set[int] = set()
    for m in re.finditer(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,(\d+))? @@", out, re.M):
        start, cnt = int(m.group(1)), int(m.group(2) or 1)
        res.update(range(start, start + cnt))
    return res


def enclosing_def(lines: list[str], idx: int) -> tuple[str, int]:
    indent = len(lines[idx]) - len(lines[idx].lstrip())
    for j in range(idx, -1, -1):
        m = re.match(r"^(\s*)(?:async\s+)?def\s+(\w+)", lines[j])
        if m and len(m.group(1)) < indent:
            return m.group(2), j
    return "", -1


def next_stmt(lines: list[str], i: int) -> int | None:
    for j in range(i + 1, min(i + 8, len(lines))):
        s = lines[j].strip()
        if s and not s.startswith("#"):
            return j
    return None


def allowed(lines: list[str], idx: int, kind: str, *extra: int) -> bool:
    """같은 줄 또는 직전 줄(extra 로 추가 줄 인덱스)에 `# harness: allow-<kind> <사유>` 마커가 있으면 True."""
    for j in (idx, idx - 1, *extra):
        if 0 <= j < len(lines):
            m = ALLOW_RE.search(lines[j])
            if m and m.group(1) == kind:
                return True
    return False


def check_static(path: str, rel: str, lines: list[str], added: set[int]) -> list[str]:
    v: list[str] = []
    is_json_helper = rel.endswith("app/core/json_helpers.py")
    realtime = any(mk in "/" + rel for mk in REALTIME_MARKERS)
    for idx, raw in enumerate(lines):
        ln = idx + 1
        s = raw.strip()
        if s.startswith("#"):
            continue
        # json.dumps
        if ln in added and not is_json_helper and re.search(r"\bjson\.dumps\(", raw) and not allowed(lines, idx, "json-dumps"):
            v.append(f"{rel}:{ln} json.dumps 사용 — core.json_helpers._dumps() 를 쓸 것 (CLAUDE.md §4)")
        # except 블록 규칙
        m = EXCEPT_RE.match(raw)
        if m:
            nxt = next_stmt(lines, idx)
            inline = m.group(1).strip()
            first = inline if inline and not inline.startswith("#") else (lines[nxt].strip() if nxt is not None else "")
            first_ln = ln if inline and not inline.startswith("#") else (nxt + 1 if nxt is not None else ln)
            touched = ln in added or first_ln in added
            if touched and re.search(r"\blog(?:ger|ging)?\.debug\(", first):
                v.append(f"{rel}:{first_ln} except 블록의 첫 문장이 logger.debug — 최소 warning 이어야 한다 (CLAUDE.md §4, §8.1-10)")
            if (
                touched
                and first == "pass"
                and re.match(r"^\s*except(\s+(Exception|BaseException)(\s+as\s+\w+)?)?\s*:", raw)
                and not allowed(lines, idx, "except-pass", first_ln - 1)
            ):
                v.append(f"{rel}:{ln} `except Exception: pass` — 예외 삼킴 금지, 최소 warning 로그 (CLAUDE.md §4, §8.1-10)")
        # 실시간 계층 DB I/O
        if realtime and ln in added:
            if ".commit(" in raw:
                v.append(f"{rel}:{ln} 실시간 계층에서 commit() — DB 쓰기는 db_task_queue 경유 (CLAUDE.md §3.1·§3.2)")
            if "AsyncSessionLocal(" in raw and not s.startswith(("def ", "class ", "from ", "import ")):
                name, di = enclosing_def(lines, idx)
                window = lines[max(0, idx - 3): idx + 1]
                if name in DB_READ_FUNCS:
                    continue
                if any("DB-READ-EXCEPTION" in w for w in window) or (di >= 0 and "DB-READ-EXCEPTION" in lines[di]):
                    continue
                if rel.endswith("daytrading/scheduler.py") and any(
                    "restore_from_db" in w for w in lines[max(0, idx - 10): idx + 10]
                ):
                    continue
                v.append(f"{rel}:{ln} 실시간 계층에서 AsyncSessionLocal( — SELECT 전용 예외면 `# DB-READ-EXCEPTION` 마커를 달고 commit 금지 (CLAUDE.md §3.1·§3.2 ③)")
    return v


def _quote_of(tok_string: str) -> str:
    """'f"' / "rb'''" 같은 문자열 시작 토큰에서 따옴표 부분('"', "'''" 등)만 뽑는다."""
    body = tok_string.lstrip("rRbBuUfF")
    for q in ('"""', "'''", '"', "'"):
        if body.startswith(q):
            return q
    return ""


def check_py311_fstring(rel: str, text: str) -> list[str]:
    """운영 이미지(python:3.11-slim)에서 SyntaxError 가 나는 PEP 701(3.12+) f-string 을 찾는다.

    3.12 의 tokenize 는 f-string 을 FSTRING_START/END 로 쪼개므로, 열린 f-string 안에서
    바깥과 같은 따옴표로 시작하는 문자열·f-string 이 나오면 3.11 에서는 바깥을 끝내 버린다
    (2026-10-02 mdd_manager 4줄이 이 이유로 운영 기동을 막을 뻔했다). 3.12 미만에서 돌면 검사 생략.
    """
    if sys.version_info < (3, 12):
        return []
    import io
    import tokenize

    out: list[str] = []
    stack: list[str] = []
    try:
        for tok in tokenize.generate_tokens(io.StringIO(text).readline):
            name = tokenize.tok_name.get(tok.type, "")
            if name in ("FSTRING_START", "STRING"):
                q = _quote_of(tok.string)
                if stack and q and any(
                    (len(outer) == 1 and q[0] == outer) or (len(outer) == 3 and q == outer) for outer in stack
                ):
                    out.append(
                        f"{rel}:{tok.start[0]} f-string 안에서 바깥과 같은 따옴표 재사용 — Python 3.12 전용 문법(PEP 701), "
                        "운영 이미지 python:3.11 에서 SyntaxError 로 기동 실패 (Dockerfile)"
                    )
                if name == "FSTRING_START":
                    stack.append(q)
            elif name == "FSTRING_END" and stack:
                stack.pop()
    except (tokenize.TokenError, SyntaxError):
        return []
    return out


def check_router_auth(root: str, rel: str, text: str) -> list[str]:
    if "/api/endpoints/" not in "/" + rel or not re.search(r"@router\.(post|put|delete|patch)\(", text):
        return []
    if "require_api_key" in text:
        return []
    init = os.path.join(root, "backend/app/api/__init__.py")
    mod = os.path.splitext(os.path.basename(rel))[0]
    try:
        body = open(init, encoding="utf-8").read()
    except OSError:
        return []
    for m in re.finditer(r"include_router\(\s*([\w.]+)\.router(.*?)\)\s*(?=\n\S|\Z|\napi_router)", body, re.S):
        if m.group(1).split(".")[-1] == mod and "require_api_key" in m.group(2):
            return []
    return [f"{rel}:1 [경고] 쓰기 엔드포인트가 있는데 require_api_key 가 라우터/api/__init__.py 등록에서 확인되지 않음 (CLAUDE.md §4)"]


def main() -> int:
    data = read_stdin_json()
    fp = (data.get("tool_input") or {}).get("file_path")
    if not isinstance(fp, str) or not fp.endswith(".py"):
        return 0
    root = os.path.realpath(project_dir(data))
    fp = os.path.realpath(fp)
    if not fp.startswith(root + "/backend/") or not os.path.isfile(fp):
        return 0
    rel = os.path.relpath(fp, root)
    errors: list[str] = []

    try:
        with tempfile.TemporaryDirectory() as td:
            py_compile.compile(fp, cfile=os.path.join(td, "x.pyc"), doraise=True)
    except py_compile.PyCompileError as e:
        errors.append(f"{rel} py_compile 실패 — {str(e).strip().splitlines()[-1]}")
    except Exception:
        pass

    in_tests = "/backend/tests/" in fp or "/backend/alembic/" in fp
    warnings: list[str] = []
    if not errors and not in_tests:
        try:
            text = open(fp, encoding="utf-8").read()
            lines = text.split("\n")
            errors += check_static(fp, rel, lines, added_lines(root, rel, text))
            errors += check_py311_fstring(rel, text)
            warnings += check_router_auth(root, rel, text)
        except Exception as e:  # noqa: BLE001
            sys.stderr.write(f"[rules_check] internal {e!r}\n")
            return 0

    if warnings and not errors:
        sys.stderr.write("\n".join(warnings) + "\n")
        return 0
    if errors:
        sys.stderr.write("[harness] CLAUDE.md 규칙 위반(이번 편집에서 추가된 라인):\n" + "\n".join(errors + warnings) + "\n")
        return 2
    return 0


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