"""兴邦金租智能评审助手 - 统一API（7 个助手完整版） 基于 2026.8.18 会议痛点的完整解决方案： 1. 立项前置风控助手 🔍 - 业务经理风险自查 2. 中诚信智能填报助手 📊 - 激活十几万投资的评级模板 3. 空窗期动态监控助手 🚨 - 17天空窗期风险重扫 4. 保单条款合规核验助手 🛡️ - OCR+条款解析 5. 评审报告辅助生成助手 📄 - 看资料+写报告提效 6. 主审员智能分配助手 🎯 - 多维度负荷优化 7. AI预审过滤+加速助手 🤖 - 团队经验预判+加速 """ import asyncio import json from datetime import datetime from typing import AsyncGenerator, Dict, List, Optional from fastapi import APIRouter, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field # 导入数据层 from data_layer.erp_connector import get_erp_connector from data_layer.history_store import get_history_store from data_layer.reviewer_profile import get_reviewer_store from data_layer.team_experience import get_team_experience_store router = APIRouter(prefix="/api/xingbang/unified", tags=["兴邦智能评审统一接口"]) # ============ Pydantic 模型 ============ class ProjectBasicInfo(BaseModel): """项目基础信息""" project_id: str = "" project_name: str client_name: str industry: str credit_amount: float # 万元 lease_object: str = "" lease_value: float = 0 # 万元 class FinancialData(BaseModel): """财务数据""" registered_capital: float = 0 established_years: int = 0 asset_liability_ratio: float = 0 current_ratio: float = 0 quick_ratio: float = 0 net_profit_2024: float = 0 net_profit_2025: float = 0 revenue_growth_rate: float = 0 class CreditReport(BaseModel): """征信报告""" overdue_count: int = 0 query_count_recent_6m: int = 0 credit_limit: float = 0 enforcement_records: int = 0 class InsurancePolicyData(BaseModel): """保单数据""" policy_number: str insurance_company: str policy_type: str coverage_amount: float exclusions: List[str] = [] policy_pdf_path: str = "" # ============ 助手 1：立项前置风控 ============ class Assistant1Request(BaseModel): """立项前置风控请求""" project: ProjectBasicInfo financial: FinancialData credit: CreditReport @router.post("/assistant1/risk-check") async def assistant1_risk_check(request: Assistant1Request): """助手1：立项前置风控自查（只给风险点，不给结论）""" async def generate(): try: # 获取团队经验 team_exp = get_team_experience_store() thresholds = team_exp.get_approval_thresholds() risk_patterns = team_exp.get_risk_patterns(request.project.industry) red_flags = team_exp.get_industry_red_flags(request.project.industry) yield f"data: {json.dumps({'type': 'start'}, ensure_ascii=False)}\n\n" # 生成 Markdown 报告 content = f"""# 立项前置风险自查清单 ## 📋 项目基本信息 - **项目名称**：{request.project.project_name} - **客户名称**：{request.project.client_name} - **行业**：{request.project.industry} - **授信金额**：{request.project.credit_amount} 万元 - **租赁物**：{request.project.lease_object or '未填写'} - **租赁物价值**：{request.project.lease_value} 万元 --- ## 🔍 风险自查结果 """ yield f"data: {json.dumps({'type': 'content', 'data': content}, ensure_ascii=False)}\n\n" # 检查高风险项 high_risks = [] medium_risks = [] low_risks = [] # 1. 客户资质检查 if request.financial.registered_capital < 1000: high_risks.append("**注册资本不足**：注册资本 {:.0f} 万元 < 1000 万元（标准要求）\n - **建议**：考虑提高授信门槛或要求集团担保".format(request.financial.registered_capital)) if request.financial.established_years < 3: medium_risks.append("**成立时间较短**：成立 {} 年 < 3 年（标准要求）\n - **建议**：补充实控人其他企业经营情况".format(request.financial.established_years)) # 2. 财务指标检查 asset_liability = request.financial.asset_liability_ratio threshold_al = thresholds.get("资产负债率", {}) if asset_liability > threshold_al.get("critical", 0.8): high_risks.append(f"**资产负债率过高**：{asset_liability:.1%} > {threshold_al.get('critical', 0.8):.0%}（警戒线）\n - **团队共识**：{threshold_al.get('team_consensus', '需要额外担保')}") elif asset_liability > threshold_al.get("warning", 0.7): medium_risks.append(f"**资产负债率偏高**：{asset_liability:.1%}（位于 {threshold_al.get('warning', 0.7):.0%}-{threshold_al.get('critical', 0.8):.0%} 区间）\n - **建议**：补充负债结构说明，关注短期偿债压力") else: low_risks.append(f"✅ **资产负债率**：{asset_liability:.1%}（良好）") current_ratio = request.financial.current_ratio threshold_cr = thresholds.get("流动比率", {}) if current_ratio < threshold_cr.get("critical", 1.0): high_risks.append(f"**流动比率不足**：{current_ratio:.2f} < {threshold_cr.get('critical', 1.0):.2f}（红线）\n - **团队共识**：{threshold_cr.get('team_consensus', '短期偿债能力堪忧')}") elif current_ratio < threshold_cr.get("warning", 1.2): medium_risks.append(f"**流动比率偏低**：{current_ratio:.2f}\n - **建议**：关注应收账款周转情况") else: low_risks.append(f"✅ **流动比率**：{current_ratio:.2f}（符合标准）") # 3. 征信检查 if request.credit.overdue_count > 0: high_risks.append(f"**征信逾期记录**：近 2 年有 {request.credit.overdue_count} 次逾期\n - **建议**：要求客户提供逾期说明及还款证明") else: low_risks.append("✅ **征信记录**：近 2 年无逾期（良好）") if request.credit.enforcement_records > 0: high_risks.append(f"**被执行记录**：存在 {request.credit.enforcement_records} 条被执行记录\n - **重要**：评审大概率退回，需先解决执行案件") else: low_risks.append("✅ **被执行记录**：无（良好）") if request.credit.query_count_recent_6m > 10: medium_risks.append(f"**征信查询频繁**：近 6 个月查询 {request.credit.query_count_recent_6m} 次\n - **建议**：了解客户融资需求是否过于迫切") # 4. 租赁物检查 if request.project.lease_value > 0: coverage_ratio = request.project.lease_value / request.project.credit_amount if coverage_ratio < 1.2: high_risks.append(f"**租赁物价值不足**：租赁物价值 {request.project.lease_value} 万元，覆盖率 {coverage_ratio:.1%} < 120%（标准要求）\n - **建议**：补充其他担保措施") else: low_risks.append(f"✅ **租赁物覆盖率**：{coverage_ratio:.1%}（符合要求）") # 5. 行业红线检查 if red_flags: content_flags = f"\n### ⚠️ {request.project.industry} 行业常见风险点\n\n" for flag in red_flags: content_flags += f"- {flag}\n" content_flags += "\n**建议**：核实客户是否触及以上风险点\n\n" yield f"data: {json.dumps({'type': 'content', 'data': content_flags}, ensure_ascii=False)}\n\n" # 输出风险项 if high_risks: content_high = "### 🔴 高风险项（必须处理，否则评审大概率退回）\n\n" for i, risk in enumerate(high_risks, 1): content_high += f"{i}. {risk}\n\n" yield f"data: {json.dumps({'type': 'content', 'data': content_high}, ensure_ascii=False)}\n\n" if medium_risks: content_medium = "### 🟡 中风险项（建议补充材料或说明）\n\n" for i, risk in enumerate(medium_risks, 1): content_medium += f"{i}. {risk}\n\n" yield f"data: {json.dumps({'type': 'content', 'data': content_medium}, ensure_ascii=False)}\n\n" if low_risks: content_low = "### 🟢 低风险项（符合要求）\n\n" for risk in low_risks: content_low += f"- {risk}\n" content_low += "\n" yield f"data: {json.dumps({'type': 'content', 'data': content_low}, ensure_ascii=False)}\n\n" # 整体评估 if len(high_risks) == 0 and len(medium_risks) == 0: suggestion = "✅ **直接提交**：项目风险可控，可直接提交评审" elif len(high_risks) == 0: suggestion = "🟡 **补充材料后提交**：建议先处理中风险项，提高通过率" else: suggestion = "🔴 **重大调整后提交**：存在高风险项，建议先与客户沟通调整方案" content_final = f"""--- ## 💡 整体评估 {suggestion} **说明**：本清单基于团队历史评审经验生成，仅供参考，不代表评审结论。最终是否通过由主审员独立判断。 """ yield f"data: {json.dumps({'type': 'content', 'data': content_final}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n" except Exception as e: yield f"data: {json.dumps({'type': 'error', 'message': str(e)}, ensure_ascii=False)}\n\n" return StreamingResponse(generate(), media_type="text/event-stream") # ============ 助手 6：主审员智能分配 ============ class Assistant6Request(BaseModel): """主审员智能分配请求""" project_industry: str project_amount: float # 万元 project_complexity: int = Field(3, ge=1, le=5, description="项目复杂度 1-5") @router.post("/assistant6/allocate") async def assistant6_allocate(request: Assistant6Request): """助手6：主审员智能分配（多维度评分）""" try: reviewer_store = get_reviewer_store() # 获取推荐列表 recommendations = reviewer_store.get_recommended_reviewers( project_industry=request.project_industry, project_complexity=request.project_complexity, top_n=3 ) # 获取所有主审员当前状态 all_profiles = reviewer_store.get_all_profiles() reviewer_status = [ { "name": name, "current_projects": profile["current_projects"], "specialty": profile["specialty"], "workload_capacity": profile["workload_capacity"], "status": "🔴 超载" if profile["current_projects"] >= profile["workload_capacity"] else ("🟡 偏高" if profile["current_projects"] >= profile["workload_capacity"] * 0.8 else "🟢 正常") } for name, profile in all_profiles.items() ] return { "reviewers": reviewer_status, "recommendations": recommendations, "project_info": { "industry": request.project_industry, "amount": request.project_amount, "complexity": request.project_complexity, }, "algorithm_version": "v1.0_multidim_scoring", } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # ============ 助手 7：AI预审过滤+加速 ============ class Assistant7Request(BaseModel): """AI预审请求""" project: ProjectBasicInfo financial: FinancialData credit: CreditReport @router.post("/assistant7/pre-screen") async def assistant7_pre_screen(request: Assistant7Request): """助手7：AI预审过滤（基于团队经验，快速判断是否值得进入正式评审）""" async def generate(): try: yield f"data: {json.dumps({'type': 'start'}, ensure_ascii=False)}\n\n" # 获取团队经验和历史案例 team_exp = get_team_experience_store() history_store = get_history_store() thresholds = team_exp.get_approval_thresholds() rejection_reasons = team_exp.get_common_rejection_reasons(top_n=5) # 相似案例检索 similar_cases = history_store.search_similar_cases( industry=request.project.industry, credit_amount=request.project.credit_amount, limit=3 ) content = f"""# AI预审报告（团队经验模式） ## 📊 项目信息 - **项目名称**：{request.project.project_name} - **客户名称**：{request.project.client_name} - **行业**：{request.project.industry} - **授信金额**：{request.project.credit_amount} 万元 --- ## 🤖 AI预审判断（基于团队 {team_exp.get_metadata().get('total_cases_analyzed', 0)} 个历史案例） """ yield f"data: {json.dumps({'type': 'content', 'data': content}, ensure_ascii=False)}\n\n" # 快速红线检查 red_line_hit = [] warning_items = [] # 1. 财务红线 if request.financial.asset_liability_ratio > thresholds.get("资产负债率", {}).get("critical", 0.8): red_line_hit.append("资产负债率超过红线") if request.financial.current_ratio < thresholds.get("流动比率", {}).get("critical", 1.0): red_line_hit.append("流动比率低于红线") # 2. 征信红线 if request.credit.enforcement_records > 0: red_line_hit.append(f"存在 {request.credit.enforcement_records} 条被执行记录") if request.credit.overdue_count > 2: red_line_hit.append(f"征信逾期次数过多（{request.credit.overdue_count} 次）") # 3. 警告项 if request.financial.revenue_growth_rate < 0: warning_items.append("营收负增长") if request.credit.query_count_recent_6m > 10: warning_items.append("征信查询频繁") # AI预审结论 if red_line_hit: conclusion = "🔴 **不建议进入正式评审**" reason_text = "\n".join([f"- {item}" for item in red_line_hit]) content_conclusion = f"""### {conclusion} **触及红线项**： {reason_text} **建议**：与客户沟通调整方案，或直接退回立项申请，避免浪费评审资源。 """ elif len(warning_items) >= 2: conclusion = "🟡 **建议谨慎评审**" reason_text = "\n".join([f"- {item}" for item in warning_items]) content_conclusion = f"""### {conclusion} **警告项**： {reason_text} **建议**：可以进入正式评审，但需重点关注以上风险点，建议增加尽调深度。 """ else: conclusion = "🟢 **建议进入正式评审**" content_conclusion = f"""### {conclusion} **初步判断**：项目风险可控，符合基本准入标准，建议正常推进评审流程。 """ yield f"data: {json.dumps({'type': 'content', 'data': content_conclusion}, ensure_ascii=False)}\n\n" # 相似案例参考 if similar_cases: content_cases = "\n---\n\n## 📚 相似案例参考\n\n" for i, case in enumerate(similar_cases, 1): result_emoji = "✅" if case.get("review_result") == "approved" else "❌" content_cases += f"""### 案例 {i}：{case.get('project_name', '未知项目')} - **行业**：{case.get('industry', '未知')} - **金额**：{case.get('credit_amount', 0)} 万元 - **评审结果**：{result_emoji} {case.get('review_result', '未知')} - **关键风险点**：{', '.join([r.get('type', '未知') for r in case.get('identified_risks', [])])} """ yield f"data: {json.dumps({'type': 'content', 'data': content_cases}, ensure_ascii=False)}\n\n" # 高频退回原因提示 content_tips = "\n---\n\n## 💡 团队经验提示（高频退回原因）\n\n" for reason in rejection_reasons: content_tips += f"- **{reason['reason']}**（{reason['count']} 次） → {reason['solution']}\n" yield f"data: {json.dumps({'type': 'content', 'data': content_tips}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n" except Exception as e: yield f"data: {json.dumps({'type': 'error', 'message': str(e)}, ensure_ascii=False)}\n\n" return StreamingResponse(generate(), media_type="text/event-stream") # ============ 数据管理接口 ============ @router.get("/data/reviewers") async def get_all_reviewers(): """获取所有主审员画像""" reviewer_store = get_reviewer_store() return reviewer_store.get_all_profiles() @router.get("/data/team-experience") async def get_team_experience(): """获取团队集体经验""" team_exp = get_team_experience_store() return { "approval_thresholds": team_exp.get_approval_thresholds(), "common_rejections": team_exp.get_common_rejection_reasons(), "metadata": team_exp.get_metadata(), } @router.get("/data/history-stats") async def get_history_statistics(): """获取历史案例统计""" history_store = get_history_store() return history_store.get_statistics()