Neo-ZQYY/apps/backend/app/ai/prompts/app5_tactics_prompt.py

"""应用 5 话术参考 Prompt 拼装。

App4 完成后串行触发，接收 App4 返回结果作为 task_suggestion。
- 数据源：fetch_assistant_info + fetch_service_history + fetch_member_consumption_data + fetch_member_notes + context.app4_result
- 输出字段：tactics 数组（每条含 scenario + script）
- system prompt 在百炼控制台配置

返回：单个 prompt 字符串。
"""

from __future__ import annotations

import asyncio
import json
import logging
from typing import Any

from app.ai.cache_service import AICacheService
from app.ai.data_fetchers import (
    fetch_assistant_info,
    fetch_member_consumption_data,
    fetch_member_notes,
    fetch_service_history,
)
from app.ai.schemas import CacheTypeEnum
from app.services.runtime_context import as_runtime_business_now_str

logger = logging.getLogger(__name__)

_MAX_PROMPT_LEN = 8000


async def build_prompt(
    context: dict,
    cache_svc: AICacheService | None = None,
) -> str:
    """构建 App5 prompt 字符串。

    Args:
        context: site_id, assistant_id, member_id, app4_result(dict|None)

    Returns:
        JSON 序列化后的 prompt 字符串
    """
    site_id = context["site_id"]
    assistant_id = context["assistant_id"]
    member_id = context["member_id"]
    task_suggestion = context.get("app4_result") or {}

    results = await asyncio.gather(
        fetch_assistant_info(site_id, assistant_id),
        fetch_service_history(site_id, assistant_id, member_id),
        fetch_member_consumption_data(site_id, member_id),
        fetch_member_notes(site_id, member_id),
        return_exceptions=True,
    )

    warnings: list[str] = []

    assistant_info = results[0] if not isinstance(results[0], Exception) else {}
    if isinstance(results[0], Exception):
        warnings.append("助教信息获取失败")
        logger.warning("App5 助教信息获取失败: %s", results[0])

    service_history = results[1] if not isinstance(results[1], Exception) else []
    if isinstance(results[1], Exception):
        warnings.append("服务历史获取失败")
        logger.warning("App5 服务历史获取失败: %s", results[1])

    if isinstance(results[2], Exception):
        member_data = _default_member_data()
        warnings.append("消费数据获取失败")
        logger.warning("App5 消费数据获取失败: %s", results[2])
    else:
        member_data = results[2]

    notes = results[3] if not isinstance(results[3], Exception) else []
    if isinstance(results[3], Exception):
        warnings.append("备注获取失败")
        logger.warning("App5 备注获取失败: %s", results[3])

    payload: dict[str, Any] = {
        "current_time": as_runtime_business_now_str(site_id, fmt="%Y-%m-%d %H:%M"),
        "assistant_id": assistant_id,
        "member_id": member_id,
        "task_suggestion": task_suggestion,
        "assistant_info": assistant_info or "⚠ 助教信息获取失败",
        "service_history": service_history or "暂无服务记录",
        "task_assignment_basis": {
            "consumption_records": member_data.get("consumption_records", []) or "该客户暂无消费记录",
            "member_cards": member_data.get("member_cards", []),
            "card_balance_total": member_data.get("card_balance_total", 0),
            "stored_value_balance_total": member_data.get("stored_value_balance_total", 0),
            "expected_visit_date": member_data.get("expected_visit_date"),
            "days_since_last_visit": member_data.get("days_since_last_visit"),
        },
        "customer_data": {
            "member_nickname": member_data.get("member_nickname", ""),
            "notes": notes or "暂无备注",
        },
        "reference": _build_reference(site_id, member_id, cache_svc),
    }

    if warnings:
        payload["_data_warnings"] = warnings

    return _truncate_payload(payload)


def _default_member_data() -> dict:
    return {
        "member_nickname": "",
        "consumption_records": [],
        "member_cards": [],
        "card_balance_total": 0,
        "stored_value_balance_total": 0,
        "expected_visit_date": None,
        "days_since_last_visit": None,
    }


def _build_reference(
    site_id: int,
    member_id: int,
    cache_svc: AICacheService | None,
) -> dict:
    """组装最近 2 条 App8 历史。"""
    if cache_svc is None:
        return {}

    ref: dict = {}
    history = cache_svc.get_history(
        CacheTypeEnum.APP8_CLUE_CONSOLIDATED.value,
        site_id,
        str(member_id),
        limit=2,
    )
    if history:
        ref["app8_history"] = [
            {"result_json": h.get("result_json"), "generated_at": h.get("created_at")}
            for h in history
        ]

    return ref


def _truncate_payload(payload: dict) -> str:
    """按优先级截断 service_history → consumption_records → notes。"""
    text = json.dumps(payload, ensure_ascii=False, default=str)
    if len(text) <= _MAX_PROMPT_LEN:
        return text

    sh = payload.get("service_history")
    if isinstance(sh, list) and len(sh) > 5:
        payload["service_history"] = sh[:5]
        payload["_truncated_service_history"] = f"服务记录已截断，原始 {len(sh)} 条"
        text = json.dumps(payload, ensure_ascii=False, default=str)
    if len(text) > _MAX_PROMPT_LEN:
        records = payload["task_assignment_basis"].get("consumption_records")
        if isinstance(records, list) and len(records) > 5:
            payload["task_assignment_basis"]["consumption_records"] = records[:5]
            payload["task_assignment_basis"]["_truncated"] = f"消费记录已截断，原始 {len(records)} 条"
            text = json.dumps(payload, ensure_ascii=False, default=str)
    if len(text) > _MAX_PROMPT_LEN:
        n = payload["customer_data"].get("notes")
        if isinstance(n, list) and len(n) > 10:
            payload["customer_data"]["notes"] = n[:10]
            payload["customer_data"]["_truncated_notes"] = f"备注已截断，原始 {len(n)} 条"
            text = json.dumps(payload, ensure_ascii=False, default=str)
    return text