263 lines
9.3 KiB
Python
263 lines
9.3 KiB
Python
"""主审员画像库
|
||
|
||
功能：
|
||
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
|