Spaces:
Sleeping
Sleeping
| """ | |
| SQL-based FDA label retriever (Option D). | |
| Replaces ChromaDB / vector search. Exact-match key lookup via SQLite. | |
| Resolution order: | |
| 1. Brand-keyed query (if brand provided) -> returns brand-specific label | |
| 2. Generic-keyed query -> returns label by generic name | |
| 3. Aliases table fallback -> resolves salt forms, common misspellings, etc. | |
| 4. Returns None if not found | |
| Usage: | |
| from sql_retriever import FDALabelRetriever | |
| r = FDALabelRetriever("fda_labels.db") | |
| label = r.fetch_label(name="atorvastatin", brand="Lipitor") | |
| section_text = r.get_section(label["set_id"], "drug_interactions") | |
| """ | |
| import sqlite3 | |
| from pathlib import Path | |
| from typing import Optional | |
| # Standard FDA SPL section column names (must match schema.sql) | |
| SPL_SECTIONS = [ | |
| "indications_and_usage", | |
| "dosage_and_administration", | |
| "dosage_forms_and_strengths", | |
| "contraindications", | |
| "warnings_and_precautions", | |
| "boxed_warning", | |
| "adverse_reactions", | |
| "drug_interactions", | |
| "use_in_specific_populations", | |
| "drug_abuse_and_dependence", | |
| "overdosage", | |
| "description", | |
| "clinical_pharmacology", | |
| "clinical_studies", | |
| ] | |
| class FDALabelRetriever: | |
| """SQL-based exact-match FDA label retrieval.""" | |
| def __init__(self, db_path: str | Path): | |
| self.db_path = Path(db_path) | |
| if not self.db_path.exists(): | |
| raise FileNotFoundError(f"FDA label database not found: {self.db_path}") | |
| self.conn = sqlite3.connect(str(self.db_path)) | |
| self.conn.row_factory = sqlite3.Row | |
| def close(self): | |
| self.conn.close() | |
| def __enter__(self): | |
| return self | |
| def __exit__(self, *args): | |
| self.close() | |
| def fetch_label(self, name: str, brand: Optional[str] = None) -> Optional[dict]: | |
| """ | |
| Fetch a single FDA label by drug name and optional brand. | |
| Resolution order: | |
| 1. Brand-keyed exact match (if brand provided) | |
| 2. Generic-keyed exact match | |
| 3. Aliases table fallback | |
| Returns the full label row as dict, or None if not found. | |
| """ | |
| # 1. Brand-keyed exact match | |
| if brand: | |
| row = self.conn.execute( | |
| "SELECT * FROM fda_labels WHERE LOWER(brand_name) = LOWER(?)", | |
| (brand,), | |
| ).fetchone() | |
| if row: | |
| return dict(row) | |
| # 1b. Brand-keyed prefix match (handles "Wellbutrin" -> "WELLBUTRIN XL", | |
| # "Tylenol" -> "TYLENOL Extra Strength", etc.) | |
| # Order by section completeness to prefer the most fully-populated label. | |
| row = self.conn.execute( | |
| """ | |
| SELECT * | |
| FROM fda_labels | |
| WHERE LOWER(brand_name) LIKE LOWER(?) || '%' | |
| ORDER BY | |
| -- prefer shortest brand_name match (closer to user's query) | |
| LENGTH(brand_name), | |
| -- then prefer most-populated label | |
| (CASE WHEN indications_and_usage IS NOT NULL THEN 1 ELSE 0 END + | |
| CASE WHEN dosage_and_administration IS NOT NULL THEN 1 ELSE 0 END + | |
| CASE WHEN adverse_reactions IS NOT NULL THEN 1 ELSE 0 END + | |
| CASE WHEN drug_interactions IS NOT NULL THEN 1 ELSE 0 END) DESC | |
| LIMIT 1 | |
| """, | |
| (brand,), | |
| ).fetchone() | |
| if row: | |
| return dict(row) | |
| # 2. Generic-keyed lookup | |
| row = self.conn.execute( | |
| "SELECT * FROM fda_labels WHERE LOWER(generic_name) = LOWER(?)", | |
| (name,), | |
| ).fetchone() | |
| if row: | |
| return dict(row) | |
| # 3. Aliases table fallback (resolves salt forms, etc.) | |
| # Pick the alias with the most metadata-complete label | |
| row = self.conn.execute( | |
| """ | |
| SELECT l.* | |
| FROM fda_labels l | |
| JOIN drug_aliases a ON l.set_id = a.canonical_set_id | |
| WHERE LOWER(a.alias) = LOWER(?) | |
| ORDER BY | |
| -- Prefer labels with brand match if user gave brand | |
| CASE WHEN ? IS NOT NULL AND LOWER(l.brand_name) = LOWER(?) THEN 0 ELSE 1 END, | |
| -- Then prefer labels with more populated sections | |
| (CASE WHEN l.indications_and_usage IS NOT NULL THEN 1 ELSE 0 END + | |
| CASE WHEN l.dosage_and_administration IS NOT NULL THEN 1 ELSE 0 END + | |
| CASE WHEN l.adverse_reactions IS NOT NULL THEN 1 ELSE 0 END + | |
| CASE WHEN l.drug_interactions IS NOT NULL THEN 1 ELSE 0 END + | |
| CASE WHEN l.warnings_and_precautions IS NOT NULL THEN 1 ELSE 0 END) DESC | |
| LIMIT 1 | |
| """, | |
| (name, brand, brand), | |
| ).fetchone() | |
| return dict(row) if row else None | |
| def get_label_by_set_id(self, set_id: str) -> Optional[dict]: | |
| """Fetch a full label row by its canonical set_id. | |
| Phase 11: the autocomplete matcher hands the UI an exact set_id for the | |
| confirmed candidate, so the confirm->fetch step retrieves by that key | |
| directly instead of re-resolving (name, brand). No ambiguity, no fallback. | |
| """ | |
| row = self.conn.execute( | |
| "SELECT * FROM fda_labels WHERE set_id = ?", | |
| (set_id,), | |
| ).fetchone() | |
| return dict(row) if row else None | |
| def get_section(self, set_id: str, section_name: str) -> Optional[str]: | |
| """ | |
| Fetch a single section's verbatim text by set_id + section name. | |
| section_name must be one of SPL_SECTIONS. | |
| """ | |
| if section_name not in SPL_SECTIONS: | |
| raise ValueError(f"Unknown section: {section_name}. Must be one of {SPL_SECTIONS}") | |
| row = self.conn.execute( | |
| f"SELECT {section_name} FROM fda_labels WHERE set_id = ?", | |
| (set_id,), | |
| ).fetchone() | |
| return row[section_name] if row and row[section_name] else None | |
| def list_available_sections(self, set_id: str) -> list[str]: | |
| """Return list of section names that have content for this set_id.""" | |
| cols = ", ".join(SPL_SECTIONS) | |
| row = self.conn.execute( | |
| f"SELECT {cols} FROM fda_labels WHERE set_id = ?", | |
| (set_id,), | |
| ).fetchone() | |
| if not row: | |
| return [] | |
| return [s for s in SPL_SECTIONS if row[s]] | |
| def search_aliases(self, query: str) -> list[dict]: | |
| """Return all aliases matching a query string (case-insensitive).""" | |
| rows = self.conn.execute( | |
| """SELECT a.alias, a.alias_type, a.source, l.generic_name, l.brand_name, a.canonical_set_id | |
| FROM drug_aliases a JOIN fda_labels l ON a.canonical_set_id = l.set_id | |
| WHERE LOWER(a.alias) = LOWER(?)""", | |
| (query,), | |
| ).fetchall() | |
| return [dict(r) for r in rows] | |
| def get_dual_brand_audit(self, generic: Optional[str] = None) -> list[dict]: | |
| """Return dual-brand audit entries, optionally filtered by generic.""" | |
| if generic: | |
| rows = self.conn.execute( | |
| "SELECT * FROM dual_brand_audit WHERE LOWER(generic_name) = LOWER(?)", | |
| (generic,), | |
| ).fetchall() | |
| else: | |
| rows = self.conn.execute( | |
| "SELECT * FROM dual_brand_audit ORDER BY generic_name, brand_name" | |
| ).fetchall() | |
| return [dict(r) for r in rows] | |
| def stats(self) -> dict: | |
| """Return database statistics.""" | |
| return { | |
| "label_count": self.conn.execute("SELECT COUNT(*) FROM fda_labels").fetchone()[0], | |
| "alias_count": self.conn.execute("SELECT COUNT(*) FROM drug_aliases").fetchone()[0], | |
| "dual_brand_count": self.conn.execute("SELECT COUNT(*) FROM dual_brand_audit").fetchone()[0], | |
| "dual_brand_matched": self.conn.execute( | |
| "SELECT COUNT(*) FROM dual_brand_audit WHERE resolution = 'distinct_label'" | |
| ).fetchone()[0], | |
| "dual_brand_missing": self.conn.execute( | |
| "SELECT COUNT(*) FROM dual_brand_audit WHERE resolution = 'needs_fetch'" | |
| ).fetchone()[0], | |
| } | |
| if __name__ == "__main__": | |
| # Quick smoke test | |
| r = FDALabelRetriever("fda_labels.db") | |
| print("Stats:", r.stats()) | |
| print("\n=== Test: dual-brand resolution (Ozempic vs Wegovy) ===") | |
| for brand in ["Ozempic", "Wegovy"]: | |
| label = r.fetch_label(name="semaglutide", brand=brand) | |
| if label: | |
| ind = label.get("indications_and_usage") or "" | |
| print(f" {brand}: brand_name={label['brand_name']}") | |
| print(f" indications excerpt: {ind[:150]}...") | |
| else: | |
| print(f" {brand}: NOT FOUND") | |
| print("\n=== Test: salt-form alias resolution (BUPROPION) ===") | |
| label = r.fetch_label(name="BUPROPION") | |
| if label: | |
| print(f" generic_name={label['generic_name']}, brand={label['brand_name']}") | |
| else: | |
| print(" NOT FOUND") | |
| print("\n=== Test: brand-only via alias (Lipitor) ===") | |
| label = r.fetch_label(name="atorvastatin") # generic name | |
| if label: | |
| print(f" generic_name={label['generic_name']}, brand={label['brand_name']}, set_id={label['set_id'][:8]}") | |
| else: | |
| print(" Not found by generic name 'atorvastatin'") | |
| # Try with calcium salt form | |
| label = r.fetch_label(name="atorvastatin calcium") | |
| if label: | |
| print(f" Found via 'atorvastatin calcium': generic={label['generic_name']}") | |
| r.close() | |