"""Comic panel search UI."""
import os
import sys
from pathlib import Path
import streamlit as st
from dotenv import load_dotenv
ROOT = Path(__file__).parent
sys.path.insert(0, str(ROOT))
load_dotenv(ROOT / ".env")
from src.storage.images import image_src
st.set_page_config(
page_title="Comic Panel Search",
page_icon="π₯",
layout="wide",
initial_sidebar_state="collapsed",
)
st.markdown("""
""", unsafe_allow_html=True)
# ββ constants & resources βββββββββββββββββββββββββββββββββββββββββββββββββββββ
NAMESPACE = os.environ.get("PINECONE_NAMESPACE", "comics-v1")
SHOW_FIELDS = ["comic_id", "book_id", "page_num", "panel_num", "source", "ocr_text"]
OPS = ["==", "!=", ">", ">=", "<", "<=", "In", "Not In", "Match phrase", "Match all", "Match any"]
_MATCH_OPS = {"Match phrase", "Match all", "Match any"}
_OP_MAP = {
"==": "$eq", "!=": "$ne",
">": "$gt", ">=": "$gte",
"<": "$lt", "<=": "$lte",
"In": "$in", "Not In": "$nin",
"Match phrase": "$match_phrase",
"Match all": "$match_all",
"Match any": "$match_any",
}
# Compact operator labels for the one-line filter summary chips.
_OP_SHORT = {
"==": "=", "!=": "β ", ">": ">", ">=": "β₯", "<": "<", "<=": "β€",
"In": "in", "Not In": "not in",
"Match phrase": "phrase", "Match all": "has all", "Match any": "has any",
}
ALL_FIELDS = ["comic_id", "book_id", "page_id", "page_num", "panel_num", "source", "ocr_text", "search_text", "image_path", "is_ad_page"]
@st.cache_resource(show_spinner="Connecting to Pineconeβ¦")
def _index():
from pinecone import Pinecone
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
return pc.preview.index(name=os.environ.get("PINECONE_INDEX_NAME", "comic-panels"))
@st.cache_resource(show_spinner="Loading CLIP modelβ¦")
def _load_clip():
from src.embeddings.embed_images import _init_model
_init_model()
# The sound index is a SEPARATE, standard dense index (classic API), unlike the
# comic-panels document-schema index above β see src/sounds/.
@st.cache_resource(show_spinner="Connecting to sound indexβ¦")
def _sound_index():
from pinecone import Pinecone
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
return pc.Index(os.environ.get("PINECONE_SOUND_INDEX_NAME", "comic-sounds"))
# ββ helpers βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def _esc(s: str) -> str:
return s.replace("&", "&").replace("<", "<").replace(">", ">")
def _coerce(s: str):
try: return int(s)
except ValueError: pass
try: return float(s)
except ValueError: pass
if s.lower() in ("true", "yes"): return True
if s.lower() in ("false", "no"): return False
return s
def _add_filter():
fid = st.session_state.filter_next_id
st.session_state.filter_rows = st.session_state.filter_rows + [fid]
st.session_state.filter_next_id += 1
st.session_state[f"fk_{fid}"] = "search_text"
def _remove_filter(fid: int):
st.session_state.filter_rows = [r for r in st.session_state.filter_rows if r != fid]
def _clear_filters():
"""Drop every filter row and its backing widget state β fresh slate for a new query."""
for fid in st.session_state.filter_rows:
for prefix in ("fk_", "fo_", "fv_"):
st.session_state.pop(f"{prefix}{fid}", None)
st.session_state.filter_rows = []
def _build_filters() -> dict:
row_conds = []
for fid in st.session_state.filter_rows:
k = st.session_state.get(f"fk_{fid}", "").strip()
op = st.session_state.get(f"fo_{fid}", "==")
val_str = st.session_state.get(f"fv_{fid}", "").strip()
if not k or not val_str:
continue
if op in ("In", "Not In"):
vals = [_coerce(v.strip()) for v in val_str.split(",")]
row_conds.append({k: {"$in" if op == "In" else "$nin": vals}})
elif op in _MATCH_OPS:
row_conds.append({k: {_OP_MAP[op]: val_str}})
else:
row_conds.append({k: {_OP_MAP[op]: _coerce(val_str)}})
combinator = st.session_state.get("filter_combinator", "And")
if len(row_conds) == 0:
user_filter = None
elif len(row_conds) == 1:
user_filter = row_conds[0]
elif combinator == "Or":
user_filter = {"$or": row_conds}
else:
user_filter = {"$and": row_conds}
exclude = st.session_state.get("exclude_ads", True)
if exclude and user_filter:
return {"$and": [{"is_ad_page": {"$eq": False}}, user_filter]}
elif exclude:
return {"is_ad_page": {"$eq": False}}
elif user_filter:
return user_filter
return {}
def _card(hit: dict, rank: int, show_fields: list):
score = hit.get("rrf_score") or hit.get("_score") or 0.0
sources = hit.get("sources") or []
hit_id = _esc(str(hit.get("_id", "β")))
tags = "".join(f'{s}' for s in sources)
img_html = ""
if src := image_src(hit.get("image_path", "")):
img_html = f''
fields = ""
for k in show_fields:
v = hit.get(k)
if v is None or str(v).strip() == "":
continue
s = str(v)
if len(s) > 500:
s = s[:500] + "β¦"
fields += f'
Try:
', unsafe_allow_html=True) for i, (col, label) in enumerate(zip(cols[1:], STRIP_EXAMPLES)): with col: st.button(label, key=f"strip_{i}", width="stretch", on_click=_use_strip, args=(label,)) with cols[6], st.popover("More examples", key="strip_more", width="stretch"): for title, caption, kind, items in EXAMPLE_SECTIONS: st.markdown(f'{title}
', unsafe_allow_html=True) st.caption(caption) for item in items: if kind == "combined": label, queries = item st.button(label, key=f"exΒ·{label}", width="stretch", on_click=_use_combined, args=(queries,)) elif kind == "filter_dense": label = item[0] st.button(label, key=f"exΒ·{label}", width="stretch", on_click=_use_filter_dense, args=item[1:]) else: label, query = item st.button(label, key=f"exΒ·{label}", width="stretch", on_click=_use_example, args=(query, kind)) # ββ two-column layout βββββββββββββββββββββββββββββββββββββββββββββββββββββββββ left, right = st.columns([1, 2.8], gap="large") with left: # ββ Search signals ββββββββββββββββββββββββββββββββββββββββββββββββββββ st.markdown('Search signals
', unsafe_allow_html=True) # Full-text fts_on = st.toggle("Full-text search", key="fts_on") if fts_on: st.selectbox( "FTS type", ["query_string", "text"], key="fts_type", label_visibility="collapsed", help="query_string: multi-field Lucene syntax (AND/OR/phrases) Β· text: BM25 on a single field", ) st.text_input( "fts_query", key="fts_q", label_visibility="collapsed", placeholder=( 'search_text:("secret formula") Β· search_text:(explosion^2 OR fire) Β· search_text:(hero NOT villain)' if st.session_state.fts_type == "query_string" else "plain text, ranked by BM25: secret formula" ), ) # Dense dense_on = st.toggle("Dense β visual similarity", key="dense_on") if dense_on: st.text_input( "dense_query", key="dense_q", label_visibility="collapsed", placeholder="describe what you see: hero flying through sky", help="Embedded with OpenCLIP ViT-B/16", ) # Sparse sparse_on = st.toggle("Sparse β keyword expansion", key="sparse_on") if sparse_on: st.text_input( "sparse_query", key="sparse_q", label_visibility="collapsed", placeholder="keywords: secret formula villain escape", help="Embedded with pinecone-sparse-english-v0", ) st.divider() # ββ Filters βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ n_active = len(st.session_state.filter_rows) fhdr_label = f"Filters Β· {n_active}" if n_active else "Filters" fhdr_l, fhdr_r = st.columns([1.4, 1.6], vertical_alignment="center") with fhdr_l: st.markdown(f'{fhdr_label}
', unsafe_allow_html=True) if st.session_state.filter_rows: with fhdr_r: st.selectbox("Combine", ["And", "Or"], key="filter_combinator", label_visibility="collapsed", help="How to combine multiple filter conditions") for fid in st.session_state.filter_rows: k = st.session_state.get(f"fk_{fid}", "search_text") op = st.session_state.get(f"fo_{fid}", "==") val = st.session_state.get(f"fv_{fid}", "").strip() summary = f"{k} Β· {_OP_SHORT.get(op, op)} Β· {val or 'β¦'}" c1, c2 = st.columns([5.6, 0.8]) with c1: with st.popover(summary, key=f"fp_{fid}", width="stretch"): st.selectbox("Field", ALL_FIELDS, key=f"fk_{fid}") st.selectbox("Operator", OPS, key=f"fo_{fid}", help="Match phrase / all / any run full-text matching server-side") st.text_input("Value", key=f"fv_{fid}", placeholder="comma-separated for In / Not In") with c2: st.button("β", key=f"fr_{fid}", on_click=_remove_filter, args=(fid,), width="stretch") st.button("+ Add filter", key="add_filter", on_click=_add_filter, width="stretch") st.checkbox("Exclude ad pages", key="exclude_ads") st.divider() # ββ Results βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ st.markdown('Results
', unsafe_allow_html=True) st.number_input("Top K", min_value=1, max_value=200, step=5, key="top_k") st.multiselect("Metadata to show", ALL_FIELDS, key="show_fields", placeholder="Fields to display in results", help="Which metadata fields appear on each result card") st.caption("Metadata fields shown on each result card.") st.write("") if st.button("Run query", type="primary", width="stretch"): st.session_state.run = True # ββ run search ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ if st.session_state.run: st.session_state.run = False fts_on = st.session_state.fts_on dense_on = st.session_state.dense_on sparse_on = st.session_state.sparse_on if not any([fts_on, dense_on, sparse_on]): st.session_state.result_meta = {"error": "Enable at least one search signal."} st.session_state.results = [] else: idx = _index() filters = _build_filters() top_k = int(st.session_state.top_k) from src.search.search_dense import search_dense from src.search.search_sparse import search_sparse from src.search.search_fts import search_fts from src.search.fusion import rrf_merge sim_vec = st.session_state.get("_similar_vec") sim_id = st.session_state.get("_similar_id") if sim_vec is not None: st.session_state["_similar_vec"] = None st.session_state["_similar_id"] = None groups, error = [], None try: if dense_on and (sim_vec is not None or st.session_state.dense_q.strip()): if sim_vec is not None: vec = sim_vec else: _load_clip() # CLIP (torch) is only needed to embed a dense *text* query from src.embeddings.embed_images import embed_text_query vec = embed_text_query(st.session_state.dense_q.strip()) r = search_dense(idx, NAMESPACE, vec, top_k=top_k, filters=filters) groups.append(("dense", r.get("result", {}).get("hits", []))) if sparse_on and st.session_state.sparse_q.strip(): from src.embeddings.embed_sparse_text import embed_sparse_query svec = embed_sparse_query(st.session_state.sparse_q.strip()) if svec: r = search_sparse(idx, NAMESPACE, svec, top_k=top_k, filters=filters) groups.append(("sparse", r.get("result", {}).get("hits", []))) if fts_on and st.session_state.fts_q.strip(): r = search_fts(idx, NAMESPACE, st.session_state.fts_q.strip(), top_k=top_k, filters=filters, fts_type=st.session_state.fts_type) groups.append(("fts", r.get("result", {}).get("hits", []))) except Exception as exc: error = str(exc) if error: st.session_state.results = [] st.session_state.result_meta = {"error": error} elif not groups: st.session_state.results = [] st.session_state.result_meta = {"info": "No query text provided for the enabled signals."} else: merged = rrf_merge(groups) src = "hybrid Β· RRF" if len(groups) > 1 else groups[0][0] if sim_id: src = f"similar to {sim_id} Β· {src}" st.session_state.results = merged[:top_k] st.session_state.result_meta = {"source": src, "top_k": top_k} # ββ results panel βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ with right: results = st.session_state.results meta = st.session_state.result_meta if meta.get("error"): st.error(meta["error"]) elif meta.get("info"): st.warning(meta["info"]) elif results is None: st.markdown('