""" AI Extractor - Intelligent data extraction using OpenAI/DeepSeek API. Handles: - Deep page content analysis for funding data - Contact information extraction and validation - Address parsing and location inference - Structured data extraction from unstructured text - Funding format normalization ($xxB → numeric) - US office detection and multi-location decision logic - LinkedIn page structure recognition - Email priority decision tree """ import json import re from typing import Optional, Dict, Any, List, Tuple import httpx # ─── US Census Region Mapping ────────────────────────────────── US_CENSUS_REGIONS = { # Northeast "CT": {"region": "Northeast", "division": "New England"}, "ME": {"region": "Northeast", "division": "New England"}, "MA": {"region": "Northeast", "division": "New England"}, "NH": {"region": "Northeast", "division": "New England"}, "RI": {"region": "Northeast", "division": "New England"}, "VT": {"region": "Northeast", "division": "New England"}, "NJ": {"region": "Northeast", "division": "Middle Atlantic"}, "NY": {"region": "Northeast", "division": "Middle Atlantic"}, "PA": {"region": "Northeast", "division": "Middle Atlantic"}, # Midwest "IL": {"region": "Midwest", "division": "East North Central"}, "IN": {"region": "Midwest", "division": "East North Central"}, "MI": {"region": "Midwest", "division": "East North Central"}, "OH": {"region": "Midwest", "division": "East North Central"}, "WI": {"region": "Midwest", "division": "East North Central"}, "IA": {"region": "Midwest", "division": "West North Central"}, "KS": {"region": "Midwest", "division": "West North Central"}, "MN": {"region": "Midwest", "division": "West North Central"}, "MO": {"region": "Midwest", "division": "West North Central"}, "NE": {"region": "Midwest", "division": "West North Central"}, "ND": {"region": "Midwest", "division": "West North Central"}, "SD": {"region": "Midwest", "division": "West North Central"}, # South "DE": {"region": "South", "division": "South Atlantic"}, "DC": {"region": "South", "division": "South Atlantic"}, "FL": {"region": "South", "division": "South Atlantic"}, "GA": {"region": "South", "division": "South Atlantic"}, "MD": {"region": "South", "division": "South Atlantic"}, "NC": {"region": "South", "division": "South Atlantic"}, "SC": {"region": "South", "division": "South Atlantic"}, "VA": {"region": "South", "division": "South Atlantic"}, "WV": {"region": "South", "division": "South Atlantic"}, "AL": {"region": "South", "division": "East South Central"}, "KY": {"region": "South", "division": "East South Central"}, "MS": {"region": "South", "division": "East South Central"}, "TN": {"region": "South", "division": "East South Central"}, "AR": {"region": "South", "division": "West South Central"}, "LA": {"region": "South", "division": "West South Central"}, "OK": {"region": "South", "division": "West South Central"}, "TX": {"region": "South", "division": "West South Central"}, # West "AZ": {"region": "West", "division": "Mountain"}, "CO": {"region": "West", "division": "Mountain"}, "ID": {"region": "West", "division": "Mountain"}, "MT": {"region": "West", "division": "Mountain"}, "NV": {"region": "West", "division": "Mountain"}, "NM": {"region": "West", "division": "Mountain"}, "UT": {"region": "West", "division": "Mountain"}, "WY": {"region": "West", "division": "Mountain"}, "AK": {"region": "West", "division": "Pacific"}, "CA": {"region": "West", "division": "Pacific"}, "HI": {"region": "West", "division": "Pacific"}, "OR": {"region": "West", "division": "Pacific"}, "WA": {"region": "West", "division": "Pacific"}, } # ─── Post-processing helpers for AI output ─────────────────────── # Minimal US state mapping (abbreviation → full name) for validation _STATE_ABBREV_TO_FULL = { "AL": "Alabama", "AK": "Alaska", "AZ": "Arizona", "AR": "Arkansas", "CA": "California", "CO": "Colorado", "CT": "Connecticut", "DE": "Delaware", "FL": "Florida", "GA": "Georgia", "HI": "Hawaii", "ID": "Idaho", "IL": "Illinois", "IN": "Indiana", "IA": "Iowa", "KS": "Kansas", "KY": "Kentucky", "LA": "Louisiana", "ME": "Maine", "MD": "Maryland", "MA": "Massachusetts", "MI": "Michigan", "MN": "Minnesota", "MS": "Mississippi", "MO": "Missouri", "MT": "Montana", "NE": "Nebraska", "NV": "Nevada", "NH": "New Hampshire", "NJ": "New Jersey", "NM": "New Mexico", "NY": "New York", "NC": "North Carolina", "ND": "North Dakota", "OH": "Ohio", "OK": "Oklahoma", "OR": "Oregon", "PA": "Pennsylvania", "RI": "Rhode Island", "SC": "South Carolina", "SD": "South Dakota", "TN": "Tennessee", "TX": "Texas", "UT": "Utah", "VT": "Vermont", "VA": "Virginia", "WA": "Washington", "WV": "West Virginia", "WI": "Wisconsin", "WY": "Wyoming", "DC": "District of Columbia", } # Reverse: full name (lower) → canonical full name _STATE_FULL_NAMES = {v.lower(): v for v in _STATE_ABBREV_TO_FULL.values()} def _normalize_state_name(raw: str) -> Optional[str]: """Convert any state reference to canonical full name.""" if not raw: return None raw = raw.strip().strip(",.;") # Abbreviation → full name upper = raw.upper() if upper in _STATE_ABBREV_TO_FULL: return _STATE_ABBREV_TO_FULL[upper] # Full name (case-insensitive) → canonical lower = raw.lower() if lower in _STATE_FULL_NAMES: return _STATE_FULL_NAMES[lower] return raw def _normalize_city_name(raw: str) -> Optional[str]: """Clean a city name: strip punctuation, trailing state/country.""" if not raw: return None raw = raw.strip().strip(",.;") if "," in raw: raw = raw.split(",")[0].strip() if len(raw) < 2 or raw.isdigit(): return None return raw def _clean_contact_output(raw: str) -> Optional[str]: """Clean a contact value: strip brackets, labels, whitespace.""" if not raw: return None raw = raw.strip() # Remove surrounding brackets raw = re.sub(r'^[\[(]\s*', '', raw) raw = re.sub(r'\s*[\])]\s*$', '', raw) raw = raw.strip() # Remove label prefix but NOT http(s) protocol if not re.match(r'^https?://', raw, re.IGNORECASE): raw = re.sub(r'^[A-Za-z][A-Za-z0-9]*:\s+', '', raw) raw = raw.strip() if not raw or raw.lower() in ("null", "none", "n/a"): return None return raw class DecisionEngine: """Decision logic for US office detection, location priority, and contact routing.""" @staticmethod def _clean_url(url: str) -> str: """Strip brackets and labels from URL, preserving protocol prefixes.""" url = url.strip() url = re.sub(r'^[\[(]\s*', '', url) url = re.sub(r'\s*[\])]\s*$', '', url) url = url.strip() if not re.match(r'^https?://', url) and not re.match(r'^ftp://', url): url = re.sub(r'^[A-Za-z][A-Za-z0-9]*:\s+', '', url) return url.strip() # ─── US Office Detection ─────────────────────────────────── @staticmethod def is_us_office(ai_data: Dict) -> bool: """Determine if a company has a US office based on AI-extracted data.""" # Direct US office indicators if ai_data.get("us_office_city"): return True # HQ is in the US hq_country = (ai_data.get("headquarters_country") or "").strip().upper() if hq_country in ("US", "USA", "UNITED STATES", "U.S.", "U.S.A."): return True # HQ state is a valid US state abbreviation hq_state = (ai_data.get("headquarters_state") or "").strip().upper() if hq_state in US_CENSUS_REGIONS: return True # Check other_locations for US addresses for loc in ai_data.get("other_locations", []): loc_upper = loc.upper() if any(s in loc_upper for s in US_CENSUS_REGIONS): return True if "USA" in loc_upper or "UNITED STATES" in loc_upper: return True return False @staticmethod def determine_best_us_location( ai_data: Dict, wellfound_location: str = "", website_addresses: List[str] = None, ) -> Tuple[Optional[str], Optional[str]]: """Determine the best US location for application. Priority order: 1. Hiring focus location (where jobs are concentrated) 2. US office city/state (if explicitly identified) 3. HQ city/state (if HQ is in US) 4. Parse from Wellfound location string 5. Parse from website addresses Returns: Tuple of (city, state_abbr) or (None, None) if no US location found """ # Priority 1: Hiring focus hiring_loc = ai_data.get("hiring_focus_location", "") if hiring_loc: city, state = DecisionEngine._parse_us_city_state(hiring_loc) if city and state: return city, state # Priority 2: US office us_city = ai_data.get("us_office_city") us_state = ai_data.get("us_office_state") if us_city: state_abbr = DecisionEngine._normalize_state(us_state) return us_city, state_abbr # Priority 3: US HQ hq_country = (ai_data.get("headquarters_country") or "").strip().upper() if hq_country in ("US", "USA", "UNITED STATES", "U.S.", "U.S.A."): hq_city = ai_data.get("headquarters_city") hq_state = ai_data.get("headquarters_state") if hq_city: state_abbr = DecisionEngine._normalize_state(hq_state) return hq_city, state_abbr # Priority 4: Wellfound location if wellfound_location: city, state = DecisionEngine._parse_us_city_state(wellfound_location) if city and state: return city, state # Priority 5: Website addresses if website_addresses: for addr in website_addresses: city, state = DecisionEngine._parse_us_city_state(addr) if city and state: return city, state return None, None @staticmethod def _parse_us_city_state(text: str) -> Tuple[Optional[str], Optional[str]]: """Parse US city and state from free-text location string.""" if not text: return None, None # Try "City, ST ZIP" pattern match = re.search( r'([A-Z][a-z]+(?:\s[A-Z][a-z]+)*),\s*([A-Z]{2})\s*\d{5}', text ) if match: return match.group(1), match.group(2) # Try "City, ST" pattern match = re.search( r'([A-Z][a-z]+(?:\s[A-Z][a-z]+)*),\s*([A-Z]{2})(?:\b|,|\s|$)', text ) if match: return match.group(1), match.group(2) # Try "City, State Name" match = re.search( r'([A-Z][a-z]+(?:\s[A-Z][a-z]+)*),\s*([A-Z][a-z]+(?:\s[A-Z][a-z]+)*)', text ) if match: city = match.group(1) state_name = match.group(2) state_abbr = DecisionEngine._state_name_to_abbr(state_name) if state_abbr: return city, state_abbr return None, None @staticmethod def _normalize_state(state_input: str) -> Optional[str]: """Normalize state to 2-letter abbreviation.""" if not state_input: return None state_input = state_input.strip().upper() # Already an abbreviation if len(state_input) == 2 and state_input in US_CENSUS_REGIONS: return state_input # Full name → abbreviation return DecisionEngine._state_name_to_abbr(state_input) @staticmethod def _state_name_to_abbr(state_name: str) -> Optional[str]: """Convert state full name to abbreviation (reverse lookup).""" name_upper = state_name.strip().upper() for abbr, info in US_CENSUS_REGIONS.items(): division = info["division"].upper().replace(" ", "") if abbr == name_upper: return abbr # Division names like "NewYork" won't match but we try anyway if division in name_upper.replace(" ", ""): continue # Common manual mappings name_map = { "CALIFORNIA": "CA", "NEW YORK": "NY", "TEXAS": "TX", "FLORIDA": "FL", "ILLINOIS": "IL", "WASHINGTON": "WA", "MASSACHUSETTS": "MA", "PENNSYLVANIA": "PA", "OHIO": "OH", "GEORGIA": "GA", "NORTH CAROLINA": "NC", "MICHIGAN": "MI", "COLORADO": "CO", "VIRGINIA": "VA", "NEW JERSEY": "NJ", "OREGON": "OR", "MINNESOTA": "MN", "ARIZONA": "AZ", "MARYLAND": "MD", "INDIANA": "IN", "MISSOURI": "MO", "WISCONSIN": "WI", "TENNESSEE": "TN", "CONNECTICUT": "CT", "NEVADA": "NV", "UTAH": "UT", "IOWA": "IA", "KANSAS": "KS", "ARKANSAS": "AR", "NEBRASKA": "NE", "MISSISSIPPI": "MS", "NEW MEXICO": "NM", "KENTUCKY": "KY", "LOUISIANA": "LA", "OKLAHOMA": "OK", "SOUTH CAROLINA": "SC", "ALABAMA": "AL", "HAWAII": "HI", "WEST VIRGINIA": "WV", "DELAWARE": "DE", "NEW HAMPSHIRE": "NH", "MAINE": "ME", "MONTANA": "MT", "RHODE ISLAND": "RI", "NORTH DAKOTA": "ND", "SOUTH DAKOTA": "SD", "ALASKA": "AK", "VERMONT": "VT", "WYOMING": "WY", "DISTRICT OF COLUMBIA": "DC", "DC": "DC", } return name_map.get(name_upper) # ─── Email Priority Decision ────────────────────────────── @staticmethod def rank_contact_email( email: Optional[str], contact_form_url: Optional[str], careers_page_url: Optional[str], ) -> str: """Rank contact methods by priority and return the best one. Priority: careers/recruiting email > contact form > careers page > None Returns: The best contact method (email address or URL) """ # Clean URL if wrapped if contact_form_url: contact_form_url = DecisionEngine._clean_url(str(contact_form_url)) if careers_page_url: careers_page_url = DecisionEngine._clean_url(str(careers_page_url)) if not email and not contact_form_url and not careers_page_url: return "" # Priority 1: HR/recruiting email if email: email_lower = email.lower() # Direct recruiting emails - highest priority recruiting_patterns = [ "careers@", "career@", "jobs@", "job@", "recruiting@", "recruit@", "talent@", "hr@", "hiring@", "people@", "recruiter@", ] for pattern in recruiting_patterns: if pattern in email_lower: return email # General emails - lower priority (still use if nothing else) general_patterns = ["info@", "contact@", "hello@", "support@", "admin@"] is_general = any(p in email_lower for p in general_patterns) if not contact_form_url and not careers_page_url: return email # Use general email if no alternatives if not is_general: return email # Non-general email is acceptable # Priority 2: Contact form URL if contact_form_url: return contact_form_url # Priority 3: Careers page if careers_page_url: return careers_page_url # Fallback: general email if email: return email return "" # ─── LinkedIn Structure Recognition ───────────────────────── @staticmethod def parse_linkedin_company_info(text: str) -> Dict[str, Any]: """Extract HQ info from LinkedIn company page text. LinkedIn typically has structured sections: - "Headquarters" → city, state, country - "Company size" → employee range - "Industry" → sector - "Locations" → list of office cities """ result = { "headquarters": None, "headquarters_city": None, "headquarters_state": None, "headquarters_country": None, "company_size": None, "industry": None, "locations": [], } lines = text.split("\n") for i, line in enumerate(lines): stripped = line.strip() # Headquarters section if "headquarters" in stripped.lower() and i + 1 < len(lines): hq_text = lines[i + 1].strip() result["headquarters"] = hq_text # Try to parse city, state, country parts = hq_text.split(",") if len(parts) >= 2: result["headquarters_city"] = parts[0].strip() state_country = parts[1].strip() if len(parts) >= 3: result["headquarters_state"] = state_country.strip() result["headquarters_country"] = parts[2].strip() else: state_upper = state_country.strip().upper() if state_upper in US_CENSUS_REGIONS: result["headquarters_state"] = state_country.strip() result["headquarters_country"] = "US" else: result["headquarters_country"] = state_country.strip() # Company size if "company size" in stripped.lower() and i + 1 < len(lines): result["company_size"] = lines[i + 1].strip() # Industry if stripped.lower().startswith("industry") and i + 1 < len(lines): result["industry"] = lines[i + 1].strip() # Locations (multiple offices) if stripped.lower() == "locations" or "offices" in stripped.lower(): j = i + 1 while j < len(lines) and j < i + 20: loc_line = lines[j].strip() if not loc_line or loc_line.lower().startswith(("see", "view", "show")): break result["locations"].append(loc_line) j += 1 return result class AIExtractor: """AI-powered data extraction using LLM APIs.""" # API endpoints PROVIDERS = { "openai": { "url": "https://api.openai.com/v1/chat/completions", "models": ["gpt-4o", "gpt-4o-mini", "gpt-4-turbo", "gpt-3.5-turbo"], }, "deepseek": { "url": "https://api.deepseek.com/v1/chat/completions", "models": ["deepseek-chat", "deepseek-reasoner"], }, } def __init__(self, provider: str = "openai", api_key: str = "", model: str = "auto"): if provider not in self.PROVIDERS: raise ValueError(f"Unknown provider: {provider}. Use 'openai' or 'deepseek'") self.provider = provider self.api_key = api_key self.base_url = self.PROVIDERS[provider]["url"] if model == "auto": self.model = self.PROVIDERS[provider]["models"][0] else: self.model = model async def analyze_funding_page( self, company_name: str, page_text: str, meta: Dict = None ) -> Dict[str, Any]: """Extract funding information from a Wellfound-like page.""" # Truncate text to avoid token limits text_sample = page_text[:8000] if page_text else "" system_prompt = """You are a precise data extraction AI. Extract funding information from company profile pages. Return ONLY valid JSON with these fields. Use null for unknown values. Do NOT guess - only extract when confident. Output format: { "valuation": "string or null - company valuation like '$50M', '$2.5B'", "total_raised": "string or null - total funding raised like '$10M'", "rounds": "integer or null - number of funding rounds", "series": "string or null - latest series like 'Series A', 'Seed', 'Series B'", "investors": ["string array - key investors mentioned"], "confidence": { "valuation": "high|medium|low", "total_raised": "high|medium|low", "rounds": "high|medium|low", "series": "high|medium|low" } }""" user_prompt = f"""Company: {company_name} Page content: {text_sample} Extract funding information from this company profile. Be precise and only return data you can confidently identify.""" response = await self._call_api(system_prompt, user_prompt) result = self._parse_json_response(response, "funding") # Normalize funding values result = self._normalize_funding(result) return result @staticmethod def _normalize_funding(data: Dict[str, Any]) -> Dict[str, Any]: """Normalize funding values: '$2.5B' → numeric, etc. Adds a 'valuation_numeric' field and standardizes format. """ def parse_money(val: Optional[str]) -> Optional[float]: if not val: return None val = str(val).strip() # Match patterns like $2.5B, $50M, $1.2K, $500K, 3B match = re.search( r'\$?\s*(\d+(?:\.\d+)?)\s*(B|Billion|bn|M|Million|m|K|Thousand|k|T|Trillion)?', val, re.IGNORECASE ) if not match: try: return float(val.replace(",", "")) except ValueError: return None number = float(match.group(1)) suffix = (match.group(2) or "") multiplier_map = { "T": 1e12, "TRILLION": 1e12, "B": 1e9, "BILLION": 1e9, "BN": 1e9, "M": 1e6, "MILLION": 1e6, "K": 1e3, "THOUSAND": 1e3, } number *= multiplier_map.get(suffix.upper(), 1) return number if data.get("valuation"): data["valuation_numeric"] = parse_money(data["valuation"]) if data.get("total_raised"): data["total_raised_numeric"] = parse_money(data["total_raised"]) return data async def analyze_company_website( self, company_name: str, page_text: str, emails: List[str], links: List[Dict], phones: List[str], addresses: List[str], wellfound_location: str = "", ) -> Dict[str, Any]: """Deep-analyze company website via AI to directly determine the three target fields: location.apply, state.apply, and contact. The AI receives aggregated multi-page content and makes the final structured decision — no post-processing address parsers or contact priority heuristics are needed. """ text_sample = page_text[:18000] if page_text else "" system_prompt = """You are a precise data-extraction AI. You will receive the FULL aggregated text from a company's website (homepage + sub-pages such as careers, contact, about, locations, team), plus extracted emails, links, phone numbers, and physical addresses. Your job is to read and deeply understand the website content, then make **final structured decisions** for exactly three fields: ## 1. location_apply (string or null) The best US city name for a job applicant to target. Decision rules: - If the company is headquartered in the US, use the HQ city. - If the company is foreign but has a US office, use the US office city. - If job listings concentrate in a particular US city (hiring_focus), use that city. - If the website clearly states a US office location, use that. - Fall back to parsing the wellfound_location string (provided below) for a US city. - Fall back to any physical address found on the website that is in the US. - **Always return ONLY a US city name.** If the company has NO US presence whatsoever, return null. - Return a clean city name only (e.g. "San Francisco"), no state, no comma. ## 2. state_apply (string or null) The full US state name (e.g. "California", "New York", "Texas") that corresponds to location_apply. - Use the US state abbreviation or full name found in the source. - Convert abbreviations to full state names (e.g. "CA" → "California"). - Must be null if location_apply is null. - Must be a valid US state name. Never return a non-US region. ## 3. contact (string or null) The single best contact method for a job application. Strict priority: Priority 1 — **Hiring/recruiting email**: careers@, hr@, jobs@, talent@, recruiting@, people@, hiring@, people-ops@ (highest priority, return immediately) Priority 2 — **Contact form URL**: a full URL like https://example.com/contact where an applicant can submit a message or inquiry. Priority 3 — **Careers page URL**: a full URL like https://example.com/careers where job listings or application instructions exist. Priority 4 — **General email**: info@, contact@, hello@, etc. Only if no URL is available. Priority 5 — null (nothing usable found). Additional contact rules: - NEVER fabricate or guess an email. Only use emails that appear in the text or emails list. - If an email is a noreply / notification / alert address, ignore it. - Return ONLY the single best item (one email OR one URL), not multiple. - Return the bare URL, never wrap it in brackets or labels. ## CRITICAL RULES - Read the FULL website text carefully. Context matters — an email mentioned near career/job/apply/hiring text is more relevant than one near a newsletter signup. - For location, the "hiring_focus" (where open positions are) is often more useful than the legal headquarters, especially for large companies with multiple offices. - If the company is remote-first with no physical US office, return null for both location_apply and state_apply. - All three fields are your FINAL ANSWER. Do not return intermediate data like headquarters_city, us_office_city etc. Just the three final fields. Return ONLY valid JSON: { "location_apply": "city name or null", "state_apply": "full state name or null", "contact": "email or URL or null", "confidence": { "location": "high|medium|low", "state": "high|medium|low", "contact": "high|medium|low" } }""" user_prompt = f"""Company: {company_name} Wellfound location string: {wellfound_location or "N/A"} FULL WEBSITE CONTENT (aggregated from homepage and sub-pages): {text_sample} ALL EMAILS FOUND ACROSS ALL PAGES: {json.dumps(emails)} ALL LINKS FOUND: {json.dumps(links[:20])} Phone numbers: {json.dumps(phones[:5])} Physical addresses found: {json.dumps(addresses[:5])} TASKS: 1. Determine the best US city for a job applicant to target (location_apply). - Consider: HQ location → US office → hiring focus → wellfound location → website addresses. - If the company has no US presence, return null. 2. Determine the corresponding full US state name (state_apply). - Convert abbreviations to full names. 3. Determine the single best contact method (contact). - Priority: hiring email > contact form URL > careers page URL > general email. - Only use emails actually found on the site. Never fabricate. Return ONLY the JSON object. No explanation, no markdown.""" response = await self._call_api(system_prompt, user_prompt) result = self._parse_json_response(response, "deep_extract") # Post-processing: validate state name state = result.get("state_apply") if state: state = _normalize_state_name(state) result["state_apply"] = state # Post-processing: clean city name city = result.get("location_apply") if city: city = _normalize_city_name(city) result["location_apply"] = city # Post-processing: clean contact contact = result.get("contact") if contact: contact = _clean_contact_output(contact) result["contact"] = contact return result async def verify_and_enhance( self, company_name: str, scraped_data: Dict, page_text: str ) -> Dict[str, Any]: """Verify and enhance extracted data with AI reasoning.""" text_sample = page_text[:6000] if page_text else "" system_prompt = """You are a data verification AI. Review scraped funding data against page content. Verify accuracy and fill in missing fields when possible. Return ONLY valid JSON. Output format: { "verified_data": { "valuation": "confirmed value or null", "total_raised": "confirmed value or null", "rounds": "confirmed integer or null", "series": "confirmed value or null" }, "corrections": ["list of corrections made"], "verification_notes": "brief note about data confidence" }""" user_prompt = f"""Company: {company_name} Scraped data: {json.dumps(scraped_data)} Page content snippet: {text_sample} Verify the scraped data against the page content. Correct any errors and fill in missing values where possible.""" response = await self._call_api(system_prompt, user_prompt) return self._parse_json_response(response, "verify") async def _call_api(self, system_prompt: str, user_prompt: str) -> str: """Make API call to the LLM provider.""" headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json", } payload = { "model": self.model, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], "temperature": 0.1, "max_tokens": 1500, "response_format": {"type": "json_object"} if self.provider == "openai" else None, } async with httpx.AsyncClient(timeout=60.0) as client: response = await client.post(self.base_url, json=payload, headers=headers) if response.status_code != 200: raise Exception(f"API error {response.status_code}: {response.text[:500]}") data = response.json() return data["choices"][0]["message"]["content"] def _parse_json_response(self, response: str, extract_type: str) -> Dict[str, Any]: """Parse JSON from API response with fallback extraction.""" # Try direct JSON parse try: return json.loads(response) except json.JSONDecodeError: pass # Try extracting JSON block json_match = re.search(r'\{[\s\S]*\}', response) if json_match: try: return json.loads(json_match.group(0)) except json.JSONDecodeError: pass # Return empty structure on failure if extract_type == "funding": return { "valuation": None, "total_raised": None, "rounds": None, "series": None, "investors": [], "confidence": {"valuation": "low", "total_raised": "low", "rounds": "low", "series": "low"} } elif extract_type == "contact" or extract_type == "deep_extract": return { "location_apply": None, "state_apply": None, "contact": None, "confidence": { "location": "low", "state": "low", "contact": "low", }, } else: return {"verified_data": {}, "corrections": [], "verification_notes": ""}