""" Degradation Manager - Fallback orchestration when AI is unavailable. Handles: - Failure detection (AI disabled, timeout, API error, invalid response) - Graceful degradation to regex-based extraction - Non-AI source tagging ("备注" column injection) - Structured error logging with timestamps, failure reasons, degradation paths - Full chain traceability """ import logging import traceback from datetime import datetime from enum import Enum from typing import Optional, Dict, Any, List, Tuple from .address_processor import AddressProcessor from .contact_finder import ContactFinder # ─── Failure Classification ────────────────────────────────── class FailureType(Enum): """Classification of AI failure scenarios.""" AI_DISABLED = "ai_disabled" # AI service not enabled by user AI_TIMEOUT = "ai_timeout" # API call exceeded timeout AI_API_ERROR = "ai_api_error" # API returned error status code AI_PARSE_ERROR = "ai_parse_error" # AI response could not be parsed as JSON AI_INVALID_RESPONSE = "ai_invalid_response" # AI returned null/empty for all fields AI_AUTH_ERROR = "ai_auth_error" # API key invalid or quota exceeded AI_NETWORK_ERROR = "ai_network_error" # Network-level failure (DNS, connection) AI_UNKNOWN_ERROR = "ai_unknown_error" # Unclassifiable error AI_NOT_NEEDED = "ai_not_needed" # Scraping itself failed, AI can't help class DegradationLogger: """Structured error logger with timestamped entries for traceability.""" def __init__(self, log_dir: str = "data/logs"): from pathlib import Path self.log_dir = Path(log_dir) self.log_dir.mkdir(parents=True, exist_ok=True) # File logger for persistent degradation records self._file_logger = logging.getLogger("degradation") self._file_logger.setLevel(logging.INFO) self._file_logger.propagate = False # Don't double-log to console # Create rotating file handler log_file = self.log_dir / f"degradation_{datetime.now().strftime('%Y%m%d')}.log" handler = logging.FileHandler(str(log_file), encoding="utf-8") handler.setFormatter(logging.Formatter( "%(asctime)s | %(levelname)-8s | %(message)s", datefmt="%Y-%m-%d %H:%M:%S" )) self._file_logger.addHandler(handler) # In-memory summary for the current processing session self._session_summary: List[Dict[str, Any]] = [] def log_degradation( self, row_idx: int, company: str, failure_type: FailureType, failure_reason: str, degradation_path: str, fields_affected: List[str], fallback_used: List[str], ): """Record a single degradation event. Args: row_idx: Excel row index where degradation occurred. company: Company name for context. failure_type: Enum classifying the failure. failure_reason: Human-readable description of what went wrong. degradation_path: e.g. "ai_extractor -> address_processor + contact_finder" fields_affected: Which target fields were filled by fallback. fallback_used: Which fallback modules were activated. """ event = { "timestamp": datetime.now().isoformat(), "row_index": row_idx, "company": company, "failure_type": failure_type.value, "failure_reason": failure_reason[:300], "degradation_path": degradation_path, "fields_affected": fields_affected, "fallback_used": fallback_used, } # Log to file self._file_logger.info( "row=%d | company=%s | type=%s | reason=%s | path=%s | fields=%s | fallback=%s", row_idx, company, failure_type.value, failure_reason[:200], degradation_path, fields_affected, fallback_used, ) # Keep in memory for session summary self._session_summary.append(event) def get_session_summary(self) -> Dict[str, Any]: """Get aggregated degradation stats for the current session.""" if not self._session_summary: return { "total_degradations": 0, "by_type": {}, "by_field": {}, "affected_rows": [], } by_type: Dict[str, int] = {} by_field: Dict[str, int] = {} affected_rows = [] for event in self._session_summary: ft = event["failure_type"] by_type[ft] = by_type.get(ft, 0) + 1 for field in event["fields_affected"]: by_field[field] = by_field.get(field, 0) + 1 affected_rows.append(event["row_index"]) return { "total_degradations": len(self._session_summary), "by_type": by_type, "by_field": by_field, "affected_rows": sorted(set(affected_rows)), } def get_recent_entries(self, limit: int = 20) -> List[Dict[str, Any]]: """Get the most recent degradation entries.""" return self._session_summary[-limit:] # ─── Failure Classifier ───────────────────────────────────── class FailureClassifier: """Classify exceptions into typed FailureType values.""" # Patterns that indicate specific failure types _TIMEOUT_PATTERNS = ["timeout", "timed out", "deadline"] _AUTH_PATTERNS = [ "401", "403", "unauthorized", "forbidden", "invalid api key", "authentication", "quota", "rate limit", "429", ] _NETWORK_PATTERNS = [ "connection", "dns", "resolve", "network", "socket", "ssl", "certificate", "connecterror", ] _PARSE_PATTERNS = ["json", "decode", "parse", "expecting", "malformed"] @classmethod def classify(cls, exc: Exception) -> FailureType: """Classify an exception into a FailureType. Args: exc: The caught exception instance. Returns: A FailureType enum value. """ msg = str(exc).lower() exc_type = type(exc).__name__.lower() # Timeout if any(p in msg for p in cls._TIMEOUT_PATTERNS): return FailureType.AI_TIMEOUT # Auth / quota if any(p in msg for p in cls._AUTH_PATTERNS) or any(p in msg for p in cls._AUTH_PATTERNS): return FailureType.AI_AUTH_ERROR # Network if any(p in msg for p in cls._NETWORK_PATTERNS) or "connecterror" in exc_type: return FailureType.AI_NETWORK_ERROR # Parse error if any(p in msg for p in cls._PARSE_PATTERNS) or "jsondecode" in exc_type: return FailureType.AI_PARSE_ERROR # API error (HTTP status codes 4xx/5xx that aren't auth) if any(c in msg for c in ["400", "404", "405", "408", "500", "502", "503"]): return FailureType.AI_API_ERROR return FailureType.AI_UNKNOWN_ERROR @classmethod def classify_api_status(cls, status_code: int) -> FailureType: """Classify an HTTP status code into a FailureType.""" if status_code == 401 or status_code == 403: return FailureType.AI_AUTH_ERROR elif status_code == 429: return FailureType.AI_AUTH_ERROR # Rate limit → auth/quota bucket elif status_code == 408 or status_code == 504: return FailureType.AI_TIMEOUT elif 500 <= status_code <= 599: return FailureType.AI_API_ERROR else: return FailureType.AI_API_ERROR # ─── Degradation Manager ───────────────────────────────────── class DegradationManager: """Orchestrates fallback when AI extraction fails. Provides: - detect_failure(): Classify what went wrong - run_fallback(): Execute contact_finder + address_processor as fallback - build_remark(): Generate the "备注" column value for traceability """ def __init__(self, logger: Optional[DegradationLogger] = None): self.address_processor = AddressProcessor() self.contact_finder = ContactFinder() self.logger = logger or DegradationLogger() # ─── Failure Detection ────────────────────────────────── @staticmethod def detect_failure( ai_extractor=None, exc: Optional[Exception] = None, ai_result: Optional[Dict[str, Any]] = None, ) -> Optional[FailureType]: """Detect and classify AI failure. Checks (in priority order): 1. AI extractor is None → AI_DISABLED 2. An exception was raised → classify by exception type 3. AI result exists but all target fields are null → AI_INVALID_RESPONSE 4. No failure detected → None Returns: FailureType if degradation is needed, None if AI succeeded. """ # Scenario 1: AI not enabled if ai_extractor is None: return FailureType.AI_DISABLED # Scenario 2: Exception during AI call if exc is not None: return FailureClassifier.classify(exc) # Scenario 3: AI returned result but all target fields are null/empty if ai_result is not None: loc = ai_result.get("location_apply") state = ai_result.get("state_apply") contact = ai_result.get("contact") if not any([loc, state, contact]): return FailureType.AI_INVALID_RESPONSE return None # ─── Fallback Execution ────────────────────────────────── def run_fallback( self, emails: List[str], links: List[Dict], addresses: List[str], homepage_html: str = "", base_url: str = "", wellfound_location: str = "", combined_text: str = "", ) -> Tuple[Dict[str, Any], List[str]]: """Execute fallback extraction using regex-based modules. Returns: Tuple of: - result dict with location_apply, state_apply, contact - list of fallback module names that were activated """ result = { "location_apply": None, "state_apply": None, "contact": None, } fallbacks_used = [] # ── Contact fallback ── if not result["contact"]: # 1. Score emails by priority best_email = self.contact_finder.find_best_contact_email( emails, combined_text ) if best_email: result["contact"] = best_email fallbacks_used.append("contact_finder.email_scoring") # 2. Try contact form URL if not result["contact"] and links and base_url: form_url = self.contact_finder.find_contact_form_url( base_url, links, homepage_html ) if form_url: result["contact"] = ContactFinder._clean_url(form_url) fallbacks_used.append("contact_finder.form_url") # 3. Try careers page URL if not result["contact"] and links and base_url: careers_url = self.contact_finder.find_careers_page_url( base_url, links, homepage_html ) if careers_url: result["contact"] = ContactFinder._clean_url(careers_url) fallbacks_used.append("contact_finder.careers_url") # ── Location fallback ── if not result["location_apply"] or not result["state_apply"]: loc, state = self.address_processor.determine_location_apply( wellfound_location=wellfound_location, ai_contact_data={}, # No AI data available scraped_addresses=addresses, ) if loc: result["location_apply"] = loc fallbacks_used.append("address_processor.location") if state: result["state_apply"] = state fallbacks_used.append("address_processor.state") return result, fallbacks_used # ─── Remark Builder ──────────────────────────────────── @staticmethod def build_remark( failure_type: FailureType, fallbacks_used: List[str], fields_filled: List[str], ) -> str: """Build the "备注" column value. Format: source: "non-ai" | reason: | fallback: Examples: source: "non-ai" | reason: ai_timeout | fallback: contact_finder.email_scoring, address_processor.location source: "non-ai" | reason: ai_disabled | fallback: address_processor.location, address_processor.state """ parts = [f'source: "non-ai"'] parts.append(f"reason: {failure_type.value}") if fallbacks_used: parts.append(f"fallback: {', '.join(fallbacks_used)}") if fields_filled: parts.append(f"fields: {', '.join(fields_filled)}") return " | ".join(parts) # ─── Full Degradation Pipeline ────────────────────────── async def handle_degradation( self, row_idx: int, company: str, ai_extractor=None, exc: Optional[Exception] = None, ai_result: Optional[Dict[str, Any]] = None, scraped_data: Optional[Dict[str, Any]] = None, wellfound_location: str = "", ) -> Dict[str, Any]: """Full degradation pipeline: detect → fallback → tag → log. Args: row_idx: Excel row index. company: Company name. ai_extractor: The AIExtractor instance (None if AI disabled). exc: Exception caught during AI call (if any). ai_result: The AI result dict (if AI ran but returned empty). scraped_data: Raw scraped data from website (emails, links, etc.). wellfound_location: Wellfound location string. Returns: Dict with keys: - location_apply, state_apply, contact (fallback values) - remark: the "备注" column value - degraded: bool, whether degradation was applied - failure_type: FailureType enum """ # Default return output = { "location_apply": None, "state_apply": None, "contact": None, "remark": None, "degraded": False, "failure_type": None, } scraped_data = scraped_data or {} # Step 1: Detect failure failure_type = self.detect_failure(ai_extractor, exc, ai_result) if failure_type is None: # No failure — use AI result directly if ai_result: output["location_apply"] = ai_result.get("location_apply") output["state_apply"] = ai_result.get("state_apply") output["contact"] = ai_result.get("contact") return output # Degradation needed output["degraded"] = True output["failure_type"] = failure_type # Step 2: Build failure reason string if exc: failure_reason = f"{type(exc).__name__}: {str(exc)[:200]}" elif failure_type == FailureType.AI_DISABLED: failure_reason = "AI service not enabled by user configuration" elif failure_type == FailureType.AI_INVALID_RESPONSE: failure_reason = "AI returned null/empty for all target fields" else: failure_reason = f"Unclassified failure: {failure_type.value}" # Step 3: Execute fallback fallback_result, fallbacks_used = self.run_fallback( emails=scraped_data.get("emails", []), links=scraped_data.get("links", []), addresses=scraped_data.get("addresses", []), homepage_html=scraped_data.get("html", ""), base_url=scraped_data.get("base_url", ""), wellfound_location=wellfound_location, combined_text=scraped_data.get("combined_text", ""), ) # Step 4: Determine which fields were filled by fallback fields_filled = [] for field in ["location_apply", "state_apply", "contact"]: if fallback_result.get(field): output[field] = fallback_result[field] fields_filled.append(field) # Step 5: Build remark if fields_filled or fallbacks_used: output["remark"] = self.build_remark( failure_type, fallbacks_used, fields_filled ) # Step 6: Build degradation path description degradation_path_parts = [] if failure_type == FailureType.AI_DISABLED: degradation_path_parts.append("ai_skipped") else: degradation_path_parts.append("ai_extractor") degradation_path_parts.append(f"({failure_type.value})") if fallbacks_used: degradation_path_parts.append("->") degradation_path_parts.append(" + ".join(fallbacks_used)) else: degradation_path_parts.append("-> no_fallback_available") degradation_path = " ".join(degradation_path_parts) # Step 7: Log degradation event self.logger.log_degradation( row_idx=row_idx, company=company, failure_type=failure_type, failure_reason=failure_reason, degradation_path=degradation_path, fields_affected=fields_filled, fallback_used=fallbacks_used, ) return output