"""评审提效工作台 API 帮助兴邦金租的项目评审流程提效： - 评审周期从16-17个工作日压缩 - 主审员负荷优化（每人3-7个项目排队） - 看资料/算测算/写报告三大块耗时降低 """ import json from typing import Any, AsyncGenerator, Optional, List, Dict from fastapi import APIRouter, Query from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field # 导入历史数据模块 from kaiju_history_data import ( get_all_projects, get_projects_filtered, get_reviewer_stats, ) router = APIRouter(prefix="/api/review-assist", tags=["评审提效工作台"]) # ============ Pydantic 模型 ============ class ReviewerInfo(BaseModel): """主审员信息""" name: str current_projects: int # 在手项目数 specialty: str # 专长领域 pipeline: Optional[List[int]] = None # 在手项目复杂度列表（可选，用于精准负荷估算） estimated_finish_days: Optional[int] = None # 预计完成天数（可选） class LoadBalanceRequest(BaseModel): """评审负荷与分配建议请求""" reviewers: list[ReviewerInfo] new_project_industry: str = "" # 新项目行业 new_project_amount: float = 0 # 新项目金额（万元） class DueDiligenceRequest(BaseModel): """尽调资料预处理请求""" project_name: str client_name: str industry: str financial_data: dict # 财务数据 credit_report: dict # 征信报告 guarantee_info: dict # 担保信息 external_risks: dict # 外部风险数据 class RentCalculationRequest(BaseModel): """租金测算辅助校验请求""" credit_amount: float # 授信金额（万元） lease_term: int # 租赁期限（月） interest_rate: float # 年利率（%） monthly_cashflow: float # 月现金流（万元） project_type: str = "对公" # 对公/零售 class ReportDraftRequest(BaseModel): """评审报告初稿生成请求""" intelligence_card: str # 情报卡内容 calculation_result: str # 测算结果 project_name: str client_name: str class PreScreeningRequest(BaseModel): """立项前置风险筛查请求""" project_name: str client_name: str industry: str document_checklist: list[str] # 立项资料清单 financial_data: dict similar_cases: list[dict] = [] # 历史相似案例（可选） class AllocateRequest(BaseModel): """智能分配请求（数据驱动）""" reviewers: List[ReviewerInfo] new_project_industry: str new_project_amount: float # 万元 new_project_complexity: int = Field(3, ge=1, le=5) # 1-5，默认3 class RecommendationItem(BaseModel): """单个推荐结果""" reviewer: str score: float # 综合评分 reasons: List[str] # 推荐理由 est_complete_days: int # 预计完成天数（从现在算起） industry_match_score: float # 行业匹配分 load_score: float # 负荷分 quality_score: float # 质量分 class AllocateResponse(BaseModel): """智能分配响应""" reviewers: List[Dict] # 主审员当前状态列表 recommendation: List[RecommendationItem] # Top-N推荐 balance_before: float # 分配前负荷方差 balance_after: float # 分配后负荷方差（假设分配给Top1） algorithm_version: str # 算法版本标识 # ============ 示例数据（供前端使用）============ SAMPLE_DATA = { "load_balance": { "reviewers": [ {"name": "李明", "current_projects": 5, "specialty": "制造业"}, {"name": "王芳", "current_projects": 3, "specialty": "新能源"}, {"name": "张强", "current_projects": 7, "specialty": "交通运输"}, {"name": "刘洋", "current_projects": 4, "specialty": "医疗健康"}, ], "new_project_industry": "光伏设备制造", "new_project_amount": 5000 }, "due_diligence": { "project_name": "阳光光伏设备融资租赁项目", "client_name": "阳光光伏设备有限公司", "industry": "光伏设备制造", "financial_data": { "registered_capital": 8000, "established_years": 6, "asset_liability_ratio": 0.58, "current_ratio": 1.45, "net_profit_2024": 1200, "net_profit_2025": 1450, "revenue_growth_rate": 0.18 }, "credit_report": { "overdue_count": 0, "query_count_recent_6m": 4, "credit_limit": 20000, "enforcement_records": 0 }, "guarantee_info": { "type": "法人连带担保+设备抵押", "guarantor": "阳光集团有限公司", "guarantor_net_assets": 50000 }, "external_risks": { "litigation_records": [], "negative_news": 0, "industry_policy": "国家支持光伏产业发展" } }, "rent_calculation": { "credit_amount": 5000, "lease_term": 36, "interest_rate": 5.8, "monthly_cashflow": 85, "project_type": "对公" }, "report_draft": { "intelligence_card": "# 客户情报卡\n\n## 基本信息\n- 客户名称：阳光光伏设备有限公司\n- 行业：光伏设备制造\n\n## 财务状况\n- 资产负债率：58%（良好）\n- 流动比率：1.45（符合标准）\n\n## 风险评估\n🟢 低风险：财务指标优良，行业前景好", "calculation_result": "# 租金测算结果\n\n- 月租金：152.8万元\n- 覆盖率：180%（远超120%标准）\n- 风险等级：低", "project_name": "阳光光伏设备融资租赁项目", "client_name": "阳光光伏设备有限公司" }, "pre_screening": { "project_name": "阳光光伏设备融资租赁项目", "client_name": "阳光光伏设备有限公司", "industry": "光伏设备制造", "document_checklist": [ "营业执照", "财务报表（近3年）", "征信报告", "租赁物清单", "担保函", "法人身份证", "银行流水（近6个月）" ], "financial_data": { "registered_capital": 8000, "asset_liability_ratio": 0.58, "net_profit_2025": 1450 }, "similar_cases": [ { "project_name": "某光伏企业融资租赁（2025年）", "rejection_reason": "担保方资产不足，要求补充第二担保人", "lesson": "光伏行业波动大，需要双重担保" } ] } } # ============ 辅助1：评审负荷与分配建议 ============ async def _generate_load_balance(request: LoadBalanceRequest) -> AsyncGenerator[str, None]: """生成评审负荷与分配建议""" # 计算平均负荷 total_projects = sum(r.current_projects for r in request.reviewers) avg_load = total_projects / len(request.reviewers) if request.reviewers else 0 # 排序：按当前项目数升序 sorted_reviewers = sorted(request.reviewers, key=lambda r: r.current_projects) # 生成 Markdown 内容 content = f"""# 评审负荷与分配建议 ## 📊 当前负荷看板 | 主审员 | 在手项目数 | 专长领域 | 负荷状态 | |--------|-----------|---------|---------| """ for r in request.reviewers: load_status = "🔴 超载" if r.current_projects >= 6 else ("🟡 偏高" if r.current_projects >= 4 else "🟢 正常") content += f"| {r.name} | {r.current_projects} | {r.specialty} | {load_status} |\n" content += f"\n**平均负荷**：{avg_load:.1f} 个项目/人\n\n" yield content # 新项目推荐 if request.new_project_industry and request.new_project_amount: content = f"""## 🎯 新项目分配推荐 **项目信息**：{request.new_project_industry} | 授信 {request.new_project_amount} 万元 ### Top 3 推荐主审员 """ # 推荐逻辑：专长匹配 > 负荷最低 recommendations = [] for r in sorted_reviewers: match_score = 10 if request.new_project_industry in r.specialty or r.specialty in request.new_project_industry else 0 load_score = (10 - r.current_projects) # 项目越少分数越高 total_score = match_score + load_score recommendations.append((r, match_score, load_score, total_score)) recommendations.sort(key=lambda x: x[3], reverse=True) for i, (r, match, load, total) in enumerate(recommendations[:3], 1): reason = [] if match > 0: reason.append(f"专长匹配（{r.specialty}）") if r.current_projects < avg_load: reason.append("负荷较低") reason_str = "、".join(reason) if reason else "负荷可承接" content += f"**{i}. {r.name}** — {reason_str}\n" content += f" - 当前负荷：{r.current_projects} 个项目\n" content += f" - 专长：{r.specialty}\n" content += f" - 综合评分：{total} 分\n\n" yield content # 瓶颈预警 bottleneck_reviewers = [r for r in request.reviewers if r.current_projects >= 6] if bottleneck_reviewers: content = f"""## ⚠️ 瓶颈预警 以下主审员负荷已达瓶颈（≥6个项目），建议暂停新项目分配： """ for r in bottleneck_reviewers: content += f"- **{r.name}**：{r.current_projects} 个项目（专长：{r.specialty}）\n" content += f"\n**建议措施**：\n" content += f"1. 优先分配给负荷较低的主审员（{sorted_reviewers[0].name}、{sorted_reviewers[1].name if len(sorted_reviewers) > 1 else ''}）\n" content += f"2. 考虑延长评审周期或增加评审人力\n" content += f"3. 对简单项目启用快速评审通道\n" yield content else: content = """## ✅ 无瓶颈预警 当前所有主审员负荷均在合理范围内，可正常分配新项目。 """ yield content @router.post("/load-balance/stream") async def load_balance_stream(request: LoadBalanceRequest): """评审负荷与分配建议（流式返回）""" async def event_generator(): try: yield f"data: {json.dumps({'type': 'start', 'message': '正在分析评审负荷...'}, ensure_ascii=False)}\n\n" async for content in _generate_load_balance(request): yield f"data: {json.dumps({'type': 'content', 'data': content}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'done', 'message': '分析完成'}, 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(event_generator(), media_type="text/event-stream") # ============ 辅助2：尽调资料预处理·情报卡 ============ async def _generate_intelligence_card(request: DueDiligenceRequest) -> AsyncGenerator[str, None]: """生成客户情报卡""" content = f"""# 📋 客户情报卡 ## 基本信息 - **项目名称**：{request.project_name} - **客户名称**：{request.client_name} - **所属行业**：{request.industry} --- ## 💰 财务状况分析 """ yield content # 财务指标分析 fd = request.financial_data content = "" # 注册资本 reg_capital = fd.get("registered_capital", 0) if reg_capital >= 1000: content += f"- **注册资本**：{reg_capital} 万元 🟢 符合标准（≥1000万）\n" else: content += f"- **注册资本**：{reg_capital} 万元 🔴 不足（要求≥1000万）\n" # 资产负债率 debt_ratio = fd.get("asset_liability_ratio", 0) if debt_ratio <= 0.70: content += f"- **资产负债率**：{debt_ratio*100:.1f}% 🟢 良好（≤70%）\n" else: content += f"- **资产负债率**：{debt_ratio*100:.1f}% 🔴 偏高（标准≤70%）\n" # 流动比率 current_ratio = fd.get("current_ratio", 0) if current_ratio >= 1.2: content += f"- **流动比率**：{current_ratio:.2f} 🟢 健康（≥1.2）\n" else: content += f"- **流动比率**：{current_ratio:.2f} 🟡 偏低（标准≥1.2）\n" # 净利润 profit_2024 = fd.get("net_profit_2024", 0) profit_2025 = fd.get("net_profit_2025", 0) if profit_2024 > 0 and profit_2025 > 0: growth = ((profit_2025 - profit_2024) / profit_2024 * 100) if profit_2024 else 0 content += f"- **净利润**：2024年 {profit_2024} 万元，2025年 {profit_2025} 万元 🟢 连续为正，增长 {growth:.1f}%\n" else: content += f"- **净利润**：2024年 {profit_2024} 万元，2025年 {profit_2025} 万元 🔴 存在亏损\n" yield content # 征信报告分析 cr = request.credit_report content = f""" --- ## 🔍 征信记录分析 """ overdue = cr.get("overdue_count", 0) if overdue == 0: content += f"- **逾期记录**：无 🟢 信用良好\n" else: content += f"- **逾期记录**：{overdue} 次 🔴 存在逾期风险\n" enforcement = cr.get("enforcement_records", 0) if enforcement == 0: content += f"- **被执行记录**：无 🟢\n" else: content += f"- **被执行记录**：{enforcement} 条 🔴 严重风险\n" query_count = cr.get("query_count_recent_6m", 0) if query_count <= 5: content += f"- **近6个月查询次数**：{query_count} 次 🟢 正常\n" else: content += f"- **近6个月查询次数**：{query_count} 次 🟡 频繁（可能存在资金压力）\n" yield content # 担保信息 gi = request.guarantee_info content = f""" --- ## 🛡️ 担保信息 - **担保方式**：{gi.get('type', '未提供')} - **担保人**：{gi.get('guarantor', '未提供')} - **担保人净资产**：{gi.get('guarantor_net_assets', 0)} 万元 """ yield content # 外部风险 er = request.external_risks content = f""" --- ## ⚠️ 外部风险评估 - **涉诉记录**：{len(er.get('litigation_records', []))} 条 - **负面舆情**：{er.get('negative_news', 0)} 条 - **行业政策**：{er.get('industry_policy', '未知')} --- ## 📝 数据完整性检查 """ # 检查缺失项 missing = [] if not fd: missing.append("财务数据缺失") if not cr: missing.append("征信报告缺失") if not gi or not gi.get('type'): missing.append("担保信息不完整") if missing: content += "### 🔴 缺失项\n" for item in missing: content += f"- {item}\n" content += "\n" else: content += "### ✅ 无缺失项\n\n" # 检查矛盾点 contradictions = [] if fd.get("net_profit_2025", 0) > 0 and fd.get("asset_liability_ratio", 0) > 0.80: contradictions.append("盈利但负债率过高，需核实资产结构") if cr.get("overdue_count", 0) > 0 and cr.get("query_count_recent_6m", 0) > 10: contradictions.append("征信逾期且查询频繁，存在资金链风险") if contradictions: content += "### 🟡 矛盾点\n" for item in contradictions: content += f"- {item}\n" else: content += "### ✅ 无明显矛盾\n" yield content @router.post("/due-diligence/stream") async def due_diligence_stream(request: DueDiligenceRequest): """尽调资料预处理·情报卡（流式返回）""" async def event_generator(): try: yield f"data: {json.dumps({'type': 'start', 'message': '正在生成情报卡...'}, ensure_ascii=False)}\n\n" async for content in _generate_intelligence_card(request): yield f"data: {json.dumps({'type': 'content', 'data': content}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'done', 'message': '情报卡生成完成'}, 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(event_generator(), media_type="text/event-stream") # ============ 辅助3：租金测算辅助校验 ============ async def _generate_calculation_check(request: RentCalculationRequest) -> AsyncGenerator[str, None]: """生成租金测算校验报告""" content = f"""# 💹 租金测算辅助校验 ## 输入参数 - **授信金额**：{request.credit_amount} 万元 - **租赁期限**：{request.lease_term} 个月 - **年利率**：{request.interest_rate}% - **月现金流**：{request.monthly_cashflow} 万元 - **项目类型**：{request.project_type} --- ## 📊 二次核算结果 """ yield content # 计算月租金（等额本息） r = request.interest_rate / 100 / 12 # 月利率 n = request.lease_term P = request.credit_amount # 等额本息公式：A = P * r * (1+r)^n / ((1+r)^n - 1) monthly_rent = P * r * ((1 + r) ** n) / (((1 + r) ** n) - 1) total_payment = monthly_rent * n total_interest = total_payment - P content = f"""- **月租金**：{monthly_rent:.2f} 万元 - **总还款额**：{total_payment:.2f} 万元 - **总利息**：{total_interest:.2f} 万元 - **实际年化利率**：{request.interest_rate}% --- ## ✅ 覆盖率验证 """ yield content # 覆盖率 = 月现金流 / 月租金 coverage_ratio = (request.monthly_cashflow / monthly_rent * 100) if monthly_rent > 0 else 0 required_ratio = 120 if request.project_type == "对公" else 150 # 对公≥120%，零售≥150% if coverage_ratio >= required_ratio: content = f"""- **现金流覆盖率**：{coverage_ratio:.1f}% 🟢 **符合标准**（{request.project_type}要求 ≥{required_ratio}%） - **月现金流**：{request.monthly_cashflow} 万元 - **月租金**：{monthly_rent:.2f} 万元 - **富余现金流**：{request.monthly_cashflow - monthly_rent:.2f} 万元 **结论**：现金流覆盖充足，还款能力较强。 """ else: content = f"""- **现金流覆盖率**：{coverage_ratio:.1f}% 🔴 **不符合标准**（{request.project_type}要求 ≥{required_ratio}%） - **月现金流**：{request.monthly_cashflow} 万元 - **月租金**：{monthly_rent:.2f} 万元 - **缺口**：{monthly_rent - request.monthly_cashflow:.2f} 万元 **结论**：现金流覆盖不足，存在还款风险。 """ yield content # 敏感性分析 content = f"""--- ## 📉 敏感性分析 ### 利率变动影响 | 年利率 | 月租金（万元） | 覆盖率 | 风险评估 | |--------|---------------|--------|---------| """ for rate_adj in [-1.0, -0.5, 0, 0.5, 1.0]: test_rate = (request.interest_rate + rate_adj) / 100 / 12 test_rent = P * test_rate * ((1 + test_rate) ** n) / (((1 + test_rate) ** n) - 1) if test_rate > 0 else P / n test_coverage = (request.monthly_cashflow / test_rent * 100) if test_rent > 0 else 0 risk = "🟢 安全" if test_coverage >= required_ratio else "🔴 风险" content += f"| {request.interest_rate + rate_adj:.1f}% | {test_rent:.2f} | {test_coverage:.1f}% | {risk} |\n" content += "\n### 现金流变动影响\n\n| 现金流变动 | 覆盖率 | 风险评估 |\n|-----------|--------|----------|\n" for cf_pct in [-20, -10, 0, 10, 20]: test_cf = request.monthly_cashflow * (1 + cf_pct / 100) test_coverage = (test_cf / monthly_rent * 100) risk = "🟢 安全" if test_coverage >= required_ratio else "🔴 风险" content += f"| {cf_pct:+d}% ({test_cf:.2f}万) | {test_coverage:.1f}% | {risk} |\n" yield content # 风险提示 content = f""" --- ## ⚠️ 风险提示 """ if coverage_ratio < required_ratio: content += f"1. 🔴 **高风险**：现金流覆盖率不足{required_ratio}%，还款压力大\n" elif coverage_ratio < required_ratio * 1.2: content += f"1. 🟡 **中风险**：现金流覆盖率接近临界值，建议增加担保措施\n" else: content += f"1. 🟢 **低风险**：现金流覆盖率充足\n" if request.lease_term > 36: content += f"2. 🟡 租赁期限较长（{request.lease_term}个月），注意长期经营风险\n" if request.interest_rate > 7.0: content += f"3. 🟡 利率较高（{request.interest_rate}%），客户成本压力大\n" content += f"\n**评审建议**：\n" if coverage_ratio >= required_ratio * 1.5: content += "- ✅ 测算合理，可通过财务审核\n" elif coverage_ratio >= required_ratio: content += "- ✅ 基本符合要求，建议关注现金流稳定性\n" else: content += "- ❌ 不符合标准，建议：\n" content += " - 降低授信金额或延长期限\n" content += " - 要求客户补充收入证明\n" content += " - 增加连带担保人\n" yield content @router.post("/rent-calculation/stream") async def rent_calculation_stream(request: RentCalculationRequest): """租金测算辅助校验（流式返回）""" async def event_generator(): try: yield f"data: {json.dumps({'type': 'start', 'message': '正在校验测算结果...'}, ensure_ascii=False)}\n\n" async for content in _generate_calculation_check(request): yield f"data: {json.dumps({'type': 'content', 'data': content}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'done', 'message': '校验完成'}, 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(event_generator(), media_type="text/event-stream") # ============ 辅助4：评审报告初稿生成 ============ async def _generate_report_draft(request: ReportDraftRequest) -> AsyncGenerator[str, None]: """生成评审报告初稿""" content = f"""# 📄 评审报告初稿 > ⚠️ **本报告为AI辅助生成的初稿，所有结论性意见需评审员人工复核修订后方可使用。** --- ## 一、项目基本信息 - **项目名称**：{request.project_name} - **申请人**：{request.client_name} - **报告生成时间**：{__import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M')} - **评审员**：【需填写】 --- ## 二、客户基本情况 {request.intelligence_card if request.intelligence_card else '【需评审员补充客户情报卡内容】'} --- ## 三、财务测算分析 {request.calculation_result if request.calculation_result else '【需评审员补充测算结果】'} --- ## 四、风险分析与控制措施 """ yield content # 从情报卡中提取风险点 content = """### 4.1 主要风险点 """ # 简单解析情报卡中的风险标记 if "🔴" in request.intelligence_card: content += "- 🔴 **高风险项**：【需评审员根据情报卡中🔴标记项补充具体分析】\n" if "🟡" in request.intelligence_card: content += "- 🟡 **中风险项**：【需评审员根据情报卡中🟡标记项补充具体分析】\n" if "🔴" not in request.intelligence_card and "🟡" not in request.intelligence_card: content += "- 🟢 **风险较低**：客户资质良好，财务状况健康\n" content += """ ### 4.2 风险控制措施 【需评审员补充】建议包括但不限于： 1. 担保措施（保证金、抵押、第三方担保） 2. 保险要求（险种、保额、第一受益人） 3. 监管措施（资金监管、租赁物监管） 4. 特殊条款（提前还款、违约处理） --- ## 五、评审意见 ### 5.1 综合评估 【需评审员填写】基于以上分析，对项目整体风险进行综合评估。 ### 5.2 评审结论 - [ ] 同意通过 - [ ] 附条件通过（条件：__________） - [ ] 不予通过 **评审员签字**：__________ **日期**：__________ --- ## 六、附件清单 1. 客户营业执照 2. 近三年财务报表 3. 征信报告 4. 担保函 5. 租赁物清单及评估报告 6. 【其他附件】 --- > 📌 **修订说明**：本报告由评审提效工作台自动生成，评审员需重点关注以下部分： > - [ ] 客户基本情况的准确性 > - [ ] 财务测算的合理性 > - [ ] 风险控制措施的完整性 > - [ ] 评审结论的专业性 """ yield content @router.post("/report-draft/stream") async def report_draft_stream(request: ReportDraftRequest): """评审报告初稿生成（流式返回）""" async def event_generator(): try: yield f"data: {json.dumps({'type': 'start', 'message': '正在生成报告初稿...'}, ensure_ascii=False)}\n\n" async for content in _generate_report_draft(request): yield f"data: {json.dumps({'type': 'content', 'data': content}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'done', 'message': '初稿生成完成'}, 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(event_generator(), media_type="text/event-stream") # ============ 辅助5：立项前置风险筛查 ============ async def _generate_pre_screening(request: PreScreeningRequest) -> AsyncGenerator[str, None]: """生成立项前置风险筛查报告""" content = f"""# 🔍 立项前置风险筛查 > **重要**：本筛查仅识别风险点，不给"通过/不通过"结论，最终决策由评审员作出。 --- ## 项目信息 - **项目名称**：{request.project_name} - **客户名称**：{request.client_name} - **所属行业**：{request.industry} --- ## 📋 资料完整性检查 """ yield content # 标准资料清单 standard_docs = [ "营业执照", "财务报表（近3年）", "征信报告", "租赁物清单", "担保函", "法人身份证", "银行流水（近6个月）", "租赁物评估报告", "保险单" ] missing_docs = [doc for doc in standard_docs if doc not in request.document_checklist] content = "### ✅ 已提交资料\n\n" for doc in request.document_checklist: content += f"- {doc}\n" if missing_docs: content += "\n### 🔴 缺失资料\n\n" for doc in missing_docs: content += f"- {doc}\n" content += "\n**风险提示**：资料不完整可能导致评审周期延长或项目退回。\n" else: content += "\n### ✅ 资料齐全\n\n所有必要资料已提交。\n" yield content # 风险点列表 content = f""" --- ## ⚠️ 风险点识别 """ risk_points = [] # 财务风险 fd = request.financial_data if fd.get("registered_capital", 0) < 1000: risk_points.append({ "category": "资质风险", "level": "🔴", "description": f"注册资本不足1000万（实际：{fd.get('registered_capital', 0)}万）", "impact": "不符合准入门槛，评审退回风险高" }) if fd.get("asset_liability_ratio", 0) > 0.70: risk_points.append({ "category": "财务风险", "level": "🟡", "description": f"资产负债率超过70%（实际：{fd.get('asset_liability_ratio', 0)*100:.1f}%）", "impact": "偿债压力较大，可能需要增加担保" }) if fd.get("net_profit_2025", 0) <= 0: risk_points.append({ "category": "盈利风险", "level": "🔴", "description": "最近一年净利润为负或为零", "impact": "盈利能力不足，还款能力存疑" }) # 行业风险 high_risk_industries = ["房地产", "金融", "矿业", "钢铁", "煤炭"] if any(ind in request.industry for ind in high_risk_industries): risk_points.append({ "category": "行业风险", "level": "🟡", "description": f"{request.industry}属于政策敏感或波动较大行业", "impact": "需重点关注行业政策变化和经营稳定性" }) # 渲染风险点 if risk_points: for i, rp in enumerate(risk_points, 1): content += f"""### {rp['level']} 风险点 {i}：{rp['category']} **描述**：{rp['description']} **影响**：{rp['impact']} """ else: content += "### 🟢 未发现明显风险点\n\n根据现有资料，客户资质和财务状况基本符合准入标准。\n" yield content # 历史相似案例 if request.similar_cases: content = f""" --- ## 📚 历史相似案例参考 """ for i, case in enumerate(request.similar_cases, 1): content += f"""### 案例 {i}：{case.get('project_name', '未命名')} - **退回原因**：{case.get('rejection_reason', '未说明')} - **经验教训**：{case.get('lesson', '无')} """ yield content # 筛查建议 content = f""" --- ## 💡 筛查建议 """ if len([r for r in risk_points if r['level'] == '🔴']) > 0: content += """**建议**：存在高风险项，建议： 1. 要求客户补充资料或说明 2. 增加风险控制措施（如提高担保比例） 3. 谨慎评估是否继续推进 """ elif len([r for r in risk_points if r['level'] == '🟡']) > 0: content += """**建议**：存在中风险项，建议： 1. 在评审报告中重点说明 2. 设置附加条件（如追加担保、增加保险） 3. 密切监控项目执行 """ else: content += """**建议**：风险可控，可正常推进评审流程。 """ content += """ > 📌 **重要提醒**：本筛查不代表评审结论，所有风险点需评审员结合实际情况判断。 """ yield content @router.post("/pre-screening/stream") async def pre_screening_stream(request: PreScreeningRequest): """立项前置风险筛查（流式返回）""" async def event_generator(): try: yield f"data: {json.dumps({'type': 'start', 'message': '正在筛查风险点...'}, ensure_ascii=False)}\n\n" async for content in _generate_pre_screening(request): yield f"data: {json.dumps({'type': 'content', 'data': content}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': 'done', 'message': '筛查完成'}, 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(event_generator(), media_type="text/event-stream") # ============ 示例数据接口 ============ @router.get("/sample-data/{data_type}") async def get_sample_data(data_type: str): """获取示例数据""" if data_type in SAMPLE_DATA: return {"success": True, "data": SAMPLE_DATA[data_type]} else: return {"success": False, "message": "未找到示例数据"} # ============ 新增1：历史评审数据查询 ============ @router.get("/history") async def get_history( reviewer: Optional[str] = Query(None, description="主审员姓名"), industry: Optional[str] = Query(None, description="行业（模糊匹配）"), ): """查询历史评审记录（支持按主审员/行业过滤）""" try: projects = get_projects_filtered(reviewer=reviewer, industry=industry) return { "success": True, "count": len(projects), "data": projects } except Exception as e: return {"success": False, "message": str(e)} @router.get("/reviewer-stats") async def get_reviewer_statistics(): """获取所有主审员的统计画像""" try: stats = get_reviewer_stats() return { "success": True, "data": stats } except Exception as e: return {"success": False, "message": str(e)} # ============ 新增2：智能分配算法（数据驱动）============ def _calculate_smart_allocation(request: AllocateRequest) -> AllocateResponse: """智能分配算法核心逻辑 算法公式： score = w1 * 行业匹配度 + w2 * 空闲度 + w3 * 质量 + w4 * 均衡度 其中： - 行业匹配度：基于历史数据的行业专长质量分（0-100） - 空闲度：基于当前负荷的空闲程度（0-100，越高越空闲） - 质量：历史平均质量分（0-100） - 均衡度：分配后是否拉平工作负担（0-100） 权重配置：w1=0.35, w2=0.35, w3=0.20, w4=0.10 """ # 获取历史统计画像 reviewer_stats = get_reviewer_stats() # 权重配置 W_INDUSTRY = 0.35 # 行业匹配 W_LOAD = 0.35 # 空闲度 W_QUALITY = 0.20 # 质量 W_BALANCE = 0.10 # 均衡度 # 计算当前平均负荷（用于均衡度计算） total_load = sum(r.current_projects for r in request.reviewers) avg_load = total_load / len(request.reviewers) if request.reviewers else 0 # 计算分配前负荷方差 loads = [r.current_projects for r in request.reviewers] balance_before = sum((load - avg_load) ** 2 for load in loads) / len(loads) if loads else 0 recommendations = [] for reviewer_info in request.reviewers: reviewer_name = reviewer_info.name # 获取该主审员的历史画像 stats = reviewer_stats.get(reviewer_name, {}) if not stats: # 如果没有历史数据，给默认值 stats = { "avg_actual_days": 10, "avg_quality_score": 80, "industry_expertise": {}, "total_projects": 0 } # 1. 行业匹配度（0-100） industry_expertise = stats.get("industry_expertise", {}) industry_match_score = 0.0 # 查找新项目行业在该主审员的历史记录中的表现 matched_industry = None for hist_industry, expertise_data in industry_expertise.items(): if request.new_project_industry in hist_industry or hist_industry in request.new_project_industry: matched_industry = hist_industry # 行业匹配分 = 该行业历史质量分 industry_match_score = expertise_data.get("avg_quality", 0) break # 如果没有直接匹配，但专长字段匹配，给60分基础分 if industry_match_score == 0 and ( request.new_project_industry in reviewer_info.specialty or reviewer_info.specialty in request.new_project_industry ): industry_match_score = 60.0 # 如果完全不匹配，给30分保底分（通用能力） if industry_match_score == 0: industry_match_score = 30.0 # 2. 空闲度（0-100，越高越空闲） # 基于当前项目数，考虑在手项目复杂度 if reviewer_info.pipeline and len(reviewer_info.pipeline) > 0: # 如果提供了在手项目复杂度，计算折算天数 avg_days_per_complexity = stats.get("avg_actual_days", 10) / 3 # 假设复杂度3的平均天数 estimated_workload_days = sum( complexity * avg_days_per_complexity for complexity in reviewer_info.pipeline ) elif reviewer_info.estimated_finish_days: # 如果直接提供了预计完成天数 estimated_workload_days = reviewer_info.estimated_finish_days else: # 简单按项目数估算 avg_days = stats.get("avg_actual_days", 10) estimated_workload_days = reviewer_info.current_projects * avg_days # 空闲度 = 100 - (当前负荷天数 / 最大合理负荷天数) * 100 # 假设最大合理负荷 = 6个复杂度3的项目 = 6 * 10天 = 60天 max_reasonable_workload = 60 load_score = max(0, 100 - (estimated_workload_days / max_reasonable_workload) * 100) # 3. 质量分（直接使用历史平均质量分） quality_score = stats.get("avg_quality_score", 80) # 4. 均衡度（分配给该主审员后，是否拉平负荷分布） # 如果当前负荷低于平均，分配给他能提高均衡度 current_load = reviewer_info.current_projects if current_load < avg_load: balance_score = 100 - abs(current_load + 1 - avg_load) / avg_load * 100 else: balance_score = max(0, 100 - abs(current_load + 1 - avg_load) / avg_load * 100) balance_score = max(0, min(100, balance_score)) # 综合评分 total_score = ( W_INDUSTRY * industry_match_score + W_LOAD * load_score + W_QUALITY * quality_score + W_BALANCE * balance_score ) # 预计完成天数（当前负荷 + 新项目） new_project_days = request.new_project_complexity * (stats.get("avg_actual_days", 10) / 3) est_complete_days = int(estimated_workload_days + new_project_days) # 生成推荐理由 reasons = [] if industry_match_score >= 85: reasons.append(f"✨ 行业专长突出：{matched_industry or reviewer_info.specialty}历史质量{industry_match_score:.0f}分") elif industry_match_score >= 70: reasons.append(f"✓ 行业经验丰富：{matched_industry or reviewer_info.specialty}") elif industry_match_score < 40: reasons.append(f"⚠ 行业匹配度较低") if load_score >= 70: reasons.append(f"✓ 当前负荷较轻：预计{estimated_workload_days:.0f}天可开工") elif load_score < 30: reasons.append(f"⚠ 当前负荷较重：预计{estimated_workload_days:.0f}天后才能开工") if quality_score >= 90: reasons.append(f"⭐ 历史质量优秀：平均{quality_score:.0f}分") elif quality_score >= 80: reasons.append(f"✓ 历史质量良好：平均{quality_score:.0f}分") if current_load < avg_load: reasons.append(f"✓ 有助于负荷均衡（当前{current_load}个 < 平均{avg_load:.1f}个）") if not reasons: reasons.append("基础推荐") recommendations.append( RecommendationItem( reviewer=reviewer_name, score=round(total_score, 2), reasons=reasons, est_complete_days=est_complete_days, industry_match_score=round(industry_match_score, 2), load_score=round(load_score, 2), quality_score=round(quality_score, 2), ) ) # 按综合评分排序 recommendations.sort(key=lambda x: x.score, reverse=True) # 计算分配后负荷方差（假设分配给Top1） if recommendations: top_reviewer_name = recommendations[0].reviewer loads_after = [ r.current_projects + 1 if r.name == top_reviewer_name else r.current_projects for r in request.reviewers ] avg_load_after = sum(loads_after) / len(loads_after) if loads_after else 0 balance_after = sum((load - avg_load_after) ** 2 for load in loads_after) / len(loads_after) if loads_after else 0 else: balance_after = balance_before # 生成主审员当前状态列表 reviewers_status = [] for r in request.reviewers: stats = reviewer_stats.get(r.name, {}) reviewers_status.append({ "name": r.name, "current_projects": r.current_projects, "specialty": r.specialty, "avg_quality": stats.get("avg_quality_score", 0), "total_history_projects": stats.get("total_projects", 0), }) return AllocateResponse( reviewers=reviewers_status, recommendation=recommendations, balance_before=round(balance_before, 2), balance_after=round(balance_after, 2), algorithm_version="v1.0-multi-dimensional" ) @router.post("/allocate") async def smart_allocate(request: AllocateRequest): """智能分配算法（JSON返回，数据驱动） 基于历史数据的多维度评分： - 行业专长匹配（历史该行业质量分） - 当前负荷空闲度（预计开工时间） - 历史质量表现 - 负荷均衡优化 返回Top-N推荐 + 详细理由 """ try: result = _calculate_smart_allocation(request) return { "success": True, "data": result.dict() } except Exception as e: return {"success": False, "message": str(e)}