tags with href found: {len(all_links)}")
+ if len(all_links) > 0:
+ logger.info(f"Sample links (first 10):")
+ for i, link in enumerate(all_links[:10]):
+ href = link.get('href', '')[:100]
+ text = link.get_text().strip()[:50]
+ logger.info(f" {i+1}. {text} -> {href}")
+
+ # Extract meeting links - eBoard uses MID parameter
+ # Look for links containing "MID=" (Meeting ID)
+ meeting_links = []
+
+ for link in soup.find_all('a', href=True):
+ href = link.get('href', '')
+ text = link.get_text().strip()
+
+ # eBoard meeting detail links contain MID parameter
+ if 'MID=' in href.upper() or 'meetingdetail' in href.lower():
+ full_url = urljoin(base_url, href)
+ meeting_links.append({
+ 'url': full_url,
+ 'text': text,
+ 'mid': re.search(r'MID=(\d+)', href, re.IGNORECASE).group(1) if re.search(r'MID=(\d+)', href, re.IGNORECASE) else None
+ })
+
+ # Also look for direct PDF links (agendas/minutes)
+ for link in soup.find_all('a', href=True):
+ href = link.get('href', '')
+ text = link.get_text().strip()
+
+ if href.lower().endswith('.pdf') and any(word in text.lower() for word in ['agenda', 'minutes', 'packet']):
+ full_url = urljoin(base_url, href)
+ meeting_links.append({
+ 'url': full_url,
+ 'text': text,
+ 'type': 'pdf'
+ })
+
+ logger.info(f"Found {len(meeting_links)} meeting/document links")
+
+ # Process each meeting (limit to prevent overwhelming)
+ for idx, meeting_info in enumerate(meeting_links[:50]):
+ try:
+ meeting_url = meeting_info['url']
+ meeting_title = meeting_info['text']
+
+ if idx > 0 and idx % 10 == 0:
+ logger.info(f" Progress: {idx}/{min(50, len(meeting_links))} meetings processed")
+
+ # CRITICAL: Randomized rate limiting to prevent Advanced Mode trigger
+ # Never use flat sleep - Incapsula detects patterns
+ wait_time = random.uniform(3.0, 7.0)
+ await asyncio.sleep(wait_time)
+
+ # Simulate human mouse movement before each action
+ await page.mouse.move(random.randint(200, 800), random.randint(200, 600))
+
+ # Handle PDF links directly
+ if meeting_info.get('type') == 'pdf':
+ try:
+ # Download PDF
+ pdf_content = await self._scrape_pdf_document(meeting_url)
+
+ if pdf_content and len(pdf_content.strip()) > 100:
+ # Extract date from title/text
+ meeting_date = None
+ try:
+ from dateutil import parser as date_parser
+ date_match = re.search(r'(\d{1,2}[-/]\d{1,2}[-/]\d{2,4})', meeting_title)
+ if not date_match:
+ date_match = re.search(r'(January|February|March|April|May|June|July|August|September|October|November|December)\s+\d{1,2},?\s+\d{4}', meeting_title, re.IGNORECASE)
+
+ if date_match:
+ meeting_date = date_parser.parse(date_match.group(0))
+ except:
+ meeting_date = datetime.now()
+
+ if not meeting_date:
+ meeting_date = datetime.now()
+
+ document_id = hashlib.md5(f"{meeting_url}{municipality}".encode()).hexdigest()
+
+ doc = MeetingDocument(
+ document_id=document_id,
+ source_url=meeting_url,
+ municipality=municipality,
+ state=state,
+ meeting_date=meeting_date,
+ meeting_type='Board Meeting',
+ title=meeting_title,
+ content=pdf_content[:50000],
+ metadata={
+ 'platform': 'eboard',
+ 'school_id': school_id,
+ 'scraped_with': 'playwright_stealth'
+ }
+ )
+
+ documents.append(doc)
+ logger.success(f" ✓ Scraped PDF: {meeting_title[:50]}")
+
+ except Exception as e:
+ logger.error(f" Error downloading PDF: {e}")
+ continue
+
+ # Handle meeting detail pages
+ else:
+ logger.debug(f" Loading meeting detail: {meeting_title[:50]}")
+
+ try:
+ # Simulate clicking on link (human-like behavior)
+ await page.mouse.move(random.randint(300, 700), random.randint(200, 500))
+ await page.goto(meeting_url, wait_until='domcontentloaded', timeout=30000)
+
+ # Random wait to appear human
+ await page.wait_for_timeout(random.randint(1500, 3000))
+
+ meeting_content = await page.content()
+ meeting_soup = BeautifulSoup(meeting_content, 'html.parser')
+
+ # Extract meeting date
+ meeting_date = None
+ for elem in meeting_soup.find_all(['h1', 'h2', 'h3', 'div', 'span']):
+ text = elem.get_text().strip()
+ date_match = re.search(r'(\d{1,2}[-/]\d{1,2}[-/]\d{2,4})', text)
+ if not date_match:
+ date_match = re.search(r'(January|February|March|April|May|June|July|August|September|October|November|December)\s+\d{1,2},?\s+\d{4}', text, re.IGNORECASE)
+
+ if date_match:
+ try:
+ from dateutil import parser as date_parser
+ meeting_date = date_parser.parse(date_match.group(0))
+ break
+ except:
+ pass
+
+ if not meeting_date:
+ meeting_date = datetime.now()
+
+ # Find document links (PDFs)
+ doc_links = []
+ for link in meeting_soup.find_all('a', href=True):
+ href = link.get('href', '')
+ link_text = link.get_text().strip()
+
+ if (href.lower().endswith('.pdf') or
+ 'agenda' in link_text.lower() or
+ 'minutes' in link_text.lower() or
+ 'packet' in link_text.lower()):
+
+ doc_url = urljoin(base_url, href)
+ doc_links.append({
+ 'url': doc_url,
+ 'text': link_text
+ })
+
+ logger.info(f" Found {len(doc_links)} documents for {meeting_title[:40]}")
+
+ # Download each document
+ for doc_info in doc_links[:5]: # Limit per meeting
+ try:
+ doc_url = doc_info['url']
+ doc_title = doc_info['text']
+
+ if doc_url.lower().endswith('.pdf'):
+ doc_content = await self._scrape_pdf_document(doc_url)
+
+ if doc_content and len(doc_content.strip()) > 100:
+ document_id = hashlib.md5(f"{doc_url}{municipality}".encode()).hexdigest()
+
+ doc = MeetingDocument(
+ document_id=document_id,
+ source_url=doc_url,
+ municipality=municipality,
+ state=state,
+ meeting_date=meeting_date,
+ meeting_type='Board Meeting',
+ title=doc_title or meeting_title,
+ content=doc_content[:50000],
+ metadata={
+ 'platform': 'eboard',
+ 'meeting_page': meeting_url,
+ 'school_id': school_id,
+ 'meeting_id': meeting_info.get('mid'),
+ 'scraped_with': 'playwright_stealth'
+ }
+ )
+
+ documents.append(doc)
+ logger.success(f" ✓ Scraped: {doc_title[:50]}")
+
+ except Exception as e:
+ logger.error(f" Error scraping document: {e}")
+ continue
+
+ except Exception as e:
+ logger.error(f" Error processing meeting {meeting_title[:40]}: {e}")
+ continue
+
+ except Exception as e:
+ logger.error(f"Error processing meeting link: {e}")
+ continue
+
+ # Close browser
+ await browser.close()
+
+ except ImportError as e:
+ logger.error(f"Missing dependency: {e}")
+ logger.error("Install with: pip install playwright-stealth && playwright install chromium")
+ return []
+ except Exception as e:
+ logger.error(f"Error scraping eBoard {url}: {e}")
+ import traceback
+ logger.error(traceback.format_exc())
+
+ logger.success(f"eBoard scrape complete: {len(documents)} documents from {municipality}")
+ return documents
+
+ async def _scrape_generic(
+ self,
+ target: Dict[str, Any],
+ date_range: Dict[str, str]
+ ) -> List[Dict[str, Any]]:
+ """
+ Scrape meeting minutes from generic municipal websites.
+
+ Args:
+ target: Target configuration
+ date_range: Date range for filtering
+
+ Returns:
+ List of meeting documents
+ """
+ url = target["url"]
+ municipality = target["municipality"]
+ state = target["state"]
+
+ documents = []
+
+ try:
+ response = await self.http_client.get(url)
+ response.raise_for_status()
+
+ soup = BeautifulSoup(response.content, "html.parser")
+
+ candidate_documents = self._extract_document_candidates(
+ page_url=url,
+ html=response.text,
+ soup=soup
+ )
+
+ # Crawl a few likely meeting pages because many sites (including JS-heavy portals)
+ # keep document links off the landing page.
+ meeting_pages = self._extract_meeting_pages(page_url=url, soup=soup)
+ for meeting_page in meeting_pages[:8]:
+ try:
+ page_response = await self.http_client.get(meeting_page)
+ if page_response.status_code >= 400:
+ continue
+
+ page_soup = BeautifulSoup(page_response.content, "html.parser")
+ page_candidates = self._extract_document_candidates(
+ page_url=meeting_page,
+ html=page_response.text,
+ soup=page_soup
+ )
+ candidate_documents.extend(page_candidates)
+ await asyncio.sleep(0.2)
+ except Exception as page_err:
+ logger.debug(f"Could not scrape meeting page {meeting_page}: {page_err}")
+
+ # De-duplicate while preserving order
+ seen_urls = set()
+ deduped_candidates = []
+ for doc_url, doc_label in candidate_documents:
+ if doc_url not in seen_urls:
+ seen_urls.add(doc_url)
+ deduped_candidates.append((doc_url, doc_label))
+
+ for doc_url, doc_label in deduped_candidates[:50]:
+ if doc_url in self.scraped_urls:
+ continue
+
+ # Prioritize meeting-related labels but still allow document URL heuristics.
+ label = (doc_label or "").lower()
+ if label and not any(keyword in label for keyword in self.meeting_keywords):
+ if not any(keyword in doc_url.lower() for keyword in self.meeting_keywords):
+ continue
+
+ doc = await self._scrape_document(
+ url=doc_url,
+ municipality=municipality,
+ state=state,
+ title=doc_label or "meeting document"
+ )
+
+ if doc:
+ documents.append(doc)
+ self.scraped_urls.add(doc_url)
+
+ await asyncio.sleep(0.2)
+
+ except Exception as e:
+ logger.error(f"Error scraping generic site {url}: {e}")
+
+ return documents
+
+ def _extract_document_candidates(
+ self,
+ page_url: str,
+ html: str,
+ soup: BeautifulSoup
+ ) -> List[tuple[str, str]]:
+ """Extract document URLs from anchors and script text."""
+ candidates: List[tuple[str, str]] = []
+
+ # Anchor/link extraction
+ for anchor in soup.find_all("a", href=True):
+ href = anchor.get("href", "")
+ if not href:
+ continue
+ full_url = urljoin(page_url, href)
+ full_url = self._normalize_document_url(full_url)
+ lowered = full_url.lower()
+ if any(ext in lowered for ext in self.document_extensions) or any(k in lowered for k in self.document_route_keywords):
+ text = anchor.get_text(" ", strip=True) or anchor.get("title", "") or "document"
+ candidates.append((full_url, text))
+
+ # Script extraction for JS-driven portals that embed links in JSON blobs.
+ url_pattern = r'(https?://[^"\'\s)]+\.(?:pdf|docx?|pptx?|xlsx?)(?:\?[^"\'\s)]*)?)'
+ rel_pattern = r'([\w/\-.]+\.(?:pdf|docx?|pptx?|xlsx?)(?:\?[^"\'\s)]*)?)'
+
+ for raw in re.findall(url_pattern, html, flags=re.IGNORECASE):
+ candidates.append((self._normalize_document_url(raw), "document"))
+
+ route_pattern = r'(["\'](?:/event/Get(?:Agenda|Minutes)File/[^"\']+)["\'])'
+ for raw in re.findall(route_pattern, html, flags=re.IGNORECASE):
+ cleaned = raw.strip("\"'")
+ candidates.append((self._normalize_document_url(urljoin(page_url, cleaned)), "document"))
+
+ for raw in re.findall(rel_pattern, html, flags=re.IGNORECASE):
+ if raw.startswith("http"):
+ continue
+ if raw.startswith("/") or "/" in raw:
+ candidates.append((self._normalize_document_url(urljoin(page_url, raw)), "document"))
+
+ return candidates
+
+ def _normalize_document_url(self, url: str) -> str:
+ """Clean common malformed URL artifacts found in embedded portal markup."""
+ normalized = url.strip()
+ normalized = normalized.replace(" %20?", "?")
+ normalized = normalized.replace("%20?", "?")
+ normalized = normalized.replace(" ?", "?")
+ return normalized
+
+ def _extract_meeting_pages(self, page_url: str, soup: BeautifulSoup) -> List[str]:
+ """Find likely meeting-related subpages to expand document discovery."""
+ pages = []
+ for anchor in soup.find_all("a", href=True):
+ href = anchor.get("href", "")
+ text = anchor.get_text(" ", strip=True).lower()
+ if not href:
+ continue
+
+ full_url = urljoin(page_url, href)
+ combined = f"{text} {full_url.lower()}"
+ if "/event/?id=" in full_url.lower() or any(keyword in combined for keyword in self.meeting_keywords):
+ pages.append(full_url)
+
+ seen = set()
+ deduped = []
+ for p in pages:
+ if p not in seen:
+ seen.add(p)
+ deduped.append(p)
+ return deduped
+
+ async def _scrape_meeting_page(
+ self,
+ url: str,
+ municipality: str,
+ state: str,
+ meeting_data: Optional[Dict[str, Any]] = None
+ ) -> Optional[Dict[str, Any]]:
+ """
+ Scrape a single meeting page.
+
+ Args:
+ url: Meeting page URL
+ municipality: Municipality name
+ state: State code
+ meeting_data: Optional pre-fetched meeting data
+
+ Returns:
+ Meeting document or None
+ """
+ try:
+ response = await self.http_client.get(url)
+ response.raise_for_status()
+
+ soup = BeautifulSoup(response.content, "html.parser")
+
+ # Extract meeting details (simplified - actual implementation would be more robust)
+ title = soup.find("h1") or soup.find("title")
+ title_text = title.get_text().strip() if title else "Untitled Meeting"
+
+ # Extract main content
+ content_div = soup.find("div", class_="meeting-content") or soup.find("main")
+ content = content_div.get_text(separator="\n").strip() if content_div else ""
+
+ # Generate document ID
+ doc_id = hashlib.md5(f"{url}{municipality}".encode()).hexdigest()
+
+ document = MeetingDocument(
+ document_id=doc_id,
+ source_url=url,
+ municipality=municipality,
+ state=state,
+ meeting_date=datetime.utcnow(), # Would parse from content
+ meeting_type="City Council", # Would determine from content
+ title=title_text,
+ content=content,
+ metadata={"platform": "web", "raw_data": meeting_data}
+ )
+
+ return document
+
+ except Exception as e:
+ logger.error(f"Error scraping meeting page {url}: {e}")
+ return None
+
+ async def _scrape_document(
+ self,
+ url: str,
+ municipality: str,
+ state: str,
+ title: str
+ ) -> Optional[Dict[str, Any]]:
+ """
+ Download and extract text from a document URL.
+
+ Args:
+ url: PDF URL
+ municipality: Municipality name
+ state: State code
+ title: Document title
+
+ Returns:
+ Meeting document or None
+ """
+ try:
+ response = await self.http_client.get(url)
+ response.raise_for_status()
+
+ content_type = (response.headers.get("content-type") or "").lower()
+ url_lower = url.lower()
+ is_pdf = ".pdf" in url_lower or "application/pdf" in content_type
+ is_image = any(ext in url_lower for ext in [".png", ".jpg", ".jpeg", ".tif", ".tiff"]) or content_type.startswith("image/")
+
+ content = "[Document content extraction unavailable]"
+ ocr_used = False
+ ocr_pages = 0
+
+ if is_pdf and PdfReader is not None:
+ try:
+ reader = PdfReader(io.BytesIO(response.content))
+ pages = []
+ for page in reader.pages[:30]:
+ pages.append(page.extract_text() or "")
+ extracted = "\n".join(pages).strip()
+ if extracted:
+ content = extracted
+ else:
+ content = "[PDF has no extractable text]"
+ except Exception as parse_error:
+ logger.debug(f"PDF parse failed for {url}: {parse_error}")
+ content = "[PDF parsing failed]"
+
+ # OCR fallback for scanned/image-based PDFs.
+ if is_pdf and content in ["[PDF has no extractable text]", "[PDF parsing failed]"]:
+ ocr_text, ocr_pages = self._ocr_pdf_bytes(response.content)
+ if ocr_text:
+ content = ocr_text
+ ocr_used = True
+
+ # OCR for image documents.
+ if is_image and content == "[Document content extraction unavailable]":
+ image_text = self._ocr_image_bytes(response.content)
+ if image_text:
+ content = image_text
+ ocr_used = True
+ ocr_pages = 1
+
+ doc_id = hashlib.md5(f"{url}{municipality}".encode()).hexdigest()
+
+ document = MeetingDocument(
+ document_id=doc_id,
+ source_url=url,
+ municipality=municipality,
+ state=state,
+ meeting_date=datetime.utcnow(),
+ meeting_type="Unknown",
+ title=title,
+ content=content,
+ metadata={
+ "platform": "document",
+ "file_size": len(response.content),
+ "content_type": response.headers.get("content-type"),
+ "is_pdf": is_pdf,
+ "is_image": is_image,
+ "ocr_used": ocr_used,
+ "ocr_pages": ocr_pages,
+ "text_extracted": content not in [
+ "[Document content extraction unavailable]",
+ "[PDF has no extractable text]",
+ "[PDF parsing failed]"
+ ]
+ }
+ )
+
+ return document
+
+ except Exception as e:
+ if isinstance(e, httpx.HTTPStatusError) and e.response.status_code == 404:
+ logger.debug(f"Document not found (404): {url}")
+ else:
+ logger.error(f"Error downloading document {url}: {e}")
+ return None
+
+ def _ocr_pdf_bytes(self, pdf_bytes: bytes) -> tuple[str, int]:
+ """OCR PDF pages when direct PDF text extraction fails."""
+ if pdfplumber is None or pytesseract is None:
+ return "", 0
+
+ try:
+ extracted_pages = []
+ with pdfplumber.open(io.BytesIO(pdf_bytes)) as pdf:
+ for page in pdf.pages[:self.ocr_max_pages]:
+ try:
+ image = page.to_image(resolution=200).original
+ text = pytesseract.image_to_string(image).strip()
+ if text:
+ extracted_pages.append(text)
+ except TesseractNotFoundError:
+ if not self._ocr_missing_tesseract_warned:
+ logger.warning("Tesseract binary not found. Install 'tesseract-ocr' to enable OCR.")
+ self._ocr_missing_tesseract_warned = True
+ return "", 0
+ except Exception as page_err:
+ logger.debug(f"OCR page failed: {page_err}")
+
+ if not extracted_pages:
+ return "", 0
+ return "\n\n".join(extracted_pages), len(extracted_pages)
+ except Exception as err:
+ logger.debug(f"OCR PDF fallback failed: {err}")
+ return "", 0
+
+ def _ocr_image_bytes(self, image_bytes: bytes) -> str:
+ """OCR text from image-based documents."""
+ if pytesseract is None or Image is None:
+ return ""
+
+ try:
+ image = Image.open(io.BytesIO(image_bytes))
+ return pytesseract.image_to_string(image).strip()
+ except TesseractNotFoundError:
+ if not self._ocr_missing_tesseract_warned:
+ logger.warning("Tesseract binary not found. Install 'tesseract-ocr' to enable OCR.")
+ self._ocr_missing_tesseract_warned = True
+ return ""
+ except Exception as err:
+ logger.debug(f"Image OCR failed: {err}")
+ return ""
+
+ async def _scrape_pdf_document(
+ self,
+ url: str,
+ municipality: str,
+ state: str,
+ title: str
+ ) -> Optional[Dict[str, Any]]:
+ """Backward-compatible wrapper for existing call sites."""
+ return await self._scrape_document(url, municipality, state, title)
diff --git a/agents/scraper_undetected.py b/agents/scraper_undetected.py
new file mode 100644
index 0000000000000000000000000000000000000000..fc2a351aee4efb48c7b1ee7638553e00a4679a5d
--- /dev/null
+++ b/agents/scraper_undetected.py
@@ -0,0 +1,261 @@
+"""
+Alternative eBoard scraper using undetected-chromedriver
+This bypasses Incapsula without manual cookies
+"""
+import asyncio
+import re
+from typing import Dict, Any, List
+from pathlib import Path
+from bs4 import BeautifulSoup
+from urllib.parse import urljoin
+from datetime import datetime
+import hashlib
+
+from loguru import logger
+
+
+class UndetectedEboardScraper:
+ """
+ Scrape eBoard using undetected-chromedriver to bypass Incapsula.
+
+ This library patches Selenium ChromeDriver to avoid detection by:
+ - Removing Selenium markers from navigator.webdriver
+ - Randomizing browser fingerprints
+ - Using real Chrome instead of ChromeDriver
+ """
+
+ async def scrape_eboard(
+ self,
+ url: str,
+ municipality: str,
+ state: str,
+ school_id: str = None
+ ) -> List[Dict[str, Any]]:
+ """
+ Scrape eBoard platform without manual cookies.
+
+ Args:
+ url: eBoard URL
+ municipality: School district name
+ state: State code
+ school_id: Optional school ID (extracted from URL if not provided)
+
+ Returns:
+ List of meeting documents
+ """
+ try:
+ import undetected_chromedriver as uc
+ from selenium.webdriver.common.by import By
+ from selenium.webdriver.support.ui import WebDriverWait
+ from selenium.webdriver.support import expected_conditions as EC
+ import time
+ import random
+ except ImportError:
+ logger.error("Missing undetected-chromedriver. Install: pip install undetected-chromedriver")
+ return []
+
+ # Extract school ID
+ if not school_id:
+ match = re.search(r'[?&]s=(\d+)', url, re.IGNORECASE)
+ school_id = match.group(1) if match else None
+
+ if not school_id:
+ logger.error(f"Could not extract school ID from URL: {url}")
+ return []
+
+ base_url = "https://simbli.eboardsolutions.com"
+ meetings_url = f"{base_url}/SB_Meetings/SB_MeetingListing.aspx?S={school_id}"
+
+ logger.info(f"Using undetected-chromedriver to bypass Incapsula")
+ logger.info(f"Target: {meetings_url}")
+
+ documents = []
+
+ try:
+ # Create undetected Chrome instance
+ options = uc.ChromeOptions()
+ # options.add_argument('--headless') # Headless may still be detected
+ options.add_argument('--no-sandbox')
+ options.add_argument('--disable-dev-shm-usage')
+ options.add_argument('--disable-blink-features=AutomationControlled')
+
+ # Create driver with version management
+ driver = uc.Chrome(options=options, version_main=None)
+
+ logger.info("Chrome launched with anti-detection patches")
+
+ # Navigate to meetings page
+ driver.get(meetings_url)
+ logger.info(f"Loaded page: {driver.title[:100]}")
+
+ # Wait for Incapsula challenge to complete
+ # The challenge usually takes 3-5 seconds
+ wait_time = random.uniform(5.0, 8.0)
+ logger.info(f"Waiting {wait_time:.1f}s for Incapsula challenge...")
+ time.sleep(wait_time)
+
+ # Check if we bypassed Incapsula
+ page_source = driver.page_source
+
+ if 'Incapsula' in page_source and len(page_source) < 10000:
+ logger.error("Still blocked by Incapsula")
+ logger.warning("Try running with headless=False or use Option 2 (Residential Proxies)")
+ driver.quit()
+ return []
+
+ logger.success(f"✓ Bypassed Incapsula! Page size: {len(page_source)} bytes")
+
+ # Parse the page
+ soup = BeautifulSoup(page_source, 'html.parser')
+
+ # Extract meeting links
+ meeting_links = []
+
+ # Method 1: Look for MID parameter
+ for link in soup.find_all('a', href=True):
+ href = link.get('href', '')
+ text = link.get_text().strip()
+
+ if 'MID=' in href.upper() or 'meetingdetail' in href.lower():
+ full_url = urljoin(base_url, href)
+ meeting_links.append({
+ 'url': full_url,
+ 'text': text,
+ 'type': 'meeting'
+ })
+ elif href.lower().endswith('.pdf'):
+ full_url = urljoin(base_url, href)
+ meeting_links.append({
+ 'url': full_url,
+ 'text': text,
+ 'type': 'pdf'
+ })
+
+ logger.info(f"Found {len(meeting_links)} meeting/document links")
+
+ # If no links found, try JavaScript execution
+ if len(meeting_links) == 0:
+ logger.warning("No links found in HTML, checking for JavaScript-rendered content...")
+
+ # Wait for dynamic content
+ try:
+ WebDriverWait(driver, 10).until(
+ EC.presence_of_element_located((By.TAG_NAME, "a"))
+ )
+ time.sleep(3) # Additional wait for JS
+
+ # Re-parse
+ page_source = driver.page_source
+ soup = BeautifulSoup(page_source, 'html.parser')
+
+ for link in soup.find_all('a', href=True):
+ href = link.get('href', '')
+ text = link.get_text().strip()
+
+ if 'MID=' in href.upper() or href.lower().endswith('.pdf'):
+ full_url = urljoin(base_url, href)
+ meeting_links.append({
+ 'url': full_url,
+ 'text': text,
+ 'type': 'pdf' if href.lower().endswith('.pdf') else 'meeting'
+ })
+
+ logger.info(f"After JS wait: Found {len(meeting_links)} links")
+ except Exception as e:
+ logger.warning(f"JS content wait failed: {e}")
+
+ # Process meeting links (limit to prevent overwhelming)
+ for idx, meeting_info in enumerate(meeting_links[:50]):
+ if idx > 0 and idx % 10 == 0:
+ logger.info(f"Progress: {idx}/{min(50, len(meeting_links))}")
+
+ # Human-like delay
+ time.sleep(random.uniform(2.0, 5.0))
+
+ try:
+ meeting_url = meeting_info['url']
+ meeting_title = meeting_info['text']
+
+ if meeting_info['type'] == 'pdf':
+ # Download PDF directly
+ logger.debug(f" Downloading PDF: {meeting_title[:50]}")
+ # TODO: Implement PDF download
+ # For now, just record the URL
+ doc = {
+ 'document_id': hashlib.md5(f"{meeting_url}{municipality}".encode()).hexdigest(),
+ 'source_url': meeting_url,
+ 'municipality': municipality,
+ 'state': state,
+ 'meeting_date': datetime.now(),
+ 'meeting_type': 'Board Meeting',
+ 'title': meeting_title,
+ 'content': '', # Would need PDF extraction
+ 'metadata': {
+ 'platform': 'eboard',
+ 'school_id': school_id,
+ 'scraped_with': 'undetected_chromedriver'
+ }
+ }
+ documents.append(doc)
+ else:
+ # Navigate to meeting detail page
+ logger.debug(f" Loading meeting: {meeting_title[:50]}")
+ driver.get(meeting_url)
+ time.sleep(random.uniform(2.0, 4.0))
+
+ meeting_soup = BeautifulSoup(driver.page_source, 'html.parser')
+
+ # Extract PDFs from meeting page
+ for link in meeting_soup.find_all('a', href=True):
+ href = link.get('href', '')
+ if href.lower().endswith('.pdf'):
+ doc_url = urljoin(base_url, href)
+ doc_title = link.get_text().strip()
+
+ doc = {
+ 'document_id': hashlib.md5(f"{doc_url}{municipality}".encode()).hexdigest(),
+ 'source_url': doc_url,
+ 'municipality': municipality,
+ 'state': state,
+ 'meeting_date': datetime.now(),
+ 'meeting_type': 'Board Meeting',
+ 'title': doc_title or meeting_title,
+ 'content': '',
+ 'metadata': {
+ 'platform': 'eboard',
+ 'meeting_page': meeting_url,
+ 'school_id': school_id,
+ 'scraped_with': 'undetected_chromedriver'
+ }
+ }
+ documents.append(doc)
+ logger.success(f" ✓ Found: {doc_title[:50]}")
+
+ except Exception as e:
+ logger.error(f"Error processing {meeting_info.get('text', 'unknown')}: {e}")
+ continue
+
+ driver.quit()
+ logger.success(f"Scraping complete: {len(documents)} documents")
+ return documents
+
+ except Exception as e:
+ logger.error(f"Error in undetected scraper: {e}")
+ import traceback
+ logger.error(traceback.format_exc())
+ return []
+
+
+# Example usage
+async def main():
+ scraper = UndetectedEboardScraper()
+ docs = await scraper.scrape_eboard(
+ url="http://simbli.eboardsolutions.com/index.aspx?s=2088",
+ municipality="Tuscaloosa City Schools",
+ state="AL"
+ )
+ print(f"Scraped {len(docs)} documents")
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/agents/sentiment.py b/agents/sentiment.py
new file mode 100644
index 0000000000000000000000000000000000000000..e3b8d948070ed36035629e59d2967e7400bc969c
--- /dev/null
+++ b/agents/sentiment.py
@@ -0,0 +1,381 @@
+"""
+Sentiment Analyzer Agent for determining policy stance and debate intensity.
+"""
+import asyncio
+from typing import List, Dict, Any, Optional
+from datetime import datetime
+from loguru import logger
+
+from agents.base import BaseAgent, AgentRole, AgentMessage, MessageType, AgentStatus
+
+
+class PolicyStance:
+ """Enumeration of policy stances."""
+ STRONGLY_SUPPORTIVE = "strongly_supportive"
+ SUPPORTIVE = "supportive"
+ NEUTRAL = "neutral"
+ OPPOSED = "opposed"
+ STRONGLY_OPPOSED = "strongly_opposed"
+ DEBATED = "debated" # When there's active debate
+
+
+class DebateIntensity:
+ """Enumeration of debate intensity levels."""
+ NONE = "none" # Passing mention
+ LOW = "low" # Brief discussion
+ MODERATE = "moderate" # Extended discussion
+ HIGH = "high" # Heated debate with multiple viewpoints
+ CRITICAL = "critical" # Vote imminent or major decision pending
+
+
+class SentimentAnalyzerAgent(BaseAgent):
+ """
+ Agent responsible for analyzing sentiment and policy stance.
+
+ Determines:
+ - Overall stance toward oral health policies
+ - Intensity of debate/discussion
+ - Key arguments for and against
+ - Likelihood of policy action
+ - Advocacy opportunities
+ """
+
+ def __init__(self, agent_id: str = "sentiment-001"):
+ """Initialize the sentiment analyzer agent."""
+ super().__init__(agent_id, AgentRole.SENTIMENT_ANALYZER)
+ self._initialize_indicators()
+
+ def _initialize_indicators(self):
+ """Initialize sentiment and debate indicators."""
+ self.supportive_indicators = [
+ "approve", "support", "favor", "endorse", "recommend",
+ "beneficial", "important", "necessary", "implement",
+ "move forward", "proceed with"
+ ]
+
+ self.opposition_indicators = [
+ "oppose", "against", "reject", "deny", "concerns about",
+ "problems with", "issues with", "delay", "postpone",
+ "table the motion", "reconsider"
+ ]
+
+ self.debate_indicators = [
+ "discussion", "debate", "motion", "vote", "amendment",
+ "public comment", "testimony", "hearing", "concerns",
+ "questions about", "divided"
+ ]
+
+ self.urgency_indicators = [
+ "urgent", "immediate", "deadline", "vote", "decision",
+ "approval needed", "time-sensitive", "pressing",
+ "second reading", "final vote"
+ ]
+
+ async def process(self, message: AgentMessage) -> List[AgentMessage]:
+ """
+ Process sentiment analysis commands.
+
+ Args:
+ message: Message containing classified documents
+
+ Returns:
+ List of messages with sentiment analysis results
+ """
+ self.update_status(AgentStatus.PROCESSING, "Analyzing policy sentiment and debate")
+
+ try:
+ documents = message.payload.get("documents", [])
+
+ analyzed_documents = []
+
+ for doc in documents:
+ analysis = await self._analyze_document(doc)
+ doc["sentiment_analysis"] = analysis
+ analyzed_documents.append(doc)
+
+ # Identify advocacy opportunities
+ opportunities = self._identify_advocacy_opportunities(analyzed_documents)
+
+ # Send to advocacy writer agent
+ response = await self.send_message(
+ AgentRole.ADVOCACY_WRITER,
+ MessageType.DATA,
+ {
+ "workflow_id": message.payload.get("workflow_id"),
+ "documents": analyzed_documents,
+ "opportunities": opportunities,
+ "count": len(analyzed_documents)
+ }
+ )
+
+ self.log_success()
+ logger.info(
+ f"Analyzed sentiment for {len(analyzed_documents)} documents, "
+ f"found {len(opportunities)} advocacy opportunities"
+ )
+
+ return [response]
+
+ except Exception as e:
+ self.log_failure(str(e))
+ error_msg = await self.send_message(
+ AgentRole.ORCHESTRATOR,
+ MessageType.ERROR,
+ {"error": str(e), "agent": self.agent_id}
+ )
+ return [error_msg]
+
+ async def _analyze_document(
+ self,
+ doc: Dict[str, Any]
+ ) -> Dict[str, Any]:
+ """
+ Analyze sentiment and policy stance for a document.
+
+ Args:
+ doc: Document to analyze
+
+ Returns:
+ Sentiment analysis results
+ """
+ text = self._get_analyzable_text(doc)
+ text_lower = text.lower()
+
+ # Count sentiment indicators
+ support_score = sum(
+ 1 for indicator in self.supportive_indicators
+ if indicator in text_lower
+ )
+
+ opposition_score = sum(
+ 1 for indicator in self.opposition_indicators
+ if indicator in text_lower
+ )
+
+ debate_score = sum(
+ 1 for indicator in self.debate_indicators
+ if indicator in text_lower
+ )
+
+ urgency_score = sum(
+ 1 for indicator in self.urgency_indicators
+ if indicator in text_lower
+ )
+
+ # Determine policy stance
+ stance = self._determine_stance(support_score, opposition_score, debate_score)
+
+ # Determine debate intensity
+ intensity = self._determine_intensity(debate_score, urgency_score, doc)
+
+ # Extract key arguments
+ arguments = self._extract_arguments(doc, text_lower)
+
+ # Calculate advocacy urgency
+ advocacy_urgency = self._calculate_advocacy_urgency(
+ stance, intensity, urgency_score
+ )
+
+ analysis = {
+ "stance": stance,
+ "debate_intensity": intensity,
+ "support_score": support_score,
+ "opposition_score": opposition_score,
+ "debate_score": debate_score,
+ "urgency_score": urgency_score,
+ "advocacy_urgency": advocacy_urgency,
+ "key_arguments": arguments,
+ "analyzed_at": datetime.utcnow().isoformat()
+ }
+
+ return analysis
+
+ def _get_analyzable_text(self, doc: Dict[str, Any]) -> str:
+ """Extract text for sentiment analysis."""
+ parts = []
+
+ # Prioritize excerpts from classification
+ for excerpt in doc.get("classification", {}).get("relevant_excerpts", []):
+ parts.append(excerpt.get("text", ""))
+
+ # Add motions (highly relevant)
+ for motion in doc.get("motions", []):
+ parts.append(motion.get("text", ""))
+
+ # Add votes
+ for vote in doc.get("votes", []):
+ parts.append(vote.get("result", ""))
+
+ # Fallback to full text if needed
+ if not parts:
+ parts.append(doc.get("full_text", ""))
+
+ return " ".join(parts)
+
+ def _determine_stance(
+ self,
+ support_score: int,
+ opposition_score: int,
+ debate_score: int
+ ) -> str:
+ """Determine overall policy stance."""
+ if debate_score >= 3 and abs(support_score - opposition_score) <= 1:
+ return PolicyStance.DEBATED
+
+ if support_score > opposition_score:
+ if support_score >= 3:
+ return PolicyStance.STRONGLY_SUPPORTIVE
+ else:
+ return PolicyStance.SUPPORTIVE
+ elif opposition_score > support_score:
+ if opposition_score >= 3:
+ return PolicyStance.STRONGLY_OPPOSED
+ else:
+ return PolicyStance.OPPOSED
+ else:
+ return PolicyStance.NEUTRAL
+
+ def _determine_intensity(
+ self,
+ debate_score: int,
+ urgency_score: int,
+ doc: Dict[str, Any]
+ ) -> str:
+ """Determine debate intensity."""
+ # Check for votes or motions (indicates high intensity)
+ has_vote = len(doc.get("votes", [])) > 0
+ has_motion = len(doc.get("motions", [])) > 0
+
+ if urgency_score >= 2 or (has_vote and has_motion):
+ return DebateIntensity.CRITICAL
+ elif debate_score >= 5 or has_vote or has_motion:
+ return DebateIntensity.HIGH
+ elif debate_score >= 3:
+ return DebateIntensity.MODERATE
+ elif debate_score >= 1:
+ return DebateIntensity.LOW
+ else:
+ return DebateIntensity.NONE
+
+ def _extract_arguments(
+ self,
+ doc: Dict[str, Any],
+ text_lower: str
+ ) -> Dict[str, List[str]]:
+ """Extract key arguments for and against."""
+ arguments = {
+ "supporting": [],
+ "opposing": []
+ }
+
+ # Extract from motions and discussion
+ for motion in doc.get("motions", []):
+ motion_text = motion.get("text", "").lower()
+
+ if any(ind in motion_text for ind in self.supportive_indicators):
+ arguments["supporting"].append(motion.get("text", ""))
+ elif any(ind in motion_text for ind in self.opposition_indicators):
+ arguments["opposing"].append(motion.get("text", ""))
+
+ return arguments
+
+ def _calculate_advocacy_urgency(
+ self,
+ stance: str,
+ intensity: str,
+ urgency_score: int
+ ) -> str:
+ """
+ Calculate how urgent advocacy action is needed.
+
+ Returns: "critical", "high", "medium", "low", or "none"
+ """
+ # Critical: Vote imminent and debated/opposed
+ if intensity == DebateIntensity.CRITICAL:
+ if stance in [PolicyStance.DEBATED, PolicyStance.OPPOSED, PolicyStance.STRONGLY_OPPOSED]:
+ return "critical"
+ return "high"
+
+ # High: Active debate with opposition
+ if intensity == DebateIntensity.HIGH:
+ if stance in [PolicyStance.OPPOSED, PolicyStance.STRONGLY_OPPOSED]:
+ return "high"
+ elif stance == PolicyStance.DEBATED:
+ return "high"
+ return "medium"
+
+ # Medium: Moderate discussion or emerging issue
+ if intensity == DebateIntensity.MODERATE:
+ return "medium"
+
+ # Low: Early stage or general mention
+ if intensity == DebateIntensity.LOW:
+ return "low"
+
+ return "none"
+
+ def _identify_advocacy_opportunities(
+ self,
+ documents: List[Dict[str, Any]]
+ ) -> List[Dict[str, Any]]:
+ """
+ Identify advocacy opportunities across all analyzed documents.
+
+ Args:
+ documents: All analyzed documents
+
+ Returns:
+ List of advocacy opportunities
+ """
+ opportunities = []
+
+ for doc in documents:
+ sentiment = doc.get("sentiment_analysis", {})
+ urgency = sentiment.get("advocacy_urgency")
+
+ # Only flag high and critical urgency items
+ if urgency in ["critical", "high"]:
+ opportunity = {
+ "document_id": doc["document_id"],
+ "municipality": doc["municipality"],
+ "state": doc["state"],
+ "meeting_date": doc["meeting_date"],
+ "source_url": doc["source_url"],
+ "topic": doc["classification"]["primary_topic"],
+ "stance": sentiment["stance"],
+ "intensity": sentiment["debate_intensity"],
+ "urgency": urgency,
+ "key_excerpts": doc["classification"].get("relevant_excerpts", []),
+ "recommended_action": self._recommend_action(sentiment, doc)
+ }
+ opportunities.append(opportunity)
+
+ # Sort by urgency
+ urgency_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
+ opportunities.sort(key=lambda x: urgency_order.get(x["urgency"], 4))
+
+ return opportunities
+
+ def _recommend_action(
+ self,
+ sentiment: Dict[str, Any],
+ doc: Dict[str, Any]
+ ) -> str:
+ """Recommend advocacy action based on analysis."""
+ stance = sentiment.get("stance")
+ intensity = sentiment.get("debate_intensity")
+
+ if intensity == DebateIntensity.CRITICAL:
+ if stance in [PolicyStance.OPPOSED, PolicyStance.STRONGLY_OPPOSED]:
+ return "URGENT: Contact officials immediately. Vote imminent."
+ elif stance == PolicyStance.DEBATED:
+ return "URGENT: Provide supporting testimony. Decision pending."
+
+ if stance == PolicyStance.DEBATED:
+ return "Engage with stakeholders. Provide educational materials."
+ elif stance in [PolicyStance.OPPOSED, PolicyStance.STRONGLY_OPPOSED]:
+ return "Initiate dialogue with decision-makers. Address concerns."
+ elif stance == PolicyStance.NEUTRAL:
+ return "Introduce topic to agenda. Build awareness."
+
+ return "Monitor situation. Prepare support materials."
diff --git a/api/__init__.py b/api/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..f36f9ec8c192765a57dc181c696be38b9a3573c2
--- /dev/null
+++ b/api/__init__.py
@@ -0,0 +1,4 @@
+"""API module for the Oral Health Policy Pulse system."""
+from api.main import app
+
+__all__ = ["app"]
diff --git a/api/app.py b/api/app.py
new file mode 100644
index 0000000000000000000000000000000000000000..a7a1b2eb50ef01d8c7951006e7ccd93a386985f8
--- /dev/null
+++ b/api/app.py
@@ -0,0 +1,711 @@
+"""
+FastAPI application optimized for Databricks Apps deployment.
+Serves React frontend and provides REST API for agent interactions.
+"""
+from typing import List, Dict, Any, Optional
+from datetime import datetime
+from pathlib import Path
+from fastapi import FastAPI, HTTPException, Query, BackgroundTasks
+from fastapi.responses import HTMLResponse, JSONResponse, FileResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel, Field
+from loguru import logger
+import os
+
+from agents.orchestrator import OrchestratorAgent
+from pipeline.delta_lake import DeltaLakePipeline
+from config import settings
+
+# Initialize FastAPI app
+app = FastAPI(
+ title="Open Navigator",
+ description="AI-powered advocacy opportunity finder",
+ version="2.0.0",
+ docs_url="/api/docs",
+ redoc_url="/api/redoc",
+ openapi_url="/api/openapi.json"
+)
+
+# Add CORS middleware
+app.add_middleware(
+ CORSMiddleware,
+ allow_origins=["*"],
+ allow_credentials=True,
+ allow_methods=["*"],
+ allow_headers=["*"],
+)
+
+# Initialize components
+orchestrator = OrchestratorAgent()
+pipeline = DeltaLakePipeline()
+
+# Pydantic models
+class WorkflowRequest(BaseModel):
+ """Request to start a new analysis workflow."""
+ targets: List[Dict[str, str]]
+ topics: Optional[List[str]] = None
+
+class OpportunityFilter(BaseModel):
+ """Filter criteria for advocacy opportunities."""
+ state: Optional[str] = None
+ topic: Optional[str] = None
+ urgency: Optional[str] = None
+ min_confidence: Optional[float] = 0.7
+
+
+# API Routes
+@app.get("/api/health")
+async def health_check():
+ """Health check endpoint."""
+ return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}
+
+
+@app.get("/api/dashboard")
+async def get_dashboard_stats():
+ """Get dashboard statistics and recent opportunities."""
+ try:
+ # Query Delta Lake for stats
+ stats = await pipeline.get_dashboard_stats()
+ return stats
+ except Exception as e:
+ logger.error(f"Dashboard stats error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/api/opportunities")
+async def get_opportunities(
+ state: Optional[str] = Query(None),
+ topic: Optional[str] = Query(None),
+ urgency: Optional[str] = Query(None),
+ limit: int = Query(100, le=1000)
+):
+ """Get advocacy opportunities with optional filters."""
+ try:
+ opportunities = await pipeline.query_opportunities(
+ state=state,
+ topic=topic,
+ urgency=urgency,
+ limit=limit
+ )
+ return {"opportunities": opportunities, "count": len(opportunities)}
+ except Exception as e:
+ logger.error(f"Query opportunities error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/api/documents")
+async def get_documents(
+ search: Optional[str] = Query(None),
+ page: int = Query(1, ge=1),
+ limit: int = Query(20, le=100)
+):
+ """Get analyzed documents with pagination."""
+ try:
+ offset = (page - 1) * limit
+ documents = await pipeline.query_documents(
+ search=search,
+ limit=limit,
+ offset=offset
+ )
+ total = await pipeline.count_documents(search=search)
+ return {
+ "documents": documents,
+ "page": page,
+ "limit": limit,
+ "total": total,
+ "total_pages": (total + limit - 1) // limit
+ }
+ except Exception as e:
+ logger.error(f"Query documents error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/api/search/")
+async def search_all(
+ q: str = Query(..., min_length=1, description="Search query"),
+ types: Optional[str] = Query("contacts,organizations,causes", description="Comma-separated types to search"),
+ state: Optional[str] = Query(None, description="Filter by state code (e.g. MA, AL)"),
+ limit: int = Query(10, ge=1, le=100),
+ page: int = Query(1, ge=1),
+ ntee_code: Optional[str] = Query(None, description="NTEE category code")
+):
+ """
+ Unified search across contacts, organizations, causes, and jurisdictions.
+ Returns results grouped by type with pagination.
+ """
+ try:
+ offset = (page - 1) * limit
+ search_types = [t.strip() for t in types.split(',') if t.strip()]
+
+ # Initialize results structure
+ results = {
+ "contacts": [],
+ "organizations": [],
+ "causes": [],
+ "meetings": [],
+ "jurisdictions": []
+ }
+
+ # Search logic (placeholder - implement actual search)
+ # For now, return mock results for demonstration
+ if "contacts" in search_types:
+ results["contacts"] = [
+ {
+ "type": "contact",
+ "title": f"Sample Contact matching '{q}'",
+ "subtitle": "Government Official",
+ "description": "Contact information for local official",
+ "url": "/contact/1",
+ "score": 0.9,
+ "metadata": {"state": state or "MA"}
+ }
+ ]
+
+ if "organizations" in search_types:
+ results["organizations"] = [
+ {
+ "type": "organization",
+ "title": f"Sample Organization matching '{q}'",
+ "subtitle": "Nonprofit Organization",
+ "description": "Community health organization",
+ "url": "/org/1",
+ "score": 0.85,
+ "metadata": {"state": state or "MA", "ntee": ntee_code}
+ }
+ ]
+
+ if "causes" in search_types:
+ results["causes"] = [
+ {
+ "type": "cause",
+ "title": f"Sample Cause matching '{q}'",
+ "subtitle": "Health & Wellness",
+ "description": "Advocacy for community health",
+ "url": "/cause/1",
+ "score": 0.8,
+ "metadata": {}
+ }
+ ]
+
+ # Calculate total results
+ total_results = sum(len(v) for v in results.values())
+ total_pages = max(1, (total_results + limit - 1) // limit)
+
+ return {
+ "query": q,
+ "total_results": total_results,
+ "results": results,
+ "pagination": {
+ "page": page,
+ "limit": limit,
+ "offset": offset,
+ "total_pages": total_pages,
+ "has_next": page < total_pages,
+ "has_prev": page > 1
+ },
+ "filters": {
+ "state": state,
+ "ntee_code": ntee_code,
+ "types": search_types
+ }
+ }
+ except Exception as e:
+ logger.error(f"Search error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/api/workflow/start")
+async def start_workflow(request: WorkflowRequest, background_tasks: BackgroundTasks):
+ """Start a new analysis workflow."""
+ try:
+ workflow_id = f"workflow_{datetime.utcnow().timestamp()}"
+
+ # Start workflow in background
+ background_tasks.add_task(
+ orchestrator.execute_pipeline,
+ workflow_id=workflow_id,
+ targets=request.targets,
+ topics=request.topics
+ )
+
+ return {
+ "workflow_id": workflow_id,
+ "status": "started",
+ "message": "Workflow started successfully"
+ }
+ except Exception as e:
+ logger.error(f"Workflow start error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/api/workflow/{workflow_id}/status")
+async def get_workflow_status(workflow_id: str):
+ """Get status of a running workflow."""
+ try:
+ status = await orchestrator.get_workflow_status(workflow_id)
+ return status
+ except Exception as e:
+ logger.error(f"Workflow status error: {e}")
+ raise HTTPException(status_code=404, detail="Workflow not found")
+
+
+@app.post("/api/advocacy/email/{opportunity_id}")
+async def generate_advocacy_email(opportunity_id: str):
+ """Generate advocacy email for an opportunity."""
+ try:
+ opportunity = await pipeline.get_opportunity(opportunity_id)
+ if not opportunity:
+ raise HTTPException(status_code=404, detail="Opportunity not found")
+
+ email_content = await orchestrator.generate_advocacy_email(opportunity)
+ return {"content": email_content}
+ except Exception as e:
+ logger.error(f"Generate email error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/api/settings")
+async def get_settings():
+ """Get current system settings."""
+ return {
+ "target_states": settings.target_states or [],
+ "policy_topics": settings.policy_topics,
+ "min_confidence": 0.7,
+ "email_notifications": False,
+ "notification_email": ""
+ }
+
+
+@app.put("/api/settings")
+async def update_settings(new_settings: Dict[str, Any]):
+ """Update system settings."""
+ try:
+ # In production, this would update configuration in Unity Catalog
+ logger.info(f"Settings update requested: {new_settings}")
+ return {"message": "Settings updated successfully"}
+ except Exception as e:
+ logger.error(f"Settings update error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/api/agents/status")
+async def get_agents_status():
+ """Get status of all agents."""
+ try:
+ return {
+ "agents": [
+ {"name": "Scraper", "status": "active", "uptime": "24h"},
+ {"name": "Classifier", "status": "active", "uptime": "24h"},
+ {"name": "Sentiment Analyzer", "status": "active", "uptime": "24h"},
+ {"name": "Advocacy Writer", "status": "active", "uptime": "24h"}
+ ]
+ }
+ except Exception as e:
+ logger.error(f"Agent status error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/api/nonprofits")
+async def search_nonprofits(
+ location: str = Query("Tuscaloosa, AL", description="City, State format"),
+ keyword: Optional[str] = Query(None, description="Service keyword (e.g., 'dental', 'health')"),
+ state: Optional[str] = Query(None, description="2-letter state code (e.g., 'AL')"),
+ ntee_code: Optional[str] = Query(None, description="NTEE code (e.g., 'E' for health)"),
+ source: Optional[str] = Query(None, description="Data source: 'propublica', 'everyorg', 'all'")
+):
+ """
+ Search for nonprofits using free open data APIs.
+
+ Integrates data from:
+ - ProPublica Nonprofit Explorer (financial data, NTEE codes)
+ - Every.org (mission statements, logos)
+ - IRS TEOS (official tax-exempt status)
+
+ Example: /api/nonprofits?location=Tuscaloosa,AL&keyword=dental&ntee_code=E
+ """
+ try:
+ from discovery.nonprofit_discovery import NonprofitDiscovery
+
+ discovery = NonprofitDiscovery()
+ results = []
+
+ # Parse location for state/city
+ location_parts = location.split(',')
+ city = location_parts[0].strip() if len(location_parts) > 0 else None
+ state_from_location = location_parts[1].strip() if len(location_parts) > 1 else None
+ state_code = state or state_from_location or "AL"
+
+ # Determine which sources to query
+ sources_to_query = ['propublica', 'everyorg'] if source == 'all' or not source else [source]
+
+ # Query ProPublica
+ if 'propublica' in sources_to_query:
+ try:
+ propublica_results = discovery.search_propublica(
+ state=state_code,
+ city=city,
+ ntee_code=ntee_code
+ )
+ results.extend(propublica_results)
+ logger.info(f"ProPublica: Found {len(propublica_results)} organizations")
+ except Exception as e:
+ logger.warning(f"ProPublica search failed: {e}")
+
+ # Query Every.org
+ if 'everyorg' in sources_to_query:
+ try:
+ causes = []
+ if keyword:
+ # Map keywords to causes
+ keyword_lower = keyword.lower()
+ if 'health' in keyword_lower or 'dental' in keyword_lower or 'medical' in keyword_lower:
+ causes.append('health')
+ if 'education' in keyword_lower or 'school' in keyword_lower:
+ causes.append('education')
+
+ everyorg_results = discovery.search_everyorg(
+ location=location,
+ causes=causes if causes else None
+ )
+ results.extend(everyorg_results)
+ logger.info(f"Every.org: Found {len(everyorg_results)} organizations")
+ except Exception as e:
+ logger.warning(f"Every.org search failed: {e}")
+
+ # Filter by keyword if provided
+ if keyword and results:
+ keyword_lower = keyword.lower()
+ filtered_results = []
+ for org in results:
+ # Search in name, description, mission, ntee_description
+ searchable_text = ' '.join([
+ str(org.get('name', '')),
+ str(org.get('description', '')),
+ str(org.get('mission', '')),
+ str(org.get('ntee_description', ''))
+ ]).lower()
+
+ if keyword_lower in searchable_text:
+ filtered_results.append(org)
+
+ results = filtered_results
+
+ return {
+ "location": location,
+ "keyword": keyword,
+ "state": state_code,
+ "ntee_code": ntee_code,
+ "count": len(results),
+ "nonprofits": results,
+ "data_sources": {
+ "propublica": "https://projects.propublica.org/nonprofits/api",
+ "everyorg": "https://www.every.org/nonprofit-api",
+ "irs_teos": "https://www.irs.gov/charities-non-profits/tax-exempt-organization-search-bulk-data-downloads"
+ }
+ }
+
+ except Exception as e:
+ logger.error(f"Nonprofit search error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/api/data/status")
+async def get_data_status():
+ """
+ Get status of all reference data ingestions.
+
+ Returns counts and last update times for:
+ - Census jurisdictions
+ - NCES school districts
+ - Nonprofit organizations
+ - Meeting datasets (MeetingBank, LocalView, etc.)
+ """
+ try:
+ from pathlib import Path
+ from datetime import datetime
+
+ status = {
+ "census_jurisdictions": {
+ "path": "data/bronze/census_jurisdictions",
+ "status": "not_ingested",
+ "count": 0,
+ "last_updated": None
+ },
+ "nces_school_districts": {
+ "path": "data/bronze/nces_school_districts",
+ "status": "not_ingested",
+ "count": 0,
+ "last_updated": None
+ },
+ "nonprofits": {
+ "path": "data/cache/nonprofits",
+ "status": "cached",
+ "count": 0,
+ "last_updated": None
+ },
+ "meeting_datasets": {
+ "meetingbank": {"status": "available", "count": 1366},
+ "city_scrapers": {"status": "available", "count": "100-500"},
+ "open_states": {"status": "available", "count": "50+"}
+ }
+ }
+
+ # Check each data directory
+ for key in ["census_jurisdictions", "nces_school_districts", "nonprofits"]:
+ data_path = Path(status[key]["path"])
+ if data_path.exists():
+ files = list(data_path.glob("**/*"))
+ status[key]["count"] = len(files)
+ status[key]["status"] = "ingested" if files else "empty"
+ if files:
+ latest_file = max(files, key=lambda f: f.stat().st_mtime if f.is_file() else 0)
+ if latest_file.is_file():
+ status[key]["last_updated"] = datetime.fromtimestamp(
+ latest_file.stat().st_mtime
+ ).isoformat()
+
+ return status
+
+ except Exception as e:
+ logger.error(f"Data status error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/api/data/ingest/census")
+async def ingest_census_data(background_tasks: BackgroundTasks):
+ """
+ Trigger Census Bureau jurisdiction data ingestion.
+
+ Downloads and processes:
+ - 3,144 counties
+ - 19,500+ municipalities
+ - 36,000+ townships
+ - 13,000+ school districts
+
+ This is a long-running operation that runs in the background.
+ """
+ try:
+ def run_census_ingestion():
+ from discovery.census_ingestion import CensusGovernmentIngestion
+ import asyncio
+
+ logger.info("Starting Census data ingestion...")
+ ingestor = CensusGovernmentIngestion()
+
+ # Run async ingestion
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ result = loop.run_until_complete(ingestor.ingest_all_jurisdictions())
+ loop.close()
+
+ logger.success(f"Census ingestion complete: {result}")
+
+ background_tasks.add_task(run_census_ingestion)
+
+ return {
+ "message": "Census data ingestion started",
+ "status": "processing",
+ "check_status": "/api/data/status"
+ }
+
+ except Exception as e:
+ logger.error(f"Census ingestion error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/api/data/ingest/nces")
+async def ingest_nces_data(background_tasks: BackgroundTasks):
+ """
+ Trigger NCES school district data ingestion.
+
+ Downloads and processes 13,000+ school districts with:
+ - District names and addresses
+ - Contact information
+ - NCES IDs
+ - Enrollment data
+ """
+ try:
+ def run_nces_ingestion():
+ from discovery.nces_ingestion import NCESSchoolDistrictIngestion
+ import asyncio
+
+ logger.info("Starting NCES data ingestion...")
+ ingestor = NCESSchoolDistrictIngestion()
+
+ # Run async ingestion
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ result = loop.run_until_complete(ingestor.download_and_process())
+ loop.close()
+
+ logger.success(f"NCES ingestion complete: {result}")
+
+ background_tasks.add_task(run_nces_ingestion)
+
+ return {
+ "message": "NCES data ingestion started",
+ "status": "processing",
+ "check_status": "/api/data/status"
+ }
+
+ except Exception as e:
+ logger.error(f"NCES ingestion error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/api/data/ingest/nonprofits")
+async def ingest_nonprofits(
+ state: str = Query(..., description="2-letter state code"),
+ ntee_codes: Optional[List[str]] = Query(None, description="NTEE codes to ingest"),
+ background_tasks: BackgroundTasks = None
+):
+ """
+ Trigger nonprofit data ingestion for a specific state.
+
+ Bulk downloads nonprofit data from ProPublica API and caches locally.
+
+ Example: POST /api/data/ingest/nonprofits?state=AL&ntee_codes=E&ntee_codes=E20
+ """
+ try:
+ from discovery.nonprofit_discovery import NonprofitDiscovery
+
+ discovery = NonprofitDiscovery()
+ ntee_list = ntee_codes or ["E"] # Default to health
+
+ total_orgs = 0
+ for ntee_code in ntee_list:
+ orgs = discovery.search_propublica(state=state, ntee_code=ntee_code)
+ total_orgs += len(orgs)
+ logger.info(f"Cached {len(orgs)} nonprofits for {state}/{ntee_code}")
+
+ return {
+ "message": f"Nonprofit data ingestion complete for {state}",
+ "state": state,
+ "ntee_codes": ntee_list,
+ "organizations_cached": total_orgs,
+ "cache_location": "data/cache/nonprofits"
+ }
+
+ except Exception as e:
+ logger.error(f"Nonprofit ingestion error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/api/jurisdictions")
+async def get_jurisdictions(
+ state: Optional[str] = Query(None, description="2-letter state code"),
+ type: Optional[str] = Query(None, description="Type: county, municipality, township"),
+ limit: int = Query(100, le=1000)
+):
+ """
+ Query ingested Census jurisdiction data.
+
+ Returns government entities with FIPS codes, coordinates, and population.
+ """
+ try:
+ # This would query the Delta Lake census tables
+ # For now, return sample data
+ return {
+ "message": "Query census jurisdiction data from Delta Lake",
+ "filters": {"state": state, "type": type},
+ "limit": limit,
+ "note": "Requires Census data ingestion first (POST /api/data/ingest/census)",
+ "example_data": [
+ {
+ "name": "Tuscaloosa County",
+ "state": "AL",
+ "type": "county",
+ "fips": "01125",
+ "population": "209355"
+ }
+ ]
+ }
+
+ except Exception as e:
+ logger.error(f"Jurisdiction query error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/api/school-districts")
+async def get_school_districts(
+ state: Optional[str] = Query(None, description="2-letter state code"),
+ limit: int = Query(100, le=1000)
+):
+ """
+ Query ingested NCES school district data.
+
+ Returns school districts with contact information and enrollment.
+ """
+ try:
+ # This would query the Delta Lake NCES tables
+ return {
+ "message": "Query NCES school district data from Delta Lake",
+ "filters": {"state": state},
+ "limit": limit,
+ "note": "Requires NCES data ingestion first (POST /api/data/ingest/nces)",
+ "example_data": [
+ {
+ "name": "Tuscaloosa City Schools",
+ "state": "AL",
+ "nces_id": "0100123",
+ "phone": "(205) 759-3500",
+ "website": "https://www.tusc.k12.al.us/"
+ }
+ ]
+ }
+
+ except Exception as e:
+ logger.error(f"School district query error: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
+# Serve React frontend
+static_dir = Path(__file__).parent / "static"
+if static_dir.exists():
+ # Mount static files (JS, CSS, images)
+ app.mount("/assets", StaticFiles(directory=static_dir / "assets"), name="assets")
+
+ # Serve index.html for all non-API routes (SPA routing)
+ @app.get("/{full_path:path}")
+ async def serve_react_app(full_path: str):
+ """Serve React app for all non-API routes."""
+ if full_path.startswith("api/"):
+ raise HTTPException(status_code=404, detail="API endpoint not found")
+
+ index_file = static_dir / "index.html"
+ if index_file.exists():
+ return FileResponse(index_file)
+ else:
+ raise HTTPException(status_code=404, detail="Frontend not built")
+
+
+@app.on_event("startup")
+async def startup_event():
+ """Initialize system on startup."""
+ logger.info("Starting Oral Health Policy Pulse application...")
+
+ # Initialize Delta Lake if not exists
+ try:
+ await pipeline.initialize_tables()
+ logger.info("Delta Lake tables initialized")
+ except Exception as e:
+ logger.warning(f"Delta Lake initialization skipped: {e}")
+
+ logger.info("Application started successfully")
+
+
+@app.on_event("shutdown")
+async def shutdown_event():
+ """Cleanup on shutdown."""
+ logger.info("Shutting down application...")
+
+
+if __name__ == "__main__":
+ import uvicorn
+ uvicorn.run(
+ "api.app:app",
+ host="0.0.0.0",
+ port=8000,
+ reload=True
+ )
diff --git a/api/auth.py b/api/auth.py
new file mode 100644
index 0000000000000000000000000000000000000000..0c2e4ab0d2db44c0565fe69c81168278d92206f2
--- /dev/null
+++ b/api/auth.py
@@ -0,0 +1,153 @@
+"""
+Authentication utilities - JWT tokens, password hashing, OAuth helpers
+"""
+import os
+import secrets
+from datetime import datetime, timedelta
+from typing import Optional, Dict, Any
+from dotenv import load_dotenv
+
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+from fastapi import HTTPException, status, Depends
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from sqlalchemy.orm import Session
+
+from api.database import get_db
+from api.models import User
+
+# Load environment variables
+load_dotenv()
+
+# Security configuration
+SECRET_KEY = os.getenv("JWT_SECRET_KEY", secrets.token_urlsafe(32))
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7 days
+
+# Password hashing
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+# HTTP Bearer token scheme
+security = HTTPBearer(auto_error=False)
+
+
+def hash_password(password: str) -> str:
+ """Hash a password using bcrypt"""
+ return pwd_context.hash(password)
+
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+ """Verify a password against its hash"""
+ return pwd_context.verify(plain_password, hashed_password)
+
+
+def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
+ """
+ Create a JWT access token
+
+ Args:
+ data: Payload to encode (usually {"sub": user_id})
+ expires_delta: Optional custom expiration time
+
+ Returns:
+ JWT token string
+ """
+ to_encode = data.copy()
+
+ if expires_delta:
+ expire = datetime.utcnow() + expires_delta
+ else:
+ expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+
+ to_encode.update({"exp": expire})
+ encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+ return encoded_jwt
+
+
+def decode_access_token(token: str) -> Dict[str, Any]:
+ """
+ Decode and validate a JWT token
+
+ Args:
+ token: JWT token string
+
+ Returns:
+ Decoded payload
+
+ Raises:
+ HTTPException: If token is invalid or expired
+ """
+ try:
+ payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+ return payload
+ except JWTError:
+ raise HTTPException(
+ status_code=status.HTTP_401_UNAUTHORIZED,
+ detail="Could not validate credentials",
+ headers={"WWW-Authenticate": "Bearer"},
+ )
+
+
+def get_current_user(
+ credentials: HTTPAuthorizationCredentials = Depends(security),
+ db: Session = Depends(get_db)
+) -> Optional[User]:
+ """
+ Get current authenticated user from JWT token
+
+ Usage:
+ @app.get("/protected")
+ def protected_route(user: User = Depends(get_current_user)):
+ return {"message": f"Hello {user.email}"}
+
+ Returns:
+ User object if authenticated, None if optional auth
+ """
+ if not credentials:
+ return None
+
+ token = credentials.credentials
+ payload = decode_access_token(token)
+
+ user_id: int = payload.get("sub")
+ if user_id is None:
+ raise HTTPException(
+ status_code=status.HTTP_401_UNAUTHORIZED,
+ detail="Invalid authentication credentials"
+ )
+
+ user = db.query(User).filter(User.id == user_id).first()
+ if user is None:
+ raise HTTPException(
+ status_code=status.HTTP_401_UNAUTHORIZED,
+ detail="User not found"
+ )
+
+ # Update last login
+ user.last_login = datetime.utcnow()
+ db.commit()
+
+ return user
+
+
+def require_auth(user: User = Depends(get_current_user)) -> User:
+ """
+ Require authentication (raises 401 if not authenticated)
+
+ Usage:
+ @app.get("/protected")
+ def protected_route(user: User = Depends(require_auth)):
+ return {"message": f"Hello {user.email}"}
+ """
+ if user is None:
+ raise HTTPException(
+ status_code=status.HTTP_401_UNAUTHORIZED,
+ detail="Authentication required",
+ headers={"WWW-Authenticate": "Bearer"},
+ )
+ return user
+
+
+def generate_state_token() -> str:
+ """Generate a secure random state token for OAuth CSRF protection"""
+ return secrets.token_urlsafe(32)
diff --git a/api/database.py b/api/database.py
new file mode 100644
index 0000000000000000000000000000000000000000..29e20907910cea545863afa78d975a80e1709ac3
--- /dev/null
+++ b/api/database.py
@@ -0,0 +1,62 @@
+"""
+Database connection and session management
+"""
+import os
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, Session
+from sqlalchemy.pool import StaticPool
+from typing import Generator
+
+from api.models import Base
+
+# Database URL from environment or default to SQLite for development
+DATABASE_URL = os.getenv(
+ "DATABASE_URL",
+ "sqlite:///./data/users.db" # Fallback to SQLite if no PostgreSQL configured
+)
+
+# Handle PostgreSQL URL format for SQLAlchemy 2.0+
+if DATABASE_URL.startswith("postgres://"):
+ DATABASE_URL = DATABASE_URL.replace("postgres://", "postgresql://", 1)
+
+# Create engine
+if "sqlite" in DATABASE_URL:
+ # SQLite needs special handling for concurrent access
+ engine = create_engine(
+ DATABASE_URL,
+ connect_args={"check_same_thread": False},
+ poolclass=StaticPool,
+ )
+else:
+ # PostgreSQL configuration
+ engine = create_engine(
+ DATABASE_URL,
+ pool_pre_ping=True,
+ pool_size=10,
+ max_overflow=20,
+ )
+
+# Create session factory
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+
+def init_db():
+ """Create all tables"""
+ Base.metadata.create_all(bind=engine)
+ print(f"✅ Database initialized at: {DATABASE_URL}")
+
+
+def get_db() -> Generator[Session, None, None]:
+ """
+ Database session dependency for FastAPI
+
+ Usage:
+ @app.get("/users")
+ def get_users(db: Session = Depends(get_db)):
+ return db.query(User).all()
+ """
+ db = SessionLocal()
+ try:
+ yield db
+ finally:
+ db.close()
diff --git a/api/errors.py b/api/errors.py
new file mode 100644
index 0000000000000000000000000000000000000000..02a10129e1f7a608850c5b0b2d55831933823502
--- /dev/null
+++ b/api/errors.py
@@ -0,0 +1,154 @@
+"""
+Structured error models for API responses.
+
+Provides user-friendly error messages with expandable technical details.
+"""
+from pydantic import BaseModel, Field
+from typing import Optional, Dict, Any
+
+
+class ErrorDetail(BaseModel):
+ """Structured error detail with user-friendly message and technical info."""
+
+ message: str = Field(
+ ...,
+ description="User-friendly error message"
+ )
+
+ error_type: str = Field(
+ ...,
+ description="Type of error (e.g., 'data_not_found', 'network_error', 'validation_error')"
+ )
+
+ technical_details: Optional[str] = Field(
+ None,
+ description="Technical error details for debugging (expandable in UI)"
+ )
+
+ suggestions: Optional[list[str]] = Field(
+ None,
+ description="Helpful suggestions for the user"
+ )
+
+ metadata: Optional[Dict[str, Any]] = Field(
+ None,
+ description="Additional context (state, dataset name, etc.)"
+ )
+
+
+def parse_error(exception: Exception, context: Optional[Dict[str, Any]] = None) -> ErrorDetail:
+ """
+ Parse an exception into a structured ErrorDetail with user-friendly message.
+
+ Args:
+ exception: The exception to parse
+ context: Additional context (state, dataset, etc.)
+
+ Returns:
+ ErrorDetail with user-friendly message and technical details
+ """
+ error_str = str(exception)
+ context = context or {}
+
+ # Parse HuggingFace dataset not found errors
+ if "HTTP 404 Not Found" in error_str and "huggingface.co/datasets" in error_str:
+ # Extract dataset name from URL
+ import re
+ match = re.search(r'datasets/([^/]+/[^/]+)/', error_str)
+ dataset_name = match.group(1) if match else "unknown"
+
+ # Extract state from dataset name or context
+ state = context.get('state', 'Unknown')
+ data_type = 'bills' if 'bills' in dataset_name else 'data'
+
+ return ErrorDetail(
+ message=f"No {data_type} data available for {state.upper()}",
+ error_type="data_not_found",
+ technical_details=f"Dataset '{dataset_name}' not found on HuggingFace.\n\nFull error: {error_str}",
+ suggestions=[
+ f"Try a different state - we have data for 50+ states",
+ f"Check /api/bills/map to see which states have {data_type} data",
+ "Contact support if you believe this data should be available"
+ ],
+ metadata={
+ "dataset": dataset_name,
+ "state": state,
+ "data_type": data_type
+ }
+ )
+
+ # Parse file not found errors (local environment)
+ elif "No such file or directory" in error_str or "FileNotFoundError" in error_str:
+ state = context.get('state', 'Unknown')
+ data_type = context.get('data_type', 'data')
+
+ return ErrorDetail(
+ message=f"No {data_type} available for {state.upper()}",
+ error_type="data_not_found",
+ technical_details=error_str,
+ suggestions=[
+ f"This state may not have {data_type} in our database yet",
+ "Try a different state or check which states have data",
+ "Data is being continuously added - check back later"
+ ],
+ metadata={
+ "state": state,
+ "data_type": data_type
+ }
+ )
+
+ # Parse DuckDB/SQL errors
+ elif "DuckDB" in error_str or "SYNTAX ERROR" in error_str or "LINE" in error_str:
+ return ErrorDetail(
+ message="Database query error - please check your search parameters",
+ error_type="query_error",
+ technical_details=error_str,
+ suggestions=[
+ "Try simplifying your search query",
+ "Check that all parameters are valid",
+ "Contact support if the issue persists"
+ ],
+ metadata=context
+ )
+
+ # Parse network/timeout errors
+ elif "timeout" in error_str.lower() or "connection" in error_str.lower():
+ return ErrorDetail(
+ message="Network request timed out - please try again",
+ error_type="network_error",
+ technical_details=error_str,
+ suggestions=[
+ "Try again in a few seconds",
+ "Check your internet connection",
+ "The server may be temporarily busy"
+ ],
+ metadata=context
+ )
+
+ # Parse validation errors
+ elif "validation" in error_str.lower() or "invalid" in error_str.lower():
+ return ErrorDetail(
+ message="Invalid request parameters",
+ error_type="validation_error",
+ technical_details=error_str,
+ suggestions=[
+ "Check that all required parameters are provided",
+ "Verify parameter formats (e.g., state codes should be 2 letters)",
+ "See API documentation for valid parameter values"
+ ],
+ metadata=context
+ )
+
+ # Generic error fallback
+ else:
+ return ErrorDetail(
+ message="An unexpected error occurred",
+ error_type="server_error",
+ technical_details=error_str,
+ suggestions=[
+ "Try again in a few moments",
+ "Contact support if the issue persists",
+ "Check the technical details for more information"
+ ],
+ metadata=context
+ )
diff --git a/api/main.py b/api/main.py
new file mode 100644
index 0000000000000000000000000000000000000000..f744c848d2f23ff2578cdbee105455e94758d708
--- /dev/null
+++ b/api/main.py
@@ -0,0 +1,1288 @@
+"""
+FastAPI application for the Oral Health Policy Pulse system.
+
+Provides REST API endpoints for:
+- Initiating policy analysis workflows
+- Querying advocacy opportunities
+- Retrieving generated materials
+- Accessing visualizations
+- System status and monitoring
+"""
+from typing import List, Dict, Any, Optional
+from datetime import datetime, date
+import sys
+import time
+from pathlib import Path
+from fastapi import FastAPI, HTTPException, Query, BackgroundTasks, Request
+from fastapi.responses import HTMLResponse, JSONResponse
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+from fastapi.openapi.docs import get_swagger_ui_html
+from pydantic import BaseModel, Field
+from loguru import logger
+import os
+
+from agents.orchestrator import OrchestratorAgent
+from agents.scraper import ScraperAgent
+from agents.parser import ParserAgent
+from agents.classifier import ClassifierAgent
+from agents.sentiment import SentimentAnalyzerAgent
+from agents.advocacy import AdvocacyWriterAgent
+from pipeline.delta_lake import DeltaLakePipeline
+from visualization.heatmap import AdvocacyHeatmap
+from config import settings
+
+# Configure logging with rotation and retention
+# Output to both file (with rotation) and stderr (for HuggingFace container logs)
+logger.remove() # Remove default handler
+
+# Add console output (shows in HuggingFace container logs)
+logger.add(
+ sys.stderr,
+ format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function} - {message}",
+ level=settings.log_level
+)
+
+# Add file output with rotation and retention
+logger.add(
+ settings.log_file,
+ rotation="500 MB", # Create new file when size exceeds 500MB
+ retention="10 days", # Delete logs older than 10 days
+ level=settings.log_level,
+ format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function} - {message}"
+)
+
+# Initialize FastAPI app
+app = FastAPI(
+ title="Open Navigator API",
+ description="Multi-agent system for analyzing local government oral health policy discussions",
+ version="1.0.0",
+ docs_url=None, # Disable default docs to use custom
+ redoc_url="/redoc", # Keep ReDoc at /redoc
+ openapi_tags=[
+ {
+ "name": "auth",
+ "description": "Authentication and user management"
+ },
+ {
+ "name": "social",
+ "description": "Social features - follow users, leaders, organizations, and causes"
+ },
+ {
+ "name": "workflows",
+ "description": "Policy analysis workflows"
+ },
+ {
+ "name": "opportunities",
+ "description": "Advocacy opportunities"
+ }
+ ]
+)
+
+# Custom OpenAPI schema with logo
+def custom_openapi():
+ if app.openapi_schema:
+ return app.openapi_schema
+
+ from fastapi.openapi.utils import get_openapi
+
+ openapi_schema = get_openapi(
+ title=app.title,
+ version=app.version,
+ description=app.description,
+ routes=app.routes,
+ )
+
+ # Add custom logo
+ openapi_schema["info"]["x-logo"] = {
+ "url": "/static/communityone_logo.svg",
+ "altText": "CommunityOne Logo"
+ }
+
+ app.openapi_schema = openapi_schema
+ return app.openapi_schema
+
+app.openapi = custom_openapi
+
+# Add CORS middleware
+app.add_middleware(
+ CORSMiddleware,
+ allow_origins=["*"],
+ allow_credentials=True,
+ allow_methods=["*"],
+ allow_headers=["*"],
+)
+
+# Request logging middleware
+@app.middleware("http")
+async def log_requests(request: Request, call_next):
+ """Log all API requests with timing and response info"""
+ start_time = time.time()
+
+ # Get client info
+ client_host = request.client.host if request.client else "unknown"
+
+ # Log incoming request
+ logger.info(f"➡️ {request.method} {request.url.path} - Client: {client_host}")
+
+ # Process request
+ try:
+ response = await call_next(request)
+
+ # Calculate duration
+ duration_ms = (time.time() - start_time) * 1000
+
+ # Get response size if available
+ response_size = response.headers.get("content-length", "unknown")
+
+ # Log response with appropriate emoji based on status
+ if response.status_code < 400:
+ logger.info(
+ f"✅ {request.method} {request.url.path} - "
+ f"Status: {response.status_code} - "
+ f"Duration: {duration_ms:.2f}ms - "
+ f"Size: {response_size} bytes"
+ )
+ elif response.status_code < 500:
+ logger.warning(
+ f"⚠️ {request.method} {request.url.path} - "
+ f"Status: {response.status_code} - "
+ f"Duration: {duration_ms:.2f}ms"
+ )
+ else:
+ logger.error(
+ f"❌ {request.method} {request.url.path} - "
+ f"Status: {response.status_code} - "
+ f"Duration: {duration_ms:.2f}ms"
+ )
+
+ return response
+
+ except Exception as e:
+ duration_ms = (time.time() - start_time) * 1000
+ logger.error(
+ f"💥 {request.method} {request.url.path} - "
+ f"Error: {str(e)} - "
+ f"Duration: {duration_ms:.2f}ms"
+ )
+ raise
+
+# Mount static files for logo
+static_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "frontend", "public")
+if os.path.exists(static_dir):
+ app.mount("/static", StaticFiles(directory=static_dir), name="static")
+else:
+ logger.warning(f"Static directory not found: {static_dir}")
+
+# Include authentication routes
+from api.routes import auth as auth_routes
+from api.routes import social as social_routes
+from api.routes import search as search_routes
+# Use Neon database for fast stats queries (500x faster than parquet)
+from api.routes import stats_neon as stats_routes # Was: stats
+from api.routes import contact as contact_routes
+# Use hybrid approach for bills: Neon for map, parquet for drill-down (saves space)
+from api.routes import bills_neon as bills_routes # Was: bills
+from api.database import init_db
+
+app.include_router(auth_routes.router)
+app.include_router(social_routes.router)
+app.include_router(search_routes.router)
+app.include_router(stats_routes.router, prefix="/api", tags=["stats"])
+app.include_router(contact_routes.router)
+app.include_router(bills_routes.router)
+
+# Custom Swagger UI with logo
+@app.get("/docs", include_in_schema=False)
+async def custom_swagger_ui_html():
+ """Custom Swagger UI with CommunityOne logo"""
+ return get_swagger_ui_html(
+ openapi_url=app.openapi_url,
+ title=f"{app.title} - API Documentation",
+ swagger_favicon_url="/static/communityone_logo_64.png",
+ swagger_ui_parameters={
+ "defaultModelsExpandDepth": -1,
+ "docExpansion": "list",
+ "filter": True,
+ "syntaxHighlight.theme": "monokai"
+ }
+ )
+
+# Initialize database on startup
+@app.on_event("startup")
+async def init_database():
+ """Initialize authentication database"""
+ try:
+ init_db()
+ logger.info("✅ Authentication database initialized")
+ except Exception as e:
+ logger.warning(f"⚠️ Database initialization skipped: {e}")
+
+# Initialize components
+orchestrator = OrchestratorAgent()
+pipeline = DeltaLakePipeline()
+heatmap_generator = AdvocacyHeatmap()
+
+
+# Pydantic models for API
+class ScrapeTarget(BaseModel):
+ """Configuration for a scraping target."""
+ url: str
+ municipality: str
+ state: str
+ platform: str = "generic"
+
+
+class WorkflowRequest(BaseModel):
+ """Request to start a new analysis workflow."""
+ scrape_targets: List[ScrapeTarget]
+ date_range: Optional[Dict[str, str]] = None
+ description: Optional[str] = None
+
+
+class WorkflowResponse(BaseModel):
+ """Response for workflow operations."""
+ workflow_id: str
+ status: str
+ message: str
+
+
+class OpportunityFilter(BaseModel):
+ """Filters for querying opportunities."""
+ state: Optional[str] = None
+ municipality: Optional[str] = None
+ topic: Optional[str] = None
+ urgency: Optional[str] = None
+ min_date: Optional[date] = None
+ max_date: Optional[date] = None
+
+
+class OpportunityResponse(BaseModel):
+ """Response containing advocacy opportunities."""
+ opportunities: List[Dict[str, Any]]
+ total_count: int
+ filters_applied: Dict[str, Any]
+
+
+class SystemStatus(BaseModel):
+ """System status information."""
+ status: str
+ active_workflows: int
+ agent_status: Dict[str, Any]
+ last_update: datetime
+
+
+# API Endpoints
+
+@app.get("/", response_class=HTMLResponse)
+async def root():
+ """Root endpoint with API information."""
+ html_content = """
+
+
+
+ Open Navigator API
+
+
+
+
+
+ 🔑 Key Endpoints:
+
+ POST /workflow/start - Start a new analysis workflow
+
+
+ GET /opportunities - Query advocacy opportunities
+
+
+ GET /heatmap - Get advocacy heatmap visualization
+
+
+ GET /status - System status and health
+
+
+ POST /auth/login/{provider} - OAuth login (HuggingFace, Google, Facebook, GitHub)
+
+
+ 
+