page_files/Review_Queue.py

"""Review Queue — flagged (status != 'ok') rows across the material tables."""
import pandas as pd
import streamlit as st

from agent import agdb
from agent import ui_theme as T
from agent.ui_common import admin_gate, status_banner

_TITLE = "Review Queue"
_SUB = ("Property rows the extractor flagged — hidden from search until a "
        "human verifies them against the source PDF and promotes them.")

# Warn tint for flag_reason cells (same look as T.status_cell_style('flag…')).
_WARN_CELL = "background-color: rgba(250,178,25,.22); color: #7a5200; font-weight: 600;"


def _warn_tint(v) -> str:
    """Styler CSS for a flag_reason cell: warn tint when non-empty (NaN-proof)."""
    try:
        if v is None or (not isinstance(v, str) and pd.isna(v)):
            return ""
    except Exception:
        return ""
    return _WARN_CELL if str(v).strip() else ""


def _num_fmt(v) -> str:
    """Styler display string for a numeric cell that mirrors Streamlit's default
    number rendering (≤4 decimals, trailing zeros trimmed, no thousands
    separator) — a Styler otherwise hands the grid pandas' 6-decimal strings
    ('535.150000', '135000000000000.000000'). NaN/None-proof."""
    try:
        if v is None or pd.isna(v):
            return ""
        f = float(v)
    except Exception:
        return "" if v is None else str(v)
    if f != f or f in (float("inf"), float("-inf")):
        return ""
    if f == 0 or abs(f) >= 1e-4:
        s = f"{f:.4f}"
    else:  # tiny values: as many decimals as the exponent (numbro default)
        s = f"{f:.{abs(int(f'{f:e}'.split('e')[1]))}f}"
    if "." in s:
        s = s.rstrip("0").rstrip(".")
    return "0" if s == "-0" else s


# .aim-card-title is uppercase/letter-spaced; anything placed in right_html
# inherits that, so wrap chips/pills in a span that resets the text style.
_RIGHT_WRAP = ('<span style="display:inline-flex;gap:6px;flex-wrap:wrap;align-items:center;'
               'text-transform:none;letter-spacing:0">')


def _right_html(inner: str) -> str:
    """Wrap pill/chip HTML for card_title(right_html=...) without inheriting
    the card title's uppercase/letter-spacing."""
    return _RIGHT_WRAP + inner + "</span>"


def _chip_row_html(items) -> str:
    """Inline chip row for card_title right_html. items: (text, kind) tuples."""
    return _right_html("".join(T.chip_html(t, k) for t, k in items))


def _nunique(s) -> int:
    """NaN-proof distinct count of a Series (0 for missing/empty)."""
    try:
        return int(s.dropna().astype(str).nunique())
    except Exception:
        return 0


# Header slot: rendered immediately (so a DB-error page still has a title) and
# re-rendered with the queue pill once the queue is known.
_hdr = st.empty()


def _header(pill=None) -> None:
    with _hdr.container():
        T.page_header(_TITLE, _SUB, right=pill)


_header()
status_banner()
admin = admin_gate()

TABLES = ("Polymers", "Fibers", "Composites_materials")
# Columns the queue wants, in display order, with the SQL type a missing one
# is NULL-cast to. The team's material tables follow the 34-column InDeS
# schema (no id column; origin/figure_id only after the migration), so the
# SELECT is built per table from information_schema: a column that is not
# there becomes a typed NULL instead of an UndefinedColumn error. The old
# hard-coded list began with `id`, failed on every table, and the swallowed
# error rendered a false "queue empty" from the first deploy (13 Aug) on.
_WANT = (("source_pdf", "text"), ("page", "integer"), ("status", "text"),
         ("flag_reason", "text"), ("material_name", "text"),
         ("material_key", "text"), ("section", "text"),
         ("property_name", "text"), ("value_raw", "text"), ("unit", "text"),
         ("unit_canonical", "text"), ("value_si", "double precision"),
         ("test_condition", "text"), ("source_quote", "text"),
         ("origin", "text"), ("figure_id", "text"),
         # processing route of the specimen (prompt 2.1; NULL on older rows)
         ("process_type", "text"), ("process_conditions", "text"),
         ("process_status", "text"))


def _queue_select(table: str, have: dict) -> str:
    parts = []
    for col, typ in _WANT:
        if col == "origin":
            parts.append("COALESCE(origin, 'text') AS origin" if col in have
                         else "'text'::text AS origin")
        elif col in have:
            parts.append(col)
        else:
            parts.append(f"NULL::{typ} AS {col}")
    order = "extracted_at DESC NULLS LAST" if "extracted_at" in have else "source_pdf"
    return (f"SELECT '{table}' AS table_name, {', '.join(parts)} FROM \"{table}\" "
            f"WHERE COALESCE(status, 'ok') <> 'ok' ORDER BY {order} LIMIT 500")


frames, read_errors, _have = [], [], {}
conn = agdb.connect()
try:
    import pg_mirror
    for t in TABLES:
        try:
            have = pg_mirror.existing_columns(conn, t)
            _have[t] = have
            if not have:
                read_errors.append(f"{t}: table not found")
                continue
            if "status" not in have:
                # Pre-migration table: nothing can be flagged yet.
                continue
            frames.append(agdb.fetch_df(conn, _queue_select(t, have)))
        except Exception as exc:
            conn.rollback()
            read_errors.append(f"{t}: {type(exc).__name__}: {exc}")
finally:
    conn.close()

queue = pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()
if read_errors:
    # An unreadable queue is not an empty queue: say so instead of the
    # green check mark.
    _header(T.pill_html("error", "queue unreadable"))
    st.error("The review queue could not be read from every table:\n\n- "
             + "\n- ".join(read_errors))
    if queue.empty:
        st.stop()
if queue.empty:
    _header(T.pill_html("ok", "queue empty"))
    T.empty_state("✅", "Review queue is empty",
                  "Every published row is status 'ok'. Rows the extractor flags "
                  "(implausible values, ungrounded quotes) will appear here for "
                  "human review before they enter search results.")
    st.stop()

_header(T.pill_html("warn", f"{len(queue):,} flagged")
        + (T.pill_html("error", f"{len(read_errors)} table(s) unreadable") if read_errors else ""))

# KPI tiles live above the filter card but describe the filtered view, so the
# slot is reserved here and filled once the filters are known.
_kpi_slot = st.container()

# --- flagged rows: filters + table + download -------------------------------
_status_counts = queue["status"].dropna().astype(str).value_counts()
_status_chips = _chip_row_html(
    [(f"{s} · {int(c):,}", "warn") for s, c in _status_counts.items()])

with st.container(border=True):
    T.card_title("Flagged rows",
                 sub="status ≠ ok across Polymers · Fibers · Composites_materials",
                 right_html=_status_chips)
    f1, f2, f3 = st.columns([2, 2, 1])
    with f1:
        statuses = sorted(queue["status"].dropna().unique())
        pick = st.multiselect("Status", statuses, default=statuses, key="rq_status")
    with f2:
        pdfs = ["(all)"] + sorted(queue["source_pdf"].dropna().unique())
        pdf_pick = st.selectbox("Source PDF", pdfs, key="rq_pdf")
    with f3:
        origin_pick = st.selectbox("Origin", ["(all)", "text", "figure"], key="rq_origin",
                                   help="figure = value read off a plot/table image "
                                        "by the figure-mining stage (an estimate)")

    view = queue[queue["status"].isin(pick)]
    if pdf_pick != "(all)":
        view = view[view["source_pdf"] == pdf_pick]
    if origin_pick != "(all)" and "origin" in view.columns:
        view = view[view["origin"] == origin_pick]

    # -- KPIs (filled into the slot above the card) ---------------------------
    _n_view, _n_queue = len(view), len(queue)
    _mat = view["material_key"] if "material_key" in view.columns else pd.Series(dtype=object)
    if "material_name" in view.columns:
        _mat = _mat.astype(object).where(_mat.notna(), view["material_name"].astype(object))
    _tables = sorted(view["table_name"].dropna().astype(str).unique()) if "table_name" in view.columns else []
    with _kpi_slot:
        T.kpi_row([
            {"label": "Flagged rows", "value": _n_view,
             "delta": (f"of {_n_queue:,} in queue" if _n_view != _n_queue else "whole queue in view"),
             "delta_kind": "warn" if _n_view else "flat",
             "help": "Rows with status ≠ 'ok' matching the current filters (max 500 per table)."},
            {"label": "Source PDFs", "value": _nunique(view["source_pdf"]),
             "delta": "awaiting a human check", "delta_kind": "flat",
             "help": "Distinct source PDFs among the flagged rows in view."},
            {"label": "Materials", "value": _nunique(_mat),
             "delta": "with flagged rows", "delta_kind": "flat",
             "help": "Distinct materials (material_key, falling back to material_name) in view."},
            {"label": "Tables", "value": len(_tables),
             "delta": " · ".join(_tables) if _tables else "—", "delta_kind": "flat",
             "help": "Material tables that contain flagged rows in view."},
        ])

    # -- table ----------------------------------------------------------------
    show = view.drop(columns=[c for c in ("id",) if c in view.columns])
    _ORDER = ["status", "origin", "flag_reason", "material_name", "section",
              "property_name", "value_raw", "unit", "value_si", "unit_canonical",
              "test_condition", "process_type", "process_conditions", "process_status",
              "source_pdf", "page", "table_name", "material_key",
              "source_quote"]
    order = [c for c in _ORDER if c in show.columns] + [c for c in show.columns if c not in _ORDER]
    styled = show.style
    if "status" in show.columns:
        styled = styled.map(T.status_cell_style, subset=["status"])
    if "flag_reason" in show.columns:
        styled = styled.map(_warn_tint, subset=["flag_reason"])
    # Numeric columns: keep the same digits the plain-DataFrame grid showed
    # (the Styler's display strings replace Streamlit's number formatting).
    _num_cols = [c for c in ("value_si", "page") if c in show.columns]
    if _num_cols:
        styled = styled.format(_num_fmt, subset=_num_cols, na_rep="")
    # Fit short queues to their content; long ones keep the previous fixed 480.
    _h = "content" if 0 < len(show) <= 12 else (480 if len(show) > 12 else 120)
    st.dataframe(
        styled, hide_index=True, width="stretch", height=_h,
        column_order=order,
        column_config={
            "status": st.column_config.Column("Status", width=140),
            "origin": st.column_config.Column(
                "Origin", width=70,
                help="text = grounded in the PDF text · figure = estimate read "
                     "off a plot/table image"),
            "figure_id": st.column_config.Column("Figure", width=110),
            "flag_reason": st.column_config.Column("Flag reason", width=220),
            "material_name": st.column_config.Column("Material", width=180),
            "section": st.column_config.Column("Section", width=100),
            "property_name": st.column_config.Column("Property", width=170),
            "value_raw": st.column_config.Column("Value (raw)", width=120),
            "unit": st.column_config.Column("Unit", width=70),
            "value_si": st.column_config.Column("Value (SI)", width=120, alignment="right"),
            "unit_canonical": st.column_config.Column("SI unit", width=80),
            "test_condition": st.column_config.Column("Test condition", width=140),
            "process_type": st.column_config.Column(
                "Process", width=150,
                help="How the tested specimen was made, as the source states it"),
            "process_conditions": st.column_config.Column("Process conditions", width=220),
            "process_status": st.column_config.Column(
                "Process check", width=120,
                help="grounded = the sentence naming the route and every number "
                     "in its conditions are in the PDF text"),
            "source_pdf": st.column_config.Column("Source PDF", width=170),
            "page": st.column_config.Column("Page", width=60, alignment="right"),
            "table_name": st.column_config.Column("Table", width=160),
            "material_key": st.column_config.Column("Material key", width=150),
            "source_quote": st.column_config.Column("Source quote", width=380),
        })
    if show.empty:
        st.caption("No rows match the current filters — widen the status or PDF filter.")

    st.download_button("Download review_queue.csv",
                       view.to_csv(index=False).encode(),
                       "review_queue.csv", "text/csv", key="rq_dl",
                       icon=":material/download:")

# --- figure crops behind the rows in view -----------------------------------
_fig_ids = []
if "figure_id" in view.columns:
    _fig_ids = sorted({str(x) for x in view["figure_id"].dropna() if str(x).strip()})
if _fig_ids:
    _GALLERY_CAP = 24
    figs_df = pd.DataFrame()
    conn = agdb.connect()
    try:
        figs_df = agdb.fetch_df(
            conn,
            "SELECT figure_id, source_pdf, page, caption, figure_kind, "
            "n_values, mining_status, image_bytes FROM figures "
            "WHERE figure_id = ANY(%s) ORDER BY source_pdf, page LIMIT %s",
            (_fig_ids, _GALLERY_CAP))
    except Exception:
        conn.rollback()
    finally:
        conn.close()
    with st.container(border=True):
        T.card_title(
            "Figure crops behind these rows",
            sub="verify the reading against the actual plot before promoting",
            right_html=_chip_row_html([(f"{len(_fig_ids):,} figure"
                                        + ("s" if len(_fig_ids) != 1 else ""), "muted")]))
        if figs_df.empty:
            st.html('<div class="aim-note">No stored crops for the rows in view '
                    '(figures harvested before crop storage, or bytes over the '
                    'size cap).</div>')
        else:
            _n_rows_by_fig = (view.groupby("figure_id").size()
                              if "figure_id" in view.columns else {})
            _cols = st.columns(3)
            for i, (_, f) in enumerate(figs_df.iterrows()):
                with _cols[i % 3]:
                    _cap = str(f.get("caption") or "").strip()
                    _n = int(_n_rows_by_fig.get(f["figure_id"], 0))
                    _head = (f"{f.get('source_pdf') or '?'} · p.{f.get('page')} · "
                             f"{f.get('figure_kind') or 'unclassified'} · "
                             f"{_n} row{'s' if _n != 1 else ''} in view")
                    _img = f.get("image_bytes")
                    if _img is not None:
                        try:
                            st.image(bytes(_img), width="stretch")
                        except Exception:
                            st.html('<div class="aim-note">crop unavailable</div>')
                    else:
                        st.html('<div class="aim-note">no stored crop</div>')
                    st.caption(_head + ((" — " + _cap[:160] + ("…" if len(_cap) > 160 else ""))
                                        if _cap else ""))
            if len(_fig_ids) > _GALLERY_CAP:
                st.caption(f"Showing the first {_GALLERY_CAP} of {len(_fig_ids)} figures — "
                           f"narrow the filters above to see the rest.")

# --- promote (admin) --------------------------------------------------------
with st.container(border=True):
    T.card_title("Promote corrected rows", sub="admin action",
                 right_html=_right_html(T.pill_html("ok", "admin unlocked") if admin
                                        else T.pill_html("off", "locked")))
    st.caption("After a human verifies a flagged value is correct, promote it to "
               "status 'ok' so it appears in search results. Same semantics as "
               "`batch_ingest.py --promote`.")
    if admin:
        p1, p2 = st.columns([3, 2])
        with p1:
            sel_pdf = st.selectbox("Promote all flagged rows from PDF",
                                   ["(choose)"] + sorted(view["source_pdf"].dropna().unique()),
                                   key="rq_promote_pdf")
        with p2:
            if sel_pdf != "(choose)":
                _n_sel = int((queue["source_pdf"] == sel_pdf).sum())
                st.html(f'<div class="aim-note" style="margin-top:2.1rem">'
                        f'{_n_sel:,} flagged row{"s" if _n_sel != 1 else ""} '
                        f'currently listed from <b>{T.esc(sel_pdf)}</b></div>')
        sure = st.checkbox("I reviewed these values against the source PDF",
                           key="rq_promote_sure")
        inc_figs = st.checkbox(
            "Also promote figure-derived estimate rows",
            key="rq_promote_figs",
            help="Figure rows are values read off plots/table images — estimates "
                 "by design. Leave this off unless you checked each reading "
                 "against its crop above.")
        if st.button("✅ Promote", disabled=(sel_pdf == "(choose)" or not sure),
                     key="rq_promote_btn", type="primary"):
            conn = agdb.connect()
            try:
                n = 0
                with conn.cursor() as cur:
                    for t in TABLES:
                        # Only tables that have the origin column can tell a
                        # figure row from a text row; before the migration
                        # every flagged row is a text row.
                        _origin_sql = ("" if inc_figs or "origin" not in _have.get(t, {})
                                       else " AND COALESCE(origin, 'text') = 'text'")
                        cur.execute(
                            f'UPDATE "{t}" SET status = \'ok\', flag_reason = \'promoted\' '
                            f"WHERE source_pdf = %s AND COALESCE(status, 'ok') <> 'ok'"
                            + _origin_sql,
                            (sel_pdf,))
                        n += cur.rowcount
                conn.commit()
                agdb.log_event(conn, f"promoted {n} rows from {sel_pdf}"
                               + (" (incl. figure estimates)" if inc_figs
                                  else " (text rows only)"),
                               node="review")
            finally:
                conn.close()
            st.success(f"Promoted {n} rows.")
            st.toast(f"Promoted {n} rows")
            st.rerun()
    else:
        st.html('<div class="aim-note">🔒 Unlock admin in the sidebar to promote rows.</div>')