"""교열 진행·결과 한 열을 그리는 화면 조각 (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 = """
"""
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'{seg}'
)
run.clear()
for t, s in toks + [("", -2)]:
if t == NL: # 입력의 줄바꿈을 그대로 - 빈 줄은
두 번
flush()
parts.append("
")
continue
if s != cur:
flush()
cur = s
if t:
run.append(t)
return (
" ".join(parts).replace("
", "
").replace("
", "
").replace("
", "
")
)
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"근거: {html.escape(c['evidence'])}
" if c["evidence"] else ""
)
parts.append(
f""
f"{STAGES[c['stage']][0]}{' · ' + c['cat'] if c['cat'] else ''}"
f"{label}
"
f"{html.escape(c['before'])} → {html.escape(c['after'])}{ev_html}
"
)
return f"제안 {len(cards)}건
" + "".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(
"",
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"✓ {name} ({state['stage_done_at'][i] - prev:.1f}초)
"
)
elif i == state["stage_i"] and not state["done"] and not state["error"]:
lines.append(
f"● {name} 진행 중 {elapsed - prev:.0f}초 · 보통 {typical:.0f}초
"
)
else:
lines.append(f"○ {name}
")
st.markdown("".join(lines), unsafe_allow_html=True)
if state["error"]:
st.markdown(
f"실패: {html.escape(state['error'])}
",
unsafe_allow_html=True,
)
elif state["done"]:
st.markdown(
f"완료 {state['done']['elapsed_s']:.1f}초
",
unsafe_allow_html=True,
)
else:
msg = state["log"]
if elapsed > expected * 1.5 + 5:
msg += f" · 평소({expected:.0f}초)보다 오래 걸립니다 - API 대기 중일 수 있습니다. 결과는 그대로 도착합니다."
st.markdown(
f"{elapsed:.0f}초 경과 · {msg}
{_render_text(state)}
", unsafe_allow_html=True)
# 카드는 한 덩어리 HTML 로 그린다 - 카드마다 st.markdown 을 부르면 카드가 줄었을 때 지난 렌더의 카드가 남는다
st.markdown(_cards_html(state), unsafe_allow_html=True)