app.py

"""Finance Atlas — a Hugging Face Space by The Bit Trading Company.

An index of finance models on the Hub: what they claim, what they document,
and what nobody has checked.

Architecture, in one paragraph: the visible layer is the design's own markup,
rendered by `src/ui/shell.py` from a plain state dict. Gradio owns the
transport and nothing that is seen -- one `gr.HTML` sink, a hidden textbox the
click bridge writes into, and a hidden button it clicks. Every interaction is
`state -> apply_action -> state -> render`, which keeps the whole UI a pure
function of state and makes it testable without a browser.

All reads come from the dataset repo. There is no Hub API call in the request
path: `atlas.parquet` is loaded once at boot and every filter, sort and lineage
walk is an in-memory operation. That is what keeps this usable on CPU basic.
"""

from __future__ import annotations

import logging
import os

import gradio as gr

from src import atlas, suggest          # noqa: F401 - sets the vendor path
from src.atlas import growth
from src.ui import chrome, shell

from bit_ui import nav as bit_nav
from bit_ui import bridge, theme

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s %(levelname)-7s %(name)s: %(message)s")
log = logging.getLogger("atlas.app")

ON_SPACE = bool(os.environ.get("SPACE_ID"))

BACKTEST_URL = "https://huggingface.co/spaces/Bit-Trading-Company/bit-backtest-lab"
DATASET_URL = f"https://huggingface.co/datasets/{atlas.DATASET_REPO}"
ORG_URL = "https://huggingface.co/Bit-Trading-Company"

# Only the destinations that exist are linked. The rest of the design's nav is
# rendered but inert -- see `shell.sidebar`.
LINKS = {
    "Backtest Lab": BACKTEST_URL,
    "Dataset": DATASET_URL,
    "Hugging Face org": ORG_URL,
}

TAPE_SIZE = 12
TRENDING_SIZE = 5

# How many rows are rendered into the table at once. The design's table is a
# scroll container, not a paginated list, so this is a payload bound rather
# than a UX choice.
#
# Measured over the real index: 300 rows is ~1.1 MB of HTML, which gzips to
# ~35 KB because the design's inline styles repeat heavily (32:1). Rendering
# takes ~5 ms. The count the UI reports is always the true number of matches,
# never this cap -- see `build_view`.
MAX_ROWS = 300

# --------------------------------------------------------------------------
# Boot
# --------------------------------------------------------------------------

# `ATLAS_LOCAL_DIR` points the app at a dataset tree on disk instead of the Hub.
# Used for local development and by the tests; unset in production.
INDEX = atlas.build_index(local_dir=os.environ.get("ATLAS_LOCAL_DIR") or None)

# The navigation is shared config, not code: it lives in a dataset every BTC
# Space reads at boot. Flipping a module from "coming soon" to live is one edit
# there and reaches every Space on its next restart -- no redeploy, no
# submodule bump. `nav.load` falls back to the built-in nav if it cannot read
# it, so an unreachable config never stops the Space from booting.
NAV = bit_nav.load(local_path=os.environ.get("ATLAS_LOCAL_NAV") or None)
# The renderers need to link back to the dataset; carrying it on the index
# keeps the shell from importing app config.
INDEX.dataset_repo = atlas.DATASET_REPO

SUGGESTIONS = suggest.Suggestions(
    local_dir=os.environ.get("ATLAS_SUGGEST_DIR", "/tmp/atlas-suggestions"),
)


# --------------------------------------------------------------------------
# Derived view
# --------------------------------------------------------------------------


def tape_rows(index):
    """The header ticker: the most downloaded models, with real change."""
    top = sorted(index.rows, key=lambda r: int(r.get("downloads_30d") or 0),
                 reverse=True)[:TAPE_SIZE]
    out = []
    for row in top:
        series = index.trends.get(row["id"])
        out.append({
            "id": row["id"],
            "downloads_30d": row.get("downloads_30d"),
            "series": series,
            "change": growth(series),
        })
    return out


def trending_rows(index):
    """Top real movers. Empty until two snapshots exist, by design."""
    scored = []
    for model_id, series in index.trends.items():
        change = growth(series)
        if change is None or change <= 0:
            continue
        scored.append((model_id, change, series))
    scored.sort(key=lambda item: item[1], reverse=True)
    return scored[:TRENDING_SIZE]


def palette_groups(index, query: str):
    """Results for the ⌘K palette, ranked and grouped.

    Three kinds of thing are searchable, in the order someone is most likely
    to want them: a specific model, a filter that narrows the table, and the
    other Spaces. Matching uses the same fields the table's own search uses,
    so the palette and the table never disagree about what "finbert" means.
    """
    from bit_ui import palette as bit_palette

    query = (query or "").strip().lower()
    labels = index.taxonomy.get("task_labels", {})
    assets = index.taxonomy.get("asset_labels", {})
    groups = []

    # -- models --------------------------------------------------------
    if query:
        scored = []
        for row in index.rows:
            model_id = (row.get("id") or "")
            lowered = model_id.lower()
            position = lowered.find(query)
            if position < 0:
                if query not in (row.get("author") or "").lower():
                    continue
                position = 50
            # Earlier match first, then popularity. An exact prefix on the
            # model name beats a big download count on a loose match.
            scored.append((position, -int(row.get("downloads_30d") or 0), row))
        scored.sort(key=lambda item: (item[0], item[1]))

        results = []
        for _, _, row in scored[:40]:
            results.append(bit_palette.Result(
                label=row["id"],
                action=f"open:{row['id']}",
                sub=labels.get(row.get("task"), row.get("task") or ""),
                meta=f"{int(row.get('downloads_30d') or 0):,}",
                glyph=shell.task_glyph(row.get("task")),
                badge="✓" if row.get("verified") else "",
            ))
        groups.append(bit_palette.Group("Models", results))

    # -- filters -------------------------------------------------------
    filters = []
    for key in index.taxonomy.get("task_chips", ()):
        label = labels.get(key, key)
        if query and query not in label.lower():
            continue
        filters.append(bit_palette.Result(
            label=f"Task: {label}", action=f"task:{key}",
            meta=f"{index.count_by('task', key):,}",
            glyph=shell.task_glyph(key)))
    for key in index.taxonomy.get("asset_chips", ()):
        label = assets.get(key, key)
        if query and query not in label.lower():
            continue
        filters.append(bit_palette.Result(
            label=f"Asset: {label}", action=f"asset:{key}",
            meta=f"{index.count_by('asset_class', key):,}",
            glyph=shell.asset_glyph(key)))
    if not query or "verified" in query:
        filters.append(bit_palette.Result(
            label="Only human-verified models", action="verified:toggle",
            meta=f"{index.verified_count:,}", glyph="✓"))
    if not query or "graveyard" in query or "unmaintained" in query:
        filters.append(bit_palette.Result(
            label="Show the graveyard (unmaintained)", action="graveyard:",
            meta=f"{index.unmaintained_count:,}", glyph="†"))
    if filters:
        groups.append(bit_palette.Group("Filters", filters))

    # -- other Spaces --------------------------------------------------
    spaces = []
    for group in NAV.get("groups") or ():
        for item in group.get("items") or ():
            if item.get("status") != "live" or not item.get("url"):
                continue
            if item["name"] == "Finance Atlas":
                continue
            if query and query not in item["name"].lower():
                continue
            spaces.append(bit_palette.Result(
                label=item["name"], href=item["url"], glyph="↗", meta="OPEN"))
    if spaces:
        groups.append(bit_palette.Group("Go to", spaces))

    return groups


def build_view(index, state) -> dict:
    """Everything derived from state, computed once per render.

    `matched` is the true number of rows passing the filters; `rows` is the
    capped slice actually rendered. Both are carried because the UI must
    report the real count -- a table that says "300 shown" when 800 matched
    is telling the user their filter is narrower than it is.
    """
    matched = index.sorted(index.filtered(state), state.get("sort", "Downloads"))
    return {
        "rows": matched[:MAX_ROWS],
        "matched": len(matched),
        "truncated": max(0, len(matched) - MAX_ROWS),
        "hidden": index.hidden_by_maintained(state),
        "tape": tape_rows(index),
        "trending": trending_rows(index),
        "links": LINKS,
        "nav": NAV,
        # Only computed when the palette is actually open -- it scans every row.
        "palette_groups": (palette_groups(index, state.get("palette_q"))
                           if state.get("palette_open") else []),
    }


def render(state) -> str:
    return shell.page(INDEX, state, build_view(INDEX, state))


# --------------------------------------------------------------------------
# Actions
# --------------------------------------------------------------------------


def _toggle_in(state, key, value, allowed):
    """Add/remove `value` in a list-valued filter, validated against `allowed`.

    Actions arrive from the DOM and are user input. An unknown value is
    dropped rather than stored, so a hand-crafted click cannot inject a filter
    term the UI never offered.
    """
    if value not in allowed:
        return state
    current = list(state.get(key) or ())
    if value in current:
        current.remove(value)
    else:
        current.append(value)
    state[key] = current
    return state


def apply_action(state: dict, raw: str) -> dict:
    """Fold one bridge action into the state. Unknown actions change nothing."""
    action = chrome.parse_action(raw)
    if action is None or action.is_noop:
        return state

    state = dict(state)
    key, value = action.key, action.value
    taxonomy = INDEX.taxonomy

    if key == "task":
        state = _toggle_in(state, "tasks", value,
                           set(taxonomy.get("task_chips", ())))
    elif key == "asset":
        state = _toggle_in(state, "assets", value,
                           set(taxonomy.get("asset_chips", ())))
    elif key == "lic":
        state = _toggle_in(state, "lic", value,
                           set(taxonomy.get("license_buckets", ())))
    elif key == "verified":
        state["verified_only"] = not state.get("verified_only")
    elif key == "maintained":
        state["maintained_only"] = not state.get("maintained_only")
    elif key == "haseval":
        state["has_eval"] = not state.get("has_eval")
    elif key == "hasdata":
        state["has_data"] = not state.get("has_data")
    elif key == "sort":
        if value in atlas.SORTS:
            state["sort"] = value
    elif key == "clear":
        state.update(q="", tasks=[], assets=[], lic=[], verified_only=False,
                     has_eval=False, has_data=False)
    elif key == "graveyard":
        state["maintained_only"] = False
    elif key == "open":
        # Only ids actually in the index can open a drawer.
        state["sel"] = value if value in INDEX.by_id else None
    elif key == "close":
        state["sel"] = None
    elif key == "q":
        state["q"] = value[:120]
    elif key == "suggestbox":
        state["suggest"] = value[:120]
        state["suggest_note"] = ""
        state["suggest_ok"] = False
    elif key == "suggest":
        state = _submit_suggestion(state)

    # ---- shared chrome ------------------------------------------------
    elif key == "palette":
        if value == "open":
            # Seed the palette with whatever is already in the search box, so
            # ⌘K continues a search rather than discarding it.
            state.update(palette_open=True,
                         palette_q=state.get("palette_q") or state.get("q") or "")
        else:
            state["palette_open"] = False
    elif key == "pq":
        state["palette_q"] = value[:120]
    elif key == "soon":
        if value == "close":
            state.update(soon_module="", notify_note="")
        elif bit_nav.find(NAV, value):
            # Only modules the shared nav actually declares. A hand-crafted
            # click cannot invent a product.
            state.update(soon_module=value, notify_note="", palette_open=False)
    elif key == "nemail":
        state["notify_email"] = value[:160]
    elif key == "notify":
        state = _register_interest(state, value)
    elif key == "contact":
        if value == "open":
            state.update(contact_open=True, contact_note="", contact_ok=False,
                         palette_open=False, soon_module="")
        elif value == "close":
            state.update(contact_open=False, contact_form_open=False)
        elif value == "form":
            state["contact_form_open"] = not state.get("contact_form_open")
        elif value == "send":
            state = _send_contact(state)
    elif key == "topic":
        topics = (NAV.get("contact") or {}).get("topics") or []
        if value in topics:
            state.update(contact_topic=value, contact_note="", contact_ok=False)
    elif key == "cname":
        state["contact_name"] = value[:120]
    elif key == "cemail":
        state["contact_email"] = value[:160]
    elif key == "cmsg":
        state["contact_message"] = value[:4000]

    return state


def _register_interest(state: dict, module: str) -> dict:
    """"Notify me when this launches" from the coming-soon dialog.

    Written to the dataset alongside suggestions. It says RECORDED rather than
    "you're on the list", because there is no mailing list yet -- only a file
    somebody has to read.
    """
    if not bit_nav.find(NAV, module):
        return state
    email = (state.get("notify_email") or "").strip()
    try:
        SUGGESTIONS.submit_interest(email, module)
    except suggest.SuggestionError as exc:
        state["notify_note"] = str(exc).upper()
        return state
    state["notify_note"] = "RECORDED — WE HAVE NO MAILING LIST YET"
    return state


def _send_contact(state: dict) -> dict:
    """Append a contact message to the dataset.

    Deliberately not "SENT · TICKET #4821 OPENED" as the design writes it:
    there is no ticketing system behind this, and inventing a ticket number
    would be the most confidently false thing on the page.
    """
    try:
        SUGGESTIONS.submit_contact(
            name=state.get("contact_name", ""),
            email=state.get("contact_email", ""),
            topic=state.get("contact_topic")
            or ((NAV.get("contact") or {}).get("topics") or ["Other"])[0],
            message=state.get("contact_message", ""),
        )
    except suggest.SuggestionError as exc:
        state.update(contact_note=str(exc).upper(), contact_ok=False)
        return state

    state.update(
        contact_note=("RECEIVED" if SUGGESTIONS.syncing
                      else "RECEIVED (NOT SYNCED)"),
        contact_ok=True, contact_message="",
    )
    return state


def _submit_suggestion(state: dict) -> dict:
    try:
        model_id = SUGGESTIONS.submit(state.get("suggest", ""))
    except suggest.SuggestionError as exc:
        state["suggest_note"] = str(exc).upper()
        state["suggest_ok"] = False
        return state

    if model_id in INDEX.by_id:
        state["suggest_note"] = "ALREADY INDEXED"
    elif SUGGESTIONS.syncing:
        state["suggest_note"] = "QUEUED FOR NEXT CRAWL"
    else:
        # Honest about the difference: accepted locally is not the same as
        # committed to the dataset.
        state["suggest_note"] = "RECEIVED (NOT SYNCED)"
    state["suggest_ok"] = True
    state["suggest"] = ""
    return state


def on_action(raw, state):
    state = apply_action(state or atlas.default_state(), raw or "")
    return render(state), state


# --------------------------------------------------------------------------
# App
# --------------------------------------------------------------------------


def build_app() -> gr.Blocks:
    with gr.Blocks(
        title="Finance Atlas — The Bit Trading Company",
        theme=theme.bit_theme(),
        # chrome.full_css(), not theme.full_css(): the latter is bit-ui's
        # shared CSS only and silently drops this Space's own -- the table
        # row styling, the sticky header, the responsive breakpoints.
        css=chrome.full_css(),
        analytics_enabled=False,
        fill_width=True,
    ) as demo:
        state = gr.State(atlas.default_state())

        # The one visible component. Everything the user sees is inside it.
        view = gr.HTML(render(atlas.default_state()), elem_id="bit-view")

        # The bridge's transport. Both must be rendered -- `visible=False`
        # removes an element from the DOM entirely, and the delegated listener
        # would then be writing into nothing. They are hidden in CSS instead.
        action = gr.Textbox("", elem_id=bridge.ACTION_ELEMENT_ID,
                            label="", show_label=False, container=False,
                            interactive=True)
        trigger = gr.Button("", elem_id=bridge.TRIGGER_ELEMENT_ID)

        # Two paths for one click: the textbox's own change event, and the
        # hidden button. Gradio always honours a real click on a real Button,
        # and the nonce on every action means a duplicate delivery folds to the
        # same state rather than acting twice.
        action.change(on_action, [action, state], [view, state],
                      show_progress="hidden")
        # `api_name` keeps this reachable from `gradio_client`, which is how CI
        # smoke-tests the deployed Space: a green build only proves the image
        # started, not that an action still renders a page.
        trigger.click(on_action, [action, state], [view, state],
                      show_progress="hidden", api_name="action")

        # Installed via its own `load`, with no outputs: Gradio treats a `js=`
        # return value as the output values, so sharing a load that has outputs
        # would wipe them. Spaces also drops `gr.Blocks(head=...)`, which is why
        # this is the path that actually works once deployed.
        demo.load(fn=None, inputs=None, outputs=None, js=bridge.BRIDGE_LOAD_JS)

    return demo


demo = build_app()

if __name__ == "__main__":
    demo.launch(
        server_name="0.0.0.0" if ON_SPACE else "127.0.0.1",
        server_port=int(os.environ.get("PORT", 7860)),
        show_api=False,
        allowed_paths=theme.static_paths(),
        # Gradio 5 defaults to SSR. The page then renders server-side and only
        # becomes interactive once the client hydrates -- and when hydration
        # does not complete, every handler stays unbound: the markup is all
        # there, buttons look fine, and nothing does anything. That is exactly
        # what happened here, and it is a silent failure with no console error.
        # This app renders its own markup and needs only the event wiring, so
        # there is nothing for SSR to win and a whole failure mode to avoid.
        ssr_mode=False,
    )