from __future__ import annotations import argparse from datetime import datetime, timezone import json import math import os from pathlib import Path import struct from typing import Any import wave import httpx ROOT = Path(__file__).resolve().parents[1] DEFAULT_OUTPUT = ROOT / ".local_stack" / "demo-seed-summary.json" def utc_now() -> datetime: return datetime.now(timezone.utc) def now_stamp() -> str: return utc_now().strftime("%Y%m%d_%H%M%S") def build_seed_plan(tag: str) -> dict[str, Any]: return { "queue_name": "Demo Queue", "customer_name": "Demo Customer", "customer_phone": "+77010000001", "voice_subject": "Demo voice escalation", "chat_subject": "Demo Telegram follow-up", "webchat_text": "Demo webchat request from the website", "email_subject": "Demo email request about account access", "email_body": "Please restore access to the portal. This message is preloaded for the management demo.", "kb_category": "Demo Knowledge", "kb_title": "Demo Knowledge Article", "kb_keyword": "demo-showcase", "telegram_text": "Demo inbound Telegram message", "voice_event_type": "call.started", "recording_file_name": "demo-call.wav", "ivr_root_prompt_kz": "\u0421\u0430\u043b\u0430\u043c\u0430\u0442\u0441\u044b\u0437 \u0431\u0430! \u049a\u0430\u0437\u0430\u049b \u0442\u0456\u043b\u0456\u043d \u0442\u0430\u04a3\u0434\u0430\u0443 \u04af\u0448\u0456\u043d \u0431\u0456\u0440 \u0446\u0438\u0444\u0440\u044b\u043d \u0442\u0435\u0440\u0456\u04a3\u0456\u0437.", "ivr_root_prompt_ru": "\u0417\u0434\u0440\u0430\u0432\u0441\u0442\u0432\u0443\u0439\u0442\u0435! \u0414\u043e\u0431\u0440\u043e \u043f\u043e\u0436\u0430\u043b\u043e\u0432\u0430\u0442\u044c \u0432 \u043a\u043e\u043d\u0442\u0430\u043a\u0442-\u0446\u0435\u043d\u0442\u0440. \u0414\u043b\u044f \u0440\u0443\u0441\u0441\u043a\u043e\u0433\u043e \u044f\u0437\u044b\u043a\u0430 \u043d\u0430\u0436\u043c\u0438\u0442\u0435 \u0446\u0438\u0444\u0440\u0443 \u0434\u0432\u0430.", "ivr_root_prompt": "Здравствуйте! Добро пожаловать в контакт-центр. Для русского языка нажмите цифру два. Саламатсыз ба! Қазақ тілін таңдау үшін бір цифрын теріңіз.", "ivr_ru_menu_prompt": "Для отдела продаж нажмите 1. Для службы поддержки нажмите 2.", "ivr_kz_menu_prompt": "Сату бөлімі үшін 1 басыңыз. Қолдау қызметі үшін 2 басыңыз.", } def _demo_ivr_flow_document( plan: dict[str, Any], *, voice_start_kz_queue_id: str, voice_start_ru_queue_id: str, ) -> dict[str, Any]: return { "nodes": [ { "node_id": "root", "prompt_text": plan["ivr_root_prompt"], "prompt_sequence": [ { "prompt_audio_key": "ivr/demo-language-ru", "prompt_text": plan["ivr_root_prompt_ru"], "language": "ru", }, { "prompt_audio_key": "ivr/demo-language-kz", "prompt_text": plan["ivr_root_prompt_kz"], "language": "kz", }, ], "is_terminal": False, "invalid_target_node_id": "root", "no_input_target_node_id": "root", "options": [ {"digit": "1", "target_node_id": "voice_start_kz"}, {"digit": "2", "target_node_id": "voice_start_ru"}, ], }, { "node_id": "voice_start_ru", "prompt_text": "Передаем звонок в русскоязычный стартовый voice-сценарий.", "is_terminal": True, "outcome_code": "voice_start_ru", "resolved_queue_id": voice_start_ru_queue_id, "resolved_queue_code": "voice_start_ru", "options": [], }, { "node_id": "voice_start_kz", "prompt_text": "Қоңырауды қазақ тіліндегі бастапқы voice-сценарийге өткізіп жатырмыз.", "is_terminal": True, "outcome_code": "voice_start_kz", "resolved_queue_id": voice_start_kz_queue_id, "resolved_queue_code": "voice_start_kz", "options": [], }, ] } def _raise_for_status(response: httpx.Response, action: str) -> None: try: response.raise_for_status() except httpx.HTTPStatusError as exc: detail = response.text.strip() raise RuntimeError(f"{action} failed: {exc.response.status_code} {detail}") from exc def _build_demo_wav(target_path: Path, duration_seconds: int = 1) -> Path: target_path.parent.mkdir(parents=True, exist_ok=True) sample_rate = 8000 amplitude = 12000 frequency = 440.0 total_frames = sample_rate * duration_seconds with wave.open(str(target_path), "wb") as handle: handle.setnchannels(1) handle.setsampwidth(2) handle.setframerate(sample_rate) frames = bytearray() for index in range(total_frames): sample = int(amplitude * math.sin(2 * math.pi * frequency * (index / sample_rate))) frames.extend(struct.pack(" Path: raw = os.getenv("CC_DATA_DIR") if raw: return Path(raw).resolve() local_dir = (ROOT / ".data_local").resolve() if local_dir.exists(): return local_dir return (ROOT / ".data").resolve() def seed_demo(base_url: str) -> dict[str, Any]: admin = {"X-User": "admin", "X-Role": "admin"} supervisor = {"X-User": "supervisor", "X-Role": "supervisor"} operator = {"X-User": "operator", "X-Role": "operator"} analyst = {"X-User": "analyst", "X-Role": "analyst"} tag = now_stamp() plan = build_seed_plan(tag) sample_audio_path = _build_demo_wav((_seed_data_dir() / "demo_assets" / plan["recording_file_name"]).resolve()) with httpx.Client(base_url=base_url, timeout=10) as client: health = client.get("/health") _raise_for_status(health, "gateway health check") queue = client.post( "/proxy/routing/queues", headers=admin, json={ "name": plan["queue_name"], "description": "Prepared automatically for a management demo", "rules": [ {"channel": "voice", "priority": 3, "strategy": "round_robin", "sla_seconds": 30}, {"channel": "telegram", "priority": 3, "strategy": "round_robin", "sla_seconds": 45}, {"channel": "webchat", "priority": 3, "strategy": "round_robin", "sla_seconds": 40}, {"channel": "email", "priority": 3, "strategy": "round_robin", "sla_seconds": 120}, ], }, ) _raise_for_status(queue, "queue create") queue_id = queue.json()["queue_id"] sales_queue = client.post( "/proxy/routing/queues", headers=admin, json={ "name": "Voice Start KZ Queue", "description": "Prepared automatically for voice-start routing demo", "rules": [ {"channel": "voice", "priority": 3, "strategy": "round_robin", "sla_seconds": 25}, ], }, ) _raise_for_status(sales_queue, "voice_start_kz queue create") sales_queue_id = sales_queue.json()["queue_id"] support_queue = client.post( "/proxy/routing/queues", headers=admin, json={ "name": "Voice Start RU Queue", "description": "Prepared automatically for voice-start routing demo", "rules": [ {"channel": "voice", "priority": 3, "strategy": "round_robin", "sla_seconds": 35}, ], }, ) _raise_for_status(support_queue, "voice_start_ru queue create") support_queue_id = support_queue.json()["queue_id"] customer = client.post( "/proxy/customer/customers", json={ "display_name": plan["customer_name"], "phones": [plan["customer_phone"]], "preferred_phone": plan["customer_phone"], "tags": ["demo", "showcase"], }, ) _raise_for_status(customer, "customer create") customer_id = customer.json()["customer_id"] voice_interaction = client.post( "/proxy/interaction/interactions", headers=operator, json={ "channel": "voice", "subject": plan["voice_subject"], "customer_id": customer_id, "queue_id": queue_id, "priority": 3, }, ) _raise_for_status(voice_interaction, "voice interaction create") voice_interaction_id = voice_interaction.json()["interaction_id"] assign = client.patch( f"/proxy/interaction/interactions/{voice_interaction_id}/assign", headers=supervisor, json={"assignee": "operator_a"}, ) _raise_for_status(assign, "voice interaction assign") escalate = client.post( f"/proxy/interaction/interactions/{voice_interaction_id}/escalate", headers=operator, json={"target_queue_id": "line2"}, ) _raise_for_status(escalate, "voice interaction escalate") close = client.patch( f"/proxy/interaction/interactions/{voice_interaction_id}/status", headers=operator, json={"status": "closed"}, ) _raise_for_status(close, "voice interaction close") active_interaction = client.post( "/proxy/interaction/interactions", headers=operator, json={ "channel": "telegram", "subject": plan["chat_subject"], "customer_id": customer_id, "queue_id": "line2", "priority": 3, }, ) _raise_for_status(active_interaction, "active interaction create") active_interaction_id = active_interaction.json()["interaction_id"] voice_event = client.post( "/proxy/voice/integrations/voice/events", headers=operator, json={ "event_type": plan["voice_event_type"], "call_id": "demo_call", "interaction_id": voice_interaction_id, "payload": {"source": "demo-seed"}, }, ) _raise_for_status(voice_event, "voice event send") voice_event_id = voice_event.json()["event_id"] recording_ready_event = client.post( "/proxy/voice/integrations/voice/events", headers=operator, json={ "event_type": "recording.ready", "call_id": "demo_call", "interaction_id": voice_interaction_id, "payload": { "source": "demo-seed", "source_path": str(sample_audio_path), "file_name": plan["recording_file_name"], "mime_type": "audio/wav", "duration_seconds": 1, "recorded_at": utc_now().replace(microsecond=0).isoformat(), }, }, ) _raise_for_status(recording_ready_event, "recording ready event send") recording_event_id = recording_ready_event.json()["event_id"] recording = client.post( f"/proxy/recording/recordings/import-from-voice-event/{recording_event_id}", headers=supervisor, ) _raise_for_status(recording, "recording import") recording_id = recording.json()["recording_id"] ivr_flow = client.post( "/proxy/ivr/ivr/flows", headers=admin, json={ "name": "Demo IVR Flow", "description": "Prepared automatically for a management demo", "queue_id": queue_id, "entry_node_id": "root", "flow_json": _demo_ivr_flow_document( plan, voice_start_kz_queue_id=sales_queue_id, voice_start_ru_queue_id=support_queue_id, ), "is_active": True, }, ) _raise_for_status(ivr_flow, "ivr flow create") ivr_flow_id = ivr_flow.json()["flow_id"] ivr_session_start = client.post( "/proxy/ivr/ivr/sessions/start", headers=admin, json={ "call_id": "demo_call_ivr", "queue_id": queue_id, "interaction_id": voice_interaction_id, }, ) _raise_for_status(ivr_session_start, "ivr session start") ivr_session_id = ivr_session_start.json()["session"]["session_id"] ivr_session_step = client.post( f"/proxy/ivr/ivr/sessions/{ivr_session_id}/dtmf", headers=admin, json={"digit": "1"}, ) _raise_for_status(ivr_session_step, "ivr language step") ivr_session_step = client.post( f"/proxy/ivr/ivr/sessions/{ivr_session_id}/dtmf", headers=admin, json={"digit": "2"}, ) _raise_for_status(ivr_session_step, "ivr dtmf step") ivr_session_payload = ivr_session_step.json() ivr_route_preview = client.post( f"/proxy/routing/queues/{queue_id}/route?channel=voice&priority=3&ivr_session_id={ivr_session_id}", headers=admin, ) _raise_for_status(ivr_route_preview, "ivr route preview") telegram = client.post( "/proxy/telegram/integrations/telegram/webhook", json={ "chat_id": "demo_chat", "text": plan["telegram_text"], "customer_external_id": customer_id, "payload": {"source": "demo-seed"}, }, ) _raise_for_status(telegram, "telegram webhook send") telegram_id = telegram.json()["message_id"] webchat = client.post( "/proxy/webchat/integrations/webchat/messages", json={ "session_id": "demo_webchat_session", "text": plan["webchat_text"], "visitor_name": "Demo Visitor", "customer_external_id": customer_id, "queue_id": "line2", "priority": 3, "payload": {"source": "demo-seed", "subject": "Demo webchat intake"}, }, ) _raise_for_status(webchat, "webchat message send") webchat_id = webchat.json()["message_id"] webchat_interaction_id = webchat.json()["interaction_id"] email = client.post( "/proxy/email/integrations/email/messages", json={ "from_email": "demo.user@example.com", "subject": plan["email_subject"], "body": plan["email_body"], "customer_external_id": customer_id, "queue_id": "line2", "priority": 3, "payload": {"source": "demo-seed", "mailbox": "support@example.com"}, }, ) _raise_for_status(email, "email message send") email_id = email.json()["message_id"] email_interaction_id = email.json()["interaction_id"] category = client.post( "/proxy/kb/knowledge/categories", headers=analyst, json={"name": plan["kb_category"], "description": "Demo search content"}, ) _raise_for_status(category, "kb category create") category_id = category.json()["category_id"] article = client.post( "/proxy/kb/knowledge/articles", headers=analyst, json={ "category_id": category_id, "title": plan["kb_title"], "body": "Use the demo-showcase article when management asks about answer prompts.", "tags": ["demo", plan["kb_keyword"]], }, ) _raise_for_status(article, "kb article create") article_id = article.json()["article_id"] search = client.get(f"/proxy/kb/knowledge/search?q={plan['kb_keyword']}") _raise_for_status(search, "kb search") agent1 = client.post( "/proxy/supervisor/supervisor/agent-states", headers=supervisor, json={"agent_id": "demo_agent_a", "state": "READY", "queue_id": "line2"}, ) _raise_for_status(agent1, "supervisor agent update 1") agent2 = client.post( "/proxy/supervisor/supervisor/agent-states", headers=supervisor, json={"agent_id": "demo_agent_b", "state": "BUSY", "queue_id": "line2"}, ) _raise_for_status(agent2, "supervisor agent update 2") queue_metrics = client.post( "/proxy/supervisor/supervisor/queue-metrics?queue_id=line2&in_queue=1&avg_wait_seconds=14", headers=supervisor, ) _raise_for_status(queue_metrics, "supervisor queue metrics") realtime = client.get("/proxy/supervisor/supervisor/realtime", headers=supervisor) _raise_for_status(realtime, "supervisor realtime") kpi_rows = [ { "queue_id": "line2", "channel": "voice", "agent_id": "demo_agent_b", "answered": True, "wait_seconds": 15, "handle_seconds": 90, "abandoned": False, "resolved_first_contact": True, }, { "queue_id": "line2", "channel": "webchat", "agent_id": "demo_agent_a", "answered": True, "wait_seconds": 25, "handle_seconds": 110, "abandoned": False, "resolved_first_contact": False, }, { "queue_id": "line2", "channel": "email", "agent_id": None, "answered": False, "wait_seconds": 12, "handle_seconds": 0, "abandoned": True, "resolved_first_contact": False, }, ] for idx, row in enumerate(kpi_rows, start=1): created = client.post("/proxy/reporting/reports/events", json=row) _raise_for_status(created, f"kpi event ingest {idx}") kpi = client.get("/proxy/reporting/reports/kpi?queue_id=line2&sl_threshold_seconds=30") _raise_for_status(kpi, "kpi query") event_bus_enabled = os.getenv("EVENT_BUS_ENABLED", "0").strip().lower() in {"1", "true", "yes", "on"} event_bus_smoke_passed = False seeded_event_id = None if event_bus_enabled: try: from scripts.event_bus_smoke import run_smoke_check smoke = run_smoke_check(base_url=base_url, auth_mode="legacy_headers", timeout_seconds=10) event_bus_smoke_passed = bool(smoke.get("passed")) seeded_event_id = smoke.get("event_id") except Exception: event_bus_smoke_passed = False return { "generated_at": utc_now().isoformat(), "base_url": base_url, "demo_login": {"username": "admin", "password": "admin123"}, "queue_id": queue_id, "customer_id": customer_id, "voice_interaction_id": voice_interaction_id, "active_interaction_id": active_interaction_id, "voice_event_id": voice_event_id, "recording_event_id": recording_event_id, "recording_id": recording_id, "recording_call_id": "demo_call", "recording_file_name": plan["recording_file_name"], "event_bus_enabled": event_bus_enabled, "event_bus_smoke_passed": event_bus_smoke_passed, "seeded_event_id": seeded_event_id, "ivr_flow_id": ivr_flow_id, "ivr_session_id": ivr_session_id, "ivr_outcome_code": ivr_session_payload["session"]["outcome_code"], "ivr_resolved_queue_id": ivr_session_payload["session"]["resolved_queue_id"], "ivr_route_preview": ivr_route_preview.json(), "telegram_message_id": telegram_id, "webchat_message_id": webchat_id, "webchat_interaction_id": webchat_interaction_id, "email_message_id": email_id, "email_interaction_id": email_interaction_id, "kb_category_id": category_id, "kb_article_id": article_id, "kb_keyword": plan["kb_keyword"], "sample_recording_path": str(sample_audio_path), "kpi_snapshot": kpi.json(), "supervisor_snapshot": realtime.json(), } def main() -> int: parser = argparse.ArgumentParser(description="Seed demo-ready data into the local MVP stack") parser.add_argument("--base-url", default="http://localhost:8080", help="Gateway base URL") parser.add_argument( "--output", default=str(DEFAULT_OUTPUT), help="Where to store the seed summary JSON", ) args = parser.parse_args() output_path = Path(args.output) if not output_path.is_absolute(): output_path = (ROOT / output_path).resolve() output_path.parent.mkdir(parents=True, exist_ok=True) summary = seed_demo(args.base_url) output_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") print("Demo data is ready.") print(f"UI: {args.base_url}/operator") print(f"Summary: {output_path}") print(f"Customer: {summary['customer_id']}") print(f"Closed interaction: {summary['voice_interaction_id']}") print(f"Active interaction: {summary['active_interaction_id']}") print(f"Recording: {summary['recording_id']}") print(f"IVR flow: {summary['ivr_flow_id']}") print(f"IVR session: {summary['ivr_session_id']}") print(f"Webchat interaction: {summary['webchat_interaction_id']}") print(f"Email interaction: {summary['email_interaction_id']}") print(f"KB keyword: {summary['kb_keyword']}") return 0 if __name__ == "__main__": raise SystemExit(main())