"""FlowDeck — AI Writing Assist (v5.9.0). Headless, tool-free writing helpers used by the editor (slash commands, inline autocomplete) and the database table (AI property suggestions). All actions share one entry point, :meth:`AIWritingService.run`, which builds a tight prompt, calls the configured LLM (or the deterministic offline mock) and returns plain Markdown. `properties` additionally returns a structured ``suggestions`` mapping so the caller can fill collection properties. The service never talks to the DB directly — the router resolves the caller's provider/key and the page context before delegating here. """ from __future__ import annotations import json import logging import re from app.services.llm_client import LLMClient logger = logging.getLogger(__name__) WRITING_ACTIONS = ("write", "summarize", "translate", "continue", "autocomplete", "properties") _MAX_CONTEXT = 20000 # Property name → (type, offline default) used by the deterministic fallback so # the feature stays useful without a connected provider. _OFFLINE_PROPERTY_DEFAULTS = ( (("status", "état", "etat", "stage"), "select", "To do"), (("priority", "priorité", "priorite"), "select", "Medium"), (("done", "terminé", "termine", "complété", "complete"), "checkbox", False), (("summary", "résumé", "resume", "description", "notes"), "text", ""), ) _SYSTEM_WRITING = ( "Tu es l'assistant d'écriture de FlowDeck. " "Réponds UNIQUEMENT avec le contenu demandé, en Markdown léger " "(paragraphes, listes à puces, titres si utile). " "N'ajoute aucun préambule, aucun commentaire, aucun bloc de code autour du texte." ) class AIWritingService: """Deterministic, provider-agnostic writing assistant.""" def __init__(self, user_id: int | None = None, provider: str | None = None, model: str | None = None, api_key: str | None = None, api_base: str | None = None): self.user_id = user_id self.provider = (provider or "").strip().lower() or None self.model = (model or "").strip() or None self._api_key = api_key self._api_base = api_base # ── LLM plumbing ── def _client(self) -> LLMClient: provider = self.provider api_key = self._api_key api_base = self._api_base if provider and self.user_id: try: from app.services.llm_config import get_user_llm_key row = get_user_llm_key(self.user_id, provider) if row and row.get("api_key"): api_key = row["api_key"] api_base = (row.get("api_base") or "").strip() or api_base except Exception: # noqa: BLE001 — never fail on key lookup pass return LLMClient(provider=provider, api_key=api_key, api_base=api_base) async def _complete(self, prompt: str, context: str = "") -> tuple[str, str, bool]: llm = self._client() offline = llm.provider == "offline" or not llm._has_credentials() user_content = prompt if context and context.strip(): user_content += "\n\n# Contexte\n" + context.strip()[:_MAX_CONTEXT] messages = [ {"role": "system", "content": _SYSTEM_WRITING}, {"role": "user", "content": user_content}, ] resp = await llm.complete(messages, model=self.model, tools=None, stream=False) return (resp.text or "").strip(), (resp.model or self.model or ""), offline # ── Public API ── async def run(self, action: str, *, prompt: str = "", context: str = "", target_language: str = "English", prefix: str = "", title: str = "", properties: list | None = None) -> dict: action = (action or "").strip().lower() if action not in WRITING_ACTIONS: raise ValueError(f"Action inconnue: {action or '(vide)'}") if action == "properties": suggestions = await self.suggest_properties( context=context, title=title, properties=properties or []) return {"ok": True, "action": action, "text": "", "suggestions": suggestions, "model": self.model or "", "offline": self._offline_hint()} prompt_text = self._build_prompt( action, prompt=prompt, context=context, target_language=target_language, prefix=prefix, title=title) # No connected provider → deterministic, dependency-free output (the raw # offline planner echoes the prompt, which is wrong for continue/autocomplete). if self._offline_hint(): return {"ok": True, "action": action, "model": "", "offline": True, "text": self._offline_text(action, prompt=prompt, context=context, target_language=target_language, prefix=prefix, title=title)} try: text, model, offline = await self._complete(prompt_text, context=context) except Exception as exc: # noqa: BLE001 — surface provider errors to the UI logger.warning("AI writing '%s' failed: %s", action, exc) return {"ok": False, "action": action, "error": str(exc), "text": "", "model": self.model or "", "offline": False} if not text: text = self._offline_text(action, prompt=prompt, context=context, target_language=target_language, prefix=prefix, title=title) offline = True return {"ok": True, "action": action, "text": text, "model": model, "offline": offline} # ── Prompt building ── def _build_prompt(self, action: str, *, prompt: str, context: str, target_language: str, prefix: str, title: str) -> str: if action == "write": subject = (prompt or title or "ce document").strip() return (f"Rédige le contenu demandé : {subject}. " "Fournis un texte structuré et directement utilisable.") if action == "summarize": return ("Résume le contenu fourni de façon structurée et concise : " "un court paragraphe d'introduction puis 3 à 5 points clés à puces.") if action == "translate": lang = (target_language or "English").strip() return (f"Traduis l'intégralité du contenu fourni en {lang}, " "en conservant fidèlement sa structure (titres, listes, paragraphes). " "Ne traduis pas les noms propres et les termes techniques.") if action == "continue": return ("Poursuis naturellement le texte fourni. " "Écris un à trois paragraphes cohérents avec le style et le sujet, " "sans répéter ce qui précède et sans introduction.") if action == "autocomplete": return (f"Complète la phrase en cours par une suite courte et pertinente " f"(maximum 20 mots). Ne répète pas le texte déjà écrit, ne mets " f"aucun préambule. Texte en cours : {prefix!r}") return prompt # ── Offline deterministic fallbacks ── def _offline_hint(self) -> bool: try: llm = self._client() return llm.provider == "offline" or not llm._has_credentials() except Exception: # noqa: BLE001 return True def _offline_text(self, action: str, *, prompt: str, context: str, target_language: str, prefix: str, title: str) -> str: if action == "summarize": return self._offline_summary(context) if action == "translate": return (f"⚠️ **Traduction hors-ligne indisponible** — aucun modèle d'IA connecté.\n\n" f"Connectez un fournisseur dans **Paramètres → Agent & IA** pour traduire " f"ce document en {target_language or 'English'}.") if action == "autocomplete": return self._offline_autocomplete(prefix) if action == "continue": return ("Suite du contenu : développez ici le point précédent avec un exemple " "concret, puis ouvrez la prochaine idée en une phrase de transition.") subject = (prompt or title or "ce document").strip() return (f"## {subject}\n" "\n" "Présentation générale du sujet : objectif, contexte et public visé en " "quelques phrases. (Contenu généré hors-ligne — connectez une clé API " "pour une rédaction complète.)\n" "\n" "## Points clés\n" "• Idée principale 1 et son argument.\n" "• Idée principale 2 avec un exemple concret.\n" "\n" "## Prochaines étapes\n" "• Relire, compléter et mettre en forme ce contenu.") @staticmethod def _offline_summary(context: str) -> str: text = (context or "").strip() if not text: return "Résumé : aucun contenu fourni à résumer." headings = re.findall(r"^#{1,4}\s+(.+)$", text, flags=re.MULTILINE) sentences = re.split(r"(?<=[.!?])\s+", re.sub(r"\s+", " ", text)) lead = next((s.strip() for s in sentences if len(s.strip()) > 40), sentences[0].strip()) out = ["**Résumé**", "", lead[:400], ""] bullets = [] if headings: bullets = [f"• {h.strip()}" for h in headings[:5]] else: for s in sentences[1:6]: s = s.strip() if len(s) > 30: bullets.append(f"• {s[:180]}") if bullets: out.append("**Points clés**") out.extend(bullets) return "\n".join(out) @staticmethod def _offline_autocomplete(prefix: str) -> str: prefix = (prefix or "").strip() if len(prefix) < 8: return "" return " Cette section détaille les points clés à retenir." # ── AI properties ── async def suggest_properties(self, *, context: str = "", title: str = "", properties: list | None = None) -> dict: """Return ``{property_name: value}`` suggestions for a collection page. ``properties`` is a list of ``{name, type}`` dicts. With a connected provider the model is asked for a JSON object; offline we derive deterministic defaults from the property names so the UI stays useful. """ properties = properties or [] if not properties: return {} names = [str(p.get("name", "")).strip() for p in properties if isinstance(p, dict)] names = [n for n in names if n] llm = self._client() offline = llm.provider == "offline" or not llm._has_credentials() if not offline: schema = {str(p.get("name")): str(p.get("type", "text")) for p in properties if isinstance(p, dict)} prompt = ( "À partir du titre et du contenu du document, propose une valeur pour " "chaque propriété. Réponds STRICTEMENT par un objet JSON " "{\"nom_propriété\": valeur} sans texte autour.\n" f"Propriétés attendues : {json.dumps(schema, ensure_ascii=False)}\n" f"Titre : {title or '(sans titre)'}" ) try: text, _, _ = await self._complete(prompt, context=context) parsed = self._parse_json_object(text) if parsed: return self._coerce_suggestions(parsed, properties) except Exception as exc: # noqa: BLE001 logger.warning("AI properties failed: %s", exc) # fall through to deterministic defaults return self._offline_suggestions(title, context, properties) @staticmethod def _parse_json_object(text: str) -> dict: text = (text or "").strip() if not text: return {} m = re.search(r"\{.*\}", text, flags=re.DOTALL) if not m: return {} try: data = json.loads(m.group(0)) except json.JSONDecodeError: return {} return data if isinstance(data, dict) else {} def _coerce_suggestions(self, parsed: dict, properties: list) -> dict: out: dict = {} by_name = {str(p.get("name", "")).strip().lower(): p for p in properties if isinstance(p, dict)} for key, value in parsed.items(): prop = by_name.get(str(key).strip().lower()) if not prop: continue out[str(prop.get("name"))] = self._coerce_value(value, prop.get("type", "text")) return out @staticmethod def _coerce_value(value, prop_type: str): ptype = (prop_type or "text").lower() if ptype == "checkbox": if isinstance(value, bool): return value return str(value).strip().lower() in ("1", "true", "yes", "oui", "vrai", "x") if ptype == "number": try: num = float(value) return int(num) if num.is_integer() else num except (TypeError, ValueError): return value if isinstance(value, (dict, list)): return json.dumps(value, ensure_ascii=False) return value @staticmethod def _offline_suggestions(title: str, context: str, properties: list) -> dict: out: dict = {} summary_text = "" if context: summary_text = re.sub(r"\s+", " ", context).strip() first = re.split(r"(?<=[.!?])\s+", summary_text) summary_text = next((s for s in first if len(s) > 40), summary_text)[:180] for prop in properties: if not isinstance(prop, dict): continue name = str(prop.get("name", "")).strip() if not name: continue low = name.lower() ptype = (prop.get("type") or "text").lower() matched = False for keys, _ptype, default in _OFFLINE_PROPERTY_DEFAULTS: if any(k in low for k in keys): if "summary" in keys or "résumé" in keys or "resume" in keys or "description" in keys or "notes" in keys: out[name] = summary_text or (title or "") else: out[name] = default matched = True break if not matched and ptype in ("text", "title"): if "name" in low or "titre" in low or "title" in low: out[name] = title or "" return out async def run_action(action: str, **kwargs) -> dict: """Module-level convenience wrapper (used by tests and simple callers).""" return await AIWritingService().run(action, **kwargs)