from __future__ import annotations from backend.ai.client import llm_client, render_prompt from backend.config import get_settings from backend.ai.prompts import ANALYZE_PROMPT, CUSTOM_TASK_PROMPT from backend.models.schemas import TaskCategory def analyze_business_description(description: str) -> dict: settings = get_settings() if settings.analyze_with_ai and llm_client.is_ready(): try: return llm_client.generate_json(render_prompt(ANALYZE_PROMPT, description=description)) except Exception: pass lower = description.lower() fully_automatable = [] ai_assisted = [] manual = [] if "order" in lower: fully_automatable.append( { "id": "task_order_processing", "name": "Order processing", "description": "Read order emails, check stock, log rows, and reply.", "time_estimate": "10-15 hours/week", "pain_level": "high", } ) if "friday" in lower or "weekly" in lower: fully_automatable.append( { "id": "task_weekly_summary", "name": "Weekly order summary report", "description": "Compile the week's orders and send a summary.", "time_estimate": "1-2 hours/week", "pain_level": "medium", } ) ai_assisted.append( { "id": "task_availability_inquiries", "name": "Availability inquiries", "description": "Draft replies using current stock and pricing.", "time_estimate": "2-3 hours/week", "pain_level": "medium", } ) ai_assisted.append( { "id": "task_low_inventory_alerts", "name": "Low inventory alerts", "description": "Watch for thin inventory and propose next steps.", "time_estimate": "30 minutes/week", "pain_level": "medium", } ) manual.append( { "id": "task_special_requests", "name": "Custom and relationship-sensitive requests", "description": "Gift boxes, delivery arrangements, and VIP relationship notes.", "reason_manual": "These need human judgment, pricing discretion, and a personal touch.", } ) return { "business_type": "Email-driven small business", "summary": "FlowPilot found repetitive inbox work that maps well to Gmail, Sheets, and scheduled automations.", "tasks": { "fully_automatable": fully_automatable, "ai_assisted": ai_assisted, "manual": manual, }, } def analyze_custom_task(business_description: str, existing_workflows: list[dict], custom_task: str) -> dict: if llm_client.is_ready(): return llm_client.generate_json( render_prompt( CUSTOM_TASK_PROMPT, business_description=business_description, existing_workflows=existing_workflows, primitives_list=[ "email_received", "schedule", "file_uploaded", "manual_trigger", "ai_extract", "ai_classify", "ai_compose", "sheets_read", "sheets_write", "sheets_update", "send_email", "notify_owner", "condition", "loop", "wait_for_input", ], custom_task=custom_task, ) ) lower = custom_task.lower() feasible = any(word in lower for word in ("flag", "priority", "sheet", "email")) category = TaskCategory.FULLY_AUTOMATABLE if feasible else TaskCategory.MANUAL return { "feasible": feasible, "category": category, "task_name": "Restaurant priority handling", "task_description": custom_task, "missing_capabilities": [] if feasible else ["A matching primitive for this request is missing."], "reason_if_manual": None if feasible else "This task needs a capability outside the current primitives.", "time_estimate": "1 hour/week", "pain_level": "medium", }