# app/ui_streamlit.py
# Ensure project root is on sys.path when Streamlit runs this as a script
import sys, pathlib
ROOT = pathlib.Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
import os, json
from pathlib import Path
from datetime import date, datetime
import streamlit as st
from app.main import get_env, ensure_index_exists
from app.search import search
# ── Page config ───────────────────────────────────────────────────────────────
st.set_page_config(page_title="Grants Discovery App By Lupo", page_icon="🧭", layout="wide")
# ── THEME / CSS — single, unified block (dark app; readable controls) ─────────
st.markdown("""
""", unsafe_allow_html=True)
# ── Hero ──────────────────────────────────────────────────────────────────────
st.markdown("""
Grants Discovery Live RAG by Lupo
Find capacity-building grants fast.
""", unsafe_allow_html=True)
# ── Environment & index ───────────────────────────────────────────────────────
_env = get_env()
ensure_index_exists(_env)
# ── Helpers ───────────────────────────────────────────────────────────────────
def _norm_list(v):
if v is None: return []
if isinstance(v, str):
parts = [p.strip() for p in v.replace(";", ",").split(",")]
return [p.lower() for p in parts if p]
if isinstance(v, (list, tuple, set)):
return [str(x).lower() for x in v]
return []
def _matches_filters(rec, geo_sel, cat_sel):
rec_geo = _norm_list(rec.get("geo") or rec.get("region") or rec.get("state"))
rec_cat = _norm_list(rec.get("categories") or rec.get("cats") or rec.get("category"))
g_ok = (not geo_sel) or (set([g.lower() for g in geo_sel]) & set(rec_geo))
c_ok = (not cat_sel) or (set([c.lower() for c in cat_sel]) & set(rec_cat))
return g_ok and c_ok
def _ministry_filter(rows):
if not rows: return rows
banned_terms = [
"broad agency announcement", "baa", "research", "r&d", "prototype",
"laboratory", "university", "sbir", "sttr",
"darpa", "office of naval research", "onr", "naval", "air force", "army",
"w911", "n00014", "fa-", "afrl", "arpa"
]
preferred_agencies = {
"FTA","HHS","ACL","USDA","USDA-FNS","USDA-RD","DOL","DOJ","OJP","OVW","EDA","HRSA","SAMHSA","CFPB","HUD"
}
terms = [
"vehicle","van","bus","paratransit","mobility",
"congregate meals","home-delivered meals","senior nutrition",
"food pantry","food bank","hunger relief","refrigeration","freezer",
"community","faith","church","ministry","nonprofit",
"reentry","workforce","case management","technical assistance","capacity"
]
def txt(r):
return " ".join([str(r.get("title","")),
str(r.get("synopsis") or r.get("summary") or ""),
str(r.get("agency") or "")]).lower()
kept=[]
for r in rows:
t = txt(r)
if any(b in t for b in banned_terms): continue
agency = (r.get("agency") or "").upper()
cats = [c.lower() for c in (r.get("categories") or [])] if isinstance(r.get("categories"), list) else []
prefer = any(agency.startswith(a) for a in preferred_agencies)
has_cue = any(term in t for term in terms) or any(
c in {"transportation","vehicle","elderly","disabled","food","community","justice","reentry","workforce"} for c in cats
)
if prefer or has_cue:
kept.append(r)
return kept
def _to_date(d):
if not d: return None
try: return datetime.fromisoformat(str(d)).date()
except Exception: return None
def _days_until(iso):
if not iso: return None
try:
d = datetime.fromisoformat(str(iso)).date()
return (d - date.today()).days
except Exception:
return None
def _deadline_badge(days_left):
if days_left is None: return "🟦 TBD"
if days_left < 0: return "⬛ Closed"
if days_left <= 14: return f"🟥 Due in {days_left}d"
if days_left <= 30: return f"🟨 {days_left}d"
return f"🟩 {days_left}d"
# ── UI: Presets & inputs ──────────────────────────────────────────────────────
st.title("Grants Discovery RAG (Capacity Building)")
preset = st.radio(
"Quick topic:",
["General", "Elderly", "Prison Ministry", "Evangelism", "Vehicles/Transport", "FTA 5310"],
horizontal=True
)
default_q = {
"General": "capacity building",
"Elderly": "capacity building for seniors and aging services",
"Prison Ministry": "capacity building for reentry and prison ministry",
"Evangelism": "capacity building for faith and community outreach",
"Vehicles/Transport": "capacity building transportation vehicles vans buses mobility",
"FTA 5310": "5310 Enhanced Mobility Seniors Individuals with Disabilities",
}.get(preset, "capacity building")
q = st.text_input("Search query", value=default_q)
geo = st.multiselect("Geo filter (optional)", options=["US", "MD", "PA"], default=[])
categories = st.multiselect(
"Category filter (optional)",
options=[
"capacity_building","elderly","prison_ministry","evangelism",
"transportation","vehicle","justice","reentry",
"victim_services","youth","women","food","workforce"
],
default=[]
)
# Fetch more so pagination is meaningful
top_k = st.slider("Fetch up to (results)", 50, 500, 200, step=50)
sort_by = st.selectbox("Sort by", ["Relevance", "Deadline (soonest first)"], index=0)
only_open = st.checkbox("Only show opportunities with a future deadline", value=True)
ministry_focus = st.checkbox("Ministry Focus (hide research/defense/academic BAAs)", value=True)
view = st.selectbox("View", ["All", "Saved", "Hidden"], index=0)
# Agencies facet from meta
try:
meta_for_agencies = json.loads(Path(get_env()["INDEX_DIR"], "meta.json").read_text())
agency_options = sorted({m.get("agency") for m in meta_for_agencies if m.get("agency")})
except Exception:
agency_options = []
sel_agencies = st.multiselect("Agency filter (optional)", options=agency_options, default=[])
backend_filters = {}
if geo: backend_filters["geo"] = geo
if categories: backend_filters["categories"] = categories
if sel_agencies: backend_filters["agency"] = sel_agencies
# Sprint 2: Save/Hide state
if "saved_ids" not in st.session_state: st.session_state.saved_ids = set()
if "hidden_ids" not in st.session_state: st.session_state.hidden_ids = set()
def _save_item(item_id: str):
st.session_state.saved_ids.add(item_id)
st.session_state.hidden_ids.discard(item_id)
st.experimental_rerun()
def _hide_item(item_id: str):
st.session_state.hidden_ids.add(item_id)
st.session_state.saved_ids.discard(item_id)
st.experimental_rerun()
# ── Search & filter pipeline (stores full result set) ─────────────────────────
c1, c2 = st.columns([1,1])
with c1:
if st.button("Search"):
try:
raw = search(q, get_env(), top_k=top_k, filters=backend_filters) # fetch many
# Geo/Category client-side fallback
if geo or categories:
base_filtered = [r for r in raw if _matches_filters(r, geo, categories)]
else:
base_filtered = raw
# Only open
def _to_date_safe(val):
if not val: return None
try: return datetime.fromisoformat(str(val)).date()
except Exception: return None
open_filtered = base_filtered
if only_open:
open_filtered = [r for r in base_filtered
if (_to_date_safe(r.get("deadline")) or date.max) >= date.today()]
# Agency
if sel_agencies:
af = set(sel_agencies)
open_filtered = [r for r in open_filtered if (r.get("agency") in af)]
# Ministry
final_results = _ministry_filter(open_filtered) if ministry_focus else open_filtered
st.session_state["results"] = final_results
st.session_state["last_query"] = q
st.session_state["last_filters"] = {
"geo": geo, "categories": categories, "only_open": only_open,
"ministry_focus": ministry_focus, "agencies": sel_agencies,
}
# RESET PAGINATION on new run
st.session_state.page = 1
st.success(f"Fetched {len(raw)} • After filters: {len(final_results)}")
except Exception as e:
st.error(str(e))
with c2:
if st.button("Export Results to CSV"):
results_for_export = st.session_state.get("results", [])
if not results_for_export:
st.warning("No results to export. Run a search first.")
else:
out_dir = get_env()["EXPORT_DIR"]
os.makedirs(out_dir, exist_ok=True)
out_path = os.path.join(out_dir, "results.csv")
import pandas as pd
pd.DataFrame(results_for_export).to_csv(out_path, index=False)
st.success(f"Exported to {out_path}")
st.markdown("---")
# ── Post-search view/sort/pagination (5.4) ────────────────────────────────────
results = st.session_state.get("results", [])
ran_search = bool(st.session_state.get("last_query"))
# View filter
if view == "Saved":
results = [r for r in results if r.get("id") in st.session_state.saved_ids]
elif view == "Hidden":
results = [r for r in results if r.get("id") in st.session_state.hidden_ids]
# Sort
if sort_by.startswith("Deadline") and results:
results.sort(
key=lambda r: (
_to_date(r.get("deadline")) is None,
_to_date(r.get("deadline")) or date.max,
)
)
# Pagination state
if "page_size" not in st.session_state:
st.session_state.page_size = 25
if "page" not in st.session_state:
st.session_state.page = 1
total = len(results)
st.caption(f"Results: {total}")
# Controls
cols = st.columns([1,1,2,2,2])
with cols[0]:
page_size = st.selectbox("Page size", [10, 25, 50, 100], index=1)
st.session_state.page_size = page_size
# compute pages
total_pages = max(1, (total + page_size - 1) // page_size)
with cols[1]:
page = st.number_input("Page", min_value=1, max_value=total_pages,
value=min(st.session_state.page, total_pages), step=1)
st.session_state.page = page
# Slice AFTER filters & sort
start = (st.session_state.page - 1) * st.session_state.page_size
end = min(start + st.session_state.page_size, total)
page_items = results[start:end]
st.caption(f"Showing {start+1 if total else 0}–{end} of {total} • Page {st.session_state.page}/{total_pages}")
# Nav buttons
prev_col, _, next_col = st.columns([1,6,1])
with prev_col:
if st.button("◀ Prev", disabled=(st.session_state.page <= 1)):
st.session_state.page = max(1, st.session_state.page - 1)
st.experimental_rerun()
with next_col:
if st.button("Next ▶", disabled=(st.session_state.page >= total_pages)):
st.session_state.page = min(total_pages, st.session_state.page + 1)
st.experimental_rerun()
# ── Render page items ─────────────────────────────────────────────────────────
def _render_card(r):
title = r.get("title", "(no title)")
url = r.get("url", "")
cats = r.get("categories") or r.get("cats") or []
geo_tags = r.get("geo") or []
_id = r.get("id") or r.get("url") or title
posted = r.get("posted_date") or ""
deadline = r.get("deadline") or ""
days_left = _days_until(deadline)
st.markdown(f"
", unsafe_allow_html=True)
st.markdown(f"### {title}")
meta = f"**Source:** {r.get('source','')} • **Geo:** {', '.join(geo_tags) if isinstance(geo_tags,list) else geo_tags} • **Categories:** {', '.join(cats) if isinstance(cats,list) else cats}"
st.markdown(f"
{meta}
", unsafe_allow_html=True)
# Link / score
if url and not url.startswith('http'):
st.caption("Note: This item may display an ID instead of a full link. Open on Grants.gov if needed.")
if url:
st.write(f"[Open Link]({url})")
if r.get("score") is not None:
st.caption(f"Score: {r.get('score', 0):.3f}")
# Deadline
st.caption(f"Posted: {posted} • Deadline: {deadline} • {_deadline_badge(days_left)}")
# Save/Hide
c1, c2, _ = st.columns([1,1,6])
if c1.button(("✅ Saved" if _id in st.session_state.saved_ids else "💾 Save"), key=f"save-{_id}"):
_save_item(_id)
if c2.button(("🙈 Hidden" if _id in st.session_state.hidden_ids else "🙈 Hide"), key=f"hide-{_id}"):
_hide_item(_id)
st.markdown("
", unsafe_allow_html=True)
if page_items:
for r in page_items:
_render_card(r)
else:
if ran_search:
st.info("No active grants match these filters right now.")
else:
st.info("Enter a query and click Search.")
st.markdown("""
""", unsafe_allow_html=True)