Blame for column_view.py

"""교열 진행·결과 한 열을 그리는 화면 조각 (Streamlit) - 로컬 3종 비교 앱(`proofread-demo/compare_app.py`)과 같은 모양.

기다리는 동안 "뒤에서 무엇이 진행되는지"가 보이도록:
  - 요약 3칸(소요 · 적용 · 확인 필요)과 단계 스테퍼(끝난 단계 ✓ 소요 시간 · 진행 중 ● 경과/보통 N초 · 대기 ○)
  - 단계가 끝날 때마다 본문 하이라이트·수정 카드가 그 자리에서 쌓인다
  - 카드·하이라이트는 결과 본문에 남은 변경만 보인다(FT 가 고치고 기사 검토가 원문으로 되돌린 것은 둘 다 뺀다)
이벤트 모양은 `runner.start_proofread` 를 따른다.
"""

from __future__ import annotations

import difflib
import html
import time
from typing import Any

import streamlit as st

# 화면에 보일 단계. (표시 이름, 그 단계가 끝났다고 보는 스텝 이름, 보통 걸리는 초)
STAGES = [
    ("표기 사전 적용", "vocabulary_sub_pre", 0.1),
    ("교열 모델(FT) 문단별", "proofread", 1),
    ("기사 전체 검토 (1회)", "review", 7),
    ("FT 교정 확인", "revert_ft", 0.1),
    ("사실·숫자·문맥 교정 적용", "apply_review", 0.1),
    ("마무리 규칙", "post_process", 0.1),
]
# 교정 목록을 적용하는 스텝: (목록을 낸 스텝, 숫자 가드, 삭제 가드 비율) - 카드 상태를 적용 스텝과 같은 가드로 계산한다
LIST_CARDS = {"apply_review": ("review", False, 0.6)}
# FT 변경 되돌림: (되돌림 스텝, 변경 목록 스텝, 판정 스텝)
REVERTS = ("revert_ft", "ft_hunks", "review")

STAGE_COLORS = [
    "rgba(120,120,120,.25)",
    "rgba(70,130,220,.28)",
    "rgba(60,170,90,.28)",
    "rgba(240,150,40,.32)",
    "rgba(170,90,200,.28)",
    "rgba(120,120,120,.25)",
]
CAT_KO = {
    "spelling": "맞춤법",
    "spacing": "띄어쓰기·사이시옷",
    "grammar": "조사·호응",
    "context": "문맥 낱말",
    "omission": "누락",
    "in-article": "기사 안 불일치",
    "inconsistency": "기사 안 불일치",
    "fact": "사실 오기",
    "calc": "숫자 검산",
    "word": "문맥 낱말",
    "unit": "단위·수치 모순",
}
CSS = """
<style>
.pane{border:1px solid rgba(128,128,128,.35);border-radius:8px;padding:12px 14px;line-height:1.95;font-size:.98rem;min-height:120px}
.shimmer{height:4px;border-radius:2px;margin:6px 0 10px;background:linear-gradient(90deg,rgba(128,128,128,.15) 25%,rgba(128,128,128,.45) 50%,rgba(128,128,128,.15) 75%);background-size:200% 100%;animation:sh 1.2s linear infinite}
@keyframes sh{0%{background-position:200% 0}100%{background-position:-200% 0}}
.stage{font-size:.9rem;margin:2px 0}
.card{border:1px solid rgba(128,128,128,.3);border-radius:6px;padding:6px 10px;margin-bottom:6px;font-size:.9rem}
.muted{color:gray;font-size:.82rem}
.stats{display:flex;gap:8px;margin:4px 0 8px}
.stat{flex:1;border:1px solid rgba(128,128,128,.3);border-radius:6px;padding:6px 8px;text-align:center}
.stat .k{font-size:.72rem;color:gray}.stat .v{font-size:1.05rem;font-weight:700}
.status{min-height:30px;margin-bottom:6px}
.chip{display:inline-block;font-size:.75rem;padding:1px 8px;border-radius:10px;margin-left:6px;font-weight:600}
.ok{background:rgba(60,170,90,.22);color:#1e7d3a}
.chk{background:rgba(240,170,40,.28);color:#8a5a00}
.rej{background:rgba(128,128,128,.22);color:#555}
</style>
"""

NL = "\n"  # 줄바꿈을 어절 하나로 다뤄 diff·하이라이트가 입력의 줄 구조를 보존하게 한다


def tokenize(text: str) -> list[str]:
    """어절 + 줄바꿈 토큰. 빈 줄은 NL 토큰이 연속으로 남아 문단 간격이 유지된다."""
    out: list[str] = []
    for line in text.split("\n"):
        out.extend(line.split())
        out.append(NL)
    return out[:-1]


def _norm(s: str) -> str:
    return " ".join(s.split())


# ───────────────────────── 상태 ─────────────────────────
def new_state(original: str) -> dict[str, Any]:
    return {
        "original": original,
        "stage_i": 0,
        "stage_done_at": {},
        "tokens": None,
        "last_text": None,
        "cards": [],
        "log": "대기 중",
        "done": None,
        "error": None,
        "started": time.monotonic(),
    }


def _label_tokens(state: dict, new_text: str, stage_i: int) -> None:
    """원문 어절에 '어느 단계에서 바뀌었나' 라벨을 누적한다. 원문과 같아진 어절은 라벨을 지운다."""
    if state["tokens"] is None:
        state["tokens"] = [(t, -1) for t in tokenize(state["last_text"])]
    old = [t for t, _ in state["tokens"]]
    new = tokenize(new_text)
    sm = difflib.SequenceMatcher(a=old, b=new, autojunk=False)
    out: list[tuple[str, int]] = []
    for op, i1, i2, j1, j2 in sm.get_opcodes():
        out.extend(state["tokens"][i1:i2] if op == "equal" else ((t, stage_i) for t in new[j1:j2]))
    orig = tokenize(state["original"])
    net = difflib.SequenceMatcher(a=orig, b=[t for t, _ in out], autojunk=False)
    for op, _i1, _i2, j1, j2 in net.get_opcodes():
        if op == "equal":
            out[j1:j2] = [(t, -1) for t, _ in out[j1:j2]]
    state["tokens"] = out
    state["last_text"] = new_text


def _item_status(
    item: dict, text_before: str, protect_digits: bool, max_shrink: float | None
) -> str:
    from solar_eval.pipelines.corrections import apply_corrections

    _, rep = apply_corrections(
        text_before, [item], max_items=1, protect_digits=protect_digits, max_shrink=max_shrink
    )
    if rep.applied:
        return "적용"
    if rep.skipped_protected:
        return "거부 (숫자·삭제 가드)"
    if rep.skipped_ambiguous or rep.skipped_not_found:
        return "확인 필요"
    return "무효"


def _diff_card(stage: int, before: str, after: str) -> dict:
    spacing = "".join(before.split()) == "".join(after.split())
    return {
        "stage": stage,
        "before": before,
        "after": after,
        "cat": "띄어쓰기" if spacing else "",
        "status": "적용",
        "evidence": "",
    }


def _add_card(state: dict, card: dict) -> None:
    """카드를 더한다. 앞 단계의 적용 카드를 그대로 되돌리는 변경이면 둘 다 지운다(결과적으로 바뀐 것이 없다)."""
    if card["status"] == "적용":
        for i, c in enumerate(state["cards"]):
            if (
                c["status"] == "적용"
                and _norm(c["before"]) == _norm(card["after"])
                and _norm(c["after"]) == _norm(card["before"])
            ):
                del state["cards"][i]
                return
    state["cards"].append(card)


def _drop_reverted(state: dict, ev: dict) -> None:
    """기사 검토가 원문으로 되돌린 FT 변경은 카드에서 지운다."""
    _step, hunks_from, judge_from = REVERTS
    decision = ev["json"].get(judge_from) or {}
    by_id = {int(h["id"]): h for h in ev["hunks"].get(hunks_from) or []}
    for raw in decision.get("revert") or []:
        h = by_id.get(int(raw)) if str(raw).isdigit() else None
        if h is None:
            continue
        pair = (_norm(h["before"]), _norm(h["after"]))
        state["cards"] = [
            c
            for c in state["cards"]
            if not (c["status"] == "적용" and (_norm(c["before"]), _norm(c["after"])) == pair)
        ]


def absorb(state: dict, ev: dict) -> None:
    """이벤트 하나를 상태에 반영한다. 단계 완료 판정·카드·하이라이트·로그."""
    if ev["kind"] == "error":
        state["error"] = ev["error"]
        state["log"] = "실패"
        return
    if ev["kind"] == "done":
        state["done"] = ev
        state["log"] = f"완료 - {ev['elapsed_s']:.1f}초"
        return
    step, text = ev["step"], ev["text"]
    stage_steps = [s[1] for s in STAGES]
    si = state["stage_i"]  # 이 스텝은 다음 미완 단계에 속한다
    if state["last_text"] is None:
        state["last_text"] = state["original"]
    pre_text = state["last_text"]
    list_card = LIST_CARDS.get(step)
    reverted = False
    if list_card and list_card[0] in ev["corrections"]:
        corr_key, protect_digits, max_shrink = list_card
        for it in ev["corrections"][corr_key]:
            cat = str(it.get("category", ""))
            _add_card(
                state,
                {
                    "stage": si,
                    "before": str(it.get("original", "")),
                    "after": str(it.get("corrected", "")),
                    "cat": CAT_KO.get(cat, cat),
                    "status": _item_status(it, pre_text, protect_digits, max_shrink),
                    "evidence": str(it.get("evidence", "")),
                },
            )
    elif step == REVERTS[0]:
        _drop_reverted(state, ev)
        reverted = True
    elif ev["changes"] and not list_card:
        for b, a in ev["changes"]:
            _add_card(state, _diff_card(si, b, a))
    if text != pre_text or state["tokens"] is None:
        _label_tokens(state, text, si)
    if step in stage_steps:
        idx = stage_steps.index(step)
        state["stage_done_at"][idx] = ev["elapsed_s"]
        state["stage_i"] = idx + 1
        n = sum(c["stage"] == idx for c in state["cards"])
        n_chk = sum(c["stage"] == idx and c["status"] == "확인 필요" for c in state["cards"])
        extra = (
            (f" - 제안 {n}건" + (f" (확인 필요 {n_chk})" if n_chk else ""))
            if n
            else " - 고칠 곳 없음"
        )
        if reverted:
            extra = ""  # 확인 단계는 지운 카드 수를 내세우지 않는다
        state["log"] = f"방금: {STAGES[idx][0]} 완료{extra}"


# ───────────────────────── 렌더 ─────────────────────────
def _render_text(state: dict) -> str:
    toks = state["tokens"] or [(t, -1) for t in tokenize(state["original"])]
    parts: list[str] = []
    run: list[str] = []
    cur = -1

    def flush() -> None:
        if run:
            seg = html.escape(" ".join(run))
            parts.append(
                seg
                if cur < 0
                else f'<span style="background:{STAGE_COLORS[cur % len(STAGE_COLORS)]};border-radius:3px">{seg}</span>'
            )
            run.clear()

    for t, s in toks + [("", -2)]:
        if t == NL:  # 입력의 줄바꿈을 그대로 - 빈 줄은 <br> 두 번
            flush()
            parts.append("<br>")
            continue
        if s != cur:
            flush()
            cur = s
        if t:
            run.append(t)
    return (
        " ".join(parts).replace(" <br> ", "<br>").replace("<br> ", "<br>").replace(" <br>", "<br>")
    )


def _cards_html(state: dict) -> str:
    cards = [c for c in state["cards"] if _norm(c["before"]) != _norm(c["after"])]
    if not cards:
        return ""
    parts = []
    for c in cards:
        color = STAGE_COLORS[c["stage"] % len(STAGE_COLORS)]
        if c["status"] == "적용":
            cls, label = "ok", "✓ 적용"
        elif c["status"] == "확인 필요":
            cls, label = "chk", "확인 필요"
        else:
            cls, label = "rej", c["status"]
        ev_html = (
            f"<div class='muted'>근거: {html.escape(c['evidence'])}</div>" if c["evidence"] else ""
        )
        parts.append(
            f"<div class='card' style='border-left:5px solid {color}'><span class='muted'>"
            f"{STAGES[c['stage']][0]}{' · ' + c['cat'] if c['cat'] else ''}</span>"
            f"<span class='chip {cls}'>{label}</span><br>"
            f"<s>{html.escape(c['before'])}</s> → <b>{html.escape(c['after'])}</b>{ev_html}</div>"
        )
    return f"<div class='muted' style='margin-top:6px'>제안 {len(cards)}건</div>" + "".join(parts)


def render(ph: Any, state: dict, now: float) -> None:
    """왼쪽: 요약 3칸 · 단계 스테퍼 · 상태 줄 / 오른쪽: 하이라이트 본문 · 수정 카드."""
    elapsed = now - state["started"]
    expected = sum(s[2] for s in STAGES)
    with ph.container():
        left, right = st.columns([2, 3], gap="large")
        with left:
            n_ok = sum(c["status"] == "적용" for c in state["cards"])
            n_chk = sum(c["status"] == "확인 필요" for c in state["cards"])
            if state["error"]:
                took = "실패"
            elif state["done"]:
                took = f"{state['done']['elapsed_s']:.1f}초"
            else:
                took = f"{elapsed:.0f}초…"
            st.markdown(
                "<div class='stats'>"
                f"<div class='stat'><div class='k'>소요</div><div class='v'>{took}</div></div>"
                f"<div class='stat'><div class='k'>적용</div><div class='v'>{n_ok}</div></div>"
                f"<div class='stat'><div class='k'>확인 필요</div><div class='v'>{n_chk}</div></div>"
                "</div>",
                unsafe_allow_html=True,
            )
            lines = []
            for i, (name, _step, typical) in enumerate(STAGES):
                prev = state["stage_done_at"].get(i - 1, 0.0)
                if i in state["stage_done_at"]:
                    lines.append(
                        f"<div class='stage'>✓ {name} <span class='muted'>({state['stage_done_at'][i] - prev:.1f}초)</span></div>"
                    )
                elif i == state["stage_i"] and not state["done"] and not state["error"]:
                    lines.append(
                        f"<div class='stage'>● <b>{name}</b> <span class='muted'>진행 중 {elapsed - prev:.0f}초 · 보통 {typical:.0f}초</span></div>"
                    )
                else:
                    lines.append(f"<div class='stage muted'>○ {name}</div>")
            st.markdown("".join(lines), unsafe_allow_html=True)
            if state["error"]:
                st.markdown(
                    f"<div class='status' style='color:#b00020'>실패: {html.escape(state['error'])}</div>",
                    unsafe_allow_html=True,
                )
            elif state["done"]:
                st.markdown(
                    f"<div class='status' style='color:#1e7d3a'>완료 {state['done']['elapsed_s']:.1f}초</div>",
                    unsafe_allow_html=True,
                )
            else:
                msg = state["log"]
                if elapsed > expected * 1.5 + 5:
                    msg += f" · 평소({expected:.0f}초)보다 오래 걸립니다 - API 대기 중일 수 있습니다. 결과는 그대로 도착합니다."
                st.markdown(
                    f"<div class='status muted'>{elapsed:.0f}초 경과 · {msg}<div class='shimmer'></div></div>",
                    unsafe_allow_html=True,
                )
        with right:
            st.markdown(f"<div class='pane'>{_render_text(state)}</div>", unsafe_allow_html=True)
            # 카드는 한 덩어리 HTML 로 그린다 - 카드마다 st.markdown 을 부르면 카드가 줄었을 때 지난 렌더의 카드가 남는다
            st.markdown(_cards_html(state), unsafe_allow_html=True)