app.py

"""RANGE text curation demo with a specialized General Economy showcase."""

from __future__ import annotations

import dataclasses
import hashlib
import html
import json
import os
import sys
import uuid
from collections.abc import Mapping, Sequence
from datetime import datetime
from enum import Enum
from functools import lru_cache
from pathlib import Path
from time import perf_counter
from typing import Any

import gradio as gr
from starlette.middleware import Middleware
from starlette.middleware.body_limit import RequestBodyLimitMiddleware
from gradio.utils import get_upload_folder

VENDORED_PACKAGE_ROOT = Path(__file__).resolve().parent / "vendor"
if VENDORED_PACKAGE_ROOT.is_dir():
    sys.path.insert(0, str(VENDORED_PACKAGE_ROOT))

try:
    import spaces
except ModuleNotFoundError:  # Local installs and the Docker image do not need it.
    spaces = None

from range_curate.examples.general_economy import RelevanceEngine, __version__
from range_curate import CurationEngine, CurationTaxonomy

from examples import ARTICLE_EXAMPLES, BATCH_ROWS, SINGLE_EXAMPLES
from curation_demo import build_curation_tabs, processing_timing


if spaces is not None:

    @spaces.GPU(duration=1)
    def _zerogpu_hosting_marker() -> None:
        """Register the free ZeroGPU runtime without moving CPU inference to a GPU."""

        return None


GRADIO_MAJOR = int(gr.__version__.split(".", 1)[0])
THEME = gr.themes.Soft(primary_hue="teal", neutral_hue="slate")
MAX_BATCH_ROWS = 100
EXPECTED_BATCH_COLUMNS = 4
MAX_ARTICLE_CHARS = 20_000
MAX_ARTICLE_BYTES = 80_000
MAX_TIMESTAMP_CHARS = 128
MAX_TIMESTAMP_BYTES = 512
MAX_ARTICLE_ID_CHARS = 256
MAX_ARTICLE_ID_BYTES = 1_024
MAX_SOURCE_CHARS = 256
MAX_SOURCE_BYTES = 1_024
MAX_BATCH_CHARS = 200_000
MAX_BATCH_BYTES = 800_000
MAX_BATCH_METADATA_CHARS = 64_000
MAX_BATCH_METADATA_BYTES = 256_000
MAX_BATCH_CONTENT_CHARS = MAX_BATCH_CHARS + MAX_BATCH_METADATA_CHARS
MAX_BATCH_CONTENT_BYTES = MAX_BATCH_BYTES + MAX_BATCH_METADATA_BYTES
MAX_HTTP_REQUEST_BYTES = 4 * 1024 * 1024
EVIDENCE_HEADERS = [
    "evidence path",
    "matched text",
    "offsets",
    "type",
    "method",
    "rule or catalog",
    "identity",
    "exposure",
]
BATCH_HEADERS = [
    "article_id",
    "status",
    "accepted",
    "relevant_to",
    "why",
]

CSS = """

html, body {width: 100%; margin: 0 !important; overflow-x: clip;}

.gradio-container {

  width: 100% !important;

  max-width: none !important;

  margin: 0 !important;

  padding: 0 0 2rem !important;

}

.hero-host {

  width: 100% !important;

  max-width: none !important;

  margin-inline: 0 !important;

  padding: 0 !important;

  border: 0 !important;

  box-sizing: border-box !important;

}

.hero {

  position: relative;

  width: 100%;

  margin: 0;

  overflow: hidden;

  box-sizing: border-box;

  color: #172b3a;

  border-block: 1px solid #dbe5eb;

  border-inline: 0;

  background:

    radial-gradient(circle at 92% 15%, rgba(13, 148, 136, 0.08), transparent 34%),

    radial-gradient(circle at 8% 88%, rgba(37, 99, 235, 0.05), transparent 30%),

    linear-gradient(145deg, #ffffff, #f0f7f8);

  box-shadow: 0 18px 48px rgba(15, 23, 42, 0.08);

}

.hero-inner {

  width: min(calc(100% - 2rem), 1180px);

  margin-inline: auto;

  padding: clamp(2rem, 4vw, 3rem) clamp(1.25rem, 4vw, 3.2rem);

  box-sizing: border-box;

  text-align: center;

}

.hero-eyebrow,

.hero h1,

.hero-expansion,

.hero-copy,

.benchmark-note {

  width: 100%;

  text-align: center !important;

}

.hero-eyebrow {

  margin: 0 0 0.8rem;

  color: #0f766e !important;

  font-size: 0.78rem;

  font-weight: 750;

  letter-spacing: 0.12em;

  text-transform: uppercase;

}

.hero h1 {

  max-width: 960px;

  margin: 0 auto;

  font-size: clamp(3.4rem, 9vw, 6.2rem);

  line-height: 0.92;

  letter-spacing: -0.065em;

}

.hero-expansion {

  max-width: 960px;

  margin: 1rem auto 0;

  font-size: clamp(1.08rem, 2.5vw, 1.55rem);

  font-weight: 720;

  line-height: 1.35;

}

.hero-copy {

  text-wrap: balance;

  max-width: 900px;

  margin: 1rem auto 0;

  font-size: clamp(1rem, 1.5vw, 1.16rem);

  line-height: 1.65;

  opacity: 0.88;

}

.speed-grid {

  display: grid;

  grid-template-columns: repeat(3, minmax(0, 1fr));

  gap: 0.75rem;

  max-width: 900px;

  margin: 1.6rem auto 0;

}

.speed-stat {

  display: flex;

  flex-direction: column;

  justify-content: center;

  align-items: center;

  min-height: 6rem;

  box-sizing: border-box;

  padding: 0.9rem 1rem;

  border: 1px solid var(--border-color-primary);

  border-radius: 14px;

  background: var(--background-fill-primary);

  text-align: center;

}

.hero h1, .hero-expansion, .hero-copy, .hero .benchmark-note, .hero .speed-stat span {color: #172b3a !important;}

.hero .speed-stat, .hero .trust-pill {background: #ffffff; border-color: #d5e2e7; color: #172b3a !important;}

.hero .package-pill {background: #d5f5e9; color: #164c3d !important; border-color: #b6e4d3;}

.hero .speed-stat strong {color: #0f766e !important;}

.speed-stat strong {display: block; font-size: clamp(1.2rem, 2.6vw, 1.75rem); line-height: 1.15;}

.speed-stat span {display: block; margin-top: 0.25rem; font-size: 0.8rem; line-height: 1.35; opacity: 0.72;}

.hero .benchmark-note {max-width: 820px; margin: 1rem auto 0 !important; text-wrap: balance; font-size: 0.86rem; line-height: 1.5; opacity: 0.72;}

.trust-row {display: flex; justify-content: center; flex-wrap: wrap; gap: 0.55rem; margin-top: 1.25rem;}

.trust-pill {

  padding: 0.36rem 0.68rem;

  border: 1px solid var(--border-color-primary);

  border-radius: 999px;

  background: var(--background-fill-primary);

  font-size: 0.82rem;

  font-weight: 650;

}

.value-grid {

  display: grid;

  grid-template-columns: repeat(3, minmax(0, 1fr));

  gap: 0.8rem;

  width: min(calc(100% - 2rem), 1120px);

  max-width: 1120px;

  margin: 1.15rem auto 1.35rem;

}

.content-shell {

  width: min(calc(100% - 2rem), 1120px) !important;

  max-width: 1120px !important;

  margin-inline: auto !important;

  align-self: center !important;

}

.value-card {

  padding: 1rem 1.05rem;

  border: 1px solid var(--border-color-primary);

  border-radius: 14px;

  background: var(--background-fill-primary);

  text-align: center;

}

.value-card strong {display: block; margin-bottom: 0.25rem; font-size: 0.95rem;}

.value-card span {font-size: 0.88rem; line-height: 1.45; opacity: 0.78;}

.status-card {

  min-height: 255px;

  padding: 1.15rem 1.25rem;

  border: 1px solid var(--border-color-primary);

  border-top: 5px solid #94a3b8;

  border-radius: 16px;

  background: var(--background-fill-primary);

  box-shadow: 0 10px 26px rgba(15, 23, 42, 0.06);

}

.status-card.status-relevant {border-top-color: #0f766e;}

.status-card.status-review {border-top-color: #d97706;}

.status-card.status-no-match {border-top-color: #64748b;}

.status-card h3 {margin: 0.5rem 0 0.45rem; font-size: 1.32rem; line-height: 1.25;}

.status-label {font-size: 0.73rem; font-weight: 750; letter-spacing: 0.1em; opacity: 0.68; text-transform: uppercase;}

.status-badge {

  display: inline-block;

  margin-left: 0.45rem;

  padding: 0.2rem 0.48rem;

  border-radius: 999px;

  background: var(--background-fill-secondary);

  font-size: 0.72rem;

  font-weight: 750;

  letter-spacing: 0.04em;

}

.result-grid {display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 0.75rem; margin: 0.9rem 0;}

.result-cell {padding: 0.72rem 0.8rem; border-radius: 11px; background: var(--background-fill-secondary);}

.result-cell strong {display: block; margin-bottom: 0.18rem; font-size: 0.78rem; opacity: 0.72;}

.result-cell span {font-weight: 650; line-height: 1.4;}

.boundary-note {margin: 0.75rem 0 0; font-size: 0.86rem; line-height: 1.45; opacity: 0.76;}

.small-note {font-size: 0.78rem; opacity: 0.64;}

.evidence-legend {display: flex; flex-wrap: wrap; gap: 0.55rem 1rem; margin: 0 0 0.65rem; font-size: 0.83rem;}

.legend-key {display: inline-flex; align-items: center; gap: 0.35rem;}

.legend-swatch {width: 0.72rem; height: 0.72rem; border-radius: 3px; border: 1px solid rgba(15, 23, 42, 0.16);}

.swatch-accepted {background: #99f6e4;}

.swatch-review {background: #fde68a;}

.swatch-rejected {background: #e2e8f0;}

.swatch-observed {background: #bfdbfe;}

mark.evidence {padding: 0.08rem 0.2rem; border-radius: 0.25rem; color: #111827;}

mark.evidence:focus {outline: 2px solid #2563eb; outline-offset: 2px;}

mark.evidence-accepted {background: #99f6e4;}

mark.evidence-review {background: #fde68a;}

mark.evidence-rejected {background: #e2e8f0; text-decoration: line-through; text-decoration-thickness: 1px;}

mark.evidence-observed {background: #bfdbfe;}

.article-preview {

  min-height: 5.5rem;

  line-height: 1.8;

  white-space: pre-wrap;

  padding: 1rem 1.05rem;

  border: 1px solid var(--border-color-primary);

  border-radius: 14px;

  background: var(--background-fill-primary);

}

.empty-preview {opacity: 0.68;}

.privacy-note {

  margin-top: 1rem;

  padding: 0.85rem 1rem;

  border-left: 4px solid #64748b;

  border-radius: 8px;

  background: var(--background-fill-secondary);

  font-size: 0.88rem;

  line-height: 1.5;

}

@media (max-width: 760px) {

  .speed-grid, .value-grid, .result-grid {grid-template-columns: 1fr;}

  .hero-inner {padding: 1.8rem 1.25rem 2rem;}

  .speed-stat {text-align: center;}

  .status-card {min-height: auto;}

}

"""

ABOUT = """

## Financial news, with the evidence attached



RANGE screens timestamped articles for economic and financial relevance, maps

supported evidence to broad-market exposures, and shows exactly why it made each

decision. It is a deterministic engine: the same text, timestamp and taxonomy

produce the same result.



### Follow every decision

1. Normalize the text while preserving its original locations.

2. Match configured phrases and identifiers.

3. Apply context checks and names valid on the publication date.

4. Return relevance, supported market connections and the complete evidence trail.



### Built for volume

The CPU engine uses compiled Rust multi-pattern matching and parallel workers.

The optional CUDA research engine distributes normalization and matching across

the GPU, overlaps transfers with work, and preserves exact CPU handling where needed.

Compact Arrow results avoid creating a large Python object for every article.



### Your taxonomy, your rules

RANGE's bundled taxonomy covers economic and financial news. Taxonomies are

versioned configuration: add topics, aliases, context rules and relationships

without retraining a language model. New taxonomies still need correctness and

performance checks; their speed is not guaranteed by the default benchmark.



### What you are trying here

This live demo runs on CPU and returns rich explanations. Its small batches are

for exploring decisions. The high-throughput benchmarks measure compact bulk

screening, including GPU transfers and CPU fallback where applicable.



A relevant result means the configured screen accepted the article. It does not

infer sentiment, predict returns, or recommend a trade. No LLM, API key or model

download is required for inference.

"""

PERFORMANCE = """

## General Economy benchmark: measured speed, stated workload.



The figures below apply to the specialized economy engine. The new generic

curation API shares native text preparation but has different output and rule

semantics; these numbers do not establish its throughput. No billion-document

or biomedical benchmark has been run. Optional NER/GLiNER is future work.



| Implementation | Articles / second | Measurement |

| --- | ---: | --- |

| CUDA research record | **1,833,059** | Qualified warm screening of 10 million articles |

| CPU block-processing engine | **548,532–582,107** | Two qualified 10-million-article runs; 24–26% faster than matched CPU controls |

| Newer retained CUDA installation | **958,155** | Separate local installed-package qualification |



Measured on an Intel i7-10750H Windows laptop with 10 CPU workers; CUDA used an

RTX 2070 Max-Q. These are separate experiments, not a same-run CPU/GPU speedup.

The CUDA peak belongs to an earlier qualified implementation; later sessions

varied substantially. Do not combine its peak with later percentage gains.



Screening timings exclude file loading, imports, initialization and verification.

CUDA timings include transfers, exact CPU fallback and result assembly.

The live demo uses a different CPU environment and rich explanations, so these

figures are not its throughput or a service guarantee.



[Read the benchmark record](https://huggingface.co/spaces/Brianferrell787/range/blob/main/BENCHMARKS.md)



## Python package coming soon



The published source and preview installation files include the retained Phase60/61

CUDA implementation behind the peak benchmark. CUDA bulk screening is currently

supported on Windows, Python 3.10, and the tested sm75 GPU architecture.

[Preview installation guide](https://huggingface.co/spaces/Brianferrell787/range/blob/main/cuda/README.md).



The interactive Space runs on CPU. A public package-index release and broader

platform support are still coming; the CUDA preview uses the existing benchmark

evidence rather than a new speed measurement.



## What we want to try next



- **Semantic relevance features:** entity recognition and model-assisted taxonomy

  matching are planned. These features are not part of the current benchmark.



- **Finish the release:** clean installs, supported platforms, documentation and

  validation of the exact wheels people will use.

- **Share work across taxonomies:** build on successful two-taxonomy experiments,

  then extend packaging and memory scheduling for more rule sets.

- **Choose CPU or CUDA by workload:** measure small batches, large batches and

  routing overhead before introducing automatic selection.

- **Target remaining matching ideas:** direct StringZilla integration and other

  CUSMART methods remain candidates, not demonstrated improvements.

- **Train optimizations on separate data:** investigate corpus-trained automata

  on larger future taxonomies. A fresh compiler-profile experiment was tested and did not qualify for adoption.



Recent alternatives did not beat the retained matcher. We keep negative results

and require exact outputs before considering a speed improvement.

"""

HERO_HTML = """

<section class="hero">

  <div class="hero-inner">

    <p class="hero-eyebrow">Your data · your rules · traceable decisions</p>

    <h1>RANGE</h1>

    <p class="hero-expansion">Rapid Annotation, Normalization, and Gating Engine</p>

    <p class="hero-copy">Your taxonomy. Your curated dataset. At speed.<br>

      Define what belongs, filter your text collections, and inspect the evidence

      behind each decision.</p>

    <div class="trust-row" aria-label="Product characteristics">

      <span class="trust-pill package-pill">Python package coming soon</span>

      <span class="trust-pill">No LLM required</span>

      <span class="trust-pill">Customizable taxonomies</span>

    </div>

    <div class="speed-grid" aria-label="Recorded bulk screening benchmarks">

      <div class="speed-stat"><strong>Up to 1.83M</strong><span>articles/s · Peak CUDA benchmark</span></div>

      <div class="speed-stat"><strong>548K–582K</strong><span>records/s · General Economy CPU benchmark</span></div>

      <div class="speed-stat"><strong>Exact evidence</strong><span>Matched text, labels and selection rules</span></div>

    </div>

    <p class="benchmark-note">Recorded screening speeds using the General Economy taxonomy on a

      10-million-article benchmark. Performance varies with your hardware, taxonomy, and text.

      This interactive demo runs on CPU.</p>

  </div>

</section>

<section class="value-grid" aria-label="What RANGE returns">

  <div class="value-card"><strong>Define what belongs</strong><span>Create a taxonomy for the text you want in your dataset.</span></div>

  <div class="value-card"><strong>Curate your corpus</strong><span>Keep matching documents and assign your own labels.</span></div>

  <div class="value-card"><strong>Inspect the evidence</strong><span>Review the matched text, rules, and taxonomy version.</span></div>

</section>

"""

INITIAL_STATUS_HTML = """

<div class="status-card status-empty" role="status" aria-live="polite">

  <span class="status-label">Screening result</span>

  <h3>Your result will appear here</h3>

  <p>RANGE reports article relevance and supported broad-market exposure as two

  separate outputs.</p>

  <div class="result-grid">

    <div class="result-cell"><strong>Article relevance</strong><span>Relevant · No configured match</span></div>

    <div class="result-cell"><strong>Related market</strong><span>Shown only when the article supports a connection</span></div>

  </div>

  <p class="boundary-note">This is a relevance screen—not sentiment, materiality, a

  return forecast, or investment advice.</p>

</div>

"""


def _text_size(value: str, *, label: str) -> tuple[int, int]:
    """Return character and UTF-8 sizes with a public-safe Unicode error."""

    characters = len(value)
    try:
        utf8_bytes = len(value.encode("utf-8"))
    except UnicodeEncodeError as exc:
        raise gr.Error(f"{label} contains invalid Unicode text.") from exc
    return characters, utf8_bytes


def _validate_article_size(text: str, *, label: str = "Article") -> tuple[int, int]:
    """Validate public-request bounds before the engine sees article text."""

    characters, utf8_bytes = _text_size(text, label=label)
    if characters > MAX_ARTICLE_CHARS:
        raise gr.Error(f"{label} exceeds the {MAX_ARTICLE_CHARS:,}-character limit.")
    if utf8_bytes > MAX_ARTICLE_BYTES:
        raise gr.Error(f"{label} exceeds the {MAX_ARTICLE_BYTES:,}-byte UTF-8 limit.")
    return characters, utf8_bytes


def _bounded_metadata(

    value: Any,

    *,

    label: str,

    max_characters: int,

    max_bytes: int,

    required: bool = False,

) -> tuple[str, int, int]:
    """Normalize one metadata value after enforcing its raw request bounds."""

    if value is None:
        raw_value = ""
    elif isinstance(value, str):
        raw_value = value
    else:
        raise gr.Error(f"{label} must be text.")

    characters, utf8_bytes = _text_size(raw_value, label=label)
    if characters > max_characters:
        raise gr.Error(f"{label} exceeds the {max_characters:,}-character limit.")
    if utf8_bytes > max_bytes:
        raise gr.Error(f"{label} exceeds the {max_bytes:,}-byte UTF-8 limit.")

    cleaned = raw_value.strip()
    if required and not cleaned:
        raise gr.Error(f"Enter {label.lower()}.")
    return cleaned, characters, utf8_bytes


def _batch_cell_text(value: Any, *, label: str) -> str:
    """Accept ordinary table scalars without stringifying nested objects."""

    if value is None:
        return ""
    if isinstance(value, str):
        return value
    if isinstance(value, (bool, int, float)):
        return str(value)
    raise gr.Error(f"{label} must be plain text or a number.")


def _validate_batch_dimensions(row_count: int, column_count: int) -> None:
    """Reject oversized or malformed tables before their cells are copied."""

    if row_count > MAX_BATCH_ROWS:
        raise gr.Error(
            f"The interactive demo is capped at {MAX_BATCH_ROWS} rows per request."
        )
    if column_count != EXPECTED_BATCH_COLUMNS:
        raise gr.Error(
            "Batch input must have exactly four columns: "
            "article_id, timestamp, source, and text."
        )


def _local_error_details_enabled() -> bool:
    """Allow detailed tracebacks only through an explicit non-Space opt-in."""

    requested = os.environ.get("RANGE_LOCAL_SHOW_ERROR", "").strip().casefold()
    if requested not in {"1", "true", "yes", "on"}:
        return False
    return not any(os.environ.get(name) for name in ("SPACE_ID", "SPACE_HOST"))


@lru_cache(maxsize=1)
def _engine() -> RelevanceEngine:
    """Create and cache the immutable CPU engine used by this process."""

    return RelevanceEngine()


def _jsonable(value: Any) -> Any:
    """Convert immutable package contracts to ordinary JSON-compatible values."""

    if dataclasses.is_dataclass(value):
        return {
            field.name: _jsonable(getattr(value, field.name))
            for field in dataclasses.fields(value)
        }
    if isinstance(value, Enum):
        return value.value
    if isinstance(value, datetime):
        return value.isoformat()
    if isinstance(value, Mapping):
        return {str(_jsonable(key)): _jsonable(item) for key, item in value.items()}
    if isinstance(value, (tuple, list, set, frozenset)):
        return [_jsonable(item) for item in value]
    if isinstance(value, Path):
        return str(value)
    attributes = getattr(value, "__dict__", None)
    if isinstance(attributes, dict):
        return {
            str(key): _jsonable(item)
            for key, item in attributes.items()
            if not str(key).startswith("_")
        }
    return value


def _download_file(payload: Any, *, suffix: str) -> str:
    cache_root = Path(get_upload_folder())
    cache_root.mkdir(parents=True, exist_ok=True)
    path = cache_root / f"relevance-{uuid.uuid4().hex}{suffix}"
    if suffix == ".jsonl":
        lines = [json.dumps(row, ensure_ascii=False, sort_keys=True) for row in payload]
        path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    else:
        path.write_text(
            json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
            encoding="utf-8",
        )
    return str(path)


def _evidence_sets(prediction) -> tuple[set[str], set[str], set[str]]:
    accepted = {
        observation_id
        for path in prediction.article_decision.accepted_evidence
        for observation_id in path.observation_ids
    }
    unresolved = {
        observation_id
        for path in prediction.article_decision.unresolved_evidence
        for observation_id in path.observation_ids
    }
    rejected = {
        observation_id
        for path in prediction.article_decision.rejected_evidence
        for observation_id in path.observation_ids
    }
    return accepted, unresolved, rejected


def _evidence_rows(prediction) -> list[list[str]]:
    accepted, unresolved, rejected = _evidence_sets(prediction)
    resolutions = {
        resolution.observation_id: resolution for resolution in prediction.resolutions
    }
    paths = (
        *prediction.article_decision.accepted_evidence,
        *prediction.article_decision.unresolved_evidence,
        *prediction.article_decision.rejected_evidence,
    )
    exposure_by_observation: dict[str, set[str]] = {}
    for path in paths:
        label = path.exposure_family.label if path.exposure_family is not None else "—"
        for observation_id in path.observation_ids:
            exposure_by_observation.setdefault(observation_id, set()).add(label)

    rows: list[list[str]] = []
    for observation in prediction.observations:
        if observation.observation_id in accepted:
            status = "accepted"
        elif observation.observation_id in unresolved:
            status = "unresolved path"
        elif observation.observation_id in rejected:
            status = "rejected"
        else:
            status = "observed"
        resolution = resolutions.get(observation.observation_id)
        identity = (
            resolution.canonical_id
            if resolution is not None and resolution.canonical_id
            else (resolution.status.value if resolution is not None else "—")
        )
        lineage = observation.rule_id or observation.catalog_entry_id or "—"
        rows.append(
            [
                status,
                observation.matched_surface,
                f"{observation.span_start}:{observation.span_end}",
                observation.tag_type or observation.entity_concept,
                observation.detection_method,
                lineage,
                identity,
                ", ".join(
                    sorted(
                        exposure_by_observation.get(observation.observation_id, {"—"})
                    )
                ),
            ]
        )
    return rows


def _highlighted_article(prediction) -> str:
    text = prediction.article.text
    accepted, unresolved, rejected = _evidence_sets(prediction)

    def evidence_status(observation_id: str) -> str:
        if observation_id in accepted:
            return "accepted"
        if observation_id in unresolved:
            return "review"
        if observation_id in rejected:
            return "rejected"
        return "observed"

    status_priority = {"accepted": 0, "review": 1, "rejected": 2, "observed": 3}
    candidates = sorted(
        prediction.observations,
        key=lambda row: (
            row.span_start,
            status_priority[evidence_status(row.observation_id)],
            -(row.span_end - row.span_start),
            row.observation_id,
        ),
    )
    selected = []
    cursor = -1
    for observation in candidates:
        if observation.span_start < cursor:
            continue
        selected.append(observation)
        cursor = observation.span_end

    pieces: list[str] = []
    cursor = 0
    for observation in selected:
        pieces.append(html.escape(text[cursor : observation.span_start]))
        status = evidence_status(observation.observation_id)
        status_label = {
            "accepted": "Accepted evidence",
            "review": "Unresolved path",
            "rejected": "Rejected match",
            "observed": "Observed match",
        }[status]
        title = html.escape(
            f"{status_label} · {observation.tag_type or observation.entity_concept} · "
            f"{observation.detection_method} · "
            f"{observation.span_start}:{observation.span_end}"
        )
        surface = html.escape(text[observation.span_start : observation.span_end])
        aria_label = html.escape(f"{status_label}: {observation.matched_surface}")
        pieces.append(
            f'<mark class="evidence evidence-{status}" title="{title}" '
            f'tabindex="0" aria-label="{aria_label}">{surface}</mark>'
        )
        cursor = observation.span_end
    pieces.append(html.escape(text[cursor:]))
    legend = (
        '<div class="evidence-legend" aria-label="Evidence highlight legend">'
        '<span class="legend-key"><span class="legend-swatch swatch-accepted"></span>Accepted evidence</span>'
        '<span class="legend-key"><span class="legend-swatch swatch-review"></span>Unresolved path</span>'
        '<span class="legend-key"><span class="legend-swatch swatch-rejected"></span>Rejected match</span>'
        '<span class="legend-key"><span class="legend-swatch swatch-observed"></span>Observed match</span>'
        "</div>"
    )
    return legend + '<div class="article-preview">' + "".join(pieces) + "</div>"


def _status_card(prediction) -> str:
    card_class, badge, heading = {
        "relevant": (
            "status-relevant",
            "Relevant",
            "Article passed the configured relevance screen",
        ),
        "needs_review": (
            "status-review",
            "Needs review",
            "Possible evidence needs review",
        ),
        "not_relevant": (
            "status-no-match",
            "No configured match",
            "No accepted evidence in the bundled taxonomy",
        ),
    }[prediction.status]
    exposures = ", ".join(prediction.relevant_to)
    exposure_value = exposures or "None"
    has_unresolved_path = bool(prediction.article_decision.unresolved_evidence)
    if exposures:
        exposure_note = ""
    elif prediction.status == "relevant" and has_unresolved_path:
        exposure_note = (
            '<p class="boundary-note">The article passed at the article level, but '
            "its highlighted relationship evidence does not complete a configured "
            "path to a broad-market exposure and is shown as unresolved.</p>"
        )
    elif prediction.status == "relevant":
        exposure_note = (
            '<p class="boundary-note">The article passed at the article level, but '
            "no accepted evidence path established one of the bundled broad-market "
            "exposures.</p>"
        )
    else:
        exposure_note = (
            '<p class="boundary-note">Article relevance and exposure mapping are '
            "separate results.</p>"
        )
    return (
        f'<div class="status-card {card_class}" role="status" aria-live="polite">'
        '<span class="status-label">Screening result</span>'
        f'<span class="status-badge">{html.escape(badge)}</span>'
        f"<h3>{html.escape(heading)}</h3>"
        '<div class="result-grid">'
        '<div class="result-cell"><strong>Article relevance</strong>'
        f"<span>{html.escape(badge)}</span></div>"
        '<div class="result-cell"><strong>Related market</strong>'
        f"<span>{html.escape(exposure_value)}</span></div>"
        "</div>"
        f"<p><strong>Why RANGE returned this:</strong> {html.escape(prediction.why)}</p>"
        f"{exposure_note}"
        '<p class="boundary-note">Relevant means the article passed the configured '
        "screen. It does not mean bullish, bearish, material, or predictive.</p>"
        f'<p class="small-note">RANGE {html.escape(__version__)} · '
        f"taxonomy {html.escape(_engine().taxonomy.version)} · "
        f"CPU workers {_engine().workers}</p>"
        "</div>"
    )


def _audit_payload(prediction) -> dict[str, Any]:
    decision = _jsonable(prediction)
    article = decision.get("article") if isinstance(decision, dict) else None
    if not isinstance(article, dict):
        raise RuntimeError("RANGE produced an invalid evidence record.")

    article_text = prediction.article.text
    encoded_text = article_text.encode("utf-8")
    article.pop("text", None)
    article.update(
        {
            "text_sha256": hashlib.sha256(encoded_text).hexdigest(),
            "text_characters": len(article_text),
            "text_utf8_bytes": len(encoded_text),
            "text_normalization": "leading_and_trailing_whitespace_removed",
        }
    )
    return {
        "summary": prediction.summary(),
        "package_version": __version__,
        "backend": _jsonable(_engine().backend_info),
        "taxonomy": _engine().taxonomy.describe(),
        "decision": decision,
    }


def analyze_article(

    text: str,

    timestamp: str,

    article_id: str,

    source: str,

):
    """Screen one timestamped article for configured financial-news relevance.



    Returns a plain-language result, highlighted evidence, an application-ready

    summary, rule path, detailed evidence data, Mermaid source, and a download.

    """

    if not isinstance(text, str):
        raise gr.Error("Article text must be text.")
    _validate_article_size(text)
    cleaned_text = text.strip()
    if not cleaned_text:
        raise gr.Error("Enter non-empty article text.")
    resolved_timestamp, _, _ = _bounded_metadata(
        timestamp,
        label="Publication timestamp",
        max_characters=MAX_TIMESTAMP_CHARS,
        max_bytes=MAX_TIMESTAMP_BYTES,
        required=True,
    )
    resolved_id, _, _ = _bounded_metadata(
        article_id,
        label="Article ID",
        max_characters=MAX_ARTICLE_ID_CHARS,
        max_bytes=MAX_ARTICLE_ID_BYTES,
    )
    resolved_source, _, _ = _bounded_metadata(
        source,
        label="Source",
        max_characters=MAX_SOURCE_CHARS,
        max_bytes=MAX_SOURCE_BYTES,
    )

    try:
        started = perf_counter()
        prediction = _engine().predict(
            cleaned_text,
            timestamp=resolved_timestamp,
            article_id=resolved_id or None,
            source=resolved_source or "demo",
        )
        elapsed = perf_counter() - started
    except (TypeError, ValueError) as exc:
        raise gr.Error(
            "The article or publication timestamp is invalid. "
            "Use an ISO 8601 timestamp such as 2025-01-15T09:30:00-05:00."
        ) from exc

    audit = _audit_payload(prediction)
    timing, timing_text = processing_timing(elapsed, 1)
    audit["processing_timing"] = timing
    summary = prediction.summary()
    summary["processing_timing"] = timing
    explanation = _engine().explain(prediction, format="text", detail="details")
    diagram = _engine().explain(prediction, format="mermaid", detail="details")
    return (
        _status_card(prediction) + '<div class="status-card"><strong>'
        + html.escape(timing_text.split('\n\n')[0].strip('*'))
        + '</strong><p>' + html.escape(timing["timing_scope"]) + '</p></div>',
        _highlighted_article(prediction),
        summary,
        _evidence_rows(prediction),
        explanation,
        json.dumps(audit, ensure_ascii=False, indent=2, sort_keys=True),
        diagram,
        _download_file(audit, suffix=".json"),
    )


def _rows(value: Any) -> list[list[Any]]:
    """Return a small four-column table after validating its shape up front."""

    if value is None:
        return []

    shape = getattr(value, "shape", None)
    if shape is not None:
        try:
            dimensions = tuple(int(part) for part in shape)
        except (TypeError, ValueError) as exc:
            raise gr.Error("Batch input must be a two-dimensional table.") from exc
        if len(dimensions) != 2:
            raise gr.Error("Batch input must be a two-dimensional table.")
        _validate_batch_dimensions(dimensions[0], dimensions[1])
        if hasattr(value, "values") and hasattr(value.values, "tolist"):
            converted = value.values.tolist()
        elif hasattr(value, "tolist"):
            converted = value.tolist()
        else:
            raise gr.Error("Batch input must be a two-dimensional table.")
        if not isinstance(converted, list):
            raise gr.Error("Batch input must be a two-dimensional table.")
        rows: Sequence[Any] = converted
    else:
        if not isinstance(value, Sequence) or isinstance(
            value, (str, bytes, bytearray)
        ):
            raise gr.Error("Batch input must be a two-dimensional table.")
        _validate_batch_dimensions(len(value), EXPECTED_BATCH_COLUMNS)
        rows = value

    for index, row in enumerate(rows):
        if not isinstance(row, Sequence) or isinstance(row, (str, bytes, bytearray)):
            raise gr.Error(f"Batch row {index + 1} must contain four cells.")
        _validate_batch_dimensions(len(rows), len(row))
    return [list(row) for row in rows]


def analyze_batch(data: Any):
    """Screen up to 100 timestamped articles and return summaries plus JSONL.



    This hosted-demo endpoint uses the rich prediction path. Use the package's compact

    or Arrow APIs for high-throughput datasets.

    """

    normalized: list[tuple[str, str, str, str]] = []
    total_text_characters = 0
    total_text_bytes = 0
    total_metadata_characters = 0
    total_metadata_bytes = 0
    for index, row in enumerate(_rows(data)):
        raw_article_id, raw_timestamp, raw_source, raw_text = (
            _batch_cell_text(cell, label=f"Batch row {index + 1}, column {offset + 1}")
            for offset, cell in enumerate(row)
        )
        article_id, id_characters, id_bytes = _bounded_metadata(
            raw_article_id,
            label=f"Batch row {index + 1} article ID",
            max_characters=MAX_ARTICLE_ID_CHARS,
            max_bytes=MAX_ARTICLE_ID_BYTES,
        )
        timestamp, timestamp_characters, timestamp_bytes = _bounded_metadata(
            raw_timestamp,
            label=f"Batch row {index + 1} publication timestamp",
            max_characters=MAX_TIMESTAMP_CHARS,
            max_bytes=MAX_TIMESTAMP_BYTES,
        )
        source, source_characters, source_bytes = _bounded_metadata(
            raw_source,
            label=f"Batch row {index + 1} source",
            max_characters=MAX_SOURCE_CHARS,
            max_bytes=MAX_SOURCE_BYTES,
        )
        text_characters, text_bytes = _validate_article_size(
            raw_text, label=f"Batch row {index + 1} article"
        )
        text = raw_text.strip()

        metadata_characters = id_characters + timestamp_characters + source_characters
        metadata_bytes = id_bytes + timestamp_bytes + source_bytes
        total_metadata_characters += metadata_characters
        total_metadata_bytes += metadata_bytes
        total_text_characters += text_characters
        total_text_bytes += text_bytes
        total_content_characters = total_metadata_characters + total_text_characters
        total_content_bytes = total_metadata_bytes + total_text_bytes

        if total_metadata_characters > MAX_BATCH_METADATA_CHARS:
            raise gr.Error(
                "Batch metadata exceeds the "
                f"{MAX_BATCH_METADATA_CHARS:,}-character aggregate limit."
            )
        if total_metadata_bytes > MAX_BATCH_METADATA_BYTES:
            raise gr.Error(
                "Batch metadata exceeds the "
                f"{MAX_BATCH_METADATA_BYTES:,}-byte aggregate UTF-8 limit."
            )
        if total_text_characters > MAX_BATCH_CHARS:
            raise gr.Error(
                f"Batch article text exceeds the {MAX_BATCH_CHARS:,}-character "
                "aggregate limit."
            )
        if total_text_bytes > MAX_BATCH_BYTES:
            raise gr.Error(
                f"Batch article text exceeds the {MAX_BATCH_BYTES:,}-byte "
                "aggregate UTF-8 limit."
            )
        if total_content_characters > MAX_BATCH_CONTENT_CHARS:
            raise gr.Error(
                "Combined batch article and metadata content exceeds the "
                f"{MAX_BATCH_CONTENT_CHARS:,}-character limit."
            )
        if total_content_bytes > MAX_BATCH_CONTENT_BYTES:
            raise gr.Error(
                "Combined batch article and metadata content exceeds the "
                f"{MAX_BATCH_CONTENT_BYTES:,}-byte UTF-8 limit."
            )

        if not any((article_id, timestamp, source, text)):
            continue
        if not text:
            raise gr.Error(f"Batch row {index + 1} has no article text.")
        if not timestamp:
            raise gr.Error(f"Batch row {index + 1} has no timestamp.")
        normalized.append(
            (article_id or f"row-{index}", timestamp, source or "demo", text)
        )

    if not normalized:
        raise gr.Error("Add at least one complete batch row.")
    try:
        started = perf_counter()
        predictions = _engine().predict_batch(
            [row[3] for row in normalized],
            timestamps=[row[1] for row in normalized],
            article_ids=[row[0] for row in normalized],
            sources=[row[2] for row in normalized],
        )
        elapsed = perf_counter() - started
    except (TypeError, ValueError) as exc:
        raise gr.Error(
            "A batch row is invalid. Check that every publication timestamp is "
            "valid ISO 8601 text."
        ) from exc

    summaries = [prediction.summary() for prediction in predictions]
    table = [
        [
            summary["article_id"],
            summary["status"],
            str(summary["accepted"]),
            ", ".join(summary["relevant_to"]),
            summary["why"],
        ]
        for summary in summaries
    ]
    relevant = sum(summary["status"] == "relevant" for summary in summaries)
    review = sum(summary["status"] == "needs_review" for summary in summaries)
    no_match = len(summaries) - relevant - review
    timing, timing_text = processing_timing(elapsed, len(summaries))
    metrics = {
        **timing,
        "rows": len(summaries),
        "relevant": relevant,
        "needs_review": review,
        "no_configured_match": no_match,
        "taxonomy_fingerprint": _engine().taxonomy.fingerprint,
    }
    return table, metrics, _download_file(summaries, suffix=".jsonl"), timing_text


def build_demo() -> gr.Blocks:
    """Build the human-facing UI and named Gradio API endpoints."""

    blocks_options: dict[str, Any] = {
        "title": "RANGE",
        "fill_width": True,
        "delete_cache": (60, 300),
    }
    if GRADIO_MAJOR < 6:
        blocks_options.update(css=CSS, theme=THEME)

    def build_content() -> None:
        build_curation_tabs()
        with gr.Tab("Example taxonomy: General Economy"):
            gr.Markdown(
                "### Example taxonomy: General Economy\n"
                "This example filters economic and financial news, then shows supported market connections. "
                "Paste the text and add its publication time. Time is required so "
                "RANGE can apply time-valid names and relationships; it is not "
                "used to forecast prices."
            )
            with gr.Row():
                with gr.Column(scale=5):
                    article = gr.Textbox(
                        label="Article text",
                        lines=13,
                        value=SINGLE_EXAMPLES[0][0],
                        placeholder="Paste a headline, excerpt, or full article...",
                        max_length=MAX_ARTICLE_CHARS,
                    )
                    with gr.Row():
                        timestamp = gr.Textbox(
                            label="Publication timestamp",
                            value=SINGLE_EXAMPLES[0][1],
                            info="Required · ISO 8601; include an offset when known.",
                            max_length=MAX_TIMESTAMP_CHARS,
                        )
                    with gr.Accordion("Optional article metadata", open=False):
                        gr.Markdown(
                            "Article ID and source make exported records easier to join "
                            "back to your data. They do not change text matching."
                        )
                        with gr.Row():
                            article_id = gr.Textbox(
                                label="Article ID",
                                value=SINGLE_EXAMPLES[0][2],
                                max_length=MAX_ARTICLE_ID_CHARS,
                            )
                            source = gr.Textbox(
                                label="Source",
                                value=SINGLE_EXAMPLES[0][3],
                                max_length=MAX_SOURCE_CHARS,
                            )
                    analyze = gr.Button("Check relevance", variant="primary")
                with gr.Column(scale=4):
                    status = gr.HTML(INITIAL_STATUS_HTML)
                    with gr.Accordion("Developer output and download", open=False):
                        summary = gr.JSON(label="Application-ready summary")
                        result_file = gr.File(
                            label="Download evidence record (article text omitted)"
                        )

            example_options: dict[str, Any] = {
                "examples": SINGLE_EXAMPLES,
                "inputs": [article, timestamp, article_id, source],
                "label": "Try a representative scenario",
                "cache_examples": False,
            }
            if GRADIO_MAJOR >= 6:
                example_options["example_labels"] = [row[0] for row in ARTICLE_EXAMPLES]
            gr.Examples(**example_options)

            gr.Markdown(
                "### Evidence in the article\n"
                "Highlights distinguish evidence that supported the result from "
                "unresolved evidence paths and matches rejected by context checks."
            )
            highlighted = gr.HTML(
                '<div class="article-preview empty-preview">Run the screen to see '
                "matched evidence in its original context.</div>"
            )
            evidence = gr.Dataframe(
                headers=EVIDENCE_HEADERS,
                datatype=["str"] * len(EVIDENCE_HEADERS),
                interactive=False,
                label="Evidence and rule path",
            )
            with gr.Accordion("Why RANGE returned this result", open=True):
                explanation = gr.Textbox(lines=16, label=None)
            with gr.Accordion("Evidence record (article text omitted)", open=False):
                audit_json = gr.Code(language="json", lines=24, label=None)
            with gr.Accordion("Mermaid evidence-flow source (developer)", open=False):
                diagram = gr.Code(language="markdown", lines=20, label=None)

            analyze.click(
                fn=analyze_article,
                inputs=[article, timestamp, article_id, source],
                outputs=[
                    status,
                    highlighted,
                    summary,
                    evidence,
                    explanation,
                    audit_json,
                    diagram,
                    result_file,
                ],
                api_name="analyze",
                scroll_to_output=True,
                concurrency_limit=1,
                concurrency_id="range-cpu",
            )

        with gr.Tab("General Economy: batch demo"):
            gr.Markdown(
                "This is the rich, explainable demo path—not the package's fastest "
                "compact benchmark path. Up to 100 rows; 20,000 characters / "
                "80,000 UTF-8 bytes per article; "
                "200,000 characters / 800,000 bytes per batch. Package APIs are designed "
                "for larger batches."
            )
            batch_input = gr.Dataframe(
                headers=["article_id", "timestamp", "source", "text"],
                datatype=["str", "str", "str", "str"],
                value=BATCH_ROWS,
                **(
                    {
                        "row_count": len(BATCH_ROWS),
                        "column_count": 4,
                    }
                    if GRADIO_MAJOR >= 6
                    else {
                        "row_count": (len(BATCH_ROWS), "dynamic"),
                        "col_count": (4, "fixed"),
                    }
                ),
                type="array",
                interactive=True,
                label="Timestamped articles",
            )
            screen = gr.Button("Screen batch", variant="primary")
            batch_timing = gr.Markdown()
            with gr.Row():
                batch_results = gr.Dataframe(
                    headers=BATCH_HEADERS,
                    datatype=["str"] * len(BATCH_HEADERS),
                    interactive=False,
                    label="Screening results",
                )
                batch_metrics = gr.JSON(label="Batch summary")
            batch_file = gr.File(label="Download summaries as JSONL")
            screen.click(
                fn=analyze_batch,
                inputs=batch_input,
                outputs=[batch_results, batch_metrics, batch_file, batch_timing],
                api_name="batch",
                concurrency_limit=1,
                concurrency_id="range-cpu",
            )

        with gr.Tab("General Economy: taxonomy guide"):
            gr.Markdown(ABOUT)
            gr.Markdown(
                "The named API endpoints are `/analyze` and `/batch`. After deployment "
                "as a compatible Hugging Face Space, the Space also exposes its generated "
                "API documentation and `agents.md` instructions for coding agents."
            )

        with gr.Tab("Performance & roadmap"):
            gr.Markdown(PERFORMANCE)

        gr.Markdown(
            "---\n"
            '<div class="privacy-note"><strong>Research alpha · privacy notice.</strong> '
            "RANGE records configured evidence; selection is not a guarantee of factual "
            "or scientific validity. In a hosted Space, submitted text and generated "
            "downloads are processed temporarily on the Space host. Do not submit "
            "private, licensed, or confidential text unless you have permission.</div>"
        )

    with gr.Blocks(**blocks_options) as demo:
        gr.HTML(HERO_HTML, elem_classes=["hero-host"])
        with gr.Column(elem_classes=["content-shell"]):
            build_content()
    return demo


demo = build_demo()


if __name__ == "__main__":
    launch_options: dict[str, Any] = {
        "server_name": os.environ.get("GRADIO_SERVER_NAME", "0.0.0.0"),
        "server_port": int(os.environ.get("GRADIO_SERVER_PORT", "7860")),
        "show_error": _local_error_details_enabled(),
        "app_kwargs": {
            "middleware": [
                Middleware(
                    RequestBodyLimitMiddleware,
                    max_body_size=MAX_HTTP_REQUEST_BYTES,
                )
            ]
        },
    }
    if GRADIO_MAJOR >= 6:
        launch_options.update(css=CSS, theme=THEME)
    demo.queue(api_open=False, max_size=32).launch(**launch_options)