"""ObsiGate AI — Multi-provider AI service for editor enhancement. Supports: DeepSeek, OpenRouter, Google Gemini. Configured via environment variables. """ import os import json import logging import httpx from typing import Optional, Literal logger = logging.getLogger("obsigate.ai") ProviderName = Literal["deepseek", "openrouter", "gemini"] # Provider configurations PROVIDERS = { "deepseek": { "api_key": os.getenv("DEEPSEEK_API_KEY", "").strip(), "base_url": "https://api.deepseek.com/v1", "model": os.getenv("DEEPSEEK_MODEL", "deepseek-chat"), "auth_header": "Bearer {api_key}", }, "openrouter": { "api_key": os.getenv("OPENROUTER_API_KEY", "").strip(), "base_url": "https://openrouter.ai/api/v1", "model": os.getenv("OPENROUTER_MODEL", "openai/gpt-4o-mini"), "auth_header": "Bearer {api_key}", }, "gemini": { "api_key": os.getenv("GEMINI_API_KEY", "").strip(), "base_url": "https://generativelanguage.googleapis.com/v1beta", "model": os.getenv("GEMINI_MODEL", "gemini-2.0-flash"), "auth_header": None, # Uses query param ?key= }, } DEFAULT_PROVIDER: ProviderName = os.getenv("AI_DEFAULT_PROVIDER", "deepseek") # type: ignore def _get_provider_config(provider: Optional[ProviderName] = None) -> dict: """Get provider config, falling back to default if requested provider unavailable.""" p = provider or DEFAULT_PROVIDER if p not in PROVIDERS: p = DEFAULT_PROVIDER cfg = PROVIDERS[p] if not cfg["api_key"]: # Try next available provider for alt in PROVIDERS: if PROVIDERS[alt]["api_key"]: p = alt cfg = PROVIDERS[alt] break return {"name": p, **cfg} async def _call_deepseek_openrouter(prompt: str, system: str, provider: Optional[ProviderName] = None, temperature: float = 0.7, max_tokens: int = 2048) -> str: """Call OpenAI-compatible API (DeepSeek, OpenRouter).""" cfg = _get_provider_config(provider) # Debug: log masked key to diagnose 401 key_preview = cfg["api_key"][:8] + "..." + cfg["api_key"][-4:] if len(cfg["api_key"]) > 12 else "***" logger.info(f"AI call: provider={cfg['name']} model={cfg['model']} key={key_preview}") headers = { "Authorization": cfg["auth_header"].format(api_key=cfg["api_key"]), "Content-Type": "application/json", } payload = { "model": cfg["model"], "messages": [ {"role": "system", "content": system}, {"role": "user", "content": prompt}, ], "temperature": temperature, "max_tokens": max_tokens, } async with httpx.AsyncClient(timeout=60.0) as client: resp = await client.post( f"{cfg['base_url']}/chat/completions", headers=headers, json=payload, ) resp.raise_for_status() data = resp.json() return data["choices"][0]["message"]["content"].strip() async def _call_gemini(prompt: str, system: str, temperature: float = 0.7, max_tokens: int = 2048) -> str: """Call Google Gemini API.""" cfg = PROVIDERS["gemini"] url = f"{cfg['base_url']}/models/{cfg['model']}:generateContent?key={cfg['api_key']}" payload = { "system_instruction": {"parts": [{"text": system}]}, "contents": [{"parts": [{"text": prompt}]}], "generationConfig": { "temperature": temperature, "maxOutputTokens": max_tokens, }, } async with httpx.AsyncClient(timeout=60.0) as client: resp = await client.post(url, json=payload) resp.raise_for_status() data = resp.json() return data["candidates"][0]["content"]["parts"][0]["text"].strip() async def ai_complete(prompt: str, provider: Optional[ProviderName] = None) -> str: """Generic AI completion. Routes to appropriate provider.""" cfg = _get_provider_config(provider) if cfg["name"] == "gemini": return await _call_gemini(prompt, "You are a helpful assistant.") return await _call_deepseek_openrouter(prompt, "You are a helpful assistant.", provider) # ── Specialized AI actions ── SYSTEM_PROMPT = """You are an AI assistant integrated into ObsiGate, a knowledge management tool. Your responses should be direct and concise. When editing text, return ONLY the modified text, no explanations or markdown fences.""" async def ai_improve_writing(text: str, provider: Optional[ProviderName] = None) -> str: """Improve writing quality while preserving meaning.""" return await _call_deepseek_openrouter( f"Improve the following text. Fix grammar, clarity, and flow. Preserve the original language and meaning.\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.3, ) async def ai_fix_spelling(text: str, provider: Optional[ProviderName] = None) -> str: """Fix spelling and grammar errors.""" return await _call_deepseek_openrouter( f"Fix all spelling and grammar errors in this text. Return only the corrected text.\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.1, ) async def ai_make_shorter(text: str, provider: Optional[ProviderName] = None) -> str: """Make text more concise.""" return await _call_deepseek_openrouter( f"Make this text shorter and more concise while preserving the key information.\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.3, ) async def ai_make_longer(text: str, provider: Optional[ProviderName] = None) -> str: """Expand text with more detail.""" return await _call_deepseek_openrouter( f"Expand this text with more detail, examples, or explanation while keeping the same tone.\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.7, max_tokens=4096, ) async def ai_simplify(text: str, provider: Optional[ProviderName] = None) -> str: """Simplify language.""" return await _call_deepseek_openrouter( f"Simplify this text. Use clearer, more straightforward language. Avoid jargon.\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.3, ) async def ai_change_tone(text: str, tone: str, provider: Optional[ProviderName] = None) -> str: """Change the tone of the text.""" return await _call_deepseek_openrouter( f"Rewrite this text in a {tone} tone. Preserve the original meaning.\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.5, ) async def ai_translate(text: str, target_lang: str, provider: Optional[ProviderName] = None) -> str: """Translate text to target language.""" # Gemini is better at translation if provider is None and PROVIDERS["gemini"]["api_key"]: return await _call_gemini( f"Translate the following text to {target_lang}. Return only the translation.\n\n{text}", "You are a professional translator. Translate accurately and naturally.", temperature=0.1, ) return await _call_deepseek_openrouter( f"Translate the following text to {target_lang}. Return only the translation.\n\n{text}", "You are a professional translator. Translate accurately and naturally.", provider, temperature=0.1, ) async def ai_explain(text: str, provider: Optional[ProviderName] = None) -> str: """Explain the selected text.""" return await _call_deepseek_openrouter( f"Explain the following text clearly and concisely:\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.3, ) async def ai_summarize(text: str, provider: Optional[ProviderName] = None) -> str: """Summarize the selected text.""" return await _call_deepseek_openrouter( f"Summarize the following text concisely:\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.3, ) async def ai_continue_writing(text: str, provider: Optional[ProviderName] = None) -> str: """Continue writing from the selected text.""" return await _call_deepseek_openrouter( f"Continue writing from where this text leaves off. Match the style and tone:\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.7, max_tokens=4096, ) async def ai_custom_rewrite(text: str, instruction: str, provider: Optional[ProviderName] = None) -> str: """Rewrite text based on a custom instruction.""" return await _call_deepseek_openrouter( f"Rewrite the following text according to this instruction: {instruction}\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.5, ) async def ai_convert_to_list(text: str, provider: Optional[ProviderName] = None) -> str: """Convert paragraph text to a markdown list.""" return await _call_deepseek_openrouter( f"Convert this text into a well-organized markdown bullet list. Extract key points.\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.2, ) async def ai_convert_to_table(text: str, provider: Optional[ProviderName] = None) -> str: """Convert text to a markdown table.""" return await _call_deepseek_openrouter( f"Convert this information into a markdown table. Choose appropriate columns.\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.2, ) async def ai_generate_frontmatter(text: str, provider: Optional[ProviderName] = None) -> str: """Generate YAML frontmatter for a markdown document.""" return await _call_deepseek_openrouter( f"Generate YAML frontmatter for this markdown document. Include: titre, tags (as list), catégorie, statut, date. Return ONLY the YAML between --- markers.\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.3, ) async def ai_inline_complete(text: str, provider: Optional[ProviderName] = None) -> str: """Inline completion — suggest continuation.""" return await _call_deepseek_openrouter( f"Complete this text naturally. Return only the completion (just the new text, no repetition):\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.3, max_tokens=512, ) async def ai_convert_to_canvas(text: str, provider: Optional[ProviderName] = None) -> str: """Convert text to a Mermaid diagram or canvas representation.""" return await _call_deepseek_openrouter( f"Convert this content into a Mermaid.js diagram if applicable, or a structured outline. Choose the best format.\n\n{text}", SYSTEM_PROMPT, provider, temperature=0.3, max_tokens=4096, )