baodan/api/insurance/recommend/service.py

"""推荐服务：调用 BaoDan Workflow 生成推荐方案。"""
import os
import json
import re
import uuid
import requests
from flask import current_app
from insurance.db.compat import db
from insurance.models.recommendation import RecommendationRecord


class RecommendService:
    """产品推荐业务逻辑。"""

    def generate(self, user_id: str, data: dict) -> dict:
        """提交推荐方案生成任务。"""
        customer = data.get("customer", {})

        # 保存推荐记录
        record = RecommendationRecord(
            user_id=user_id,
            customer_name=customer.get("name", ""),
            customer_age=customer.get("age"),
            customer_gender=customer.get("gender", ""),
            health_status=customer.get("health_status", ""),
            occupation=customer.get("occupation", ""),
            annual_income=customer.get("annual_income", 0),
            monthly_budget=customer.get("monthly_budget", 0),
            insurance_types=json.dumps(data.get("insurance_types", []), ensure_ascii=False),
            coverage_amount=data.get("coverage_amount", 0),
            coverage_period=data.get("coverage_period", ""),
            existing_policies=json.dumps(data.get("existing_policies", []), ensure_ascii=False),
            status="processing",
        )
        db.session.add(record)
        try:
            db.session.commit()
        except Exception:
            db.session.rollback()
            raise

        # 调用 BaoDan Workflow API
        api_key = os.environ.get("BAODAN_WORKFLOW_API_KEY", "")
        # 统一使用 DIFY_BASE_URL 环境变量（与 app.py 保持一致）
        base_url = os.environ.get("DIFY_BASE_URL", os.environ.get("BAODAN_API_URL", "http://localhost:5001"))

        # Workflow 期望全部 string 类型输入，insurance_types 为逗号分隔字符串
        insurance_types = data.get("insurance_types", [])
        if isinstance(insurance_types, list):
            insurance_types_str = ",".join(insurance_types)
        else:
            insurance_types_str = str(insurance_types)

        try:
            resp = requests.post(
                f"{base_url}/v1/workflows/run",
                json={
                    "inputs": {
                        "age": str(customer.get("age", "")),
                        "gender": customer.get("gender", ""),
                        "occupation": customer.get("occupation", ""),
                        "annual_income": str(customer.get("annual_income", 0)),
                        "monthly_budget": str(customer.get("monthly_budget", 0)),
                        "insurance_types": insurance_types_str,
                        "coverage_amount": str(data.get("coverage_amount", 0)),
                        "coverage_period": data.get("coverage_period", ""),
                    },
                    "response_mode": "blocking",
                    "user": f"user_{user_id}",
                },
                headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                timeout=120,
            )
            result = resp.json()

            if result.get("data", {}).get("status") == "succeeded":
                output = result["data"].get("outputs", {})
                # Workflow 输出 result（Markdown 字符串）
                recommendation_md = output.get("result", "") or output.get("recommendation", "")

                if recommendation_md:
                    record.generated_plan = recommendation_md
                    record.plan_variants = json.dumps(
                        self._parse_markdown_to_plans(recommendation_md),
                        ensure_ascii=False,
                    )
                    record.status = "done"
                else:
                    record.status = "failed"
                    record.error_message = "方案生成失败：无输出"
            else:
                record.status = "failed"
                error_msg = result.get("message", "Workflow 执行失败")
                record.error_message = error_msg

        except Exception as e:
            record.status = "failed"
            record.error_message = str(e)

        try:
            db.session.commit()
        except Exception:
            db.session.rollback()
            raise

        from insurance.utils.audit import log_operation
        log_operation(user_id, "generate", "proposal", str(record.id), {
            "customer_name": customer.get("name", ""),
            "insurance_types": data.get("insurance_types", []),
        })

        return {
            "code": 0,
            "data": {
                "task_id": str(record.id),
                "status": record.status,
            },
        }

    def get_status(self, task_id: str) -> dict:
        """查询推荐任务状态。"""
        record = db.session.query(RecommendationRecord).filter_by(id=int(task_id)).first()
        if not record:
            return {"code": 1005, "message": "任务不存在", "data": None}

        data = {
            "task_id": str(record.id),
            "status": record.status,
            # 客户信息
            "customer_name": record.customer_name or "",
            "customer_age": record.customer_age,
            "customer_gender": record.customer_gender or "",
            "health_status": record.health_status or "",
            "occupation": record.occupation or "",
            "annual_income": record.annual_income,
            "monthly_budget": record.monthly_budget,
            "insurance_types": record.insurance_types or "[]",
            "coverage_amount": record.coverage_amount,
            "coverage_period": record.coverage_period or "",
            "created_at": str(record.created_at) if record.created_at else None,
        }

        if record.status == "done":
            # 返回 Markdown 原文 + 解析后的结构化方案
            plans = json.loads(record.plan_variants) if record.plan_variants else []
            data["proposal"] = {
                "id": str(record.id),
                "recommendation": record.generated_plan or "",
                "plans": plans,
            }
        elif record.status == "failed":
            data["error_message"] = record.error_message

        return {"code": 0, "data": data}

    @staticmethod
    def _parse_markdown_to_plans(md_text: str) -> list:
        """解析 LLM 输出的 Markdown 为结构化方案列表。

        解析格式：
        ### 基础方案（年保费约 XXXX 元）
        |产品名称|所属保险公司|险种|保额|年保费|推荐理由|
        ...
        方案总结：...

        ### 均衡方案（年保费约 XXXX 元）
        ...
        """
        plans = []
        # 按 ### 分割方案
        sections = re.split(r'(?=###\s)', md_text)

        for section in sections:
            section = section.strip()
            if not section.startswith('###'):
                continue

            # 提取方案名称和年保费
            title_match = re.match(r'###\s*(.+?)(?:（.*?）)?$', section, re.MULTILINE)
            plan_name = title_match.group(1).strip() if title_match else "未知方案"

            # 提取年保费
            premium_match = re.search(r'年保费约\s*([\d,]+)\s*元', section)
            total_premium = premium_match.group(1).replace(',', '') if premium_match else ""

            # 提取表格行
            items = []
            table_lines = re.findall(r'\|(.+)\|', section)
            for line in table_lines:
                cells = [c.strip() for c in line.split('|')]
                # 跳过表头和分隔行
                if len(cells) < 6 or cells[0] in ('产品名称', '---------', '---', ''):
                    continue
                items.append({
                    "product_name": cells[0],
                    "company": cells[1],
                    "insurance_type": cells[2],
                    "coverage": cells[3],
                    "premium": cells[4],
                    "reason": cells[5],
                })

            # 提取方案总结
            summary_match = re.search(r'方案总结[：:]\s*(.+?)(?=###|\Z)', section, re.DOTALL)
            summary = summary_match.group(1).strip() if summary_match else ""

            plans.append({
                "name": plan_name,
                "total_premium": total_premium,
                "items": items,
                "summary": summary,
            })

        return plans

    def delete(self, proposal_id: str) -> dict:
        """删除推荐方案。"""
        record = db.session.query(RecommendationRecord).filter_by(id=int(proposal_id)).first()
        if not record:
            return {"code": 1005, "message": "方案不存在", "data": None}

        db.session.delete(record)
        db.session.commit()
        return {"code": 0, "message": "删除成功", "data": None}

    def share(self, proposal_id: str, expire_hours: int) -> dict:
        """生成分享链接。"""
        if expire_hours < 1 or expire_hours > 720:
            return {"code": 1001, "message": "有效期应在 1-720 小时之间", "data": None}

        record = db.session.query(RecommendationRecord).filter_by(id=int(proposal_id)).first()
        if not record:
            return {"code": 1005, "message": "方案不存在", "data": None}

        # 生成分享 Token
        share_token = str(uuid.uuid4())
        from insurance.db.compat import redis_client
        redis_client.setex(f"share:{share_token}", expire_hours * 3600, proposal_id)

        domain = current_app.config.get("DOMAIN", "localhost")
        return {
            "code": 0,
            "data": {
                "share_url": f"https://{domain}/shared/{proposal_id}?token={share_token}",
                "expire_at": "",
            },
        }

    def list_proposals(self, user_id: str, params: dict) -> dict:
        """推荐历史列表（带数据权限）。"""
        from insurance.middleware.auth_middleware import get_data_scope
        scope = get_data_scope()

        query = db.session.query(RecommendationRecord)

        # 根据数据权限过滤
        if scope["scope"] == "self":
            # 销售：只看自己的
            query = query.filter(RecommendationRecord.user_id == user_id)
        elif scope["scope"] == "team":
            # 主管：看本部门的
            from insurance.models.wecom_user import WeComUserMapping
            query = query.filter(
                RecommendationRecord.user_id.in_(
                    db.session.query(WeComUserMapping.id).filter(
                        WeComUserMapping.department == scope["department"]
                    )
                )
            )
        # admin/super_admin：看全部

        if params.get("customer_name"):
            query = query.filter(RecommendationRecord.customer_name.ilike(f"%{params['customer_name']}%"))
        if params.get("insurance_type"):
            query = query.filter(RecommendationRecord.insurance_types.ilike(f"%{params['insurance_type']}%"))

        total = query.count()
        items = (
            query.order_by(RecommendationRecord.created_at.desc())
            .offset((params["page"] - 1) * params["page_size"])
            .limit(params["page_size"])
            .all()
        )

        return {
            "code": 0,
            "data": {
                "total": total,
                "items": [
                    {
                        "id": str(r.id),
                        "customer_name": r.customer_name or "",
                        "customer_age": r.customer_age,
                        "customer_gender": r.customer_gender or "",
                        "occupation": r.occupation or "",
                        "insurance_types": r.insurance_types or "[]",
                        "coverage_amount": r.coverage_amount,
                        "coverage_period": r.coverage_period or "",
                        "monthly_budget": r.monthly_budget,
                        "status": r.status,
                        "created_at": str(r.created_at) if r.created_at else None,
                    }
                    for r in items
                ],
            },
        }