Spaces:

jun2024
/

fc

Paused

fc / apps /documents /services /generation /folder_generation_service.py
jun2024
Upload 2241 files
b4f2238 verified 7 months ago
33.7 kB
"""
文件夹生成服务
负责根据合同类型匹配文件夹模板,生成文件夹结构,并将合同文书放置到指定位置.
Requirements: 2.1, 2.6, 2.7, 3.1, 4.1
"""

from __future__ import annotations

import logging
import zipfile
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast

from django.utils.translation import gettext_lazy as _

from apps.core.exceptions import NotFoundError, ValidationException
from apps.core.models.enums import CaseType

if TYPE_CHECKING:
    from apps.core.interfaces import IContractService
    from apps.documents.models import DocumentTemplate, FolderTemplate

logger = logging.getLogger(__name__)


@dataclass
class DocumentPlacement:
    """文书放置配置"""

    document_template: DocumentTemplate
    folder_path: str  # 相对于根目录的路径
    file_name: str  # 生成的文件名


class FolderGenerationService:
    """
    文件夹生成服务
    负责生成包含完整文件夹结构和合同文书的ZIP压缩包.
    使用 ServiceLocator 获取跨模块依赖,遵循四层架构规范.
    """

    def __init__(
        self,
        contract_service: IContractService | None = None,
        folder_binding_service: Any | None = None,
    ) -> None:
        """
        初始化服务(依赖注入)
        Args:
            contract_service: 合同服务接口(可选,延迟获取)
        """
        self._contract_service = contract_service
        self._folder_binding_service = folder_binding_service
        self._last_extract_path: str | None = None

    @property
    def contract_service(self) -> IContractService:
        """
        延迟获取合同服务
        Returns:
            IContractService 实例
        """
        if self._contract_service is None:
            raise RuntimeError("FolderGenerationService.contract_service 未注入")
        return self._contract_service

    @property
    def folder_binding_service(self) -> Any | None:
        return self._folder_binding_service

    def find_matching_folder_template(self, case_type: str) -> FolderTemplate | None:
        """
        根据合同类型查找匹配的文件夹模板
        Args:
            case_type: 合同类型(如 'civil', 'criminal' 等)
        Returns:
            匹配的 FolderTemplate 或 None
        """
        from .pipeline import TemplateMatcher

        return cast("FolderTemplate | None", TemplateMatcher().match_folder_template(case_type))

    def format_root_folder_name(self, contract: Any) -> str:
        """
        格式化根目录文件夹名称
        格式:{日期}-[{合同类型显示名}]{合同名称}
        示例:2026.01.02-[民商事]奥创公司案件
        Args:
            contract: 合同数据(Contract 实例或 ContractDataWrapper)
        Returns:
            格式化后的文件夹名称
        """
        # 获取当前日期
        today = date.today().strftime("%Y.%m.%d")

        # 获取合同类型中文显示名
        case_type = getattr(contract, "case_type", None)
        case_type_display = dict(CaseType.choices).get(case_type, case_type or "未知类型")  # type: ignore[no-any-return]
        # 获取合同名称
        contract_name = getattr(contract, "name", None) or "未命名合同"

        # 组合格式化名称
        return f"{today}-[{case_type_display}]{contract_name}"

    def generate_folder_structure(self, template: FolderTemplate, root_name: str) -> dict[str, Any]:
        """
        根据模板生成文件夹结构
        模板结构格式:
        - 标准格式:{"children": [...]} - 只有子节点,根目录名称由 root_name 提供
        - 带名称格式:{"name": "xxx", "children": [...]} - 有根目录名称
        Args:
            template: 文件夹模板
            root_name: 根目录名称(用于替换或创建根目录)
        Returns:
            文件夹结构字典,包含 name 和 children
        """
        structure = template.structure.copy() if template.structure else {}

        # 如果模板结构有根目录名称，则替换
        if structure and "name" in structure:
            structure["name"] = root_name
        else:
            # 模板结构只有 children，创建包含根目录的结构
            children = structure.get("children", [])
            structure = {"name": root_name, "children": children}

        return cast(dict[str, Any], structure)

    def get_document_placements(self, contract: Any, folder_template: FolderTemplate) -> list[DocumentPlacement]:
        """
        获取文书放置配置
        查询所有绑定到该文件夹模板的文件模板，并根据合同类型进行匹配
        Args:
            contract: 合同数据(Contract 实例或 ContractDataWrapper)
            folder_template: 文件夹模板
        Returns:
            文书放置配置列表
        """
        from apps.documents.models import DocumentTemplateFolderBinding

        placements: list[Any] = []
        case_type = getattr(contract, "case_type", None)

        # 查询所有绑定到该文件夹模板的文件模板
        bindings = DocumentTemplateFolderBinding.objects.filter(
            folder_template=folder_template, is_active=True
        ).select_related("document_template")

        for binding in bindings:
            template = binding.document_template

            # 检查模板是否启用
            if not template.is_active:
                continue

            # 检查模板是否匹配合同类型
            contract_types = template.contract_types or []
            if case_type not in contract_types and "all" not in contract_types:
                continue

            # 使用绑定配置的路径
            folder_path = binding.folder_node_path or ""

            logger.info(
                "找到绑定配置: 模板=%s, 节点ID=%s, 路径=%s",
                template.name,
                binding.folder_node_id,
                folder_path,
                extra={
                    "template_name": template.name,
                    "folder_node_id": binding.folder_node_id,
                    "folder_path": folder_path,
                },
            )

            # 生成文件名
            file_name = self._generate_document_filename(contract, template)

            placements.append(
                DocumentPlacement(document_template=template, folder_path=folder_path, file_name=file_name)
            )

        return placements

    def _find_contract_folder_path(self, folder_template: FolderTemplate) -> str:
        """
        在文件夹模板中查找"1-合同"文件夹的路径
        Args:
            folder_template: 文件夹模板
        Returns:
            合同文件夹路径,如 "顾问案件/1-律师资料/1-合同"
        """
        structure = folder_template.structure
        if not structure:
            return ""

        # 递归查找名称包含"合同"的文件夹
        path = self._find_folder_by_name(structure.get("children", []), "合同", [])
        return "/".join(path) if path else ""

    def _find_folder_by_name(self, children: list[Any], target_name: str, current_path: list[Any]) -> list[Any]:
        """
        递归查找包含指定名称的文件夹
        Args:
            children: 子文件夹列表
            target_name: 目标名称(部分匹配)
            current_path: 当前路径
        Returns:
            找到的文件夹路径列表
        """
        for child in children:
            child_name = child.get("name", "")
            child_path = current_path + [child_name]

            # 检查是否匹配(名称包含目标名称,且不包含"补充协议")
            if target_name in child_name and "补充协议" not in child_name:
                return child_path

            # 递归查找子文件夹
            result = self._find_folder_by_name(child.get("children", []), target_name, child_path)
            if result:
                return result

        return []

    def create_zip_package(self, folder_structure: dict[str, Any], documents: list[tuple[str, bytes, str]]) -> bytes:
        """
        创建ZIP打包
        Args:
            folder_structure: 文件夹结构字典
            documents: 文书列表 [(folder_path, content, filename), ...]
        Returns:
            ZIP文件内容
        """
        from .pipeline import ZipPackager

        return ZipPackager().create(folder_structure, documents)

    def generate_folder_with_documents(self, contract_id: int) -> tuple[bytes | None, str | None, str | None]:
        """
        生成包含文书的文件夹ZIP包
        Args:
            contract_id: 合同ID
        Returns:
            Tuple[ZIP内容, 文件名, 错误信息]
        Requirements: 2.6, 2.7
        """
        # 延迟导入,避免循环依赖
        from .contract_generation_service import ContractDataWrapper, ContractGenerationService

        contract_data = self.contract_service.get_contract_with_details_internal(contract_id)
        if not contract_data:
            raise NotFoundError("合同不存在")

        # 包装为类似对象的访问方式
        contract = ContractDataWrapper(contract_data)

        # 2. 查找匹配的文件夹模板
        folder_template = self.find_matching_folder_template(contract.case_type)
        if not folder_template:
            raise ValidationException(
                message=_("请先配置文件夹模板"),
                code="NO_FOLDER_TEMPLATE",
                errors={"case_type": f"合同类型 {contract.case_type} 没有匹配的文件夹模板"},
            )

        # 3. 获取文书放置配置(检查是否有匹配的文书模板)
        document_placements = self.get_document_placements(contract, folder_template)

        # 4. 生成根目录名称
        root_name = self.format_root_folder_name(contract)

        # 5. 生成文件夹结构
        folder_structure = self.generate_folder_structure(folder_template, root_name)

        # 6. 生成文书
        from .pipeline import DocxRenderer, PipelineContextBuilder

        documents: list[Any] = []

        # 获取合同数据用于构建上下文
        contract_model = self.contract_service.get_contract_model_internal(contract_id)
        if not contract_model:
            raise NotFoundError("合同不存在")

        for placement in document_placements:
            try:
                # 检查模板文件是否存在
                file_location = placement.document_template.get_file_location()
                if not file_location or not Path(file_location).exists():
                    logger.warning(
                        "模板文件不存在: %s",
                        placement.document_template.name,
                        extra={"template_name": placement.document_template.name},
                    )
                    continue

                # 构建上下文
                context = PipelineContextBuilder().build_contract_context(contract_model)

                # 渲染模板
                content = DocxRenderer().render(file_location, context)

                if content:
                    documents.append((placement.folder_path, content, placement.file_name))
                    logger.info(
                        "文书生成成功: %s, 放置路径: %s/%s",
                        placement.document_template.name,
                        placement.folder_path,
                        placement.file_name,
                        extra={
                            "contract_id": contract_id,
                            "template_name": placement.document_template.name,
                            "folder_path": placement.folder_path,
                            "file_name": placement.file_name,
                        },
                    )
            except Exception as e:
                logger.warning(
                    "生成文书异常: %s - %s",
                    placement.document_template.name,
                    e,
                    extra={"template_name": placement.document_template.name, "error": str(e)},
                )

        # 7. 创建ZIP包
        try:
            zip_content = self.create_zip_package(folder_structure, documents)
            zip_filename = f"{root_name}.zip"

            # 8. 检查绑定并自动解压到绑定文件夹
            self._last_extract_path = self._extract_to_bound_folder_if_exists(contract_id, zip_content)

            return zip_content, zip_filename, None
        except Exception as e:
            logger.exception("创建ZIP包失败")
            raise ValidationException(f"文件夹打包失败: {e!s}") from e

    def generate_folder_with_documents_result(
        self, contract_id: int
    ) -> tuple[bytes | None, str | None, str | None, str | None]:
        zip_content, zip_filename, error = self.generate_folder_with_documents(contract_id)
        return zip_content, zip_filename, self._last_extract_path, error

    def generate_case_folder_with_documents(
        self,
        case: Any,
        folder_template: FolderTemplate,
        root_name: str,
        *,
        wrap_folder_name: str | None = None,
    ) -> bytes:
        """
        生成案件文件夹（包含绑定文档和特殊文件夹内容）。
        特殊文件夹处理：
        - "身份证明"文件夹：放入本案所有当事人的证件材料（ClientIdentityDoc）
        - "委托材料"文件夹：放入本案指派律师的执业证文件（Lawyer.license_pdf）
        - "执行依据及生效证明"文件夹：放入本案已生效的案号裁判文书（CaseNumber.document_file）
        Args:
            case: Case 实例
            folder_template: 文件夹模板
            root_name: 根目录名称
            wrap_folder_name: 若提供，则将所有内容包裹在此文件夹下（合同有多个案件时使用）
        Returns:
            ZIP 文件内容（bytes）
        """
        from pathlib import Path

        from apps.documents.models import DocumentTemplateFolderBinding
        from apps.documents.services.generation.pipeline import DocxRenderer
        from apps.documents.services.placeholders import EnhancedContextBuilder

        # 1. 生成文件夹结构
        folder_structure = self.generate_folder_structure(folder_template, root_name)
        # raw_structure 是不经单子节点优化的结构，根为 root_name（案件文件夹的根）
        raw_structure = {"name": root_name, "children": folder_structure.get("children", [])}
        # 案件文件夹用 raw_structure 作为 ZIP 结构（不以"执行"等单子节点为根）
        folder_structure = raw_structure

        # 2. 查询绑定到该文件夹模板的文档模板
        bindings = DocumentTemplateFolderBinding.objects.filter(
            folder_template=folder_template,
            is_active=True,
        ).select_related("document_template")

        # 3. 构建案件上下文并渲染文档
        documents: list[tuple[str, bytes, str]] = []
        context = EnhancedContextBuilder().build_context({"case": case, "case_id": case.id})  # type: ignore[typeddict-unknown-key]
        # raw_structure 的根才是案件文件夹真正的根（root_name），用于路径剥离
        root_folder_name = raw_structure.get("name", "")

        for binding in bindings:
            template = binding.document_template
            if not template.is_active:
                continue

            file_location = template.get_file_location()
            if not file_location or not Path(file_location).exists():
                logger.warning(
                    "案件模板文件不存在: %s",
                    template.name,
                    extra={"template_name": template.name},
                )
                continue

            folder_path = binding.folder_node_path or ""
            if folder_path.startswith(f"{root_folder_name}/"):
                folder_path = folder_path[len(f"{root_folder_name}/") :]

            # 法定代表人身份证明书：每个我方法人各生成一份
            if "法定代表人身份证明书" in template.name:
                from apps.client.models import Client
                from apps.documents.services.generation.authorization_material_generation_service import (
                    AuthorizationMaterialGenerationService,
                )

                our_legal_entities = list(
                    case.parties.select_related("client")
                    .filter(client__is_our_client=True, client__client_type=Client.LEGAL)
                    .select_related("client")
                )
                if not our_legal_entities:
                    logger.info(
                        "案件文件夹 - 跳过法定代表人身份证明书（我方无法人）",
                        extra={"case_id": case.id, "template_name": template.name},
                    )
                    continue

                for party in our_legal_entities:
                    try:
                        # 使用与单独点击"法定代表人身份证明书"相同的上下文构建逻辑
                        auth_service = AuthorizationMaterialGenerationService()
                        client_context = auth_service._build_context(case=case, client=party.client)
                        content = DocxRenderer().render(file_location, client_context)
                        # 使用与单独点击"法定代表人身份证明书"相同的文件名生成逻辑
                        # 日期使用今日日期
                        from django.utils import timezone

                        date_str = timezone.now().strftime("%Y%m%d")
                        filename = f"法定代表人身份证明书({party.client.name})V1_{date_str}.docx"
                        documents.append((folder_path, content, filename))
                        logger.info(
                            "案件文件夹 - 法定代表人身份证明书已生成: %s → %s",
                            party.client.name,
                            folder_path,
                            extra={"case_id": case.id, "client": party.client.name},
                        )
                    except Exception as e:
                        logger.warning(
                            "案件文件夹 - 法定代表人身份证明书渲染异常: %s - %s",
                            party.client.name,
                            e,
                            extra={"client": party.client.name, "error": str(e)},
                        )
                continue

            # 授权委托书：使用 AuthorizationMaterialGenerationService 的 _build_power_of_attorney_context
            # 构建上下文（包含 selected_clients），然后用绑定的模板文件渲染
            if "授权委托书" in template.name:
                from apps.documents.services.generation.authorization_material_generation_service import (
                    AuthorizationMaterialGenerationService,
                )
                from apps.documents.services.generation.pipeline import DocxRenderer

                our_parties = [
                    p
                    for p in case.parties.select_related("client").all()
                    if getattr(getattr(p, "client", None), "is_our_client", False)
                ]
                if not our_parties:
                    logger.info(
                        "案件文件夹 - 跳过授权委托书（无我方当事人）",
                        extra={"case_id": case.id, "template_name": template.name},
                    )
                    continue

                try:
                    auth_service = AuthorizationMaterialGenerationService()
                    # _build_power_of_attorney_context 在调用 EnhancedContextBuilder 之前就把
                    # selected_clients 放入 context_data，保证 PowerOfAttorneyPlaceholderService 能读到
                    ctx = auth_service._build_power_of_attorney_context(  # type: ignore[attr-defined]
                        case=case,
                        selected_clients=[p.client for p in our_parties],
                    )
                    content = DocxRenderer().render(file_location, ctx)
                    # 使用与单独点击"授权委托书"相同的文件名生成逻辑
                    filename = auth_service._build_power_of_attorney_filename(  # type: ignore[attr-defined]
                        case=case,
                        selected_clients=[p.client for p in our_parties],
                    )
                    documents.append((folder_path, content, filename))
                    logger.info(
                        "案件文件夹 - 授权委托书已生成: %s",
                        folder_path,
                        extra={"case_id": case.id, "template_name": template.name, "client_count": len(our_parties)},
                    )
                except Exception as e:
                    logger.warning(
                        "案件文件夹 - 授权委托书渲染异常: %s - %s",
                        template.name,
                        e,
                        extra={"template_name": template.name, "error": str(e)},
                    )
                continue

            # 所函：使用 AuthorizationMaterialGenerationService 的文件名生成逻辑
            if "所函" in template.name:
                from apps.documents.services.generation.authorization_material_generation_service import (
                    AuthorizationMaterialGenerationService,
                )

                try:
                    auth_service = AuthorizationMaterialGenerationService()
                    # 先渲染模板（使用与单独点击"所函"相同的上下文构建逻辑）
                    ctx = auth_service._build_context(case=case)
                    content = DocxRenderer().render(file_location, ctx)
                    # 使用与单独点击"所函"相同的文件名生成逻辑
                    filename = auth_service._build_authority_letter_filename(  # type: ignore[attr-defined]
                        case_name=case.name or "案件",
                    )
                    documents.append((folder_path, content, filename))
                    logger.info(
                        "案件文件夹 - 所函已生成: %s",
                        folder_path,
                        extra={"case_id": case.id, "template_name": template.name},
                    )
                    continue
                except Exception as e:
                    logger.warning(
                        "案件文件夹 - 所函渲染异常: %s - %s",
                        template.name,
                        e,
                        extra={"template_name": template.name, "error": str(e)},
                    )
                    continue

            try:
                content = DocxRenderer().render(file_location, context)
                # 生成文件名：模板名称(案件名称)V1_日期.docx
                # 日期优先使用 specified_date，否则使用今日日期
                from django.utils import timezone

                if case.specified_date:
                    date_str = case.specified_date.strftime("%Y%m%d")
                else:
                    date_str = timezone.now().strftime("%Y%m%d")
                case_name = case.name or "案件"
                filename = f"{template.name}({case_name})V1_{date_str}.docx"
                documents.append((folder_path, content, filename))
                logger.info(
                    "案件文件夹 - 文书生成成功: %s → %s",
                    template.name,
                    folder_path,
                    extra={"case_id": case.id, "template_name": template.name, "folder_path": folder_path},
                )
            except Exception as e:
                logger.warning(
                    "案件文件夹 - 文书渲染异常: %s - %s",
                    template.name,
                    e,
                    extra={"template_name": template.name, "error": str(e)},
                )

        # 4. 查找特殊文件夹路径（基于原始结构，返回相对于根的路径）
        special_paths = self._find_special_folder_paths(raw_structure)

        identity_paths: list[str] = special_paths.get("身份证明", [])
        attorney_paths: list[str] = special_paths.get("委托材料", [])
        execution_paths: list[str] = special_paths.get("执行依据及生效证明", [])

        # 去掉特殊文件夹路径中与 ZIP 根目录重复的前缀
        def strip_root_prefix(paths: list[str], root: str) -> list[str]:
            prefix = f"{root}/"
            return [p[len(prefix) :] if p.startswith(prefix) else p for p in paths]

        identity_paths = strip_root_prefix(identity_paths, root_folder_name)
        attorney_paths = strip_root_prefix(attorney_paths, root_folder_name)
        execution_paths = strip_root_prefix(execution_paths, root_folder_name)

        # 5. 放入当事人证件材料 → "身份证明"文件夹
        from apps.core.services.storage_service import to_media_abs

        for identity_path in identity_paths:
            for party in case.parties.select_related("client"):
                for identity_doc in party.client.identity_docs.all():
                    if not identity_doc.file_path:
                        continue
                    try:
                        abs_path = to_media_abs(identity_doc.file_path)
                        if abs_path.exists():
                            content = abs_path.read_bytes()
                            suffix = abs_path.suffix
                            # 证件类型显示名称：去掉 "/" 避免路径问题，统一改为 "法定代表人身份证"
                            doc_type_display = (
                                identity_doc.get_doc_type_display()
                                .replace("/", "身份证")
                                .replace("负责人身份证", "法定代表人身份证")
                            )
                            filename = f"{party.client.name}_{doc_type_display}{suffix}"
                            documents.append((identity_path, content, filename))
                            logger.info(
                                "案件文件夹 - 证件材料已添加: %s → %s",
                                filename,
                                identity_path,
                            )
                    except Exception as e:
                        logger.warning("读取证件文件失败: %s - %s", identity_doc.file_path, e)

        # 6. 放入律师执业证 → "委托材料"文件夹
        for attorney_path in attorney_paths:
            for assignment in case.assignments.select_related("lawyer"):
                lawyer = assignment.lawyer
                if not lawyer.license_pdf:
                    continue
                try:
                    # license_pdf 是 FileField，.path 就是完整的文件系统路径
                    abs_path = Path(lawyer.license_pdf.path)
                    if abs_path.exists():
                        content = abs_path.read_bytes()
                        filename = f"{lawyer.real_name or lawyer.username}_执业证.pdf"
                        documents.append((attorney_path, content, filename))
                        logger.info(
                            "案件文件夹 - 律师执业证已添加: %s → %s",
                            filename,
                            attorney_path,
                        )
                except Exception as e:
                    logger.warning("读取律师执业证失败: %s - %s", lawyer.username, e)

        # 7. 放入已生效案号裁判文书 → "执行依据及生效证明"文件夹
        for exec_path in execution_paths:
            for case_number in case.case_numbers.filter(is_active=True).exclude(document_file=""):
                if not case_number.document_file:
                    continue
                try:
                    # document_file 是 FileField，使用 .path 获取实际文件路径
                    abs_path = Path(case_number.document_file.path)
                    if abs_path.exists():
                        content = abs_path.read_bytes()
                        filename = abs_path.name
                        if not filename.lower().endswith(".pdf"):
                            filename = f"{filename}.pdf"
                        documents.append((exec_path, content, filename))
                        logger.info(
                            "案件文件夹 - 执行依据已添加: %s → %s",
                            filename,
                            exec_path,
                        )
                except Exception as e:
                    logger.warning("读取案号裁判文书失败: %s - %s", case_number.number, e)

        # 8. 若有包裹文件夹，需要将所有文档路径加上包裹前缀
        if wrap_folder_name:
            wrapped_documents: list[tuple[str, bytes, str]] = []
            for doc_folder_path, content, filename in documents:
                new_path = f"{wrap_folder_name}/{doc_folder_path}" if doc_folder_path else wrap_folder_name
                wrapped_documents.append((new_path, content, filename))
            documents = wrapped_documents

        # 9. 创建 ZIP 包
        return self.create_zip_package(folder_structure, documents)

    def _find_special_folder_paths(self, structure: dict[str, Any], parent_path: str = "") -> dict[str, list[str]]:
        """
        递归查找特殊文件夹路径。
        Args:
            structure: 文件夹结构
            parent_path: 父路径
        Returns:
            { "身份证明": [路径1, 路径2], "委托材料": [...], ... }
        """
        result: dict[str, list[str]] = {
            "身份证明": [],
            "委托材料": [],
            "执行依据及生效证明": [],
        }
        if not structure:
            return result

        folder_name = structure.get("name", "")
        current_path = f"{parent_path}/{folder_name}" if parent_path else folder_name

        children = structure.get("children", [])

        # 检查是否匹配特殊文件夹名称（部分匹配）
        for keyword in result:
            if keyword in folder_name:
                result[keyword].append(current_path)

        # 递归处理子文件夹
        for child in children:
            child_result = self._find_special_folder_paths(child, current_path)
            for key, paths in child_result.items():
                result[key].extend(paths)

        return result

    def _generate_document_filename(self, contract: Any, template: DocumentTemplate) -> str:
        """
        生成文书文件名
        Args:
            contract: 合同数据(Contract 实例或 ContractDataWrapper)
            template: 文书模板
        Returns:
            文件名
        """
        # 使用现有的合同生成服务的文件名生成逻辑
        from .contract_generation_service import ContractGenerationService

        service = ContractGenerationService(
            contract_service=self.contract_service,
            folder_binding_service=self.folder_binding_service,
        )
        return service.generate_filename(contract, template)

    def _create_folders_in_zip(self, zip_file: zipfile.ZipFile, structure: dict[str, Any], parent_path: str) -> None:
        """
        在ZIP文件中递归创建文件夹结构
        Args:
            zip_file: ZipFile 对象
            structure: 文件夹结构
            parent_path: 父路径
        """
        if not structure:
            return

        folder_name = structure.get("name", "")
        if not folder_name:
            return

        # 构建当前文件夹路径
        current_path = f"{parent_path}/{folder_name}" if parent_path else folder_name

        # 创建文件夹(在ZIP中添加以/结尾的条目)
        zip_file.writestr(f"{current_path}/", "")

        # 递归处理子文件夹
        children = structure.get("children", [])
        for child in children:
            self._create_folders_in_zip(zip_file, child, current_path)

    def _extract_to_bound_folder_if_exists(self, contract_id: int, zip_content: bytes) -> Any:
        """
        如果合同已绑定文件夹,自动解压ZIP到绑定文件夹
        Args:
            contract_id: 合同ID
            zip_content: ZIP文件内容
        """
        if self.folder_binding_service is None:
            return None

        try:
            extract_path = self.folder_binding_service.extract_zip_to_bound_folder(
                contract_id=contract_id,
                zip_content=zip_content,
            )

            if extract_path:
                logger.info(
                    "文件夹ZIP已自动解压到绑定文件夹: %s",
                    extract_path,
                    extra={
                        "contract_id": contract_id,
                        "extract_path": extract_path,
                        "action": "auto_extract_folder_zip",
                    },
                )
            return extract_path
        except Exception:
            logger.exception("auto_extract_folder_zip_failed", extra={"contract_id": contract_id})
            return None