""" Basic rule-based formatting of API validation results without AI. """ import logging from typing import Dict, List, Any, Optional logger = logging.getLogger(__name__) class BasicFormatter: """ Formats API validation results using simple rules without AI. Provides clean, display-friendly output based on API scores and matches. """ def __init__(self): """Initialize the basic formatter.""" pass def format_results(self, api_results: List[Dict[str, Any]]) -> Dict[str, Any]: """ Formats API results using rule-based logic without AI. Args: api_results: The results from the API validators. Returns: A dictionary containing formatted results. """ formatted_results = [] for result in api_results: formatted = self._format_single_result(result) formatted_results.append(formatted) logger.info(f"Formatted {len(formatted_results)} results using basic rules") return {"results": formatted_results, "candidates_grounding_metadata": []} def _format_single_result(self, api_result: Dict[str, Any]) -> Dict[str, Any]: """ Formats a single API result. Args: api_result: Single API validation result. Returns: Formatted result dictionary. """ query = api_result.get("query", {}) exists = api_result.get("exists", False) best_match = api_result.get("best_match", {}) # Score is stored in best_match as overall_score score = best_match.get("overall_score", 0) if best_match else 0 source = api_result.get("source", "unknown") # Check if there was an API error (critical difference from legitimate "not found") error_type = api_result.get("error") # e.g., "rate_limit", "max_retries_exceeded", etc. error_message = api_result.get("message", "") # Extract data from query or best match title = best_match.get("title") or query.get("title", "Unknown") # Handle authors list - get first author authors = best_match.get("authors", []) first_author = authors[0] if authors else query.get("first_author", "Unknown") year = best_match.get("year") or query.get("year", "Unknown") journal = best_match.get("journal") or query.get("journal", "Unknown") link = best_match.get("link") or best_match.get("url") or best_match.get("doi", "") # Build explanation and identify issues (now includes error detection) explanation, issues = self._build_explanation_and_issues( exists, score, source, query, best_match, error_type, error_message ) return { "title": title, "exists": exists, "link": link, "explanation": explanation, "first_author": first_author, "year": str(year), "journal": journal, "issues": issues if issues else None } def _build_explanation_and_issues( self, exists: bool, score: float, source: str, query: Dict[str, Any], best_match: Dict[str, Any], error_type: Optional[str] = None, error_message: str = "" ) -> tuple[str, List[str]]: """ Builds explanation and identifies issues based on API results. Args: exists: Whether reference exists in API score: Match score source: API source name query: Original query data best_match: Best match from API error_type: Type of error if API failed (e.g., "rate_limit", "max_retries_exceeded") error_message: Human-readable error message Returns: Tuple of (explanation string, list of issues) """ issues = [] # Handle API errors differently from legitimate "not found" if not exists: if error_type: # API error occurred - don't claim "not found", explain the error if error_type == "rate_limit": explanation = "API rate limit exceeded - validation failed" issues.append("API rate limit hit - try again later") elif error_type == "max_retries_exceeded": explanation = "API unavailable after retries - validation failed" issues.append("API temporarily unavailable") elif error_type == "API timeout": explanation = "API request timed out - validation failed" issues.append("Network timeout") else: # Generic error explanation = f"Validation error: {error_message}" issues.append(f"API error: {error_type}") return explanation, issues else: # No error - legitimate "not found" return "Not found in any API database", [] # Build explanation if score >= 90: explanation = f"Confirmed via {source.title()}" elif score >= 70: explanation = f"Likely match via {source.title()} (score: {score:.0f}%)" issues.append("Medium confidence match") else: explanation = f"Weak match via {source.title()} (score: {score:.0f}%)" issues.append("Low confidence match") # Check for discrepancies query_year = str(query.get("year", "")) match_year = str(best_match.get("year", "")) if query_year and match_year and query_year != match_year: issues.append(f"Year mismatch: {query_year} vs {match_year}") query_title = query.get("title", "").lower() match_title = best_match.get("title", "").lower() if query_title and match_title: # Simple title similarity check if query_title not in match_title and match_title not in query_title: # Check if at least some words match query_words = set(query_title.split()) match_words = set(match_title.split()) common_words = query_words & match_words if len(common_words) < min(3, len(query_words) // 2): issues.append("Title variation detected") query_author = query.get("first_author", "").lower() # Get first author from authors list match_authors = best_match.get("authors", []) match_author = match_authors[0].lower() if match_authors else "" if query_author and match_author and query_author != match_author: # Check if last names match (simple check) query_last = query_author.split()[-1] if query_author else "" match_last = match_author.split()[-1] if match_author else "" if query_last != match_last: issues.append(f"Author mismatch: {query_author} vs {match_author}") return explanation, issues