app.py

"""조선일보 본문 교열 데모 (Streamlit / HF Space).

v35 파이프라인(prod 7-스텝 + 화이트리스트 복원 + 율/률 규칙 + 문서 단위 chunk) × combo 프롬프트를
solar-eval 엔진 그대로 실행한다. 로직은 전부 runner.py — 이 파일은 UI 만 담당.
사용자 입력과 결과는 logbook.py 가 Supabase 에 남긴다 (시크릿 없으면 조용히 건너뜀).
"""

from __future__ import annotations

import difflib
import html
import os

import logbook
import runner
import streamlit as st

PANE_STYLE = (
    "border:1px solid rgba(128,128,128,.35);border-radius:8px;"
    "padding:14px 16px;line-height:1.9;font-size:1.02rem;min-height:120px"
)

SAMPLE_TEXT = (
    "정부의 강력한 부동산 규제에도 불구하고 수도권 아파트 매매 거래 비률은 "
    "오히려 증가세를 보였다. 시중 은행의 주택 담보 대출 금리가 낮아지면서 "
    "실수요자의 구매 심리가 되살아났기 때문이라는 분석이 나온다."
)

# 스텝 이름 → 사용자에게 보여줄 라벨 (pipeline_dev_v34 기준. 모르는 스텝은 이름 그대로)
STEP_LABELS = {
    "vocabulary_sub_pre": "용어 사전 치환",
    "proofread": "1차 교열 (교열 전용 모델)",
    "basic_correction": "기본 교정 (자기일관성 검증)",
    "context_correction": "문맥 교정",
    "style_correction": "문체 교정",
    "vocabulary_reapply": "용어 사전 재적용",
    "compound_whitelist_restore": "복합어 화이트리스트 복원",
    "yul_ryul_normalize": "율/률 규칙 적용",
    "post_process": "후처리",
}


def diff_panes(src: str, dst: str) -> tuple[str, str, int]:
    """어절 단위 diff 를 좌(원문)/우(교정) 패널 HTML 로. 바뀐 세그먼트 수도 반환."""
    a, b = src.split(), dst.split()
    sm = difflib.SequenceMatcher(a=a, b=b, autojunk=False)
    left: list[str] = []
    right: list[str] = []
    n_changed = 0
    for op, i1, i2, j1, j2 in sm.get_opcodes():
        before, after = html.escape(" ".join(a[i1:i2])), html.escape(" ".join(b[j1:j2]))
        if op == "equal":
            left.append(before)
            right.append(after)
            continue
        n_changed += 1
        if before:
            left.append(f'<span style="background:rgba(230,80,80,.20)">{before}</span>')
        if after:
            right.append(f'<span style="background:rgba(60,170,90,.24)">{after}</span>')
    return " ".join(left), " ".join(right), n_changed


class ProgressView:
    """스텝 진행을 st.status 안에 그린다 — 단계별 출력은 보여주지 않고 '어디까지 왔나' 만.

    문서 단위 chunk 모드에서는 chunk 들이 **병렬**로 돌아 이벤트가 섞여 들어오므로, chunk 마다 줄을 하나씩
    두고(몇 단계까지 왔나) 전체 진행 바는 (끝난 단계 수) / (chunk 수 × 단계 수) 로 잡는다.
    """

    def __init__(self, names: list[str]) -> None:
        self.names = names
        self.total_steps = len(names)
        self.n_chunks = 1
        self.done: dict[int, int] = {1: 0}  # chunk 번호 → 끝난 단계 수
        self.current: dict[int, str] = {}  # chunk 번호 → 진행 중 단계 라벨
        self.status = st.status("교열 중입니다…", expanded=True)
        self.bar = self.status.progress(0.0, text=f"0 / {self.total_steps} 단계")
        self.lines = self.status.empty()
        self._render()

    def on_step(self, event: str, index: int, total: int, name: str, doc_chunk=None) -> None:
        ci = doc_chunk[0] if doc_chunk else 1
        if doc_chunk:
            self.n_chunks = doc_chunk[1]
        self.done.setdefault(ci, 0)
        label = STEP_LABELS.get(name, name)
        if event == "start":
            self.current[ci] = label
        elif event == "chunk":
            self.current[ci] = f"{label} · 조각 {index}/{total}"
        elif event == "done":
            self.done[ci] = index
            self.current.pop(ci, None)
        finished = sum(self.done.values())
        grand = self.n_chunks * self.total_steps
        unit = "단계" if self.n_chunks == 1 else f"단계 (문서 조각 {self.n_chunks}개 병렬)"
        self.bar.progress(min(finished / grand, 1.0), text=f"{finished} / {grand} {unit}")
        self._render()

    def _render(self) -> None:
        if self.n_chunks == 1:
            done = self.done.get(1, 0)
            rows = []
            for i, name in enumerate(self.names, 1):
                label = STEP_LABELS.get(name, name)
                if i <= done:
                    rows.append(f"✅ {label}")
                elif i == done + 1:
                    rows.append(f"⏳ **{self.current.get(1, label)}**")
                else:
                    rows.append(f"▫️ {label}")
        else:
            rows = []
            for ci in range(1, self.n_chunks + 1):
                done = self.done.get(ci, 0)
                if done >= self.total_steps:
                    rows.append(f"✅ 조각 {ci}: 완료")
                elif ci in self.current:
                    rows.append(f"⏳ 조각 {ci}: **{self.current[ci]}** ({done}/{self.total_steps})")
                else:
                    rows.append(f"▫️ 조각 {ci}: 대기")
        self.lines.markdown("  \n".join(rows))

    def finish(self, elapsed_s: float) -> None:
        self.status.update(label=f"교열 완료 · {elapsed_s:.1f}초", state="complete", expanded=False)

    def fail(self) -> None:
        self.status.update(label="교열 실패", state="error", expanded=True)


@st.cache_resource
def get_pipeline():
    return runner.build_pipeline()


st.set_page_config(page_title="조선일보 본문 교열 데모", page_icon="📰", layout="wide")
st.title("📰 조선일보 본문 교열 데모")

with st.sidebar:
    st.subheader("설정")
    if not os.environ.get("UPSTAGE_API_KEY"):
        key = st.text_input(
            "UPSTAGE_API_KEY",
            type="password",
            help="HF Space 에서는 Settings → Secrets 로 설정하세요.",
        )
        if key:
            os.environ["UPSTAGE_API_KEY"] = key
    model = st.selectbox("모델 (step1–3 주입)", ["solar-pro4", "solar-pro2"], index=0)

text = st.text_area("기사 본문", value=SAMPLE_TEXT, height=320, max_chars=4000)

if st.button("교열 실행", type="primary", disabled=not text.strip()):
    if not os.environ.get("UPSTAGE_API_KEY"):
        st.error(
            "UPSTAGE_API_KEY 가 필요합니다 — 사이드바에 입력하거나 Space Secrets 에 설정하세요."
        )
        st.stop()

    source = text.strip()
    pipeline = get_pipeline()
    # 입력은 실행 전에 남긴다 — 실행이 실패해도 무엇이 들어왔는지는 알 수 있게
    article_id = logbook.save_article(source)

    progress = ProgressView(runner.step_names(pipeline))
    try:
        result = runner.run_proofread(pipeline, source, model=model, on_step=progress.on_step)
    except Exception as e:  # noqa: BLE001 — 데모 표면에서는 원인 요약만 보여준다
        progress.fail()
        st.error(f"실행 실패: {type(e).__name__}: {e}")
        st.stop()
    progress.finish(result["elapsed_s"])

    logbook.save_run(
        article_id,
        pipeline_key=result["pipeline_key"],
        prompt_key=result["prompt_key"],
        model=result["model"],
        output=result["output"],
        processing_time_s=result["elapsed_s"],
    )
    if logbook.is_configured() and logbook.last_error():
        st.caption(f"기록 실패 (결과에는 영향 없음): {logbook.last_error()}")

    left, right, _n_changed = diff_panes(source, result["output"])

    c1, c2 = st.columns(2, gap="medium")
    with c1:
        st.subheader("원문")
        st.markdown(f'<div style="{PANE_STYLE}">{left}</div>', unsafe_allow_html=True)
    with c2:
        st.subheader("교열 후")
        st.markdown(f'<div style="{PANE_STYLE}">{right}</div>', unsafe_allow_html=True)