| """Business logic services.""" |
|
|
| from __future__ import annotations |
|
|
| import re |
| from typing import Any |
|
|
| from apps.documents.models import FolderTemplate |
| from apps.documents.usecases.folder_template.folder_template_usecases import FolderTemplateUsecases |
|
|
|
|
| class FolderTemplateService: |
| """ |
| 文件夹模板服务 |
| |
| 负责文件夹模板的创建、更新、查询、验证和ID唯一性检查. |
| """ |
|
|
| |
| INVALID_CHARS = r'[/\\:*?"<>|]' |
| INVALID_CHARS_PATTERN = re.compile(INVALID_CHARS) |
|
|
| def __init__(self, *, usecases: FolderTemplateUsecases) -> None: |
| self.usecases = usecases |
|
|
| def validate_and_fix_structure_ids( |
| self, structure: dict[str, Any], template_id: int | None = None |
| ) -> tuple[bool, dict[str, Any], list[str]]: |
| """ |
| 验证并自动修复文件夹结构中的重复ID |
| |
| Args: |
| structure: 文件夹结构JSON |
| template_id: 当前模板ID(更新时使用,排除自身) |
| |
| Returns: |
| (is_fixed, fixed_structure, fix_messages) |
| """ |
| return self.usecases.validate_and_fix_structure_ids(structure=structure, template_id=template_id) |
|
|
| def validate_structure_ids( |
| self, structure: dict[str, Any], template_id: int | None = None |
| ) -> tuple[bool, list[str]]: |
| """ |
| 验证文件夹结构中的ID唯一性 |
| |
| Args: |
| structure: 文件夹结构JSON |
| template_id: 当前模板ID(更新时使用,排除自身) |
| |
| Returns: |
| (is_valid, error_messages) |
| """ |
| return self.usecases.validate_structure_ids(structure=structure, template_id=template_id) |
|
|
| def get_duplicate_id_report(self) -> dict[str, Any]: |
| return self.usecases.get_duplicate_id_report() |
|
|
| def create_template( |
| self, |
| name: str, |
| case_type: str, |
| case_stage: str, |
| structure: dict[str, Any], |
| is_default: bool = False, |
| is_active: bool = True, |
| **kwargs: Any, |
| ) -> FolderTemplate: |
| return self.usecases.create_template( |
| name=name, |
| case_type=case_type, |
| case_stage=case_stage, |
| structure=structure, |
| is_default=is_default, |
| is_active=is_active, |
| **kwargs, |
| ) |
|
|
| def update_structure(self, template_id: int, structure: dict[str, Any]) -> FolderTemplate: |
| """ |
| 更新文件夹结构 |
| |
| Args: |
| template_id: 模板 ID |
| structure: 新的文件夹结构 |
| |
| Returns: |
| 更新后的 FolderTemplate 实例 |
| |
| Raises: |
| NotFoundError: 模板不存在 |
| ValidationException: 结构验证失败 |
| """ |
| return self.usecases.update_structure(template_id=template_id, structure=structure) |
|
|
| def get_template_for_case(self, case_type: str, case_stage: str) -> FolderTemplate | None: |
| """ |
| 获取匹配的文件夹模板 |
| |
| 返回匹配指定案件类型和阶段的最新更新的模板. |
| |
| Args: |
| case_type: 案件类型 |
| case_stage: 案件阶段 |
| |
| Returns: |
| 匹配的 FolderTemplate 实例,如果没有匹配则返回 None |
| |
| Requirements: 1.4 |
| """ |
| return self.usecases.get_template_for_case(case_type=case_type, case_stage=case_stage) |
|
|
| def get_template_by_id(self, template_id: int) -> FolderTemplate: |
| """ |
| 根据 ID 获取模板 |
| |
| Args: |
| template_id: 模板 ID |
| |
| Returns: |
| FolderTemplate 实例 |
| |
| Raises: |
| NotFoundError: 模板不存在 |
| """ |
| return self.usecases.get_template_by_id(template_id=template_id) |
|
|
| def validate_structure(self, structure: dict[str, Any]) -> tuple[bool, str]: |
| """ |
| 验证文件夹结构 |
| |
| 检查: |
| 1. 循环引用(Requirements: 1.2) |
| 2. 无效文件名字符(Requirements: 1.8) |
| |
| Args: |
| structure: 文件夹结构 JSON |
| |
| Returns: |
| (is_valid, error_message) 元组 |
| """ |
| return self.usecases.validate_structure(structure=structure) |
|
|
| def _check_circular_reference( |
| self, structure: dict[str, Any], visited_ids: set[Any] | None = None, path: list[Any] | None = None |
| ) -> tuple[bool, str]: |
| """ |
| 检测循环引用 |
| """ |
| if visited_ids is None: |
| visited_ids = set() |
| current_path: list[Any] = [] if path is None else path |
|
|
| children = structure.get("children", []) |
| if not isinstance(children, list): |
| return False, "" |
|
|
| for child in children: |
| if not isinstance(child, dict): |
| continue |
|
|
| node_id = child.get("id") |
| node_name = child.get("name", "unknown") |
| child_path = current_path + [node_name] |
|
|
| if node_id is not None: |
| if node_id in visited_ids: |
| return True, " -> ".join(child_path) |
| visited_ids.add(node_id) |
|
|
| has_cycle, cycle_path = self._check_circular_reference(child, visited_ids.copy(), child_path) |
| if has_cycle: |
| return True, cycle_path |
|
|
| return False, "" |
|
|
| def _check_invalid_chars(self, structure: dict[str, Any], path: list[Any] | None = None) -> tuple[bool, str]: |
| """ |
| 检测无效文件名字符 |
| """ |
| current_path: list[Any] = [] if path is None else path |
|
|
| children = structure.get("children", []) |
| if not isinstance(children, list): |
| return False, "" |
|
|
| for child in children: |
| if not isinstance(child, dict): |
| continue |
|
|
| name = child.get("name", "") |
| child_path = current_path + [name] |
|
|
| if name and self.INVALID_CHARS_PATTERN.search(name): |
| invalid_chars = self.INVALID_CHARS_PATTERN.findall(name) |
| return True, f"'{name}' 包含无效字符: {invalid_chars}" |
|
|
| has_invalid, invalid_info = self._check_invalid_chars(child, child_path) |
| if has_invalid: |
| return True, invalid_info |
|
|
| return False, "" |
|
|
| def list_templates( |
| self, case_type: str | None = None, case_stage: str | None = None, is_active: bool | None = None |
| ) -> list[FolderTemplate]: |
| """ |
| 列出文件夹模板 |
| |
| Args: |
| case_type: 按案件类型过滤 |
| case_stage: 按案件阶段过滤 |
| is_active: 按启用状态过滤 |
| |
| Returns: |
| FolderTemplate 列表 |
| """ |
| return self.usecases.list_templates(case_type=case_type, case_stage=case_stage, is_active=is_active) |
|
|
| def delete_template(self, template_id: int) -> bool: |
| """ |
| 删除模板(软删除) |
| |
| Args: |
| template_id: 模板 ID |
| |
| Returns: |
| 是否成功 |
| |
| Raises: |
| NotFoundError: 模板不存在 |
| """ |
| return self.usecases.delete_template(template_id=template_id) |
|
|
| def create_template_from_dict(self, data: dict[str, Any]) -> FolderTemplate: |
| """ |
| 从字典数据创建文件夹模板 |
| |
| 用于 API 层调用,接收 Schema 转换后的字典数据. |
| |
| Args: |
| data: 包含模板字段的字典 |
| |
| Returns: |
| 创建的 FolderTemplate 实例 |
| |
| Raises: |
| ValidationException: 验证失败 |
| |
| Requirements: 1.1 |
| """ |
| return self.usecases.create_template_from_dict(data=data) |
|
|
| def update_template_from_dict(self, template_id: int, data: dict[str, Any]) -> FolderTemplate: |
| """ |
| 从字典数据更新文件夹模板 |
| |
| 用于 API 层调用,接收 Schema 转换后的字典数据. |
| |
| Args: |
| template_id: 模板 ID |
| data: 包含更新字段的字典 |
| |
| Returns: |
| 更新后的 FolderTemplate 实例 |
| |
| Raises: |
| NotFoundError: 模板不存在 |
| ValidationException: 验证失败 |
| |
| Requirements: 1.1 |
| """ |
| return self.usecases.update_template_from_dict(template_id=template_id, data=data) |
|
|
