dl-organizer/src/vlm/review_tui.py

"""Textual TUI for reviewing high-risk plan operations."""

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path


from vlm.plan_review import save_review_csv
from vlm.review_display import (
    build_csv_rows,
    format_paths_for_detail,
    review_row_status_symbol,
    risk_flags_to_labels,
)
from vlm.utils import format_size

FILTER_CHOICES = ("all", "manual_review", "sample_source", "spot_check", "conflict", "duplicate")


def _row_matches_filter(row: dict[str, str], filter_key: str) -> bool:
    if filter_key == "all":
        return True
    flags = row.get("risk_flags", "")
    if filter_key in flags.split("|"):
        return True
    if filter_key == "duplicate" and row.get("duplicate_group_id"):
        return True
    return False


def _change_label(row: dict[str, str]) -> str:
    src = row.get("source_name") or Path(row.get("source_path", "")).name
    dst = row.get("dest_name") or ""
    if dst:
        return f"{src} → {dst}"
    return src


@dataclass(frozen=True)
class ReviewTUIContext:
    """Inputs for the plan review TUI."""

    rows: list[dict[str, str]]
    counters: dict[str, int]
    library_root: Path
    output_csv: Path
    plan_input: Path
    summary_text: str
    path_to_quality: dict[str, dict] = field(default_factory=dict)


try:
    from textual import on
    from textual.app import App, ComposeResult
    from textual.binding import Binding
    from textual.containers import Container, Horizontal, ScrollableContainer
    from textual.geometry import Size
    from textual.screen import ModalScreen, Screen
    from textual.widgets import DataTable, Footer, Static
except ImportError as exc:  # pragma: no cover
    TEXTUAL_IMPORT_ERROR: ImportError | None = exc
else:
    TEXTUAL_IMPORT_ERROR = None

MISSING_TEXTUAL_MESSAGE = 'Textual is not installed. Install with: uv pip install -e ".[tui]"'

if TEXTUAL_IMPORT_ERROR is None:

    def _index_from_row_key(row_key) -> int:  # noqa: ANN001 - RowKey | str
        if isinstance(row_key, str):
            return int(row_key)
        val = getattr(row_key, "value", None)
        if val is None:
            return int(row_key)
        return int(val)

    class SummaryScreen(Screen):
        """Migration summary; Enter continues, q aborts."""

        BINDINGS = [
            Binding("enter", "continue_", "继续", show=True),
            Binding("q", "quit", "退出", show=True),
        ]

        def __init__(self, ctx: ReviewTUIContext) -> None:
            super().__init__()
            self._ctx = ctx

        def compose(self) -> ComposeResult:
            stats_lines = [
                "---",
                "计划统计",
                f"  总操作: {self._ctx.counters['total_operations']}",
                f"  高危: {self._ctx.counters['high_risk_operations']}",
                f"  manual_review: {self._ctx.counters['manual_review']}",
                f"  sample_source: {self._ctx.counters['sample_source']}",
                f"  high_season: {self._ctx.counters['high_season']}",
                f"  high_episode: {self._ctx.counters['high_episode']}",
                f"  conflicts: {self._ctx.counters['conflicts']}",
                "",
                f"将要写入: {self._ctx.output_csv}",
                "",
                "高危操作写入 CSV；可选 --sample-safe 追加安全抽检行。",
                "筛选: 1全部 2需人工 3样片路径 4安全抽检 5冲突 6重复 · g 驳回整组重复",
                "",
                "Enter 进入审核 · q 退出",
            ]
            body = self._ctx.summary_text + "\n\n" + "\n".join(stats_lines)
            yield ScrollableContainer(Static(body, id="summary_body"))
            yield Footer()

        def action_continue_(self) -> None:
            self.dismiss(True)

        def action_quit(self) -> None:
            self.dismiss(False)


    class ConfirmDiscardScreen(ModalScreen[bool]):
        """Confirm discarding unsaved edits."""

        BINDINGS = [
            Binding("y", "yes", show=False),
            Binding("n", "no", show=False),
        ]

        def compose(self) -> ComposeResult:
            yield Container(
                Static("未保存的修改将丢失。放弃？ (y / n)", id="confirm_text"),
                id="confirm_box",
            )

        def action_yes(self) -> None:
            self.dismiss(True)

        def action_no(self) -> None:
            self.dismiss(False)

        DEFAULT_CSS = """
        ConfirmDiscardScreen {
            align: center middle;
        }
        #confirm_box {
            width: auto;
            height: auto;
            padding: 1 2;
            border: thick $primary;
            background: $surface;
        }
        """


    class ReviewMainScreen(Screen):
        """High-risk table + detail pane."""

        BINDINGS = [
            Binding("up", "cursor_up", show=False),
            Binding("down", "cursor_down", show=False),
            Binding("k", "cursor_up", show=False),
            Binding("j", "cursor_down", show=False),
            Binding("a", "keep_row", "保留", show=True),
            Binding("r", "reject_row", "驳回", show=True),
            Binding("u", "undo_row", "撤销", show=True),
            Binding("g", "reject_group", "驳回组", show=True),
            Binding("s", "save", "保存", show=True),
            Binding("q", "request_quit", "退出", show=True),
            Binding("1", "filter_all", show=False),
            Binding("2", "filter_manual", show=False),
            Binding("3", "filter_sample", show=False),
            Binding("4", "filter_spot_check", show=False),
            Binding("5", "filter_conflict", show=False),
            Binding("6", "filter_duplicate", show=False),
        ]

        def __init__(self, ctx: ReviewTUIContext) -> None:
            super().__init__()
            self.ctx = ctx
            self._by_index: dict[int, dict[str, str]] = {
                int(r["index"]): r for r in ctx.rows
            }
            self.initial_op_by_index: dict[int, str] = {
                int(r["index"]): r["operation_type"] for r in ctx.rows
            }
            self.op_by_index: dict[int, str] = dict(self.initial_op_by_index)
            self._filter = "all"

        def compose(self) -> ComposeResult:
            plan_s = str(self.ctx.plan_input)
            if len(plan_s) > 72:
                plan_s = plan_s[:35] + "…" + plan_s[-34:]
            hdr = (
                f"{plan_s}  ·  高危 {self.ctx.counters['high_risk_operations']}"
                f" / {self.ctx.counters['total_operations']}"
            )
            yield Static(hdr, id="header_line")
            yield Static("", id="filter_line")
            with Horizontal(id="body"):
                yield DataTable(id="review_table", cursor_type="row", zebra_stripes=True)
                with ScrollableContainer(id="detail_scroll"):
                    yield Static("", id="detail_text")
            yield Static(
                "1-6 筛选 · ↑↓ j/k · a 保留 · r 驳回 · g 驳回重复组 · u 撤销 · s 保存 · q 退出",
                id="footer_line",
            )
            yield Footer()

        DEFAULT_CSS = """
        #header_line {
            dock: top;
            padding: 0 1;
            background: $primary-darken-2;
            color: $text;
        }
        #filter_line {
            dock: top;
            padding: 0 1;
            background: $panel;
            color: $text-muted;
        }
        #footer_line {
            dock: bottom;
            padding: 0 1;
            background: $panel;
            color: $text-muted;
        }
        #body {
            layout: horizontal;
            height: 1fr;
        }
        #body.vertical-split {
            layout: vertical;
            height: 1fr;
        }
        #review_table {
            width: 1fr;
            min-height: 5;
        }
        #body.vertical-split #review_table {
            height: 40%;
        }
        #detail_scroll {
            width: 1fr;
            min-height: 5;
            border-left: solid $primary-darken-3;
            padding: 0 1;
        }
        #body.vertical-split #detail_scroll {
            border-left: none;
            border-top: solid $primary-darken-3;
            height: 1fr;
        }
        """

        def _filtered_rows(self) -> list[dict[str, str]]:
            return [
                r
                for r in self.ctx.rows
                if _row_matches_filter(r, self._filter)
            ]

        def _update_filter_line(self) -> None:
            labels = {
                "all": "全部",
                "manual_review": "需人工",
                "sample_source": "样片路径",
                "spot_check": "安全抽检",
                "conflict": "冲突",
                "duplicate": "重复",
            }
            visible = len(self._filtered_rows())
            self.query_one("#filter_line", Static).update(
                f"筛选: {labels.get(self._filter, self._filter)} · 显示 {visible}/{len(self.ctx.rows)} 条"
            )

        def _rebuild_table(self) -> None:
            table = self._table()
            table.clear(columns=True)
            table.add_column(" ", key="sym", width=3)
            table.add_column("#", key="idx", width=4)
            table.add_column("类型", key="op", width=10)
            table.add_column("风险", key="risk", width=14)
            table.add_column("变更", key="change")

            for r in self._filtered_rows():
                idx = int(r["index"])
                table.add_row(
                    self._symbol_for(idx),
                    str(idx),
                    self.op_by_index[idx],
                    risk_flags_to_labels(r["risk_flags"]),
                    _change_label(r),
                    key=str(idx),
                )
            self._update_filter_line()
            if table.row_count > 0:
                table.focus()
                self._refresh_detail(_index_from_row_key(table.ordered_rows[0].key))
            else:
                self.query_one("#detail_text", Static).update(
                    "当前筛选无条目。按 1 显示全部，或 s 保存 CSV。"
                )

        def on_mount(self) -> None:
            self._rebuild_table()
            self._apply_body_layout(self.app.size)

        def _table(self) -> DataTable:
            return self.query_one("#review_table", DataTable)

        def _symbol_for(self, index: int) -> str:
            return review_row_status_symbol(
                self.op_by_index,
                self.initial_op_by_index,
                index,
            )

        def _apply_body_layout(self, app_size: Size) -> None:
            body = self.query_one("#body", Horizontal)
            if app_size.width < 100:
                body.add_class("vertical-split")
            else:
                body.remove_class("vertical-split")

        def on_resize(self, event) -> None:  # noqa: ANN001 - textual Resize
            self._apply_body_layout(self.app.size)

        def _current_index(self) -> int | None:
            table = self._table()
            if table.row_count == 0:
                return None
            row_index = table.cursor_coordinate.row
            row = table.ordered_rows[row_index]
            return _index_from_row_key(row.key)

        @on(DataTable.RowHighlighted)  # type: ignore[misc]
        def on_row_highlighted(self, event: DataTable.RowHighlighted) -> None:
            if event.data_table.id != "review_table":
                return
            idx = _index_from_row_key(event.row_key)
            self._refresh_detail(idx)

        def _format_quality_block(self, index: int) -> str:
            r = self._by_index[index]
            gid = r.get("duplicate_group_id", "")
            if not gid:
                return ""

            lines = [f"重复组: {gid}"]
            group_paths = [
                other
                for other in self.ctx.rows
                if other.get("duplicate_group_id") == gid
            ]
            for other in group_paths:
                oidx = int(other["index"])
                path = other.get("source_path", "")
                qc = self.ctx.path_to_quality.get(path, {})
                if qc:
                    res = qc.get("resolution", "?")
                    size = format_size(int(qc.get("size_bytes", 0) or 0))
                    mark = "★" if self.op_by_index.get(oidx) != "no-op" else " "
                    lines.append(f"  {mark} [{oidx}] {Path(path).name}: {res} {size}")
                else:
                    hint = other.get("quality_hint", "")
                    mark = "★" if self.op_by_index.get(oidx) != "no-op" else " "
                    lines.append(f"  {mark} [{oidx}] {Path(path).name}: {hint or '(no metadata)'}")
            return "\n".join(lines)

        def _refresh_detail(self, index: int) -> None:
            r = self._by_index[index]
            op = self.op_by_index[index]
            paths = format_paths_for_detail(
                r["source_path"],
                r.get("destination_path", ""),
                self.ctx.library_root,
            )
            risk_cn = risk_flags_to_labels(r["risk_flags"], max_len=120)
            title = r.get("title", "")
            summary = (
                f"#{index} · {op} · {_change_label(r)}"
                + (f" · {title}" if title else "")
                + (f" · {risk_cn}" if risk_cn else "")
            )
            quality_block = self._format_quality_block(index)
            text = (
                f"{summary}\n\n"
                f"变更\n{paths}\n\n"
                f"依据\n{r['reason']}\n\n"
                f"标记\n{r['risk_flags']}"
            )
            if quality_block:
                text += f"\n\n画质对比\n{quality_block}"
            self.query_one("#detail_text", Static).update(text)

        def _refresh_row_cells(self, index: int) -> None:
            table = self._table()
            key = str(index)
            if key not in [str(_index_from_row_key(r.key)) for r in table.ordered_rows]:
                return
            table.update_cell(key, "sym", self._symbol_for(index))
            table.update_cell(key, "op", self.op_by_index[index])

        def _update_dirty_header(self) -> None:
            dirty = self._is_dirty()
            hdr = self.query_one("#header_line", Static)
            plan_s = str(self.ctx.plan_input)
            if len(plan_s) > 72:
                plan_s = plan_s[:35] + "…" + plan_s[-34:]
            star = " *" if dirty else ""
            hdr.update(
                f"{plan_s}{star}  ·  高危 {self.ctx.counters['high_risk_operations']}"
                f" / {self.ctx.counters['total_operations']}"
            )

        def _is_dirty(self) -> bool:
            return self.op_by_index != self.initial_op_by_index

        def action_cursor_up(self) -> None:
            if self._table().row_count:
                self._table().action_cursor_up()

        def action_cursor_down(self) -> None:
            if self._table().row_count:
                self._table().action_cursor_down()

        def action_keep_row(self) -> None:
            idx = self._current_index()
            if idx is None:
                return
            self.op_by_index[idx] = self.initial_op_by_index[idx]
            self._refresh_row_cells(idx)
            self._refresh_detail(idx)
            self._update_dirty_header()

        def action_reject_row(self) -> None:
            idx = self._current_index()
            if idx is None:
                return
            self.op_by_index[idx] = "no-op"
            self._refresh_row_cells(idx)
            self._refresh_detail(idx)
            self._update_dirty_header()

        def action_reject_group(self) -> None:
            idx = self._current_index()
            if idx is None:
                return
            gid = self._by_index[idx].get("duplicate_group_id", "")
            if not gid:
                self.action_reject_row()
                return
            for r in self.ctx.rows:
                if r.get("duplicate_group_id") == gid:
                    self.op_by_index[int(r["index"])] = "no-op"
            self._rebuild_table()
            self._update_dirty_header()

        def action_undo_row(self) -> None:
            self.action_keep_row()

        def action_filter_all(self) -> None:
            self._filter = "all"
            self._rebuild_table()

        def action_filter_manual(self) -> None:
            self._filter = "manual_review"
            self._rebuild_table()

        def action_filter_sample(self) -> None:
            self._filter = "sample_source"
            self._rebuild_table()

        def action_filter_spot_check(self) -> None:
            self._filter = "spot_check"
            self._rebuild_table()

        def action_filter_conflict(self) -> None:
            self._filter = "conflict"
            self._rebuild_table()

        def action_filter_duplicate(self) -> None:
            self._filter = "duplicate"
            self._rebuild_table()

        def action_save(self) -> None:
            out_rows = build_csv_rows(self.ctx.rows, self.op_by_index)
            save_review_csv(out_rows, self.ctx.output_csv)
            self.dismiss("saved")

        def action_request_quit(self) -> None:
            if not self._is_dirty():
                self.dismiss("aborted")
                return

            def after_confirm(confirmed: bool | None) -> None:
                if confirmed:
                    self.dismiss("aborted")

            self.app.push_screen(ConfirmDiscardScreen(), callback=after_confirm)


    class PlanReviewApp(App):
        """Application shell: summary screen then review screen."""

        def __init__(self, ctx: ReviewTUIContext) -> None:
            super().__init__()
            self.ctx = ctx

        def on_mount(self) -> None:
            self.push_screen(SummaryScreen(self.ctx), self._after_summary)

        def _after_summary(self, result: bool | None) -> None:
            if not result:
                self.exit(return_code=1)
                return
            self.push_screen(ReviewMainScreen(self.ctx), self._after_main)

        def _after_main(self, result: str | None) -> None:
            if result == "saved":
                self.exit(return_code=0)
            else:
                self.exit(return_code=1)


    def run_plan_review_tui(ctx: ReviewTUIContext) -> int:
        """Block until the user finishes the TUI. Returns process exit code."""
        app = PlanReviewApp(ctx)
        app.run()
        code = app.return_code
        return 0 if code is None else code

else:

    def run_plan_review_tui(ctx: ReviewTUIContext) -> int:
        """Raise a friendly error when the optional Textual dependency is missing."""
        raise RuntimeError(MISSING_TEXTUAL_MESSAGE) from TEXTUAL_IMPORT_ERROR