"""ARQ background jobs for the AI Proactive plugin. Deep analysis job runs after context-change for deeper analysis: - Mail-thread summary generation - Finding similar contacts via unified_search - Analyzing open tasks - Generating extended suggestion with more context - Pushing additional suggestion if confidence > threshold """ from __future__ import annotations import os import json import logging import uuid from typing import Any import litellm from sqlalchemy import select from app.core.db import create_db_session from app.core.notifications import create_notification from app.models.audit import AuditLog from app.models.contact import Contact from app.plugins.builtins.ai_proactive.models import ProactiveSuggestion, ProactiveSettings from app.plugins.builtins.ai_proactive.services import ( _serialize_row, generate_suggestion, get_user_settings, push_suggestion, ) from app.plugins.builtins.mail.models import Mail logger = logging.getLogger(__name__) OLLAMA_API_KEY = os.environ.get('API_KEY_OLLAMA_CLOUD', '') DEEP_SYSTEM_PROMPT = """Du bist ein proaktiver KI-Assistent für ein CRM. Du führst eine Tiefenanalyse durch. Analysiere den erweiterten Kontext und generiere einen detaillierten Vorschlag. Antworte mit JSON: { "suggestion_type": "info|warning|action|insight", "title": "Kurzer Titel (max 200 Zeichen)", "content": "Detaillierte Beschreibung mit konkreten Empfehlungen", "confidence": 0.0-1.0, "actions": [{"method": "GET|POST|PUT|DELETE", "path": "/api/v1/...", "body": {}, "description": "..."}] } Fokussiere auf: - Muster in der Kommunikationshistorie - Beziehungen zwischen Entities - Handlungsempfehlungen mit hoher Relevanz - Antworte nur mit gültigem JSON""" async def deep_analysis( ctx: dict[str, Any], entity_type: str, entity_id: str, user_id: str, tenant_id: str, ) -> None: """Deep Analysis Background-Job. Enqueued after context-change for deeper analysis: - Mail-thread summary generation - Similar contacts via unified_search - Open tasks analysis - Extended suggestion generation with more context - If confidence > threshold: save + push additional suggestion """ try: eid = uuid.UUID(entity_id) uid = uuid.UUID(user_id) tid = uuid.UUID(tenant_id) except (ValueError, TypeError): logger.warning("deep_analysis: invalid UUID parameters") return async with create_db_session(tid) as db: # Get settings settings = await get_user_settings(db, tid, uid) if not settings.enabled: return # Gather extended context (more mails, more activities) extended_context: dict[str, Any] = { "entity_type": entity_type, "entity_id": entity_id, "deep_analysis": True, } if entity_type == "contact": # Extended: last 30 mails mail_result = await db.execute( select(Mail) .where(Mail.contact_id == eid) .where(Mail.tenant_id == tid) .order_by(Mail.received_at.desc()) .limit(30) ) extended_context["mails"] = [ _serialize_row(m) for m in mail_result.scalars().all() ] # Extended: last 50 audit entries audit_result = await db.execute( select(AuditLog) .where(AuditLog.entity_id == eid) .where(AuditLog.tenant_id == tid) .order_by(AuditLog.timestamp.desc()) .limit(50) ) extended_context["activities"] = [ _serialize_row(a) for a in audit_result.scalars().all() ] # Contact data contact_result = await db.execute( select(Contact) .where(Contact.id == eid) .where(Contact.tenant_id == tid) .limit(1) ) contact = contact_result.scalar_one_or_none() extended_context["contact"] = _serialize_row(contact) if contact else None elif entity_type == "company": from app.models.company import Company, CompanyContact comp_result = await db.execute( select(Company) .where(Company.id == eid) .where(Company.tenant_id == tid) .limit(1) ) company = comp_result.scalar_one_or_none() extended_context["company"] = _serialize_row(company) if company else None # All mails for company mail_result = await db.execute( select(Mail) .where(Mail.company_id == eid) .where(Mail.tenant_id == tid) .order_by(Mail.received_at.desc()) .limit(30) ) extended_context["mails"] = [ _serialize_row(m) for m in mail_result.scalars().all() ] elif entity_type == "mail": mail_result = await db.execute( select(Mail) .where(Mail.id == eid) .where(Mail.tenant_id == tid) .limit(1) ) mail = mail_result.scalar_one_or_none() extended_context["mail"] = _serialize_row(mail) if mail else None if mail and mail.thread_id: thread_result = await db.execute( select(Mail) .where(Mail.thread_id == mail.thread_id) .where(Mail.tenant_id == tid) .order_by(Mail.received_at.asc()) .limit(50) ) extended_context["thread"] = [ _serialize_row(m) for m in thread_result.scalars().all() ] # Similar entities via unified_search try: from app.plugins.builtins.unified_search.search_engine import ( find_similar_all_types, ) extended_context["similar"] = await find_similar_all_types( db, entity_type, eid, tid, limit=5 ) except Exception: extended_context["similar"] = {} # Generate extended suggestion with deep analysis prompt model = settings.model or "ollama/deepseek-v4" # Get API key from DB (like ai_assistant does) or fall back to env from app.plugins.builtins.ai_proactive.services import _get_llm_api_key api_key, api_base = await _get_llm_api_key(db, tid) if not api_key: api_key = os.environ.get('API_KEY_OLLAMA_CLOUD', '') or None litellm_kwargs: dict[str, Any] = dict( model=model, messages=[ {"role": "system", "content": DEEP_SYSTEM_PROMPT}, { "role": "user", "content": json.dumps( extended_context, default=str, ensure_ascii=False ), }, ], temperature=0.3, max_tokens=800, response_format={"type": "json_object"}, ) if api_key: litellm_kwargs["api_key"] = api_key if api_base: litellm_kwargs["api_base"] = api_base try: response = await litellm.acompletion(**litellm_kwargs) content = response.choices[0].message.content if not content: return suggestion_data = json.loads(content) if not suggestion_data.get("title") or not suggestion_data.get("content"): return if not isinstance(suggestion_data.get("actions"), list): suggestion_data["actions"] = [] confidence = suggestion_data.get("confidence", 0.5) try: confidence = float(confidence) except (TypeError, ValueError): confidence = 0.5 suggestion_data["confidence"] = max(0.0, min(1.0, confidence)) valid_types = {"info", "warning", "action", "insight"} if suggestion_data.get("suggestion_type") not in valid_types: suggestion_data["suggestion_type"] = "insight" except Exception: logger.exception("deep_analysis: LLM call failed") return # Confidence threshold check if suggestion_data["confidence"] < settings.confidence_threshold: logger.debug( "deep_analysis: confidence %s below threshold %s", suggestion_data["confidence"], settings.confidence_threshold, ) return # Save extended suggestion suggestion = ProactiveSuggestion( tenant_id=tid, user_id=uid, entity_type=entity_type, entity_id=eid, suggestion_type=suggestion_data["suggestion_type"], title=f"[Tiefenanalyse] {suggestion_data['title']}", content=suggestion_data["content"], confidence=suggestion_data["confidence"], actions=suggestion_data["actions"], context_snapshot=extended_context, ) db.add(suggestion) await db.flush() suggestion_dict = { "id": str(suggestion.id), "entity_type": suggestion.entity_type, "entity_id": str(suggestion.entity_id) if suggestion.entity_id else None, "suggestion_type": suggestion.suggestion_type, "title": suggestion.title, "content": suggestion.content, "confidence": suggestion.confidence, "actions": suggestion.actions, "created_at": suggestion.created_at.isoformat() if suggestion.created_at else None, "is_dismissed": suggestion.is_dismissed, "is_acted_upon": suggestion.is_acted_upon, } # Push via SSE await push_suggestion(str(uid), suggestion_dict) # Notification for urgent suggestions if suggestion.suggestion_type == "warning": try: await create_notification( db, tid, uid, type="ai_suggestion_urgent", title=suggestion.title, body=suggestion.content[:200], ) except Exception: logger.exception("deep_analysis: notification failed") await db.commit() logger.info( "deep_analysis: generated suggestion %s for %s/%s", suggestion.id, entity_type, entity_id, )