FastapiAdmin/backend/app/plugin/module_generator/gencode/service.py

import io
import os
import re
import zipfile
from collections.abc import Callable
from typing import Any

import anyio
import sqlglot
from sqlglot.expressions import (
    Add,
    Alter,
    Create,
    Delete,
    Drop,
    Insert,
    Table,
    TruncateTable,
    Update,
)

from app.api.v1.module_system.auth.schema import AuthSchema
from app.config.path_conf import BASE_DIR
from app.config.setting import settings
from app.core.exceptions import CustomException
from app.core.logger import log

from .crud import GenTableColumnCRUD, GenTableCRUD
from .schema import (
    GenTableColumnOutSchema,
    GenTableColumnSchema,
    GenTableOutSchema,
    GenTableQueryParam,
    GenTableSchema,
)
from .tools.gen_util import GenUtils
from .tools.jinja2_template_util import Jinja2TemplateUtil


def handle_service_exception(func: Callable) -> Callable:
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except CustomException:
            raise
        except Exception as e:
            raise CustomException(msg=f"{func.__name__}执行失败: {e!s}")

    return wrapper


_MENU_TYPE_CATALOG = 1  # 与 sys_menu.type、前端 MenuTypeEnum.CATALOG 一致
_MENU_TYPE_MENU = 2


class GenTableService:
    """代码生成业务表服务层"""

    @classmethod
    def _is_example_style(cls, package_name: str | None, module_name: str | None) -> bool:
        """委托 ``Jinja2TemplateUtil.is_example_style``，保证与模板/路径逻辑一致。"""
        return Jinja2TemplateUtil.is_example_style(package_name, module_name)

    @classmethod
    async def _assert_parent_menu_is_catalog(cls, auth: AuthSchema, parent_menu_id: int | None) -> None:
        """上级菜单仅允许目录：与前端树只展示目录一致，避免挂到菜单/按钮下。"""
        if parent_menu_id is None:
            return
        from app.api.v1.module_system.menu.crud import MenuCRUD

        m = await MenuCRUD(auth).get_by_id_crud(parent_menu_id)
        if not m:
            raise CustomException(msg="上级菜单不存在")
        if m.type != _MENU_TYPE_CATALOG:
            raise CustomException(msg="上级菜单须选择目录类型")

    @classmethod
    def _menu_route_first_segment(
        cls, parent_catalog_id: int | None, package_name: str, module_name: str | None
    ) -> str:
        """前端页面路由首段（与菜单 ``route_path`` 第一段一致）。

        - **有上级目录**：使用 ``package_name``（如 ``/gencode/业务``）。
        - **无上级**：使用 ``module_name``（如 ``module_gencode`` → ``/module_gencode/业务``），与侧栏第一层目录一致。

        说明：后端 HTTP API 仍由动态路由 ``module_xxx → /xxx`` 映射，与此前端路径可不同。
        """
        pn = (package_name or "").strip()
        mn = (module_name or "").strip()
        is_ex = cls._is_example_style(pn, mn)
        if parent_catalog_id is not None:
            if not pn:
                raise CustomException(msg="包名不能为空")
            return pn
        # 无上级：示例模式首段为插件包目录 module_xxx；旧模式首段为 module_xxx（原 module_name 字段）
        if is_ex:
            if not pn:
                raise CustomException(msg="包名不能为空")
            return pn if pn.startswith("module_") else f"module_{pn}"
        if mn:
            return mn
        if not pn:
            raise CustomException(msg="包名或模块名至少填写一项以生成路由")
        if pn.startswith("module_"):
            return pn
        return f"module_{pn}"

    @classmethod
    def _catalog_menu_dir_key(
        cls, parent_catalog_id: int | None, package_name: str, module_name: str | None
    ) -> str:
        """菜单上「模块目录」节点的 name（与路由第一段 package 独立）。

        - 有上级目录：``包名（不带 module_ 前缀）``，侧栏为 上级 → 短包名 → 功能菜单 → 按钮。
        - 无上级：``module_包名``，与 ``plugin/module_xxx`` 目录一致，侧栏为 module_包名 → 功能 → 按钮。
        """
        pn = (package_name or "").strip()
        mn = (module_name or "").strip()
        is_ex = cls._is_example_style(pn, mn)
        if parent_catalog_id is not None:
            if not pn:
                raise CustomException(msg="包名不能为空")
            base = pn[len("module_") :] if pn.startswith("module_") else pn
            # 示例模式：同一包下多模块须区分目录节点，避免 name 都为短包名导致串单
            if is_ex and mn:
                return f"{base}_{mn}"
            if pn.startswith("module_"):
                return base
            return pn
        if is_ex:
            if not pn or not mn:
                raise CustomException(msg="包名、模块名不能为空")
            # 根侧栏：按「包+模块」唯一，避免多模块共用一个目录菜单
            return f"{pn}_{mn}"
        if mn:
            return mn
        if not pn:
            raise CustomException(msg="包名或模块名至少填写一项以生成菜单目录")
        if pn.startswith("module_"):
            return pn
        return f"module_{pn}"

    @classmethod
    async def _get_or_create_package_directory_menu(
        cls,
        menu_crud: Any,
        parent_catalog_id: int | None,
        package_name: str,
        module_name: str | None,
        business_name: str,
    ) -> int:
        """创建或复用 type=1 模块目录；固定为「目录 → 菜单 → 按钮」中的第一层目录。"""
        from app.api.v1.module_system.menu.schema import MenuCreateSchema
        from app.utils.common_util import CamelCaseUtil

        pn = (package_name or "").strip()
        if not pn:
            raise CustomException(msg="包名不能为空")
        mn = (module_name or "").strip()
        bn = (business_name or "").strip()
        is_ex = cls._is_example_style(pn, mn)
        dir_key = cls._catalog_menu_dir_key(parent_catalog_id, pn, module_name)

        if parent_catalog_id is not None:
            existing = await menu_crud.get(
                name=dir_key, type=_MENU_TYPE_CATALOG, parent_id=parent_catalog_id
            )
        else:
            existing = await menu_crud.get(
                name=dir_key, type=_MENU_TYPE_CATALOG, parent_id=("None", None)
            )
        if existing:
            log.info(
                f"代码生成：复用模块目录菜单 id={existing.id} name={dir_key!r} parent={parent_catalog_id!r}"
            )
            return int(existing.id)

        route_first = cls._menu_route_first_segment(parent_catalog_id, pn, module_name)
        # 示例模式：目录菜单对应「包/模块」路径；默认跳到模块根，避免多模块共用一个 redirect 指向某一业务
        if is_ex:
            catalog_route_path = f"/{route_first}/{mn}"
            redirect = f"/{route_first}/{mn}"
        else:
            catalog_route_path = f"/{route_first}"
            redirect = f"/{route_first}/{bn}" if bn else f"/{route_first}"
        created = await menu_crud.create(
            MenuCreateSchema(
                name=dir_key,
                type=_MENU_TYPE_CATALOG,
                order=9999,
                permission=None,
                icon="menu",
                route_name=CamelCaseUtil.snake_to_camel(route_first),
                route_path=catalog_route_path,
                component_path=None,
                redirect=redirect,
                hidden=False,
                keep_alive=True,
                always_show=False,
                title=dir_key,
                params=None,
                affix=False,
                parent_id=parent_catalog_id,
                status="0",
                description="模块目录（代码生成）",
            )
        )
        log.info(
            f"代码生成：新建模块目录菜单 id={created.id} name={dir_key!r} under_parent={parent_catalog_id!r}"
        )
        return int(created.id)

    @classmethod
    def normalize_and_validate_master_sub(cls, data: GenTableSchema) -> None:
        """主子表业务规则：两字段同填或同空；子表表名不得与主表相同。"""
        sn = data.sub_table_name
        fk = data.sub_table_fk_name
        if bool(sn) ^ bool(fk):
            raise CustomException(msg="子表表名与子表外键列须同时填写或同时留空")
        tn = (data.table_name or "").strip()
        if sn and fk and sn == tn:
            raise CustomException(msg="子表表名不能与主表表名相同")

    @classmethod
    @handle_service_exception
    async def get_gen_table_detail_service(cls, auth: AuthSchema, table_id: int) -> dict:
        """获取详细信息。

        参数:
        - auth (AuthSchema): 认证信息。
        - table_id (int): 业务表ID。

        返回:
        - dict: 包含业务表详细信息的字典。
        """
        gen_table = await cls.get_gen_table_by_id_service(auth, table_id)
        return gen_table.model_dump()

    @classmethod
    @handle_service_exception
    async def get_gen_table_list_service(
        cls, auth: AuthSchema, search: GenTableQueryParam
    ) -> list[dict]:
        """
        获取代码生成业务表列表信息。

        参数:
        - auth (AuthSchema): 认证信息。
        - search (GenTableQueryParam): 查询参数模型。

        返回:
        - list[dict]: 包含业务表列表信息的字典列表。
        """
        gen_table_list_result = await GenTableCRUD(auth=auth).get_gen_table_list(search)
        return [GenTableOutSchema.model_validate(obj).model_dump() for obj in gen_table_list_result]

    @classmethod
    @handle_service_exception
    async def get_gen_table_page_service(
        cls,
        auth: AuthSchema,
        page_no: int,
        page_size: int,
        search: GenTableQueryParam,
        order_by: list[dict[str, str]] | None = None,
    ) -> dict:
        """分页查询代码生成业务表（数据库 OFFSET/LIMIT）。"""
        offset = (page_no - 1) * page_size
        order = order_by or [{"created_time": "desc"}]
        return await GenTableCRUD(auth=auth).page(
            offset=offset,
            limit=page_size,
            order_by=order,
            search=search.__dict__,
            out_schema=GenTableOutSchema,
        )

    @classmethod
    @handle_service_exception
    async def get_gen_db_table_list_service(
        cls, auth: AuthSchema, search: GenTableQueryParam
    ) -> list[Any]:
        """获取数据库表列表。

        参数:
        - auth (AuthSchema): 认证信息。
        - search (GenTableQueryParam): 查询参数模型。

        返回:
        - list[Any]: 包含数据库表列表信息的任意类型列表。
        """
        gen_db_table_list_result = await GenTableCRUD(auth=auth).get_db_table_list(search)
        return gen_db_table_list_result

    @classmethod
    @handle_service_exception
    async def get_gen_db_table_list_by_name_service(
        cls, auth: AuthSchema, table_names: list[str]
    ) -> list[GenTableOutSchema]:
        """根据表名称组获取数据库表信息。

        参数:
        - auth (AuthSchema): 认证信息。
        - table_names (list[str]): 业务表名称列表。

        返回:
        - list[GenTableOutSchema]: 包含业务表详细信息的模型列表。
        """
        gen_db_table_list_result = await GenTableCRUD(auth).get_db_table_list_by_names(table_names)

        # 修复：将GenDBTableSchema对象转换为字典后再传递给GenTableOutSchema
        result = [
            GenTableOutSchema(**gen_table.model_dump()) for gen_table in gen_db_table_list_result
        ]

        return result

    @classmethod
    @handle_service_exception
    async def import_gen_table_service(
        cls, auth: AuthSchema, gen_table_list: list[GenTableOutSchema]
    ) -> bool:
        """导入表结构到生成器。

        参数:
        - auth (AuthSchema): 认证信息。
        - gen_table_list (list[GenTableOutSchema]): 包含业务表详细信息的模型列表。

        返回:
        - bool: 成功时返回True，失败时抛出异常。
        """
        # 检查是否有表需要导入
        if not gen_table_list:
            raise CustomException(msg="导入的表结构不能为空")
        try:
            for table in gen_table_list:
                _row = {
                    k: v
                    for k, v in table.model_dump().items()
                    if k in GenTableSchema.model_fields
                }
                cls.normalize_and_validate_master_sub(GenTableSchema.model_validate(_row))
                table_name = table.table_name
                # 检查表是否已存在
                existing_table = await GenTableCRUD(auth).get_gen_table_by_name(table_name)
                if existing_table:
                    raise CustomException(msg=f"以下表已存在，不能重复导入: {table_name}")
                GenUtils.init_table(table)
                if not table.columns:
                    table.columns = []
                add_gen_table = await GenTableCRUD(auth).add_gen_table(
                    GenTableSchema.model_validate(table.model_dump())
                )
                gen_table_columns = await GenTableColumnCRUD(auth).get_gen_db_table_columns_by_name(
                    table_name
                )
                if len(gen_table_columns) > 0:
                    table.id = add_gen_table.id
                    for column in gen_table_columns:
                        column_schema = GenTableColumnSchema(
                            table_id=table.id,
                            column_name=column.column_name,
                            column_comment=column.column_comment,
                            column_type=column.column_type,
                            column_length=column.column_length,
                            column_default=column.column_default,
                            is_pk=column.is_pk,
                            is_increment=column.is_increment,
                            is_nullable=column.is_nullable,
                            is_unique=column.is_unique,
                            sort=column.sort,
                            python_type=column.python_type,
                            python_field=column.python_field,
                        )
                        GenUtils.init_column_field(column_schema, table)
                        await GenTableColumnCRUD(auth).create_gen_table_column_crud(column_schema)
            return True
        except Exception as e:
            raise CustomException(msg=f"导入失败, {e!s}")

    @classmethod
    @handle_service_exception
    async def create_table_service(cls, auth: AuthSchema, sql: str) -> bool | None:
        """创建表结构并导入至代码生成模块。

        参数:
        - auth (AuthSchema): 认证信息。
        - sql (str): 包含`CREATE TABLE`语句的SQL字符串。

        返回:
        - bool | None: 成功时返回True，失败时抛出异常。
        """
        # 验证SQL非空
        if not sql or not sql.strip():
            raise CustomException(msg="SQL语句不能为空")
        try:
            # 解析SQL语句
            sql_statements = sqlglot.parse(sql, dialect=settings.DATABASE_TYPE)
            if not sql_statements:
                raise CustomException(msg="无法解析SQL语句，请检查SQL语法")

            # 校验sql语句是否为合法的建表语句
            validate_create = [
                isinstance(sql_statement, Create) for sql_statement in sql_statements
            ]
            validate_forbidden_keywords = [
                isinstance(
                    sql_statement,
                    (Add, Alter, Delete, Drop, Insert, TruncateTable, Update),
                )
                for sql_statement in sql_statements
            ]
            if not any(validate_create) or any(validate_forbidden_keywords):
                raise CustomException(msg="sql语句不是合法的建表语句")

            # 获取要创建的表名
            table_names = []
            for sql_statement in sql_statements:
                if isinstance(sql_statement, Create):
                    table = sql_statement.find(Table)
                    if table and table.name:
                        table_names.append(table.name)
            table_names = list(set(table_names))

            # 创建CRUD实例
            gen_table_crud = GenTableCRUD(auth=auth)

            # 检查每个表是否已存在
            for table_name in table_names:
                # 检查数据库中是否已存在该表
                if await gen_table_crud.check_table_exists(table_name):
                    raise CustomException(msg=f"表 {table_name} 已存在，请检查并修改表名后重试")

                # 检查代码生成模块中是否已导入该表
                existing_table = await gen_table_crud.get_gen_table_by_name(table_name)
                if existing_table:
                    raise CustomException(
                        msg=f"表 {table_name} 已在代码生成模块中存在，请检查并修改表名后重试"
                    )

            # 表不存在，执行SQL语句创建表
            for sql_statement in sql_statements:
                if not isinstance(sql_statement, Create):
                    continue
                exc_sql = sql_statement.sql(dialect=settings.DATABASE_TYPE)
                log.info(f"执行SQL语句: {exc_sql}")
                if not await gen_table_crud.execute_sql(exc_sql):
                    raise CustomException(msg=f"执行SQL语句 {exc_sql} 失败，请检查数据库")
            return True

        except Exception as e:
            raise CustomException(msg=f"创建表结构失败: {e!s}")

    @classmethod
    @handle_service_exception
    async def update_gen_table_service(
        cls, auth: AuthSchema, data: GenTableSchema, table_id: int
    ) -> dict[str, Any]:
        """编辑业务表信息。

        参数:
        - auth (AuthSchema): 认证信息。
        - data (GenTableSchema): 包含业务表详细信息的模型。
        - table_id (int): 业务表ID。

        返回:
        - dict[str, Any]: 更新后的业务表信息。
        """
        # 处理params为None的情况
        gen_table_info = await cls.get_gen_table_by_id_service(auth, table_id)
        if gen_table_info.id:
            try:
                cls.normalize_and_validate_master_sub(data)
                await cls._assert_parent_menu_is_catalog(auth, data.parent_menu_id)
                # 直接调用edit_gen_table方法，它会在内部处理排除嵌套字段的逻辑
                result = await GenTableCRUD(auth).edit_gen_table(table_id, data)
                if not result:
                    raise CustomException(msg="更新业务表信息失败")

                # 处理data.columns为None的情况
                if data.columns:
                    for gen_table_column in data.columns:
                        # 确保column有id字段
                        if hasattr(gen_table_column, "id") and gen_table_column.id:
                            column_schema = GenTableColumnSchema(**gen_table_column.model_dump())
                            await GenTableColumnCRUD(auth).update_gen_table_column_crud(
                                gen_table_column.id, column_schema
                            )
                # 重新获取带有预加载关系的对象，避免懒加载导致的MissingGreenlet错误
                updated_gen_table = await GenTableCRUD(auth).get_gen_table_by_id(table_id)
                out = GenTableOutSchema.model_validate(updated_gen_table)
                await cls.set_pk_column(out)
                await cls.hydrate_sub_table(auth, out)
                return out.model_dump()
            except CustomException:
                raise
            except Exception as e:
                raise CustomException(msg=str(e))
        else:
            raise CustomException(msg="业务表不存在")

    @classmethod
    @handle_service_exception
    async def delete_gen_table_service(cls, auth: AuthSchema, ids: list[int]) -> None:
        """删除业务表信息（先删字段，再删表）。

        参数:
        - auth (AuthSchema): 认证信息。
        - ids (list[int]): 业务表ID列表。

        返回:
        - None
        """
        # 验证ID列表非空
        if not ids:
            raise CustomException(msg="ID列表不能为空")

        try:
            # 先删除相关的字段信息
            await GenTableColumnCRUD(auth=auth).delete_gen_table_column_by_table_id_crud(ids)
            # 再删除表信息
            await GenTableCRUD(auth=auth).delete_gen_table(ids)
        except Exception as e:
            raise CustomException(msg=str(e))

    @classmethod
    @handle_service_exception
    async def get_gen_table_by_id_service(
        cls, auth: AuthSchema, table_id: int
    ) -> GenTableOutSchema:
        """获取需要生成代码的业务表详细信息。

        参数:
        - auth (AuthSchema): 认证信息。
        - table_id (int): 业务表ID。

        返回:
        - GenTableOutSchema: 业务表详细信息模型。
        """
        gen_table = await GenTableCRUD(auth=auth).get_gen_table_by_id(table_id)
        if not gen_table:
            raise CustomException(msg="业务表不存在")

        result = GenTableOutSchema.model_validate(gen_table)
        await cls.set_pk_column(result)
        await cls.hydrate_sub_table(auth, result)
        return result

    @classmethod
    @handle_service_exception
    async def get_gen_table_all_service(cls, auth: AuthSchema) -> list[GenTableOutSchema]:
        """获取所有业务表信息（列表）。

        参数:
        - auth (AuthSchema): 认证信息。

        返回:
        - list[GenTableOutSchema]: 业务表详细信息模型列表。
        """
        gen_table_all = await GenTableCRUD(auth=auth).get_gen_table_all() or []
        result = []
        for gen_table in gen_table_all:
            try:
                table_out = GenTableOutSchema.model_validate(gen_table)
                result.append(table_out)
            except Exception as e:
                log.error(f"转换业务表时出错: {e!s}")
                continue
        return result

    @classmethod
    @handle_service_exception
    async def preview_code_service(cls, auth: AuthSchema, table_id: int) -> dict[str, Any]:
        """
        预览代码（根据模板渲染内存结果）。

        参数:
        - auth (AuthSchema): 认证信息。
        - table_id (int): 业务表ID。

        返回:
        - dict[str, Any]: 文件名到渲染内容的映射。
        """
        raw = await GenTableCRUD(auth).get_gen_table_by_id(table_id)
        if not raw:
            raise CustomException(msg="业务表不存在")
        gen_table = GenTableOutSchema.model_validate(raw)
        await cls.set_pk_column(gen_table)
        await cls.hydrate_sub_table(auth, gen_table)
        cls._assert_master_sub_config_valid(gen_table)
        env = Jinja2TemplateUtil.get_env()
        context = Jinja2TemplateUtil.prepare_context(gen_table)
        template_list = Jinja2TemplateUtil.get_template_list()
        preview_code_result: dict[str, Any] = {}
        for template in template_list:
            try:
                render_content = await env.get_template(template).render_async(**context)
                out_key = Jinja2TemplateUtil.get_file_name(template, gen_table)
                preview_code_result[out_key] = render_content
            except Exception as e:
                log.error(f"渲染模板 {template} 时出错: {e!s}")
                out_key = Jinja2TemplateUtil.get_file_name(template, gen_table)
                preview_code_result[out_key] = f"渲染错误: {e!s}"
        if gen_table.sub and gen_table.sub_table:
            sub_ctx = Jinja2TemplateUtil.prepare_sub_render_context(gen_table, gen_table.sub_table)
            sub_table = gen_table.sub_table
            for template in template_list:
                try:
                    render_content = await env.get_template(template).render_async(**sub_ctx)
                    out_key = Jinja2TemplateUtil.get_file_name(template, sub_table)
                    preview_code_result[out_key] = render_content
                except Exception as e:
                    log.error(f"渲染子表模板 {template} 时出错: {e!s}")
                    out_key = Jinja2TemplateUtil.get_file_name(template, sub_table)
                    preview_code_result[out_key] = f"渲染错误: {e!s}"
        return preview_code_result

    @classmethod
    @handle_service_exception
    async def generate_code_service(cls, auth: AuthSchema, table_name: str) -> bool:
        """生成代码至指定路径（安全写入+可跳过覆盖）。

        菜单固定为 **目录(type=1) + 菜单(type=2) + 按钮(type=3)**：
        - **有上级目录**：``上级目录 / 包名(无 module_ 前缀) / 功能菜单 / 按钮``；页面路由 ``/包名/业务名``。
        - **无上级**：``module_包名 / 功能菜单 / 按钮``；页面路由 ``/module_包名/业务名``（与侧栏第一层一致）。
        - 后端 HTTP 接口仍为动态路由前缀 ``/短名``（``module_xxx``→``/xxx``），与前端页面路由可不同。

        参数:
        - auth (AuthSchema): 认证信息。
        - table_name (str): 业务表名。

        返回:
        - bool: 生成是否成功。
        """
        # 验证表名非空
        if not table_name or not table_name.strip():
            raise CustomException(msg="表名不能为空")
        env = Jinja2TemplateUtil.get_env()
        render_info = await cls.__get_gen_render_info(auth, table_name)
        gen_table_schema: GenTableOutSchema = render_info[3]

        from app.api.v1.module_system.menu.crud import MenuCRUD
        from app.api.v1.module_system.menu.schema import MenuCreateSchema
        from app.utils.common_util import CamelCaseUtil

        is_ex = cls._is_example_style(gen_table_schema.package_name, gen_table_schema.module_name)
        if is_ex:
            if not (gen_table_schema.module_name or "").strip():
                raise CustomException(msg="模块名称不能为空")
            pn = (gen_table_schema.package_name or "").strip()
            mn = (gen_table_schema.module_name or "").strip()
            raw_bn = (gen_table_schema.business_name or "").strip()
            perm_segs = [pn, mn]
            if raw_bn:
                perm_segs.extend(s for s in raw_bn.split("/") if s)
            permission_prefix = ":".join(perm_segs)
            _slug_src = raw_bn or mn
            _business_route_slug = Jinja2TemplateUtil.business_name_to_slug(_slug_src)
        else:
            if not gen_table_schema.business_name:
                raise CustomException(msg="业务名称不能为空")
            _bn_perm = (gen_table_schema.business_name or "").strip().replace("/", ":")
            permission_prefix = f"{gen_table_schema.module_name}:{_bn_perm}"
            _business_route_slug = Jinja2TemplateUtil.business_name_to_slug(gen_table_schema.business_name)
        # 创建菜单 CRUD 实例
        menu_crud = MenuCRUD(auth)
        if not gen_table_schema.function_name:
            raise CustomException(msg="功能名称不能为空")
        if not gen_table_schema.package_name:
            raise CustomException(msg="包名不能为空")
        await cls._assert_parent_menu_is_catalog(auth, gen_table_schema.parent_menu_id)
        # 1. 目录 + 菜单 + 按钮：先取/建模块目录（名称规则见 _catalog_menu_dir_key）
        dir_menu_id = await cls._get_or_create_package_directory_menu(
            menu_crud,
            gen_table_schema.parent_menu_id,
            gen_table_schema.package_name,
            gen_table_schema.module_name,
            gen_table_schema.business_name or "",
        )

        # 检查同一模块目录下是否已有同名功能菜单（避免与其它模块下的同名功能冲突）
        existing_func_menu = await menu_crud.get(
            name=gen_table_schema.function_name,
            type=_MENU_TYPE_MENU,
            parent_id=dir_menu_id,
        )
        if existing_func_menu:
            raise CustomException(
                msg=f"该模块目录下功能菜单「{gen_table_schema.function_name}」已存在，不能重复创建"
            )
        route_seg = cls._menu_route_first_segment(
            gen_table_schema.parent_menu_id,
            gen_table_schema.package_name or "",
            gen_table_schema.module_name,
        )
        if is_ex:
            _pn = (gen_table_schema.package_name or "").strip()
            _mn = (gen_table_schema.module_name or "").strip()
            _bn = (gen_table_schema.business_name or "").strip()
            _rsegs = [route_seg, _mn]
            if _bn:
                _rsegs.extend(s for s in _bn.split("/") if s)
            _route_path = "/" + "/".join(_rsegs)
            _component_path = f"{_pn}/{_mn}" + (f"/{_bn}" if _bn else "") + "/index"
        else:
            _route_path = f"/{route_seg}/{gen_table_schema.business_name}"
            _component_path = f"{gen_table_schema.module_name}/{gen_table_schema.business_name}/index"
        # 创建功能菜单（类型=2：菜单）
        parent_menu = await menu_crud.create(
            MenuCreateSchema(
                name=gen_table_schema.function_name,
                type=_MENU_TYPE_MENU,
                order=9999,
                permission=f"{permission_prefix}:query",
                icon="menu",
                route_name=CamelCaseUtil.snake_to_camel(_business_route_slug),
                route_path=_route_path,
                component_path=_component_path,
                redirect=None,
                hidden=False,
                keep_alive=True,
                always_show=False,
                title=gen_table_schema.function_name,
                params=None,
                affix=False,
                parent_id=dir_menu_id,  # 使用目录菜单ID或用户指定的parent_menu_id作为父ID
                status="0",
                description=f"{gen_table_schema.function_name}功能菜单",
            )
        )
        # 创建按钮权限（类型=3：按钮/权限）
        buttons = [
            {
                "name": f"{gen_table_schema.function_name}查询",
                "permission": f"{permission_prefix}:query",
                "order": 1,
            },
            {
                "name": f"{gen_table_schema.function_name}详情",
                "permission": f"{permission_prefix}:detail",
                "order": 2,
            },
            {
                "name": f"{gen_table_schema.function_name}新增",
                "permission": f"{permission_prefix}:create",
                "order": 3,
            },
            {
                "name": f"{gen_table_schema.function_name}修改",
                "permission": f"{permission_prefix}:update",
                "order": 4,
            },
            {
                "name": f"{gen_table_schema.function_name}删除",
                "permission": f"{permission_prefix}:delete",
                "order": 5,
            },
            {
                "name": f"{gen_table_schema.function_name}批量状态修改",
                "permission": f"{permission_prefix}:patch",
                "order": 6,
            },
            {
                "name": f"{gen_table_schema.function_name}导出",
                "permission": f"{permission_prefix}:export",
                "order": 7,
            },
            {
                "name": f"{gen_table_schema.function_name}导入",
                "permission": f"{permission_prefix}:import",
                "order": 8,
            },
            {
                "name": f"{gen_table_schema.function_name}下载导入模板",
                "permission": f"{permission_prefix}:download",
                "order": 9,
            },
        ]
        for button in buttons:
            # 检查按钮权限是否已存在
            await menu_crud.create(
                MenuCreateSchema(
                    name=button["name"],
                    type=3,
                    order=button["order"],
                    permission=button["permission"],
                    icon=None,
                    route_name=None,
                    route_path=None,
                    component_path=None,
                    redirect=None,
                    hidden=False,
                    keep_alive=True,
                    always_show=False,
                    title=button["name"],
                    params=None,
                    affix=False,
                    parent_id=parent_menu.id,
                    status="0",
                    description=f"{gen_table_schema.function_name}功能按钮",
                )
            )
            log.info(f"成功创建按钮权限: {button['name']}")
        log.info(f"成功创建{gen_table_schema.function_name}菜单及按钮权限")

        # 2. 菜单创建成功后，再生成页面代码（主表 + 可选子表）
        async def _write_templates(
            templates: list[str], ctx: dict[str, Any], table_schema: GenTableOutSchema
        ) -> None:
            for template in templates:
                try:
                    render_content = await env.get_template(template).render_async(**ctx)
                    file_name = Jinja2TemplateUtil.get_file_name(template, table_schema)
                    full_path = BASE_DIR.parent.joinpath(file_name)
                    gen_path = str(full_path)
                    if not gen_path:
                        raise CustomException(msg="【代码生成】生成路径为空")
                    os.makedirs(os.path.dirname(gen_path), exist_ok=True)
                    await anyio.Path(gen_path).write_text(render_content, encoding="utf-8")
                    plugin_root = (
                        (table_schema.package_name or "").strip()
                        if cls._is_example_style(table_schema.package_name, table_schema.module_name)
                        else (table_schema.module_name or "").strip()
                    )
                    if plugin_root:
                        module_init_path = BASE_DIR.parent.joinpath(
                            f"backend/app/plugin/{plugin_root}/__init__.py"
                        )
                        if not module_init_path.exists():
                            os.makedirs(os.path.dirname(module_init_path), exist_ok=True)
                            await anyio.Path(module_init_path).write_text(
                                "# -*- coding: utf-8 -*-", encoding="utf-8"
                            )
                except Exception as e:
                    raise CustomException(
                        msg=f"渲染模板失败，表名：{table_schema.table_name}，详细错误信息：{e!s}"
                    )

        await _write_templates(render_info[0], render_info[2], gen_table_schema)
        if gen_table_schema.sub and gen_table_schema.sub_table:
            sub_ctx = Jinja2TemplateUtil.prepare_sub_render_context(
                gen_table_schema, gen_table_schema.sub_table
            )
            await _write_templates(render_info[0], sub_ctx, gen_table_schema.sub_table)

        return True

    @classmethod
    @handle_service_exception
    async def batch_gen_code_service(cls, auth: AuthSchema, table_names: list[str]) -> bytes:
        """
        批量生成代码并打包为ZIP。
        - 备注：内存生成并压缩，兼容多模板类型；供下载使用。

        参数:
        - auth (AuthSchema): 认证信息。
        - table_names (list[str]): 业务表名列表。

        返回:
        - bytes: 包含所有生成代码的ZIP文件内容。
        """
        valid_names = [t.strip() for t in table_names if t and str(t).strip()]
        if not valid_names:
            raise CustomException(msg="表名列表不能为空")
        zip_buffer = io.BytesIO()
        file_count = 0
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zip_file:
            for table_name in valid_names:
                try:
                    env = Jinja2TemplateUtil.get_env()
                    render_info = await cls.__get_gen_render_info(auth, table_name)
                    gen_tbl = render_info[3]
                    for template_file, output_file in zip(
                        render_info[0], render_info[1], strict=False
                    ):
                        render_content = await env.get_template(template_file).render_async(
                            **render_info[2]
                        )
                        zip_file.writestr(output_file, render_content)
                        file_count += 1
                    if gen_tbl.sub and gen_tbl.sub_table:
                        sub_ctx = Jinja2TemplateUtil.prepare_sub_render_context(
                            gen_tbl, gen_tbl.sub_table
                        )
                        sub_tbl = gen_tbl.sub_table
                        for template_file in render_info[0]:
                            render_content = await env.get_template(template_file).render_async(
                                **sub_ctx
                            )
                            out_path = Jinja2TemplateUtil.get_file_name(template_file, sub_tbl)
                            zip_file.writestr(out_path, render_content)
                            file_count += 1
                except Exception as e:
                    log.error(f"批量生成代码时处理表 {table_name} 出错: {e!s}")
                    # 继续处理其他表，不中断整个过程
                    continue
        zip_data = zip_buffer.getvalue()
        zip_buffer.close()
        if file_count == 0:
            raise CustomException(
                msg="未能生成任何代码文件：请检查所选表是否存在于代码生成配置中，或主子表、字段配置是否正确"
            )
        return zip_data

    @classmethod
    @handle_service_exception
    async def sync_db_service(cls, auth: AuthSchema, table_name: str) -> None:
        """
        同步数据库表结构到业务表。

        参数:
        - auth (AuthSchema): 认证信息。
        - table_name (str): 业务表名。

        返回:
        - None
        """
        # 验证表名非空
        if not table_name or not table_name.strip():
            raise CustomException(msg="表名不能为空")
        gen_table = await GenTableCRUD(auth).get_gen_table_by_name(table_name)
        if not gen_table:
            raise CustomException(msg="业务表不存在")
        table = GenTableOutSchema.model_validate(gen_table)
        if not table.id:
            raise CustomException(msg="业务表ID不能为空")
        table_columns = table.columns or []
        table_column_map = {column.column_name: column for column in table_columns}
        # 确保db_table_columns始终是列表类型，避免None值
        db_table_columns = (
            await GenTableColumnCRUD(auth).get_gen_db_table_columns_by_name(table_name) or []
        )
        db_table_columns = [col for col in db_table_columns if col is not None]
        db_table_column_names = [column.column_name for column in db_table_columns]
        try:
            for column in db_table_columns:
                GenUtils.init_column_field(column, table)
                if column.column_name in table_column_map:
                    prev_column = table_column_map[column.column_name]
                    if hasattr(prev_column, "id") and prev_column.id:
                        column.id = prev_column.id
                    if hasattr(prev_column, "dict_type") and prev_column.dict_type:
                        column.dict_type = prev_column.dict_type
                    if hasattr(prev_column, "query_type") and prev_column.query_type:
                        column.query_type = prev_column.query_type
                    if hasattr(prev_column, "html_type") and prev_column.html_type:
                        column.html_type = prev_column.html_type
                    is_pk_bool = False
                    if hasattr(prev_column, "is_pk"):
                        is_pk_bool = (
                            prev_column.is_pk
                            if isinstance(prev_column.is_pk, bool)
                            else str(prev_column.is_pk) == "1"
                        )
                    if hasattr(prev_column, "is_nullable") and not is_pk_bool:
                        column.is_nullable = prev_column.is_nullable
                    if hasattr(prev_column, "python_field"):
                        column.python_field = prev_column.python_field or column.python_field
                    if hasattr(column, "id") and column.id:
                        await GenTableColumnCRUD(auth).update_gen_table_column_crud(
                            column.id, column
                        )
                    else:
                        await GenTableColumnCRUD(auth).create_gen_table_column_crud(column)
                else:
                    # 设置table_id以确保新字段能正确关联到表
                    column.table_id = table.id
                    await GenTableColumnCRUD(auth).create_gen_table_column_crud(column)
            del_columns = [
                column
                for column in table_columns
                if column.column_name not in db_table_column_names
            ]
            if del_columns:
                for column in del_columns:
                    if hasattr(column, "id") and column.id:
                        await GenTableColumnCRUD(auth).delete_gen_table_column_by_column_id_crud([
                            column.id
                        ])
        except Exception as e:
            raise CustomException(msg=f"同步失败: {e!s}")

    @classmethod
    async def hydrate_sub_table(cls, auth: AuthSchema, gen_table: GenTableOutSchema) -> None:
        """从数据库加载子表列结构，填充 ``sub_table`` 并设置 ``sub``。"""
        gen_table.master_sub_hint = None
        sub_name_raw = (gen_table.sub_table_name or "").strip()
        fk_raw = (gen_table.sub_table_fk_name or "").strip()
        if not sub_name_raw and not fk_raw:
            gen_table.sub = False
            gen_table.sub_table = None
            return
        if sub_name_raw and not fk_raw:
            gen_table.sub = False
            gen_table.sub_table = None
            gen_table.master_sub_hint = "已填写子表表名，请同时填写「子表外键列」后再保存"
            return
        if fk_raw and not sub_name_raw:
            gen_table.sub = False
            gen_table.sub_table = None
            gen_table.master_sub_hint = "已填写子表外键列，请同时填写「子表表名」后再保存"
            return
        if sub_name_raw == (gen_table.table_name or "").strip():
            gen_table.sub = False
            gen_table.sub_table = None
            gen_table.master_sub_hint = "子表表名不能与主表表名相同"
            return
        try:
            gen_table_columns = await GenTableColumnCRUD(auth).get_gen_db_table_columns_by_name(
                sub_name_raw
            )
        except Exception as e:
            log.warning(f"获取子表 {sub_name_raw} 字段失败: {e!s}")
            gen_table.sub = False
            gen_table.sub_table = None
            gen_table.master_sub_hint = f"无法读取子表结构：{e!s}"
            return
        if not gen_table_columns:
            gen_table.sub = False
            gen_table.sub_table = None
            gen_table.master_sub_hint = (
                f"当前数据库中不存在表「{sub_name_raw}」或该表无列，请先建表再配置主子表"
            )
            return
        fk_names = {c.column_name for c in gen_table_columns if c.column_name}
        if fk_raw not in fk_names:
            gen_table.sub = False
            gen_table.sub_table = None
            gen_table.master_sub_hint = (
                f"子表「{sub_name_raw}」中不存在名为「{fk_raw}」的列，请核对外键列名"
            )
            return
        table_comment = await GenTableCRUD(auth).get_db_table_comment(sub_name_raw)
        sub = GenTableOutSchema(
            id=-1,
            table_name=sub_name_raw,
            table_comment=table_comment or None,
            class_name=GenUtils.convert_class_name(sub_name_raw),
            package_name=gen_table.package_name,
            module_name=gen_table.module_name,
            business_name=sub_name_raw,
            function_name=re.sub(r"(?:表|测试)", "", table_comment or "") or sub_name_raw,
            sub_table_name=None,
            sub_table_fk_name=None,
            parent_menu_id=gen_table.parent_menu_id,
            columns=[],
            sub=False,
            sub_table=None,
        )
        for column in gen_table_columns:
            col_dump = column.model_dump()
            col_dump["table_id"] = -1
            col_schema = GenTableColumnSchema.model_validate(col_dump)
            GenUtils.init_column_field(col_schema, sub)
            sub.columns.append(GenTableColumnOutSchema(**col_schema.model_dump()))
        await cls.set_pk_column(sub)
        gen_table.sub = True
        gen_table.sub_table = sub
        gen_table.master_sub_hint = None

    @classmethod
    def _assert_master_sub_config_valid(cls, gen_table: GenTableOutSchema) -> None:
        """预览/生成前校验主子表配置是否可用。"""
        sn = (gen_table.sub_table_name or "").strip()
        fk = (gen_table.sub_table_fk_name or "").strip()
        if not sn and not fk:
            return
        if not sn or not fk:
            raise CustomException(
                msg=gen_table.master_sub_hint
                or "子表表名与子表外键列须同时填写或同时留空"
            )
        if not gen_table.sub_table:
            raise CustomException(
                msg=gen_table.master_sub_hint
                or "无法生成主子表代码：请确认子表已在当前数据库中存在，且外键列名正确"
            )

    @classmethod
    async def set_pk_column(cls, gen_table: GenTableOutSchema) -> None:
        """设置主键列信息（主表/子表）。
        - 备注：同时兼容`pk`布尔与`is_pk == '1'`字符串两种标识。

        参数:
        - gen_table (GenTableOutSchema): 业务表详细信息模型。

        返回:
        - None
        """
        if gen_table.columns:
            for column in gen_table.columns:
                is_pk = getattr(column, "is_pk", False)
                if bool(is_pk) if isinstance(is_pk, bool) else str(is_pk) == "1":
                    gen_table.pk_column = column
                    break
        # 如果没有找到主键列且有列存在，使用第一个列作为主键
        if gen_table.pk_column is None and gen_table.columns:
            gen_table.pk_column = gen_table.columns[0]

    @classmethod
    async def __get_gen_render_info(cls, auth: AuthSchema, table_name: str) -> list[Any]:
        """
        获取生成代码渲染模板相关信息。

        参数:
        - auth (AuthSchema): 认证对象。
        - table_name (str): 业务表名称。

        返回:
        - list[Any]: [模板列表, 输出文件名列表, 渲染上下文, 业务表对象]。

        异常:
        - CustomException: 当业务表不存在或数据转换失败时抛出。
        """
        gen_table_model = await GenTableCRUD(auth=auth).get_gen_table_by_name(table_name)
        # 检查表是否存在
        if gen_table_model is None:
            raise CustomException(msg=f"业务表 {table_name} 不存在")
        gen_table = GenTableOutSchema.model_validate(gen_table_model)
        await cls.set_pk_column(gen_table)
        await cls.hydrate_sub_table(auth, gen_table)
        cls._assert_master_sub_config_valid(gen_table)
        context = Jinja2TemplateUtil.prepare_context(gen_table)
        template_list = Jinja2TemplateUtil.get_template_list()
        output_files = [
            Jinja2TemplateUtil.get_file_name(template, gen_table) for template in template_list
        ]
        return [template_list, output_files, context, gen_table]


class GenTableColumnService:
    """代码生成业务表字段服务层"""

    @classmethod
    @handle_service_exception
    async def get_gen_table_column_list_by_table_id_service(
        cls, auth: AuthSchema, table_id: int
    ) -> list[dict[str, Any]]:
        """获取业务表字段列表信息（输出模型）。

        参数:
        - auth (AuthSchema): 认证信息。
        - table_id (int): 业务表ID。

        返回:
        - list[dict[str, Any]]: 业务表字段列表，每个元素为字段详细信息字典。
        """
        gen_table_column_list_result = await GenTableColumnCRUD(auth).list_gen_table_column_crud({
            "table_id": table_id
        })
        result = [
            GenTableColumnOutSchema.model_validate(gen_table_column).model_dump()
            for gen_table_column in gen_table_column_list_result
        ]
        return result