"""Allowlisted deterministic business executors, never dynamic imports/eval.""" from decimal import Decimal, InvalidOperation, ROUND_HALF_UP EXECUTORS = {"quality", "ratio", "overdue", "compare", "required", "duplicate", "model"} def number(value): if value is None or isinstance(value, bool): raise ValueError("缺少有效数值") try: result = Decimal(str(value)) except (InvalidOperation, ValueError, TypeError): raise ValueError("金额必须是十进制数") from None if not result.is_finite() or abs(result) > Decimal("1e18"): raise ValueError("数值超出支持范围") return result def ratio(numerator, denominator): n, d = number(numerator), number(denominator) if d <= 0: raise ValueError("分母必须大于零") return str((n / d).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)) def applicability_checks(skill, values, sources): """One predicate implementation for deterministic and model Skills (BR-66).""" fields = [] def matches(condition): field = condition['field']; fields.append(field) value = values.get(field) if value is None: return None expected = condition.get('threshold', condition.get('value')) op = condition['operator'] if op == 'eq' and isinstance(expected, bool): return value == expected try: a, b = number(value), number(expected) return {'eq': a == b, 'lt': a < b, 'lte': a <= b, 'gt': a > b, 'gte': a >= b}[op] except (ValueError, KeyError): return None conditions = [matches(c) for c in skill.get('conditions', [])] exceptions = [matches(c) for c in skill.get('exceptions', [])] exception_field = skill.get('exception_field') if exception_field: fields.append(exception_field) exceptions.append(values.get(exception_field) not in (None, '', False)) result = ('not_applicable' if False in conditions or True in exceptions else 'unknown' if None in conditions or None in exceptions else None) if result is None: return None return [{'code': skill.get('code', skill['id'] + '.applicability'), 'title': skill.get('title', skill['name']) + '适用性', 'result': result, 'fields': fields, 'evidence_ids': [r['id'] for f in fields for r in sources.get(f, [])], 'basis_ids': [skill['id'] + '@' + skill['version']], 'severity': 'attention', 'suggested_action': '补充适用条件依据' if result == 'unknown' else '本次不适用，保留判定依据'}] def execute_skill(skill, values, sources): engine = skill.get("engine") if engine not in EXECUTORS or engine == "model": raise ValueError("不支持的确定性执行器") gate = applicability_checks(skill, values, sources) if gate is not None: return {}, gate checks, outputs = [], {} fields = skill.get("inputs", []) def check(code, title, result, refs, action="请核实材料与口径"): checks.append({"code": code, "title": title, "result": result, "fields": refs, "suggested_action": action, "evidence_ids": [r["id"] for f in refs for r in sources.get(f, [])], "basis_ids": [skill["id"] + "@" + skill["version"]], "severity": "attention"}) if engine in ("quality", "required"): for field in (list(values) if "*" in fields else fields): check(skill.get("code", "missing." + field), skill.get("title", field + "完整性"), "unknown" if values.get(field) is None else "pass", [field], "补充对应材料和来源依据") elif engine == "compare": a, b = skill["left"], skill["right"] result = "unknown" if values.get(a) is None or values.get(b) is None else ("pass" if number(values[a]) == number(values[b]) else "issue") refs = sources.get(a, []) + sources.get(b, []) if not sources.get(a) or not sources.get(b) or any(not r.get(k) for r in refs for k in ("currency", "unit", "scope", "period")) or any(len({r.get(k) for r in refs}) != 1 for k in ("currency", "unit", "scope", "period")): result = "unknown" check(skill["code"], skill["name"], result, [a, b]) elif engine == "duplicate": value = values.get(skill["field"]) check(skill["code"], skill["name"], "unknown" if value is None else ("issue" if number(value) > 0 else "pass"), [skill["field"]], "核对设备编号并提交去重清单") elif engine == "ratio": a, b = skill["numerator"], skill["denominator"] refs = sources.get(a, []) + sources.get(b, []) if not sources.get(a) or not sources.get(b) or any(not r.get(k) for r in refs for k in ("currency", "unit", "scope")): raise ValueError("缺少币种、单位或口径依据，不能计算") for dimension in ("currency", "unit", "scope"): if len({r.get(dimension) for r in refs if r.get(dimension)}) > 1: raise ValueError(dimension + "不一致，不能交叉计算") if a.startswith("financial.") and b.startswith("financial.") and len({r.get("period") for r in refs}) > 1: raise ValueError("期间不一致，不能交叉计算") outputs[skill["output"]] = ratio(values.get(a), values.get(b)) else: for field in fields: check("input." + field, field + "输入检查", "unknown" if values.get(field) is None else "pass", [field]) return outputs, checks def execute_rules(rule, values, sources): checks = [] for index, condition in enumerate(rule.get("conditions", [])): field = condition["field"] applicable = condition.get("when", {}) if any(values.get(k) != v for k, v in applicable.items()): result = "not_applicable" else: try: value, threshold = number(values.get(field)), number(condition["threshold"]) hit = {"gt": value > threshold, "lt": value < threshold, "eq": value == threshold, "gte": value >= threshold, "lte": value <= threshold}[condition["operator"]] result = "issue" if hit else "pass" except ValueError: result = "unknown" checks.append({"code": condition.get("code", rule["id"] + "." + str(index)), "title": condition.get("title", field), "result": result, "fields": [field], "value": values.get(field), "evidence_ids": [r["id"] for r in sources.get(field, [])], "basis_ids": [rule["id"] + "@" + rule["version"]], "suggested_action": condition.get("suggestion", "请核实"), "severity": "attention"}) return checks