spb/uqo-research-db

1# ============================================================2# SPID — structured & keyword search over the project DB3# ============================================================4from __future__ import annotations56from spid.store import Store789def search_projects(store: Store, *, keyword: str | None = None,10                    project_type: str | None = None, sector: str | None = None,11                    technology: str | None = None, location: str | None = None,12                    ticker: str | None = None, min_amount: float | None = None,13                    year_from: int | None = None, year_to: int | None = None,14                    status: str | None = None, limit: int = 25) -> list[dict]:15    w, p = [], []16    if keyword:17        w.append("(lower(project_name) LIKE ? OR lower(description) LIKE ?)")18        p += [f"%{keyword.lower()}%", f"%{keyword.lower()}%"]19    if project_type:20        w.append("project_type = ?"); p.append(project_type)21    if sector:22        w.append("lower(sector) LIKE ?"); p.append(f"%{sector.lower()}%")23    if technology:24        w.append("list_contains(technologies, ?)"); p.append(technology)25    if location:26        w.append("lower(canonical_location) LIKE ?"); p.append(f"%{location.lower()}%")27    if ticker:28        w.append("ticker = ?"); p.append(ticker.upper())29    if min_amount is not None:30        w.append("total_amount_usd >= ?"); p.append(float(min_amount))31    if year_from is not None:32        w.append("last_year >= ?"); p.append(int(year_from))33    if year_to is not None:34        w.append("first_year <= ?"); p.append(int(year_to))35    if status:36        w.append("status = ?"); p.append(status)37    where = (" WHERE " + " AND ".join(w)) if w else ""38    sql = (f"SELECT project_id, ticker, company_name, sector, project_type, project_name, "39           f"canonical_location, technologies, status, total_amount_usd, n_filings, "40           f"first_seen, last_seen, avg_confidence, description "41           f"FROM projects{where} "42           f"ORDER BY coalesce(total_amount_usd,0) DESC, n_filings DESC LIMIT {int(limit)}")43    return store.con.execute(sql, p).to_arrow_table().to_pylist()444546def get_project(store: Store, project_id: str) -> dict | None:47    rows = store.con.execute(48        "SELECT * FROM projects WHERE project_id = ?", [project_id]49    ).to_arrow_table().to_pylist()50    if not rows:51        return None52    proj = rows[0]53    proj["timeline"] = store.con.execute(54        "SELECT filing_date, form_type, status, amount_usd, confidence, snippet "55        "FROM project_timeline WHERE project_id = ? ORDER BY filing_date", [project_id]56    ).to_arrow_table().to_pylist()57    return proj58