File size: 8,197 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 | """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)
|