""" 分类策略和目录路径的纯校验，汇总可发布性错误与警告。 """ import json import re import unicodedata from collections.abc import Iterable, Mapping, Sequence from dataclasses import dataclass, field from typing import Any, Final, cast from app.domain.classification.fields import ( ALL_MEDIA_TYPES, VALUE_TYPE_OPERATORS, standard_field_definitions, ) from app.schemas.category import ( ClassificationCondition, ClassificationConditionGroup, ClassificationConditionNode, ClassificationFactScalar, ClassificationFieldDefinition, ClassificationMediaType, ClassificationPolicy, ClassificationRule, ClassificationValidationIssue, ClassificationValidationResult, ) MAX_CATEGORY_DEPTH: Final[int] = 4 MAX_CATEGORY_SEGMENT_LENGTH: Final[int] = 64 MAX_CATEGORY_PATH_LENGTH: Final[int] = 240 MAX_CONDITION_DEPTH: Final[int] = 3 MAX_CONDITIONS_PER_RULE: Final[int] = 30 MAX_RULES: Final[int] = 1000 MAX_TOTAL_CONDITIONS: Final[int] = MAX_RULES * MAX_CONDITIONS_PER_RULE _STABLE_ID_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]{0,127}$") _SOURCE_ID_PATTERN = re.compile(r"^[a-z][a-z0-9._-]{0,63}$") _WINDOWS_RESERVED_NAMES = { "CON", "PRN", "AUX", "NUL", *(f"COM{index}" for index in range(1, 10)), *(f"LPT{index}" for index in range(1, 10)), } _ILLEGAL_PATH_CHARACTERS = frozenset('<>:"/\\|?*') @dataclass(frozen=True, slots=True) class _CategoryPathViolation: """ 描述分类路径中的一个结构化违规位置和消息。 :param message: 可展示给用户的分类路径违规说明 :param segment_index: 违规路径段的零基序号；整体路径错误时为 None """ message: str segment_index: int | None = None def validate_classification_category_path( segments: Sequence[str], ) -> tuple[str, ...]: """ 校验并冻结相对于媒体类型根目录的分类路径段。 """ violations = _category_path_violations(segments) if violations: raise ValueError(violations[0].message) return tuple(segments) @dataclass class _ValidationCollector: """ 按检查顺序收集分类策略的错误和警告。 :param issues: 按检查顺序累积的错误与警告，默认从空列表开始 """ issues: list[ClassificationValidationIssue] = field(default_factory=list) def error(self, code: str, message: str, path: list[str | int]) -> None: """ 追加阻止策略发布的结构化错误。 """ self.issues.append( ClassificationValidationIssue( severity="error", code=code, message=message, path=path, ) ) def warning(self, code: str, message: str, path: list[str | int]) -> None: """ 追加允许策略发布但需要用户确认的结构化警告。 """ self.issues.append( ClassificationValidationIssue( severity="warning", code=code, message=message, path=path, ) ) class ClassificationPolicyValidator: """ 媒体分类策略纯校验器。 校验器只检查策略、标准字段目录和调用方提供的扩展字段，不访问配置、插件或持久化状态。 """ @classmethod def validate( cls, policy: ClassificationPolicy, extra_fields: Iterable[ClassificationFieldDefinition] = (), ) -> ClassificationValidationResult: """ 校验分类策略是否满足发布约束。 :param policy: 待发布的完整分类策略 :param extra_fields: 当前已注册来源声明的扩展字段 :return: 包含 error 和 warning 的结构化校验结果 """ collector = _ValidationCollector() field_definitions = (*standard_field_definitions(), *tuple(extra_fields)) cls._validate_fields(field_definitions, collector) fields = {definition.id: definition for definition in field_definitions} categories = cls._validate_categories(policy, collector) cls._validate_rules(policy, categories, fields, collector) cls._validate_fallbacks(policy, categories, collector) cls._validate_aliases(policy, fields, collector) cls._warn_overlapping_rules(policy, collector) return ClassificationValidationResult( valid=not any(issue.severity == "error" for issue in collector.issues), issues=collector.issues, ) @classmethod def _validate_fields( cls, extra_fields: Sequence[ClassificationFieldDefinition], collector: _ValidationCollector, ) -> None: """ 校验扩展字段的 ID、类型、操作符和来源命名空间。 """ standard_ids = {definition.id for definition in standard_field_definitions()} seen_ids: set[str] = set() for index, definition in enumerate(extra_fields): path: list[str | int] = ["extra_fields", index] if definition.id in seen_ids: collector.error( "duplicate_field_id", f"字段 ID {definition.id} 重复或覆盖了标准字段", [*path, "id"], ) seen_ids.add(definition.id) if definition.value_type not in VALUE_TYPE_OPERATORS: collector.error( "unsupported_value_type", f"字段 {definition.id} 使用了不支持的值类型 {definition.value_type}", [*path, "value_type"], ) else: allowed = set(VALUE_TYPE_OPERATORS[definition.value_type]) unsupported = [operator for operator in definition.operators if operator not in allowed] if unsupported: collector.error( "unsupported_field_operator", f"字段 {definition.id} 声明了不兼容操作符：{', '.join(unsupported)}", [*path, "operators"], ) cls._validate_media_types( definition.media_types, [*path, "media_types"], collector, ) if not definition.id.startswith("extensions."): if definition.id not in standard_ids: collector.error( "invalid_extension_field_id", f"非标准字段 {definition.id} 必须位于 extensions 命名空间", [*path, "id"], ) continue extension_sources = [ source for source, support in definition.source_support.items() if support == "extension" ] if len(extension_sources) != 1: collector.error( "invalid_extension_source", f"扩展字段 {definition.id} 必须声明且只能声明一个 extension 来源", [*path, "source_support"], ) continue source = extension_sources[0] prefix = f"extensions.{source}." if not _SOURCE_ID_PATTERN.fullmatch(source) or not definition.id.startswith(prefix): collector.error( "extension_namespace_mismatch", f"扩展字段 {definition.id} 必须位于 {prefix} 命名空间", [*path, "id"], ) suffix = definition.id[len(prefix) :] if not suffix or any(not _SOURCE_ID_PATTERN.fullmatch(segment) for segment in suffix.split(".")): collector.error( "invalid_extension_field_id", f"扩展字段 {definition.id} 的来源内字段路径无效", [*path, "id"], ) @classmethod def _validate_categories( cls, policy: ClassificationPolicy, collector: _ValidationCollector, ) -> dict[str, Any]: """ 校验分类 ID、名称、路径和同媒体类型路径唯一性。 """ categories: dict[str, Any] = {} paths: dict[tuple[str, tuple[str, ...]], str] = {} for index, category in enumerate(policy.categories): path: list[str | int] = ["categories", index] if category.id in categories: collector.error( "duplicate_category_id", f"分类 ID {category.id} 重复", [*path, "id"], ) else: categories[category.id] = category if not _STABLE_ID_PATTERN.fullmatch(category.id): collector.error( "invalid_category_id", f"分类 ID {category.id} 只能包含小写字母、数字、点、下划线和连字符", [*path, "id"], ) if not category.name.strip(): collector.error( "empty_category_name", "分类名称不能为空", [*path, "name"], ) cls._validate_category_path(category.path, path, collector) normalized_path = tuple(_normalize_path_segment(item) for item in category.path) path_key = (category.media_type, normalized_path) existing = paths.get(path_key) if normalized_path and existing: collector.error( "duplicate_category_path", f"分类 {category.id} 与 {existing} 在同一媒体类型下使用了相同路径", [*path, "path"], ) elif normalized_path: paths[path_key] = category.id return categories @classmethod def _validate_category_path( cls, segments: Sequence[str], base_path: list[str | int], collector: _ValidationCollector, ) -> None: """ 校验分类相对路径的层级、段长度和跨平台文件名安全性。 """ for violation in _category_path_violations(segments): path = [*base_path, "path"] if violation.segment_index is not None: path.append(violation.segment_index) collector.error( "invalid_category_path", violation.message, path, ) @classmethod def _validate_rules( cls, policy: ClassificationPolicy, categories: Mapping[str, Any], fields: Mapping[str, ClassificationFieldDefinition], collector: _ValidationCollector, ) -> None: """ 校验规则身份、目标引用、条件复杂度和字段契约。 """ if len(policy.rules) > MAX_RULES: collector.error( "max_rules_exceeded", f"策略最多允许 {MAX_RULES} 条规则", ["rules"], ) seen_ids: set[str] = set() total_conditions = 0 for index, rule in enumerate(policy.rules): path: list[str | int] = ["rules", index] if rule.id in seen_ids: collector.error( "duplicate_rule_id", f"规则 ID {rule.id} 重复", [*path, "id"], ) seen_ids.add(rule.id) if not _STABLE_ID_PATTERN.fullmatch(rule.id): collector.error( "invalid_rule_id", f"规则 ID {rule.id} 只能包含小写字母、数字、点、下划线和连字符", [*path, "id"], ) if not rule.name.strip(): collector.error("empty_rule_name", "规则名称不能为空", [*path, "name"]) cls._validate_media_types(rule.media_types, [*path, "media_types"], collector) for source_index, source in enumerate(rule.sources): if not _SOURCE_ID_PATTERN.fullmatch(source): collector.error( "invalid_rule_source", f"规则来源 {source} 不是合法数据源标识", [*path, "sources", source_index], ) cls._validate_target(rule, categories, path, collector) condition_count = cls._validate_condition_node( rule.when, rule, fields, [*path, "when"], 1, collector, ) total_conditions += condition_count if condition_count == 0: collector.error( "empty_rule_condition", "普通规则必须至少包含一个叶子条件；兜底分类请使用 fallbacks", [*path, "when"], ) if condition_count > MAX_CONDITIONS_PER_RULE: collector.error( "max_conditions_exceeded", f"每条规则最多允许 {MAX_CONDITIONS_PER_RULE} 个叶子条件", [*path, "when"], ) if total_conditions > MAX_TOTAL_CONDITIONS: collector.error( "too_many_policy_conditions", f"策略最多允许 {MAX_TOTAL_CONDITIONS} 个叶子条件", ["rules"], ) @classmethod def _validate_target( cls, rule: ClassificationRule, categories: Mapping[str, Any], base_path: list[str | int], collector: _ValidationCollector, ) -> None: """ 校验分类规则和标签规则各自允许的目标结构。 """ category_id = rule.target.category_id if rule.kind == "label": if category_id: collector.error( "label_rule_category_target", "标签规则不能指定目标分类", [*base_path, "target", "category_id"], ) if not rule.target.labels: collector.error( "empty_label_target", "标签规则至少需要输出一个标签", [*base_path, "target", "labels"], ) return if not category_id: collector.error( "missing_category_target", "分类规则必须指定目标分类 ID", [*base_path, "target", "category_id"], ) return category = categories.get(category_id) if not category: collector.error( "target_category_not_found", f"规则引用了不存在的分类 {category_id}", [*base_path, "target", "category_id"], ) return if not category.enabled: collector.error( "disabled_category_target", f"规则引用的分类 {category_id} 已禁用", [*base_path, "target", "category_id"], ) unsupported_types = [media_type for media_type in rule.media_types if media_type != category.media_type] if unsupported_types: collector.error( "target_media_type_mismatch", f"规则媒体类型与目标分类 {category_id} 的媒体类型不一致", [*base_path, "target", "category_id"], ) @classmethod def _validate_condition_node( cls, node: ClassificationConditionNode, rule: ClassificationRule, fields: Mapping[str, ClassificationFieldDefinition], path: list[str | int], depth: int, collector: _ValidationCollector, ) -> int: """ 递归校验条件树并返回叶子条件数量。 """ if isinstance(node, ClassificationCondition): cls._validate_condition(node, rule, fields, path, collector) return 1 if depth > MAX_CONDITION_DEPTH: collector.error( "max_depth_exceeded", f"条件组最多允许 {MAX_CONDITION_DEPTH} 层", path, ) group_name, children = _group_children(node) count = 0 for index, child in enumerate(children): count += cls._validate_condition_node( child, rule, fields, [*path, group_name, index] if group_name != "not" else [*path, "not"], depth + 1, collector, ) return count @classmethod def _validate_condition( cls, condition: ClassificationCondition, rule: ClassificationRule, fields: Mapping[str, ClassificationFieldDefinition], path: list[str | int], collector: _ValidationCollector, ) -> None: """ 校验叶子条件的字段、操作符、媒体类型、来源范围和值结构。 """ definition = fields.get(condition.field) if not definition: collector.error( "unknown_field", f"条件引用了未登记字段 {condition.field}", [*path, "field"], ) return if condition.operator not in definition.operators: collector.error( "unsupported_operator", f"字段 {condition.field} 不支持操作符 {condition.operator}", [*path, "operator"], ) unsupported_types = [media_type for media_type in rule.media_types if media_type not in definition.media_types] if unsupported_types: collector.error( "field_media_type_mismatch", f"字段 {condition.field} 不适用于媒体类型：{', '.join(unsupported_types)}", [*path, "field"], ) cls._validate_condition_value(condition, definition, path, collector) cls._validate_extension_scope(condition.field, rule, definition, path, collector) cls._warn_partial_support(condition.field, rule, definition, path, collector) cls._validate_unavailable_support(condition.field, rule, definition, path, collector) @classmethod def _validate_condition_value( cls, condition: ClassificationCondition, definition: ClassificationFieldDefinition, path: list[str | int], collector: _ValidationCollector, ) -> None: """ 按字段值类型校验条件值和范围结构。 """ operator = condition.operator value = condition.value if operator in {"exists", "not_exists", "is_true", "is_false"}: if value is not None: collector.error( "unexpected_condition_value", f"操作符 {operator} 不接受条件值", [*path, "value"], ) return if operator in {"in", "not_in"}: if not _is_non_empty_scalar_list(value): collector.error( "empty_membership_value", f"操作符 {operator} 需要非空标量列表", [*path, "value"], ) else: members = cast(list[ClassificationFactScalar], value) if all(_valid_scalar_type(definition.value_type, item) for item in members): return collector.error( "invalid_condition_value", f"字段 {condition.field} 的成员值必须符合 {definition.value_type} 类型", [*path, "value"], ) return if operator in {"contains_any", "contains_all", "contains_none"}: if not _is_non_empty_scalar_list(value): collector.error( "invalid_condition_value", f"操作符 {operator} 需要非空字符串列表", [*path, "value"], ) return members = cast(list[ClassificationFactScalar], value) if not all(isinstance(item, str) for item in members): collector.error( "invalid_condition_value", f"操作符 {operator} 需要非空字符串列表", [*path, "value"], ) return if operator == "between": if not isinstance(value, list) or len(value) != 2 or not all(_is_number(item) for item in value): collector.error( "invalid_condition_value", "between 需要包含两个数字的闭区间", [*path, "value"], ) return bounds = cast(list[int | float], value) if bounds[0] > bounds[1]: collector.error( "invalid_between_range", "between 的范围起点不能大于终点", [*path, "value"], ) return valid = _valid_scalar_type(definition.value_type, value) if not valid: collector.error( "invalid_condition_value", f"条件值与字段 {condition.field} 的类型 {definition.value_type} 不匹配", [*path, "value"], ) if definition.value_type == "enum" and valid: options = {_option_value(item) for item in definition.options} if options and value not in options: collector.error( "unknown_enum_value", f"条件值 {value} 不在字段 {condition.field} 的可选值中", [*path, "value"], ) @classmethod def _validate_extension_scope( cls, field_id: str, rule: ClassificationRule, definition: ClassificationFieldDefinition, path: list[str | int], collector: _ValidationCollector, ) -> None: """ 确保来源扩展字段只被其声明来源范围内的规则引用。 """ extension_sources = [source for source, support in definition.source_support.items() if support == "extension"] if not field_id.startswith("extensions.") or not extension_sources: return source = extension_sources[0] if rule.sources and source not in rule.sources: collector.error( "extension_namespace_mismatch", f"字段 {field_id} 属于来源 {source}，但规则未包含该来源", [*path, "field"], ) elif not rule.sources: collector.warning( "extension_field_all_sources", f"字段 {field_id} 只由来源 {source} 提供，其它来源下条件不会命中", [*path, "field"], ) @classmethod def _warn_partial_support( cls, field_id: str, rule: ClassificationRule, definition: ClassificationFieldDefinition, path: list[str | int], collector: _ValidationCollector, ) -> None: """ 提示规则范围内只提供部分事实覆盖的数据源。 """ scoped_sources = rule.sources or list(definition.source_support) partial_sources = [source for source in scoped_sources if definition.source_support.get(source) == "partial"] if partial_sources: collector.warning( "partial_field_support", f"字段 {field_id} 在以下来源仅部分可用：{', '.join(partial_sources)}", [*path, "field"], ) @classmethod def _validate_unavailable_support( cls, field_id: str, rule: ClassificationRule, definition: ClassificationFieldDefinition, path: list[str | int], collector: _ValidationCollector, ) -> None: """ 阻止规则限定到全部无法提供该字段的数据源。 """ scoped_sources = rule.sources or list(definition.source_support) if not scoped_sources: return unavailable = [ source for source in scoped_sources if definition.source_support.get(source, "unavailable") == "unavailable" ] if len(unavailable) == len(scoped_sources): collector.error( "unavailable_field_support", f"字段 {field_id} 在所选来源中均不可用：{', '.join(unavailable)}", [*path, "field"], ) @classmethod def _validate_fallbacks( cls, policy: ClassificationPolicy, categories: Mapping[str, Any], collector: _ValidationCollector, ) -> None: """ 确保每种媒体类型的全局兜底引用同类型的可用分类。 """ for media_type in ALL_MEDIA_TYPES: category_id = policy.fallbacks.get(cast(ClassificationMediaType, media_type)) path: list[str | int] = ["fallbacks", media_type] if not category_id: collector.error( "missing_fallback", f"媒体类型 {media_type} 未配置兜底分类", path, ) continue category = categories.get(category_id) if not category: collector.error( "unknown_fallback_category", f"兜底分类 {category_id} 不存在", path, ) elif category.media_type != media_type: collector.error( "fallback_media_type_mismatch", f"兜底分类 {category_id} 不属于媒体类型 {media_type}", path, ) elif not category.enabled: collector.error( "disabled_fallback_category", f"兜底分类 {category_id} 已禁用", path, ) @classmethod def _validate_aliases( cls, policy: ClassificationPolicy, fields: Mapping[str, ClassificationFieldDefinition], collector: _ValidationCollector, ) -> None: """ 校验字段别名只引用已登记字段且不形成直接或间接循环。 """ for field_id, aliases in policy.field_aliases.items(): path: list[str | int] = ["field_aliases", field_id] if field_id not in fields: collector.error( "unknown_alias_field", f"字段别名引用了未登记字段 {field_id}", path, ) for alias, canonical in aliases.items(): if not alias or not canonical: collector.error( "empty_field_alias", "字段别名和规范值均不能为空", [*path, alias], ) elif _alias_has_cycle(alias, aliases): collector.error( "cyclic_field_alias", f"字段 {field_id} 的别名 {alias} 形成循环映射", [*path, alias], ) @classmethod def _validate_media_types( cls, media_types: Sequence[str], path: list[str | int], collector: _ValidationCollector, ) -> None: """ 校验媒体类型集合非空、不重复且属于分类体系支持范围。 """ if not media_types: collector.error("empty_media_types", "媒体类型不能为空", path) return if len(set(media_types)) != len(media_types): collector.error("duplicate_media_types", "媒体类型不能重复", path) unsupported = [item for item in media_types if item not in ALL_MEDIA_TYPES] if unsupported: collector.error( "unsupported_media_type", f"不支持的媒体类型：{', '.join(unsupported)}", path, ) @classmethod def _warn_overlapping_rules( cls, policy: ClassificationPolicy, collector: _ValidationCollector, ) -> None: """ 对条件完全相同且适用范围重叠的主分类规则给出顺序提示。 """ seen: dict[str, ClassificationRule] = {} for index, rule in enumerate(policy.rules): if not rule.enabled or rule.kind != "category": continue fingerprint = json.dumps( rule.when.model_dump(mode="json", by_alias=True), ensure_ascii=False, sort_keys=True, ) previous = seen.get(fingerprint) if previous and _rules_overlap(previous, rule): collector.warning( "overlapping_category_rules", f"规则 {rule.id} 与 {previous.id} 条件相同，将按列表顺序首条命中", ["rules", index, "when"], ) else: seen[fingerprint] = rule def _group_children( group: ClassificationConditionGroup, ) -> tuple[str, list[ClassificationConditionNode]]: """ 读取严格条件组中的操作符和子节点。 """ if group.all is not None: return "all", group.all if group.any is not None: return "any", group.any if group.not_ is None: return "not", [] return "not", [group.not_] def _path_segment_is_illegal(segment: str) -> bool: """ 判断路径段是否触发目录穿越、控制字符或跨平台非法文件名约束。 """ if segment in {".", ".."} or segment != segment.strip(): return True if segment.endswith((".", " ")): return True if any(character in _ILLEGAL_PATH_CHARACTERS for character in segment): return True if any(ord(character) < 32 for character in segment): return True basename = segment.split(".", 1)[0].upper() return basename in _WINDOWS_RESERVED_NAMES def _category_path_violations( segments: Sequence[str], ) -> tuple[_CategoryPathViolation, ...]: """ 返回分类路径的全部违规，供发布校验和运行时防御复用。 """ if isinstance(segments, (str, bytes)): return (_CategoryPathViolation("分类目录路径必须使用路径段数组"),) values = tuple(segments) if not values: return (_CategoryPathViolation("分类目录路径不能为空"),) violations: list[_CategoryPathViolation] = [] if len(values) > MAX_CATEGORY_DEPTH: violations.append( _CategoryPathViolation( f"分类目录最多允许 {MAX_CATEGORY_DEPTH} 级" ) ) for index, segment in enumerate(values): if not isinstance(segment, str) or not segment or not segment.strip(): violations.append(_CategoryPathViolation("目录路径段不能为空", index)) continue if len(segment) > MAX_CATEGORY_SEGMENT_LENGTH: violations.append( _CategoryPathViolation( f"目录路径段最多允许 {MAX_CATEGORY_SEGMENT_LENGTH} 个字符", index, ) ) if _path_segment_is_illegal(segment): violations.append( _CategoryPathViolation( f"目录路径段 {segment} 包含目录穿越或平台非法文件名", index, ) ) if all(isinstance(segment, str) for segment in values) and len( "/".join(values) ) > MAX_CATEGORY_PATH_LENGTH: violations.append( _CategoryPathViolation( f"分类目录总长度最多允许 {MAX_CATEGORY_PATH_LENGTH} 个字符" ) ) return tuple(violations) def _normalize_path_segment(segment: str) -> str: """ 按 Unicode NFC 和大小写折叠构造跨平台路径唯一键。 """ return unicodedata.normalize("NFC", segment).casefold() def _valid_scalar_type(value_type: str, value: Any) -> bool: """ 为标量条件与成员列表复用严格类型规则，布尔值不能冒充数字。 :param value_type: 字段目录声明的值类型 :param value: 待校验的标量值 :return: 值符合字段类型时返回 True """ if value_type in {"integer", "year"}: return isinstance(value, int) and not isinstance(value, bool) if value_type == "number": return _is_number(value) if value_type in {"string", "enum"}: return isinstance(value, str) return value is not None def _is_number(value: Any) -> bool: """ 判断值是否为分类规则支持的整数或浮点数，布尔值不属于数字。 """ return isinstance(value, (int, float)) and not isinstance(value, bool) def _is_non_empty_scalar_list(value: Any) -> bool: """ 判断值是否为不含 null 或嵌套对象的非空标量列表。 """ return isinstance(value, list) and bool(value) and all(isinstance(item, (str, int, float, bool)) for item in value) def _option_value(option: Any) -> Any: """ 兼容简单枚举值和包含 value 字段的前端选项目录。 """ if isinstance(option, Mapping): return option.get("value") if hasattr(option, "value"): return option.value return option def _alias_has_cycle(alias: str, aliases: Mapping[str, str]) -> bool: """ 检测从一个别名出发是否会回到已经访问过的别名。 """ visited: set[str] = set() current = alias while current in aliases: if current in visited: return True visited.add(current) current = aliases[current] return False def _rules_overlap(left: ClassificationRule, right: ClassificationRule) -> bool: """ 判断两条规则的媒体类型和来源范围是否存在交集。 """ if not set(left.media_types).intersection(right.media_types): return False if not left.sources or not right.sources: return True return bool(set(left.sources).intersection(right.sources))