class SkillMatcher:    """技能匹配引擎 - 处理等级、衰减、多技能组合"""    # 技能等级权重映射    SKILL_LEVEL_WEIGHT = {        "初级": 0.4, "中级": 0.6, "高级": 0.8, "技师": 0.9, "高级技师": 1.0    }    # 技能衰减系数：每月未使用衰减2%    DECAY_RATE_PER_MONTH = 0.02    MAX_DECAY = 0.5  # 最多衰减50%    def calculate_skill_score(self, order_skills: dict, worker: dict) -> float:        """        order_skills: {"电工": "中级以上", "登高": "必须"}        worker["skills"]: [{"name": "电工", "level": "高级", "last_used": "2024-06"}]        """        total_score = 0.0        for skill_name, required in order_skills.items():            worker_skill = self._find_worker_skill(worker, skill_name)            if not worker_skill:                return 0.0  # 核心技能缺失，直接0分            # 等级分            level_score = self._calc_level_score(worker_skill, required)            if level_score == 0:                return 0.0  # 等级不达标，硬约束            # 衰减分            decay = self._calc_decay(worker_skill.get("last_used"))            total_score += level_score * (1 - decay)        return total_score / len(order_skills)    def _calc_decay(self, last_used_str: str) -> float:        """计算技能衰减：距离上次使用的时间越长，衰减越大"""        if not last_used_str:            return self.MAX_DECAY        last_used = datetime.strptime(last_used_str, "%Y-%m")        months_diff = (datetime.now().year - last_used.year) * 12 + \                      (datetime.now().month - last_used.month)        decay = min(self.MAX_DECAY, months_diff * self.DECAY_RATE_PER_MONTH)        return decay

class CertificateValidator:    """证书校验引擎 - 处理发证机构、跨区域、备案"""    def validate(self, order: dict, worker: dict) -> tuple[bool, str]:        """        返回 (是否通过, 不通过原因)        """        for cert_req in order.get("required_certificates", []):            cert_name = cert_req["name"]            worker_cert = worker.get("certificates", {}).get(cert_name)            if not worker_cert:                return False, f"缺少证书:{cert_name}"            # 1. 有效期检查            if worker_cert["expire_date"] < datetime.now():                return False, f"证书过期:{cert_name}"            # 2. 发证机构白名单            issuer = worker_cert["issuer"]            if issuer not in cert_req.get("approved_issuers", []):                # 3. 跨区域互认检查                if not self._check_cross_region_recognition(                    issuer, cert_req.get("region"), worker_cert.get("region")                ):                    return False, f"发证机构不在白名单且未互认:{issuer}"            # 4. 平台备案状态（某些高危工种必须在监管平台备案）            if cert_req.get("require_registration"):                if not self._check_platform_registration(worker["id"], cert_name):                    return False, f"证书未在监管平台备案:{cert_name}"        return True, "OK"    def _check_cross_region_recognition(self, issuer_region: str,                                         order_region: str,                                         cert_region: str) -> bool:        """检查跨区域互认协议"""        # 查询互认协议表：比如长三角互认、京津冀互认        # 这里简化，实际是查数据库或缓存        recognition_map = {            ("上海", "江苏"): True, ("上海", "浙江"): True,            ("北京", "天津"): True, ("广东", "深圳"): True,        }        key = (cert_region, order_region)        return recognition_map.get(key, False)

class TimeConflictChecker:    """时间冲突检测 - 连续工时+通勤+跨天"""    MAX_DAILY_HOURS = 8  # 单日最大工时    MIN_REST_HOURS = 10  # 连续工作后最少休息10小时（夜班后）    def check(self, order: dict, worker: dict, distance_matrix: dict) -> bool:        """        distance_matrix: 地点间距离矩阵，用于计算通勤时间        """        new_start = self._parse_time(order["start_time"])        new_end = self._parse_time(order["end_time"])        new_location = order["location_id"]        # 检查已有订单        for existing in worker.get("existing_orders", []):            exist_start = self._parse_time(existing["start_time"])            exist_end = self._parse_time(existing["end_time"])            exist_location = existing["location_id"]            # 1. 直接时间重叠            if new_start < exist_end and new_end > exist_start:                return False            # 2. 连续工时检查：两个订单之间是否有足够休息            if exist_end <= new_start:  # 已有订单在前                rest_hours = (new_start - exist_end).total_seconds() / 3600                commute_hours = self._get_commute_time(                    exist_location, new_location, distance_matrix                )                # 休息时间 = 间隔 - 通勤时间                effective_rest = rest_hours - commute_hours                if effective_rest < self.MIN_REST_HOURS:                    return False            if new_end <= exist_start:  # 新订单在前                rest_hours = (exist_start - new_end).total_seconds() / 3600                commute_hours = self._get_commute_time(                    new_location, exist_location, distance_matrix                )                effective_rest = rest_hours - commute_hours                if effective_rest < self.MIN_REST_HOURS:                    return False        # 3. 单日累计工时检查        daily_hours = self._calc_daily_hours(worker, new_start.date())        if daily_hours + (new_end - new_start).total_seconds() / 3600 > self.MAX_DAILY_HOURS:            return False        return True    def _get_commute_time(self, from_loc: str, to_loc: str, matrix: dict) -> float:        """从距离矩阵获取通勤时间（小时）"""        # 实际中这是实时路况API，这里简化        key = (from_loc, to_loc)        return matrix.get(key, 999) / 60  # 假设速度60km/h

class DynamicWeightConfig:    """动态权重配置 - 根据订单类型调整"""    BASE_WEIGHTS = {        "skill_match": 0.30, "distance": 0.20,         "rating": 0.20, "accept_rate": 0.15, "price": 0.15    }    ORDER_TYPE_ADJUSTMENTS = {        "emergency": {"skill_match": +0.10, "distance": -0.10},        "long_term": {"accept_rate": +0.10, "price": -0.05, "skill_match": -0.05},        "batch": {"price": +0.10, "rating": -0.05, "skill_match": -0.05},    }    def get_weights(self, order_type: str) -> dict:        weights = self.BASE_WEIGHTS.copy()        adjustments = self.ORDER_TYPE_ADJUSTMENTS.get(order_type, {})        for k, v in adjustments.items():            weights[k] = max(0, min(1, weights[k] + v))  # 限制在0-1        # 归一化        total = sum(weights.values())        return {k: v/total for k, v in weights.items()}

def add_team_constraints(self, prob, x, orders, candidates):    """班组完整性约束：班组长和组员必须同时分配到同一个订单"""    for order in orders:        if order.get("require_team"):            team_leader_skill = order["team_leader_skill"]            member_skill = order["member_skill"]            team_size = order["team_size"]  # 包括组长            # 找出所有班组长候选和组员候选            leaders = [c for c in candidates if team_leader_skill in c["skills"]]            members = [c for c in candidates if member_skill in c["skills"]]            # 约束：每个订单必须恰好分配1个组长和(team_size-1)个组员            leader_vars = [x[order["id"]][c["id"]] for c in leaders]            prob += pulp.lpSum(leader_vars) == 1, f"TeamLeader_{order['id']}"            member_vars = [x[order["id"]][c["id"]] for c in members]            prob += pulp.lpSum(member_vars) == team_size - 1, f"TeamMembers_{order['id']}"            # 约束：班组长和组员不能是同一个人            for c in candidates:                if c in leaders and c in members:                    # 如果既是组长又是组员，不能同时分配                    pass  # 实际中会在候选池里区分

def add_continuity_constraints(self, prob, x, multi_day_orders, candidates):    """连续性约束：同一工人在连续多天的订单中，要么全接，要么全不接"""    for worker in candidates:        for order_group in multi_day_orders:  # 同一项目的多天订单            # 如果工人接了其中一天，就必须接所有天            vars_for_worker = [x[order["id"]][worker["id"]] for order in order_group]            # 所有变量必须相等：要么全1，要么全0            for i in range(len(vars_for_worker) - 1):                prob += vars_for_worker[i] == vars_for_worker[i+1], \                        f"Continuity_{worker['id']}_{order_group[0]['id']}_{i}"

class ComplexGeneticScheduler:    """复杂遗传算法调度器 - 多目标+公平性+技能衰减"""    def _fitness(self, individual, orders, candidates, distance_matrix):        """适应度 = 匹配分 - 惩罚项 + 公平性奖励 - 技能衰减惩罚"""        total_score = 0        penalty = 0        worker_assign_count = {}  # 工人被分配的总次数        worker_daily_hours = {}   # 工人每天的工作时长        for order_idx, assignment in enumerate(individual):            order = orders[order_idx]            for worker_id in assignment:                worker = candidates[worker_id]                # 1. 匹配分（考虑技能衰减）                match_score = self._calc_match_with_decay(order, worker)                total_score += match_score                # 2. 统计分配次数                worker_assign_count[worker_id] = worker_assign_count.get(worker_id, 0) + 1                # 3. 统计每日工时                day = order["start_time"][:10]  # 日期                hours = self._calc_hours(order)                key = (worker_id, day)                worker_daily_hours[key] = worker_daily_hours.get(key, 0) + hours        # 惩罚项1：同一工人时间冲突（重叠）        penalty += self._calc_conflict_penalty(individual, orders, candidates)        # 惩罚项2：工人负载不均衡（用基尼系数）        if worker_assign_count:            counts = list(worker_assign_count.values())            gini = self._calc_gini(counts)            penalty += gini * 1000  # 基尼系数越大，惩罚越重        # 惩罚项3：超时工作        for (worker_id, day), hours in worker_daily_hours.items():            if hours > 8:                penalty += (hours - 8) * 500  # 超时惩罚        # 惩罚项4：技能衰减过大（工人被分配去做很久没做的活）        penalty += self._calc_decay_penalty(individual, orders, candidates)        # 公平性奖励：技能高的工人不应该总被分配去做低级活        fairness_bonus = self._calc_skill_fairness(individual, orders, candidates)        total_score += fairness_bonus        return total_score - penalty    def _calc_gini(self, values):        """计算基尼系数，衡量分配公平性"""        if not values or sum(values) == 0:            return 0        sorted_values = sorted(values)        n = len(sorted_values)        cumsum = [sum(sorted_values[:i+1]) for i in range(n)]        return (2 * sum((i+1) * sorted_values[i] for i in range(n)) / (n * sum(sorted_values))) - (n+1)/n    def _calc_decay_penalty(self, individual, orders, candidates):        """技能衰减惩罚：如果工人被分配去做很久没用的技能，扣分"""        penalty = 0        for order_idx, assignment in enumerate(individual):            order = orders[order_idx]            for worker_id in assignment:                worker = candidates[worker_id]                for skill in order.get("required_skills", []):                    last_used = worker.get("skills", {}).get(skill, {}).get("last_used")                    if last_used:                        months = self._months_since(last_used)                        if months > 12:  # 超过1年没用                            penalty += (months - 12) * 10  # 每多一个月扣10分        return penalty

def _crossover_pmx(self, parent1, parent2):    """部分匹配交叉（PMX），适合排列编码"""    size = len(parent1)    cx1, cx2 = sorted(random.sample(range(size), 2))    child1 = [-1] * size    child2 = [-1] * size    # 复制交叉段    for i in range(cx1, cx2 + 1):        child1[i] = parent1[i]        child2[i] = parent2[i]    # 映射关系    mapping1 = {parent1[i]: parent2[i] for i in range(cx1, cx2 + 1)}    mapping2 = {parent2[i]: parent1[i] for i in range(cx1, cx2 + 1)}    # 填充剩余位置    for i in range(size):        if i < cx1 or i > cx2:            # 处理child1            val1 = parent2[i]            while val1 in child1:                val1 = mapping1.get(val1, val1)            child1[i] = val1            # 处理child2            val2 = parent1[i]            while val2 in child2:                val2 = mapping2.get(val2, val2)            child2[i] = val2    return child1, child2