"""AI Summary Service using Anthropic/Z.AI API (GLM).""" import logging import os from typing import Optional import requests logger = logging.getLogger(__name__) class AISummaryService: """Service for AI-powered text summarization using Anthropic/Z.AI API.""" def __init__( self, auth_token: Optional[str] = None, base_url: Optional[str] = None, model: Optional[str] = None, timeout: int = 120, ) -> None: """Initialize the AI Summary Service. Args: auth_token: API authentication token. Defaults to ANTHROPIC_AUTH_TOKEN env var. base_url: API base URL. Defaults to ANTHROPIC_BASE_URL env var. model: Model identifier. Defaults to ANTHROPIC_MODEL env var. timeout: Request timeout in seconds. Defaults to 120. """ self.auth_token = auth_token or os.getenv("ANTHROPIC_AUTH_TOKEN") # Normalize base_url: remove /anthropic suffix if present raw_base_url = base_url or os.getenv("ANTHROPIC_BASE_URL") if raw_base_url and raw_base_url.endswith("/anthropic"): raw_base_url = raw_base_url[:-len("/anthropic")] self.base_url = raw_base_url self.model = model or os.getenv("ANTHROPIC_MODEL", "glm-4") self.timeout = timeout self._available = bool(self.auth_token and self.base_url) if self._available: logger.info( "AISummaryService initialized with model=%s, base_url=%s", self.model, self.base_url, ) else: logger.debug("AISummaryService: no configuration found, running in silent mode") @property def is_available(self) -> bool: """Check if the service is properly configured.""" return self._available def summarize(self, text: str, prompt_template: Optional[str] = None) -> str: """Summarize the given text using the AI API. Args: text: The text to summarize. prompt_template: Optional custom prompt template. If None, uses default. Returns: The summarized text. Raises: RuntimeError: If the service is not configured. requests.RequestException: If the API call fails. """ if not self._available: logger.debug("AISummaryService not configured, returning original text") return text default_prompt = "Resume el siguiente texto de manera clara y concisa:" prompt = prompt_template.format(text=text) if prompt_template else f"{default_prompt}\n\n{text}" payload = { "model": self.model, "messages": [{"role": "user", "content": prompt}], "max_tokens": 2048, "temperature": 0.7, } headers = { "Authorization": f"Bearer {self.auth_token}", "Content-Type": "application/json", } try: logger.debug("Calling AI API for summarization (text length: %d)", len(text)) response = requests.post( f"{self.base_url}/v1/chat/completions", json=payload, headers=headers, timeout=self.timeout, ) response.raise_for_status() result = response.json() summary = result.get("choices", [{}])[0].get("message", {}).get("content", "") logger.info("Summarization completed successfully (output length: %d)", len(summary)) return summary except requests.Timeout: logger.error("AI API request timed out after %d seconds", self.timeout) raise requests.RequestException(f"Request timed out after {self.timeout}s") from None except requests.RequestException as e: logger.error("AI API request failed: %s", str(e)) raise def fix_latex(self, text: str) -> str: """Fix LaTeX formatting issues in the given text. Args: text: The text containing LaTeX to fix. Returns: The text with corrected LaTeX formatting. """ if not self._available: logger.debug("AISummaryService not configured, returning original text") return text prompt = ( "Corrige los errores de formato LaTeX en el siguiente texto. " "Mantén el contenido pero corrige la sintaxis de LaTeX:\n\n" f"{text}" ) payload = { "model": self.model, "messages": [{"role": "user", "content": prompt}], "max_tokens": 4096, "temperature": 0.3, } headers = { "Authorization": f"Bearer {self.auth_token}", "Content-Type": "application/json", } try: logger.debug("Calling AI API for LaTeX fixing (text length: %d)", len(text)) response = requests.post( f"{self.base_url}/v1/chat/completions", json=payload, headers=headers, timeout=self.timeout, ) response.raise_for_status() result = response.json() fixed = result.get("choices", [{}])[0].get("message", {}).get("content", "") logger.info("LaTeX fixing completed successfully") return fixed except requests.RequestException as e: logger.error("LaTeX fixing failed: %s", str(e)) return text