"""Database — material-first browser over everything the agent's DB holds.

Materials tab: one expandable entry per material (across Polymers, Fibers and
Composites_materials), broken down into its properties — value, canonical SI,
status, source and DOI. Raw tables tab: the classic row-level browser with
filters, pagination and CSV export.

Columns are discovered dynamically from information_schema, so the page works
unchanged against the fresh `aim_agent` clone and against the team's
`AIMDatabase` if DB_NAME is ever pointed back at it.
"""
import pandas as pd
import streamlit as st

from agent import ui_theme as T
from agent.ui_common import status_banner, df_or_empty

T.page_header("Database",
              "Every material the agent has ingested, expandable into its "
              "properties — plus a row-level browser over the raw tables.")
status_banner()

MATERIAL_TABLES = ("Polymers", "Fibers", "Composites_materials")
BROWSABLE = MATERIAL_TABLES + ("sources", "figures")
TABLE_ICON = {"Polymers": "🧪", "Fibers": "🧵", "Composites_materials": "🧱"}

# Columns the union view exposes (aliased to NULL where a table lacks one).
UNION_COLS = [
    ("material_key", "text"), ("material_abbreviation", "text"),
    ("trade_grade", "text"), ("manufacturer", "text"), ("matrix", "text"),
    ("fiber", "text"), ("material_class", "text"), ("section", "text"),
    ("property_name", "text"), ("test_condition", "text"), ("value", "text"),
    ("unit", "text"), ("value_raw", "text"), ("unit_canonical", "text"),
    ("value_si", "double precision"), ("qualifier", "text"),
    ("status", "text"), ("source_pdf", "text"), ("page", "integer"),
    ("origin", "text"), ("figure_id", "text"),
    # processing route of the specimen (prompt 2.1; NULL on older rows)
    ("process_type", "text"), ("process_name", "text"),
    ("process_conditions", "text"), ("process_status", "text"),
    ("extracted_at", "text"),
]
PREFERRED_COLS = [
    "material_abbreviation", "material_key", "material_class", "trade_grade",
    "manufacturer", "matrix", "fiber", "section", "property_name",
    "test_condition", "process_type", "process_conditions", "value", "value_raw",
    "unit", "unit_canonical", "value_si", "qualifier", "status", "process_status",
    "origin", "source_pdf", "page", "doi_url",
    "extracted_at",
]
SEARCH_COLS = [
    "material_abbreviation", "material_key", "trade_grade", "manufacturer",
    "matrix", "fiber", "section", "property_name", "test_condition",
    "process_type", "process_name", "process_conditions",
    "source_pdf", "english", "value", "value_raw", "doi_url",
    "pdf_filename", "title", "doi", "url",
    "caption", "figure_kind", "figure_id",
]

_q = lambda ident: '"' + ident.replace('"', "") + '"'

# Markdown-special characters that must be backslash-escaped when a DB string
# is placed in a widget label (labels are markdown, not HTML — a material
# called ":red[x]" or "*a*" must render literally).
_MD_SPECIAL = set("\\`*_{}[]()#+!|<>~:$^")


def _md(x) -> str:
    """NaN/None-proof markdown-escaped string for widget labels."""
    s = "" if x is None or (isinstance(x, float) and pd.isna(x)) else str(x)
    return "".join(("\\" + ch) if ch in _MD_SPECIAL else ch for ch in s)


def _plural(n: int, word: str, plural: str = "") -> str:
    """'1 source' / '3 sources' / '2 properties'."""
    n = int(n)
    return f"{n:,} {word if n == 1 else (plural or word + 's')}"


def _count_line(n: int, noun: str, page: int, pages: int) -> None:
    """'4 materials match · page 1 of 1' as an inline count line."""
    st.html(f'<div class="aim-inline-count" style="padding-bottom:9px"><b>{int(n):,}</b> '
            f'{noun if int(n) == 1 else noun + "s"} match · '
            f'page {int(page):,} of {int(pages):,}</div>')


# Property-table columns: fixed sensible widths (values stay exactly as built).
PROP_COL_CONFIG = {
    "Section": st.column_config.TextColumn("Section", width=105),
    "Property": st.column_config.TextColumn("Property", width=215),
    "Condition": st.column_config.TextColumn("Condition", width=100),
    "Process": st.column_config.TextColumn(
        "Process", width=250,
        help="How the tested specimen was made (processing route, as the "
             "source states it). ⚠ = the sentence naming the route, or a "
             "number in its conditions, was not found in the PDF text."),
    "Process conditions": st.column_config.TextColumn(
        "Process conditions", width=230,
        help="Processing parameters as printed in the source"),
    "Value": st.column_config.TextColumn("Value", width=145),
    "SI": st.column_config.TextColumn("SI", width=125,
                                      help="Canonical SI value (when known)"),
    "": st.column_config.TextColumn("Status", width=70, alignment="center",
                                    help="✓ verified · 🚩 flagged (Review Queue) · "
                                         "📈 figure-derived estimate (quarantined)"),
    "Source": st.column_config.TextColumn("Source", width=205,
                                          help="Source PDF · page"),
    "DOI": st.column_config.LinkColumn("DOI", display_text=r"https://doi\.org/(.*)"),
}


@st.cache_data(ttl=60)
def columns_of(table: str) -> list[str]:
    df = df_or_empty(
        "SELECT column_name FROM information_schema.columns "
        "WHERE table_schema = 'public' AND table_name = %s "
        "ORDER BY ordinal_position", (table,))
    return df["column_name"].tolist() if not df.empty else []


def union_cte() -> str:
    """WITH u AS (...) — all material tables unified, mat_key computed."""
    selects = []
    for t in MATERIAL_TABLES:
        have = set(columns_of(t))
        if not have:
            continue
        cols = ", ".join(
            _q(c) if c in have else f"NULL::{typ} AS {_q(c)}"
            for c, typ in UNION_COLS)
        selects.append(f"SELECT '{t}' AS tbl, {cols} FROM {_q(t)}")
    if not selects:
        return ""
    return ("WITH raw AS (" + " UNION ALL ".join(selects) + "), "
            "u AS (SELECT *, COALESCE(NULLIF(material_key, ''), "
            "NULLIF(material_abbreviation, ''), NULLIF(trade_grade, ''), "
            "'(unlabeled)') AS mat_key FROM raw)")


CTE = union_cte()

tab_mat, tab_raw = st.tabs(["🧪 Materials", "📋 Raw tables"])

# ===========================================================================
# MATERIALS — one expandable entry per material, properties inside
# ===========================================================================
with tab_mat:
    if not CTE:
        st.error("No material tables found in this database.")
        st.stop()

    totals = df_or_empty(
        f"{CTE} SELECT count(DISTINCT (tbl, mat_key)) AS materials, "
        f"count(*) AS props, "
        f"count(*) FILTER (WHERE COALESCE(status,'ok')='ok') AS verified, "
        f"count(DISTINCT source_pdf) FILTER (WHERE source_pdf IS NOT NULL) AS docs, "
        f"count(*) FILTER (WHERE COALESCE(process_type,'') <> '') AS with_process "
        f"FROM u")
    n_mat = 0
    if not totals.empty:
        n_mat = int(totals["materials"].iloc[0])
        n_props = int(totals["props"].iloc[0])
        n_verified = int(totals["verified"].iloc[0])
        n_flagged = max(0, n_props - n_verified)
        T.kpi_row([
            {"label": "Materials", "value": n_mat,
             "help": "Distinct materials across Polymers, Fibers and Composites"},
            {"label": "Property rows", "value": n_props},
            {"label": "Verified", "value": n_verified,
             "delta": (f"{n_flagged:,} flagged for review" if n_flagged else
                       ("all rows verified" if n_props else None)),
             "delta_kind": "warn" if n_flagged else "up",
             "help": "Grounded in the PDF, unit-sane and plausible"},
            {"label": "Source documents", "value": int(totals["docs"].iloc[0])},
            {"label": "With processing route",
             "value": int(totals["with_process"].iloc[0]),
             "delta": (f"{100 * int(totals['with_process'].iloc[0]) / n_props:.0f}% of rows"
                       if n_props else None),
             "delta_kind": "flat",
             "help": "Rows that carry the process type and conditions of the "
                     "specimen. Extracted since prompt 2.1; a row stays empty "
                     "when its source does not say how the material was made."},
        ])
    else:
        T.kpi_row([{"label": "Materials", "value": "—"},
                   {"label": "Property rows", "value": "—"},
                   {"label": "Verified", "value": "—"},
                   {"label": "Source documents", "value": "—"},
                   {"label": "With processing route", "value": "—"}])

    with st.container(border=True):
        T.card_title("Find materials", "search, filter and sort the material list")
        f1, f2, f3, f5, f4 = st.columns([3, 2, 2, 2, 2])
        m_search = f1.text_input(
            "Search materials", key="mat_search",
            placeholder="name, key, grade, manufacturer, matrix, fiber…")
        m_class = f2.selectbox("Type", ["(all)"] + list(MATERIAL_TABLES),
                               key="mat_class")
        props_all = df_or_empty(
            f"{CTE} SELECT property_name, count(*) AS n FROM u "
            f"WHERE property_name IS NOT NULL AND property_name <> '' "
            f"GROUP BY 1 ORDER BY n DESC, 1 LIMIT 300")
        m_prop = f3.selectbox(
            "Has property",
            ["(any)"] + (props_all["property_name"].tolist() if not props_all.empty else []),
            key="mat_prop")
        procs_all = df_or_empty(
            f"{CTE} SELECT process_type, count(*) AS n FROM u "
            f"WHERE COALESCE(process_type, '') <> '' "
            f"GROUP BY 1 ORDER BY n DESC, 1 LIMIT 50")
        m_proc = f5.selectbox(
            "Process",
            ["(any)"] + (procs_all["process_type"].tolist() if not procs_all.empty else []),
            key="mat_proc",
            help="Materials with at least one property measured on a "
                 "specimen made by this processing route")
        m_sort = f4.selectbox("Sort by", ["Most properties", "Name A→Z",
                                          "Recently updated"], key="mat_sort")

        where, params = [], []
        if m_class != "(all)":
            where.append("tbl = %s")
            params.append(m_class)
        if m_search:
            name_cols = ["mat_key", "material_abbreviation", "trade_grade",
                         "manufacturer", "matrix", "fiber"]
            where.append("(" + " OR ".join(
                f"COALESCE({c}, '') ILIKE %s" for c in name_cols) + ")")
            params.extend([f"%{m_search}%"] * len(name_cols))
        if m_prop != "(any)":
            where.append("(tbl, mat_key) IN "
                         "(SELECT tbl, mat_key FROM u WHERE property_name = %s)")
            params.append(m_prop)
        if m_proc != "(any)":
            where.append("(tbl, mat_key) IN "
                         "(SELECT tbl, mat_key FROM u WHERE process_type = %s)")
            params.append(m_proc)
        where_sql = (" WHERE " + " AND ".join(where)) if where else ""

        order_sql = {"Most properties": "props DESC, display_name ASC",
                     "Name A→Z": "display_name ASC",
                     "Recently updated": "updated DESC NULLS LAST"}[m_sort]

        mat_count = df_or_empty(
            f"{CTE} SELECT count(DISTINCT (tbl, mat_key)) AS n FROM u{where_sql}",
            tuple(params))
        n_materials = int(mat_count["n"].iloc[0]) if not mat_count.empty else 0

        p1, p2, p3 = st.columns([1, 1, 4], vertical_alignment="bottom")
        m_psize = p1.selectbox("Materials / page", [10, 20, 50], index=1,
                               key="mat_psize")
        m_pages = max(1, -(-n_materials // m_psize))
        m_page = p2.number_input("Page", 1, m_pages, 1, key="mat_page")
        with p3:
            _count_line(n_materials, "material", m_page, m_pages)

    mats = df_or_empty(
        f"{CTE} SELECT tbl, mat_key, "
        f"  min(COALESCE(NULLIF(material_abbreviation,''), NULLIF(trade_grade,''), mat_key)) AS display_name, "
        f"  string_agg(DISTINCT NULLIF(material_class,''), ', ') AS mclass, "
        f"  string_agg(DISTINCT NULLIF(trade_grade,''), ', ') AS grades, "
        f"  string_agg(DISTINCT NULLIF(manufacturer,''), ', ') AS makers, "
        f"  string_agg(DISTINCT NULLIF(matrix,''), ', ') AS matrices, "
        f"  string_agg(DISTINCT NULLIF(fiber,''), ', ') AS fibers, "
        f"  string_agg(DISTINCT NULLIF(process_type,''), ', ') AS processes, "
        f"  count(*) AS props, count(DISTINCT property_name) AS distinct_props, "
        f"  count(*) FILTER (WHERE COALESCE(status,'ok')='ok') AS verified, "
        f"  count(DISTINCT source_pdf) FILTER (WHERE source_pdf IS NOT NULL) AS docs, "
        f"  max(extracted_at) AS updated "
        f"FROM u{where_sql} GROUP BY tbl, mat_key "
        f"ORDER BY {order_sql} LIMIT %s OFFSET %s",
        tuple(params) + (m_psize, (m_page - 1) * m_psize))

    if mats.empty:
        if n_mat == 0:
            T.empty_state("🗄️", "The database is empty so far",
                          "Every material the agent ingests will appear here, "
                          "expandable into its properties, values, SI "
                          "conversions, sources and DOIs.")
        else:
            T.empty_state("🔎", "No materials match the current filters",
                          "Try a shorter search term, another type, or clear "
                          "the property filter.")
    else:
        # ONE query for every property row on this page, split in pandas.
        pair_sql = " OR ".join(["(tbl = %s AND mat_key = %s)"] * len(mats))
        pair_params = [x for _, r in mats.iterrows() for x in (r["tbl"], r["mat_key"])]
        detail = df_or_empty(
            f"{CTE} SELECT * FROM u WHERE {pair_sql} "
            f"ORDER BY section NULLS LAST, property_name, test_condition",
            tuple(pair_params))

        prov = df_or_empty("SELECT filename, doi, title FROM agent_doi_seen")
        if not detail.empty and not prov.empty:
            detail = detail.merge(prov, how="left",
                                  left_on="source_pdf", right_on="filename")
        if "doi" not in detail.columns:
            detail["doi"] = ""

        def _s(x) -> str:
            """NaN/None-proof string."""
            return "" if x is None or (isinstance(x, float) and pd.isna(x)) else str(x)

        for _, m in mats.iterrows():
            icon = TABLE_ICON.get(m["tbl"], "🔬")
            mclass = _s(m["mclass"]) or m["tbl"].replace("_", " ").lower()
            n_dp, n_rows = int(m["distinct_props"]), int(m["props"])
            n_ok, n_docs = int(m["verified"]), int(m["docs"])
            n_flag = max(0, n_rows - n_ok)

            # Expander labels are markdown (colored text + badges), not HTML.
            bits = [_plural(n_dp, "property", "properties"), f"{n_ok} ✓"]
            if n_flag:
                bits.append(f"{n_flag} 🚩")
            if n_docs:
                bits.append(_plural(n_docs, "source"))
            label = (f"{icon} **{_md(m['display_name'])}** "
                     f":blue-background[{_md(mclass)}] "
                     f":gray[{' · '.join(bits)}]")

            with st.expander(label):
                # Rich header row (HTML) — name, class, status, provenance.
                meta_bits = [f"<b>{n_rows:,}</b> property row{'s' if n_rows != 1 else ''}"]
                if n_docs:
                    meta_bits.append(f"<b>{n_docs:,}</b> source document{'s' if n_docs != 1 else ''}")
                else:
                    meta_bits.append("no source document recorded")
                if _s(m["updated"]):
                    meta_bits.append("updated " + T.esc(_s(m["updated"])[:16].replace("T", " ")))
                head = [f'<span class="name">{T.esc(m["display_name"])}</span>',
                        T.chip_html(f"{n_ok:,} verified", "good" if n_ok else "muted",
                                    "grounded in the PDF, unit-sane, plausible")]
                if n_flag:
                    head.append(T.chip_html(f"{n_flag:,} flagged", "warn",
                                            "quarantined in the Review Queue"))
                head.append(f'<span class="meta">{" · ".join(meta_bits)}</span>')
                st.html(f'<div class="aim-mat">{"".join(head)}</div>')

                meta = []
                if m["mat_key"] != m["display_name"]:
                    meta.append(("key", _s(m["mat_key"]), "muted"))
                if _s(m["grades"]):
                    meta.append(("grade", _s(m["grades"]), "muted"))
                if _s(m["makers"]):
                    meta.append(("manufacturer", _s(m["makers"]), "muted"))
                if _s(m["matrices"]):
                    meta.append(("matrix", _s(m["matrices"]), "muted"))
                if _s(m["fibers"]):
                    meta.append(("fiber", _s(m["fibers"]), "muted"))
                if _s(m["processes"]):
                    meta.append(("process", _s(m["processes"]), "muted"))
                if meta:
                    st.html(T.chips_html(meta))

                rows = detail[(detail["tbl"] == m["tbl"])
                              & (detail["mat_key"] == m["mat_key"])].copy()
                if rows.empty:
                    st.html('<div class="aim-note">No property rows.</div>')
                    continue

                raw_val = rows.get("value_raw", pd.Series("", index=rows.index)).fillna("")
                plain = (rows.get("value", pd.Series("", index=rows.index)).fillna("").astype(str)
                         + " " + rows.get("unit", pd.Series("", index=rows.index)).fillna("").astype(str)).str.strip()
                value_col = raw_val.astype(str).where(raw_val.astype(str) != "", plain)
                qual = rows.get("qualifier", pd.Series("", index=rows.index)).fillna("")
                value_col = (qual.astype(str) + " " + value_col).str.strip()

                si = pd.Series("", index=rows.index, dtype=object)
                if "value_si" in rows and "unit_canonical" in rows:
                    ok_si = rows["value_si"].notna()
                    if ok_si.any():  # empty float64 selection would stay float64
                        si.loc[ok_si] = (
                            rows.loc[ok_si, "value_si"].astype(float).map(lambda v: f"{v:g}")
                            + " " + rows.loc[ok_si, "unit_canonical"].fillna("").astype(str)
                        ).str.strip()

                pg = rows.get("page", pd.Series(index=rows.index, dtype=object))
                src = rows.get("source_pdf", pd.Series("", index=rows.index)).fillna("")
                src = src.astype(str) + pg.map(
                    lambda p: f" · p.{int(p)}" if pd.notna(p) else "")

                # Processing route: normalized type, with the printed name
                # when it says more; ⚠ when the route did not ground.
                def _col(name):
                    return rows.get(name, pd.Series("", index=rows.index)).fillna("").astype(str)
                p_type, p_name, p_stat = _col("process_type"), _col("process_name"), _col("process_status")
                proc_col = [
                    "" if not t_ else
                    (t_ + (f" ({n_})" if n_ and n_.lower() not in t_.lower() else "")
                     + ("" if s_ in ("grounded", "grounded_off_page") else " ⚠"))
                    for t_, n_, s_ in zip(p_type, p_name, p_stat)]

                out = pd.DataFrame({
                    "Section": rows.get("section", pd.Series("", index=rows.index)).fillna(""),
                    "Property": rows["property_name"].fillna(""),
                    "Condition": rows.get("test_condition", pd.Series("", index=rows.index)).fillna(""),
                    "Process": proc_col,
                    "Process conditions": _col("process_conditions"),
                    "Value": value_col,
                    "SI": si,
                    "": [("✓" if _s(st_) in ("", "ok") else
                          ("📈" if _s(og) == "figure" else "🚩"))
                         for st_, og in zip(
                             rows.get("status", pd.Series("ok", index=rows.index)).fillna("ok"),
                             rows.get("origin", pd.Series("text", index=rows.index)).fillna("text"))],
                    "Source": src,
                    "DOI": rows["doi"].fillna("").map(
                        lambda d: f"https://doi.org/{d}" if d else ""),
                })
                if (out["Section"] == "").all():
                    out = out.drop(columns=["Section"])
                # Rows from before prompt 2.1, or sources that state no route.
                if (out["Process"] == "").all():
                    out = out.drop(columns=["Process", "Process conditions"])
                st.dataframe(
                    out,
                    column_config={k: v for k, v in PROP_COL_CONFIG.items()
                                   if k in out.columns},
                    hide_index=True, width="stretch",
                    height=min(37 + 35 * len(out), 420))

        st.html('<div class="aim-note">✓ = verified (grounded in the PDF, '
                'unit-sane, plausible) · 🚩 = flagged, quarantined in the '
                'Review Queue · 📈 = figure-derived estimate (read off a '
                'plot/table image; quarantined until promoted). '
                'Process = how the tested specimen was made, with its '
                'processing conditions as printed in the source; empty when '
                'the source does not say.</div>')

# ===========================================================================
# RAW TABLES — classic row-level browser
# ===========================================================================
with tab_raw:
    counts = {}
    for t in BROWSABLE:
        df = df_or_empty(f"SELECT count(*) AS n FROM {_q(t)}")
        counts[t] = int(df["n"].iloc[0]) if not df.empty else 0
    T.kpi_row([{"label": t.replace("_", " "), "value": counts[t]}
               for t in BROWSABLE], small=True)

    with st.container(border=True):
        T.card_title("Browse rows", "pick a table, then filter and page through it")
        tc, fc1, fc2, fc3, fc4 = st.columns([2, 3, 2, 2, 2])
        table = tc.selectbox("Table", BROWSABLE, index=0, key="raw_table")
        have = columns_of(table)
        if not have:
            st.error(f"Table {table} not found in this database.")
            st.stop()

        search = fc1.text_input("Search", key="raw_search",
                                placeholder="material, property, PDF, DOI…",
                                help="Case-insensitive match across the text columns")

        where, params = [], []

        if "property_name" in have:
            props = df_or_empty(
                f"SELECT property_name, count(*) AS n FROM {_q(table)} "
                f"WHERE property_name IS NOT NULL AND property_name <> '' "
                f"GROUP BY 1 ORDER BY n DESC, 1 LIMIT 300")
            prop_opts = ["(all)"] + props["property_name"].tolist() if not props.empty else ["(all)"]
            prop = fc2.selectbox("Property", prop_opts, index=0, key="raw_prop")
            if prop != "(all)":
                where.append("property_name = %s")
                params.append(prop)

        if "material_class" in have:
            classes = df_or_empty(
                f"SELECT DISTINCT material_class FROM {_q(table)} "
                f"WHERE material_class IS NOT NULL AND material_class <> '' "
                f"ORDER BY 1 LIMIT 100")
            cls_opts = ["(all)"] + classes["material_class"].tolist() if not classes.empty else ["(all)"]
            cls = fc3.selectbox("Class", cls_opts, index=0, key="raw_class")
            if cls != "(all)":
                where.append("material_class = %s")
                params.append(cls)

        if "status" in have:
            stat = fc4.selectbox("Status", ["(all)", "ok (verified)", "flagged"],
                                 index=0, key="raw_status")
            if stat == "ok (verified)":
                where.append("COALESCE(status, 'ok') = 'ok'")
            elif stat == "flagged":
                where.append("COALESCE(status, 'ok') <> 'ok'")

        searchable = [c for c in SEARCH_COLS if c in have]
        if search and searchable:
            where.append("(" + " OR ".join(f"{_q(c)}::text ILIKE %s" for c in searchable) + ")")
            params.extend([f"%{search}%"] * len(searchable))

        where_sql = (" WHERE " + " AND ".join(where)) if where else ""

        total_df = df_or_empty(f"SELECT count(*) AS n FROM {_q(table)}{where_sql}",
                               tuple(params))
        total = int(total_df["n"].iloc[0]) if not total_df.empty else 0

        pc1, pc2, pc3 = st.columns([1, 1, 4], vertical_alignment="bottom")
        page_size = pc1.selectbox("Rows / page", [25, 50, 100, 250], index=1,
                                  key="raw_psize")
        n_pages = max(1, -(-total // page_size))
        page_no = pc2.number_input("Page", 1, n_pages, 1, key="raw_page")
        with pc3:
            _count_line(total, "row", page_no, n_pages)

    order = [f"{_q(c)} DESC NULLS LAST" for c in ("extracted_at",) if c in have]
    if "id" in have:
        order.append(f"{_q('id')} DESC")
    order_sql = (" ORDER BY " + ", ".join(order)) if order else ""

    # Binary columns (figure crop bytes) never enter the dataframe — show the
    # stored size instead. Everything else keeps SELECT * semantics.
    _BINARY_COLS = {"image_bytes"}
    if _BINARY_COLS & set(have):
        select_sql = ", ".join(
            f"(octet_length({_q(c)}) / 1024)::int AS {_q(c[:-6] + '_kb')}"
            if c in _BINARY_COLS else _q(c)
            for c in have)
    else:
        select_sql = "*"

    rows = df_or_empty(
        f"SELECT {select_sql} FROM {_q(table)}{where_sql}{order_sql} LIMIT %s OFFSET %s",
        tuple(params) + (page_size, (page_no - 1) * page_size))

    if rows.empty:
        if counts.get(table, 0) == 0:
            T.empty_state("📋", f"{table.replace('_', ' ')} has no rows yet",
                          "Nothing has been ingested into this table so far — "
                          "start a cycle from Run Control or check back after "
                          "the next scheduled run.")
        else:
            T.empty_state("🔎", "No rows match the current filters",
                          "Try a shorter search term or reset the property, "
                          "class and status filters.")
    else:
        _FIG_PREFERRED = ["source_pdf", "page", "figure_kind", "caption",
                          "mining_status", "n_values", "material_key", "route",
                          "image_kb", "width_px", "height_px", "figure_id"]
        _pref = _FIG_PREFERRED if table == "figures" else PREFERRED_COLS
        default_cols = [c for c in _pref if c in rows.columns] or list(rows.columns)
        with st.expander("Columns", expanded=False):
            chosen = st.multiselect("Show columns", list(rows.columns),
                                    default=default_cols, key="raw_cols")
        show = rows[chosen] if chosen else rows[default_cols]

        col_config = {}
        if "doi_url" in show.columns:
            col_config["doi_url"] = st.column_config.LinkColumn(
                "DOI", display_text=r"https://doi\.org/(.*)")
        if "extracted_at" in show.columns:
            show = show.assign(
                extracted_at=show["extracted_at"].astype(str).str.slice(0, 16).str.replace("T", " "))
        st.dataframe(show, column_config=col_config, hide_index=True,
                     width="stretch", height=min(37 + 35 * len(show), 520))

        EXPORT_CAP = 5000
        export = df_or_empty(
            f"SELECT {select_sql} FROM {_q(table)}{where_sql}{order_sql} LIMIT {EXPORT_CAP}",
            tuple(params))
        st.download_button(
            f"⬇️ Download filtered CSV ({min(total, EXPORT_CAP):,} rows"
            + (f", capped at {EXPORT_CAP:,}" if total > EXPORT_CAP else "") + ")",
            export.to_csv(index=False).encode(),
            file_name=f"{table.lower()}_export.csv", mime="text/csv",
            key="raw_dl")

        if "property_name" in have:
            with st.expander("Property summary (current filters)"):
                summary = df_or_empty(
                    f"SELECT property_name AS property, count(*) AS rows_, "
                    f"       count(*) FILTER (WHERE COALESCE(status,'ok')='ok') AS verified "
                    f"FROM {_q(table)}{where_sql} GROUP BY 1 ORDER BY rows_ DESC LIMIT 200",
                    tuple(params))
                if summary.empty:
                    st.html('<div class="aim-note">No properties yet.</div>')
                else:
                    st.dataframe(
                        summary.rename(columns={"rows_": "rows"}),
                        column_config={
                            "property": st.column_config.TextColumn("Property", width="large"),
                            "rows": st.column_config.NumberColumn("Rows", width="small", format="%d"),
                            "verified": st.column_config.NumberColumn("Verified", width="small", format="%d"),
                        },
                        hide_index=True, width="stretch")

    st.html('<div class="aim-note">Read-only view of the connected database '
            '(the one the DB_NAME secret points at). Rows arrive via the '
            'agent\'s ingest step or any <code>batch_ingest.py --pg</code> run.</div>')