"""主审员画像库 功能： 1. 存储主审员基础信息（专长、效率、质量评分） 2. 动态更新主审员负荷 3. 多维度评分（用于智能分配） """ import json from typing import Dict, List, Optional from pathlib import Path from datetime import datetime import logging logger = logging.getLogger(__name__) class ReviewerProfileStore: """主审员画像存储""" def __init__(self, data_dir: str = "./xingbang_data/reviewers"): self.data_dir = Path(data_dir) self.data_dir.mkdir(parents=True, exist_ok=True) self.profiles_file = self.data_dir / "reviewer_profiles.json" # 初始化默认数据 if not self.profiles_file.exists(): self._init_default_profiles() def _init_default_profiles(self): """初始化默认主审员画像""" default_profiles = { "李明": { "name": "李明", "specialty": ["制造业", "新能源"], "current_projects": 5, "avg_completion_days": 8, "quality_score": 0.85, # 0-1，基于历史评审质量 "efficiency_score": 0.78, # 0-1，基于完成速度 "workload_capacity": 7, # 最大承载项目数 "industry_experience": { "制造业": 45, # 历史评审项目数 "新能源": 32, "光伏": 18, }, "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat(), }, "王芳": { "name": "王芳", "specialty": ["新能源", "光伏", "风电"], "current_projects": 3, "avg_completion_days": 7, "quality_score": 0.92, "efficiency_score": 0.88, "workload_capacity": 6, "industry_experience": { "新能源": 56, "光伏": 38, "风电": 22, }, "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat(), }, "张强": { "name": "张强", "specialty": ["交通运输", "物流", "制造业"], "current_projects": 7, "avg_completion_days": 9, "quality_score": 0.80, "efficiency_score": 0.72, "workload_capacity": 8, "industry_experience": { "交通运输": 67, "物流": 43, "制造业": 28, }, "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat(), }, "刘洋": { "name": "刘洋", "specialty": ["医疗健康", "服务业", "消费"], "current_projects": 4, "avg_completion_days": 8, "quality_score": 0.88, "efficiency_score": 0.82, "workload_capacity": 7, "industry_experience": { "医疗健康": 51, "服务业": 34, "消费": 29, }, "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat(), }, } with open(self.profiles_file, "w", encoding="utf-8") as f: json.dump(default_profiles, f, ensure_ascii=False, indent=2) logger.info("主审员画像初始化完成") def get_all_profiles(self) -> Dict[str, Dict]: """获取所有主审员画像""" with open(self.profiles_file, "r", encoding="utf-8") as f: return json.load(f) def get_profile(self, reviewer_name: str) -> Optional[Dict]: """获取指定主审员画像""" profiles = self.get_all_profiles() return profiles.get(reviewer_name) def update_workload(self, reviewer_name: str, current_projects: int): """更新主审员工作负荷""" profiles = self.get_all_profiles() if reviewer_name in profiles: profiles[reviewer_name]["current_projects"] = current_projects profiles[reviewer_name]["updated_at"] = datetime.now().isoformat() with open(self.profiles_file, "w", encoding="utf-8") as f: json.dump(profiles, f, ensure_ascii=False, indent=2) logger.info(f"已更新 {reviewer_name} 的负荷: {current_projects} 个项目") def update_quality_score(self, reviewer_name: str, quality_score: float): """更新主审员质量评分""" profiles = self.get_all_profiles() if reviewer_name in profiles: profiles[reviewer_name]["quality_score"] = quality_score profiles[reviewer_name]["updated_at"] = datetime.now().isoformat() with open(self.profiles_file, "w", encoding="utf-8") as f: json.dump(profiles, f, ensure_ascii=False, indent=2) logger.info(f"已更新 {reviewer_name} 的质量评分: {quality_score:.2f}") def calculate_match_score( self, reviewer_name: str, project_industry: str, project_complexity: int = 3 ) -> Dict: """计算主审员与项目的匹配度（多维度评分） 评分维度： - 35% 行业匹配度（专长+历史经验） - 35% 可用性（当前负荷） - 20% 质量评分（历史评审质量） - 10% 团队平衡（避免某人长期超载） """ profile = self.get_profile(reviewer_name) if not profile: return {"total_score": 0, "breakdown": {}} # 1. 行业匹配度（35%） industry_match = 0 if any(spec in project_industry or project_industry in spec for spec in profile["specialty"]): industry_match = 0.8 # 专长匹配 # 检查历史经验 exp_count = 0 for industry_key in profile["industry_experience"]: if industry_key in project_industry or project_industry in industry_key: exp_count = profile["industry_experience"][industry_key] break if exp_count >= 30: industry_match = 1.0 elif exp_count >= 10: industry_match = max(industry_match, 0.6) elif exp_count > 0: industry_match = max(industry_match, 0.3) industry_score = industry_match * 35 # 2. 可用性（35%） current_load = profile["current_projects"] capacity = profile["workload_capacity"] if current_load >= capacity: availability = 0.1 # 超载 elif current_load >= capacity * 0.8: availability = 0.5 # 偏高 else: availability = 1.0 - (current_load / capacity) # 线性递减 availability_score = availability * 35 # 3. 质量评分（20%） quality_score = profile["quality_score"] * 20 # 4. 团队平衡（10%）- 优先分配给负荷较低的 all_profiles = self.get_all_profiles() avg_load = sum(p["current_projects"] for p in all_profiles.values()) / len(all_profiles) if current_load < avg_load: balance_score = 10 elif current_load == avg_load: balance_score = 7 else: balance_score = max(0, 10 - (current_load - avg_load) * 2) # 总分 total_score = industry_score + availability_score + quality_score + balance_score return { "total_score": round(total_score, 2), "breakdown": { "industry_match": round(industry_score, 2), "availability": round(availability_score, 2), "quality": round(quality_score, 2), "balance": round(balance_score, 2), }, "estimated_days": profile["avg_completion_days"] + current_load // 2, # 考虑负荷影响 } def get_recommended_reviewers( self, project_industry: str, project_complexity: int = 3, top_n: int = 3 ) -> List[Dict]: """获取推荐主审员列表（Top N）""" all_profiles = self.get_all_profiles() recommendations = [] for reviewer_name in all_profiles: match_result = self.calculate_match_score( reviewer_name, project_industry, project_complexity ) profile = all_profiles[reviewer_name] recommendations.append({ "reviewer": reviewer_name, "score": match_result["total_score"], "breakdown": match_result["breakdown"], "estimated_days": match_result["estimated_days"], "current_projects": profile["current_projects"], "specialty": profile["specialty"], }) # 按总分降序排序 recommendations.sort(key=lambda x: x["score"], reverse=True) return recommendations[:top_n] # 全局单例 _reviewer_store: Optional[ReviewerProfileStore] = None def get_reviewer_store() -> ReviewerProfileStore: """获取主审员画像存储单例""" global _reviewer_store if _reviewer_store is None: _reviewer_store = ReviewerProfileStore() return _reviewer_store