"""Translation endpoint for converting grant jargon into simple language.""" import logging from typing import List, Optional, Dict, Any from pydantic import BaseModel from fastapi import APIRouter, HTTPException from src.analyzer.llm_client import LLMClient from src.analyzer.config import load_config from src.database import SummaryStore logger = logging.getLogger(__name__) router = APIRouter(prefix="/translate", tags=["translation"]) # Global cache for LLM client _llm_client: Optional[LLMClient] = None _summary_store: Optional[SummaryStore] = None def get_llm_client() -> LLMClient: """Get or create LLM client for translations.""" global _llm_client if _llm_client is None: config = load_config() _llm_client = LLMClient(config) return _llm_client def get_summary_store() -> SummaryStore: """Get or create summary store.""" global _summary_store if _summary_store is None: _summary_store = SummaryStore() return _summary_store class TranslateRequest(BaseModel): """Request model for translation endpoint.""" grant_id: str force_refresh: bool = False # Force regeneration even if cached class BatchTranslateRequest(BaseModel): """Request model for batch translation endpoint.""" grant_ids: List[str] force_refresh: bool = False class TranslationResponse(BaseModel): """Response model for translation endpoint.""" grant_id: str translation: Dict[str, str] cached: bool class BatchTranslationResponse(BaseModel): """Response model for batch translation endpoint.""" translations: List[TranslationResponse] total: int cached_count: int # Translation prompt template TRANSLATION_PROMPT_TEMPLATE = """You are translating a UK grant into simple, everyday language. Grant Information: {grant_context} Please provide a clear, structured explanation with these EXACT sections (use these headings): ## WHO CAN APPLY (Eligibility) Explain who is eligible in simple terms. Use bullet points. ## WHAT YOU'LL DO (Process) Describe the application process step-by-step in simple language. ## WHEN IT HAPPENS (Timeline) Explain key dates and how long things take. ## WHAT YOU NEED (Requirements) List what applicants need to provide or have ready. Use simple, conversational language. Avoid jargon. Write as if explaining to a friend. """ def load_grant_data(grant_id: str) -> Optional[Dict[str, Any]]: """ Load grant data from snapshots. Args: grant_id: Grant ID (e.g., "competition-2058") Returns: Grant data dict or None if not found """ from pathlib import Path import json # Normalize grant ID if not grant_id.startswith("competition-"): grant_id = f"competition-{grant_id}" snapshots_dir = Path("data/snapshots") grant_file = snapshots_dir / f"{grant_id}.json" if not grant_file.exists(): logger.warning(f"Grant file not found: {grant_file}") return None try: with open(grant_file, "r", encoding="utf-8") as f: return json.load(f) except Exception as e: logger.error(f"Failed to load grant {grant_id}: {e}") return None def build_grant_context(grant: Dict[str, Any]) -> str: """ Build context for translation from grant data. Args: grant: Grant data dict Returns: Formatted context string """ from src.analyzer.summarizer_optimized import extract_minimal_context # Use the optimized context extractor context = extract_minimal_context(grant) # Add any additional fields specific to translation sections = grant.get("sections", {}) additional_info = [] # Add project details if available if "project_scope" in sections: additional_info.append(f"PROJECT SCOPE: {sections['project_scope'][:300]}") # Add funding details if available if "funding_details" in sections: additional_info.append(f"FUNDING DETAILS: {sections['funding_details'][:300]}") if additional_info: context += "\n\n" + "\n".join(additional_info) return context def translate_grant(grant_id: str, force_refresh: bool = False) -> Dict[str, Any]: """ Translate a grant into simple language with structured sections. Args: grant_id: Grant ID to translate force_refresh: Force regeneration even if cached Returns: Dict with translation sections and metadata Raises: HTTPException: If grant not found or translation fails """ store = get_summary_store() # Normalize grant ID if not grant_id.startswith("competition-"): grant_id = f"competition-{grant_id}" # Check cache first (unless force_refresh) if not force_refresh: cached_translation = store.get_summary(grant_id, summary_type="translation") if cached_translation: logger.info(f"Using cached translation for {grant_id}") return { "grant_id": grant_id, "translation": parse_translation_sections(cached_translation), "cached": True } # Load grant data grant = load_grant_data(grant_id) if not grant: raise HTTPException(status_code=404, detail=f"Grant {grant_id} not found") # Build context context = build_grant_context(grant) # Generate translation using GPT-5-mini try: llm_client = get_llm_client() prompt = TRANSLATION_PROMPT_TEMPLATE.format(grant_context=context) # Use translator model with appropriate parameters translation_text = llm_client.summarize( prompt, model_type="translator", # Use gpt-5-mini verbosity="medium", reasoning_effort="minimal", max_tokens=800, temperature=0.3 ) # Save to cache (permanent storage) store.save_summary( grant_id=grant_id, summary_type="translation", summary_text=translation_text, metadata={"model": "gpt-5-mini", "context_length": len(context)} ) logger.info(f"Generated translation for {grant_id}") return { "grant_id": grant_id, "translation": parse_translation_sections(translation_text), "cached": False } except Exception as e: logger.error(f"Translation failed for {grant_id}: {e}") raise HTTPException(status_code=500, detail=f"Translation failed: {str(e)}") def parse_translation_sections(translation_text: str) -> Dict[str, str]: """ Parse translation text into structured sections. Args: translation_text: Raw translation text with markdown headers Returns: Dict with section names as keys and content as values """ import re sections = { "eligibility": "", "process": "", "timeline": "", "requirements": "" } # Define section patterns patterns = { "eligibility": r"## WHO CAN APPLY.*?\n(.*?)(?=\n## |\Z)", "process": r"## WHAT YOU'LL DO.*?\n(.*?)(?=\n## |\Z)", "timeline": r"## WHEN IT HAPPENS.*?\n(.*?)(?=\n## |\Z)", "requirements": r"## WHAT YOU NEED.*?\n(.*?)(?=\n## |\Z)" } for key, pattern in patterns.items(): match = re.search(pattern, translation_text, re.DOTALL | re.IGNORECASE) if match: sections[key] = match.group(1).strip() # Fallback: if no sections found, return full text as eligibility if not any(sections.values()): sections["eligibility"] = translation_text return sections @router.post("/", response_model=TranslationResponse) async def translate_grant_endpoint(request: TranslateRequest): """ Translate a grant into simple, structured language. Returns a layman explanation with sections: - Eligibility: Who can apply - Process: How to apply - Timeline: Key dates and deadlines - Requirements: What you need Translations are permanently cached since grants don't change. """ result = translate_grant(request.grant_id, request.force_refresh) return TranslationResponse(**result) @router.post("/batch", response_model=BatchTranslationResponse) async def translate_grants_batch(request: BatchTranslateRequest): """ Translate multiple grants in batch. Efficiently processes multiple grant translations with caching. Returns all translations, using cached versions where available. """ translations = [] cached_count = 0 for grant_id in request.grant_ids: try: result = translate_grant(grant_id, request.force_refresh) translations.append(TranslationResponse(**result)) if result["cached"]: cached_count += 1 except HTTPException as e: logger.warning(f"Failed to translate {grant_id}: {e.detail}") # Continue with other grants continue except Exception as e: logger.error(f"Unexpected error translating {grant_id}: {e}") continue return BatchTranslationResponse( translations=translations, total=len(translations), cached_count=cached_count ) @router.get("/{grant_id}", response_model=TranslationResponse) async def get_translation(grant_id: str, force_refresh: bool = False): """ Get translation for a specific grant (GET method for convenience). Args: grant_id: Grant ID (with or without 'competition-' prefix) force_refresh: Force regeneration even if cached Returns: Structured translation with eligibility, process, timeline, requirements """ result = translate_grant(grant_id, force_refresh) return TranslationResponse(**result)