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 ( Alter, Comment, Create, Delete, Drop, Insert, Table, TruncateTable, Update, ) from app.api.v1.module_system.auth.schema import AuthSchema from app.common.constant import GenConstant 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 ( GenSyncColumnChange, GenSyncPreviewSchema, 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: """ 服务层异步方法装饰器：透传 CustomException，其余异常包装为 CustomException。 参数: - 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 async def _effective_package_name( cls, auth: AuthSchema, parent_catalog_id: int | None, package_name: str | None ) -> str: """根据「是否选择上级目录」计算最终包名（分系统根目录）。 规则（与你描述一致）： - **未选上级目录**：认为是「新分系统」，包名固定为 ``module_目录``（即 ``module_xxx``）。 - **已选上级目录**：认为是「分系统内新模块」，包名继承上级目录对应的 ``module_xxx``。 """ pn = (package_name or "").strip() # 1) 选择上级目录：从上级菜单 route_path 第一段推断 module_xxx if parent_catalog_id is not None: from app.api.v1.module_system.menu.crud import MenuCRUD m = await MenuCRUD(auth).get_by_id_crud(parent_catalog_id) if not m: raise CustomException(msg="上级菜单不存在") route_path = (getattr(m, "route_path", None) or "").strip() # 期望形如 /module_xxx 或 /module_xxx/yyy seg = route_path.strip("/").split("/", 1)[0] if route_path else "" seg = (seg or "").strip() if seg: return seg if seg.startswith("module_") else f"module_{seg}" # 路由缺失则回退包名字段 if pn: return pn if pn.startswith("module_") else f"module_{pn}" raise CustomException(msg="无法从上级目录推断分系统包名，请先完善上级目录路由") # 2) 未选上级目录：以包名字段为准，并确保 module_ 前缀 if not pn: raise CustomException(msg="包名不能为空") return pn if pn.startswith("module_") else f"module_{pn}" @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`` 第一段一致）。 统一规则：始终使用分系统包名 ``module_xxx`` 作为路由首段。 - **无上级目录**：``/module_xxx/...``（新分系统） - **有上级目录**：``/module_xxx/...``（继承上级所属分系统） """ pn = (package_name or "").strip() if not pn: raise CustomException(msg="包名不能为空") return pn if pn.startswith("module_") else 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_name``（你填写的“模块”） - 是否选择上级目录，仅影响分系统根 ``module_xxx`` 的推断方式（见 ``_effective_package_name``） """ pn = (package_name or "").strip() mn = (module_name or "").strip() if not pn: raise CustomException(msg="包名不能为空") if not mn: raise CustomException(msg="模块名不能为空") return mn @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() 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) # 目录菜单固定跳到模块根：/{module_xxx}/{module_name} catalog_route_path = f"/{route_first}/{mn}" redirect = f"/{route_first}/{mn}" # route_name 须唯一且体现「分系统+模块目录」，勿仅用 package（会与 module_example 根混淆） catalog_route_name = CamelCaseUtil.snake_to_camel(f"{route_first}_{mn}") created = await menu_crud.create( MenuCreateSchema( name=dir_key, type=_MENU_TYPE_CATALOG, order=9999, permission=None, icon="menu", route_name=catalog_route_name, 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: """ 主子表业务规则：子表表名与外键列同填或同空；子表表名不得与主表相同。 参数: - data (GenTableSchema): 主表配置。 返回: - None 异常: - CustomException: 规则不满足时抛出。 """ 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）。 参数: - auth (AuthSchema): 认证信息。 - page_no (int): 页码。 - page_size (int): 每页条数。 - search (GenTableQueryParam): 查询条件。 - order_by (list[dict[str, str]] | None): 排序。 返回: - dict: 分页结果。 """ 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_page_service( cls, auth: AuthSchema, page_no: int, page_size: int, search: GenTableQueryParam, ) -> dict[str, Any]: """ 数据库表列表分页（数据库侧 OFFSET/LIMIT）。 参数: - auth (AuthSchema): 认证信息。 - page_no (int): 页码。 - page_size (int): 每页条数。 - search (GenTableQueryParam): 查询条件。 返回: - dict[str, Any]: 含 items、total、has_next 等字段。 """ offset = (page_no - 1) * page_size items, total = await GenTableCRUD(auth=auth).get_db_table_page( search=search, offset=offset, limit=page_size ) return { "items": items, "total": total, "page_no": page_no, "page_size": page_size, "has_next": offset + page_size < total, } @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 是否为合法的建表语句集合： # - 允许：CREATE TABLE + COMMENT ON TABLE/COLUMN +（可选）ALTER TABLE ADD CONSTRAINT ... FOREIGN KEY ... # - 禁止：DROP/DELETE/INSERT/UPDATE/TRUNCATE 等破坏性语句 has_create = any(isinstance(s, Create) for s in sql_statements) if not has_create: raise CustomException(msg="sql语句不是合法的建表语句：缺少 CREATE TABLE") forbidden = (Delete, Drop, Insert, TruncateTable, Update) if any(isinstance(s, forbidden) for s in sql_statements): raise CustomException(msg="sql语句包含禁止的关键操作（DROP/DELETE/INSERT/UPDATE/TRUNCATE）") # 获取要创建的表名 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: # 只执行白名单语句：Create / Comment /（受限）Alter if not isinstance(sql_statement, (Create, Comment, Alter)): continue exc_sql = sql_statement.sql(dialect=settings.DATABASE_TYPE) log.info(f"执行SQL语句: {exc_sql}") # ALTER 仅允许添加外键约束，避免任意 ALTER 带来的破坏性 if isinstance(sql_statement, Alter): upper = exc_sql.upper() allow = ( "ALTER TABLE" in upper and "ADD" in upper and "CONSTRAINT" in upper and "FOREIGN KEY" in upper and "DROP" not in upper and "RENAME" not in upper and "SET " not in upper ) if not allow: raise CustomException( msg="仅允许 ALTER TABLE ADD CONSTRAINT ... FOREIGN KEY ...（拒绝其它 ALTER）" ) 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) # 预览回显的路径/包名规则必须与「写入本地」一致： # - 选择上级目录：继承上级目录所属 module_xxx # - 未选上级目录：使用表单包名（并补齐 module_ 前缀） gen_table.package_name = await cls._effective_package_name( auth, gen_table.parent_menu_id, gen_table.package_name ) # 子表与主表同分系统/同模块 if gen_table.sub and gen_table.sub_table: gen_table.sub_table.package_name = gen_table.package_name if not (gen_table.sub_table.module_name or "").strip(): gen_table.sub_table.module_name = gen_table.module_name 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)**： - **目录层(name)**：固定为你填写的 ``module_name``（模块） - **分系统根(package_name)**： - 选上级目录：继承上级目录所属 ``module_xxx`` - 未选上级目录：使用你填写的包名（自动补齐 ``module_`` 前缀） - **页面路由**：``/{module_xxx}/{module_name}/{business_path}`` - **组件路径**：``module_xxx/module_name/business_path/index`` - **后端 HTTP 接口前缀**：由动态路由发现容器提供 ``/xxx``（``module_xxx``→去掉 ``module_``） 参数: - 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 # 按“上级目录”规则矫正最终包名（分系统根） gen_table_schema.package_name = await cls._effective_package_name( auth, gen_table_schema.parent_menu_id, gen_table_schema.package_name ) # 统一权限前缀（对齐 module_example/demo）： # - module_xxx:module_name（操作在按钮/模板中追加 :query/:create...） pn = (gen_table_schema.package_name or "").strip() mn = (gen_table_schema.module_name or "").strip() if not mn: raise CustomException(msg="模块名不能为空") permission_prefix = ":".join([s for s in [pn, mn] if s]) # 创建菜单 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, ) _pn = (gen_table_schema.package_name or "").strip() _mn = (gen_table_schema.module_name or "").strip() if not _mn: raise CustomException(msg="模块名不能为空") # 与 Jinja2TemplateUtil.get_file_name 统一：module_xxx/{module_name} _route_path = f"/{route_seg}/{_mn}" _component_path = f"{_pn}/{_mn}/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 使用模块名（对齐 module_example/demo：/example/demo） route_name=CamelCaseUtil.snake_to_camel(_mn), 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") # Python 插件目录需保证包层级可导入：为分系统/模块目录补齐 __init__.py # 生成规则固定为 backend/app/plugin/{module_xxx}/{module_name}/... pn = (table_schema.package_name or "").strip() mn = (table_schema.module_name or "").strip() if pn and mn: plugin_base = BASE_DIR.parent.joinpath(f"backend/app/plugin/{pn}") module_base = plugin_base.joinpath(mn) for d in (plugin_base, module_base): init_path = d.joinpath("__init__.py") if not init_path.exists(): os.makedirs(str(d), exist_ok=True) await anyio.Path(str(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, _sync_sub: bool = True) -> 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: # 参考 RuoYi：同步 DB 元信息，但尽量保留用户“生成配置”字段（dict/html/query/python_field...） preserve_keys = { "dict_type", "query_type", "python_field", "python_type", "is_insert", "is_edit", "is_list", "is_query", "sort", } 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 getattr(prev_column, "id", None): column.id = prev_column.id prev_dump = prev_column.model_dump() if hasattr(prev_column, "model_dump") else {} for k in preserve_keys: if k in prev_dump and prev_dump.get(k) not in (None, ""): setattr(column, k, prev_dump.get(k)) # html_type：保留用户显式选择的非 input；若仍是默认 input，则允许按 DB 重新推断 prev_html = prev_dump.get("html_type") if prev_html not in (None, "", GenConstant.HTML_INPUT): column.html_type = prev_html # 主键强约束：避免历史配置导致主键出现在新增/编辑/列表/查询 if bool(getattr(column, "is_pk", False)): column.is_insert = False column.is_edit = False column.is_list = False column.is_query = False column.query_type = None # is_nullable：主键列以 DB 为准，其余保留用户设置 if not bool(getattr(column, "is_pk", False)) and hasattr(prev_column, "is_nullable"): column.is_nullable = prev_column.is_nullable 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 ]) # 主子表：若子表也已导入生成器，则一并同步子表配置（更接近 RuoYi 体验） sn = (table.sub_table_name or "").strip() fk = (table.sub_table_fk_name or "").strip() if _sync_sub and sn and fk: sub_cfg = await GenTableCRUD(auth).get_gen_table_by_name(sn) if sub_cfg: await cls.sync_db_service(auth, sn, _sync_sub=False) except Exception as e: raise CustomException(msg=f"同步失败: {e!s}") @classmethod async def hydrate_sub_table(cls, auth: AuthSchema, gen_table: GenTableOutSchema) -> None: """ 主子表：优先使用已导入的子表配置，否则回退为只读 DB 结构。 对齐 RuoYi：子表宜为独立 gen_table；主表仅引用。 参数: - auth (AuthSchema): 认证信息。 - gen_table (GenTableOutSchema): 主表输出模型（原地填充 sub_table、master_sub_hint 等）。 返回: - None """ 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 # 1) 若子表已作为 gen_table 导入，则使用其 columns 配置（可控、可复用） try: sub_cfg_model = await GenTableCRUD(auth).get_gen_table_by_name(sub_name_raw, preload=["columns"]) except Exception: sub_cfg_model = None if sub_cfg_model: sub_cfg = GenTableOutSchema.model_validate(sub_cfg_model) await cls.set_pk_column(sub_cfg) # 校验外键列存在于子表配置中 fk_names = {c.column_name for c in (sub_cfg.columns or []) 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 gen_table.sub = True gen_table.sub_table = sub_cfg gen_table.master_sub_hint = ( "主子表已启用：子表字段来自「已导入的子表配置」（更接近 RuoYi 的方式）。" "如需调整子表字段，请在列表中打开该子表进行配置。" ) return # 2) 回退：仅从 DB 读取结构（只读，无法配置子表字段） 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.model_validate( { "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) if sub.columns is None: sub.columns = [] 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 = ( "主子表已启用：当前子表仅从数据库结构读取（只读）。" f"若要像 RuoYi 那样可配置子表字段，请先在「导入」中把子表「{sub_name_raw}」也导入生成器。" ) @classmethod def _sync_preview_diff( cls, current_cols: list[GenTableColumnOutSchema], db_cols: list[GenTableColumnOutSchema], ) -> tuple[list[str], list[str], list[GenSyncColumnChange], int]: cur_map = {c.column_name: c for c in (current_cols or []) if c and c.column_name} db_map = {c.column_name: c for c in (db_cols or []) if c and c.column_name} cur_names = set(cur_map.keys()) db_names = set(db_map.keys()) added = sorted(db_names - cur_names) removed = sorted(cur_names - db_names) changed: list[GenSyncColumnChange] = [] unchanged = 0 keys = [ "column_type", "column_comment", "column_default", "column_length", "is_pk", "is_increment", "is_nullable", "is_unique", ] for name in sorted(cur_names & db_names): before = cur_map[name] after = db_map[name] diff_fields: list[str] = [] b_dump = before.model_dump() a_dump = after.model_dump() for k in keys: if b_dump.get(k) != a_dump.get(k): diff_fields.append(k) if diff_fields: changed.append( GenSyncColumnChange( column_name=name, change_fields=diff_fields, before={k: b_dump.get(k) for k in keys}, after={k: a_dump.get(k) for k in keys}, ) ) else: unchanged += 1 return added, removed, changed, unchanged @classmethod @handle_service_exception async def sync_db_preview_service( cls, auth: AuthSchema, table_name: str ) -> dict[str, Any]: """ 同步数据库前差异预览（主表 + 可选子表）。 参数: - auth (AuthSchema): 认证信息。 - table_name (str): 主表物理表名。 返回: - dict[str, Any]: 预览差异结构（可序列化）。 异常: - CustomException: 表名无效或业务表不存在等。 """ if not table_name or not table_name.strip(): raise CustomException(msg="表名不能为空") gen_table = await GenTableCRUD(auth).get_gen_table_by_name(table_name, preload=["columns"]) if not gen_table: raise CustomException(msg="业务表不存在") table = GenTableOutSchema.model_validate(gen_table) if not table.id: raise CustomException(msg="业务表ID不能为空") db_cols = await GenTableColumnCRUD(auth).get_gen_db_table_columns_by_name(table_name) added, removed, changed, unchanged = cls._sync_preview_diff( current_cols=table.columns or [], db_cols=db_cols or [], ) preview = GenSyncPreviewSchema( table_name=table_name, added=added, removed=removed, changed=changed, unchanged=unchanged, ) # 子表差异：如果启用主子表，则同时预览子表（无论子表是否已导入） sn = (table.sub_table_name or "").strip() fk = (table.sub_table_fk_name or "").strip() if sn and fk: preview.sub_table_name = sn # 优先取“已导入的子表配置”，否则用 DB 结构（只读） sub_cfg = await GenTableCRUD(auth).get_gen_table_by_name(sn, preload=["columns"]) if sub_cfg: cur_sub_cols = GenTableOutSchema.model_validate(sub_cfg).columns or [] else: cur_sub_cols = [] db_sub_cols = await GenTableColumnCRUD(auth).get_gen_db_table_columns_by_name(sn) s_added, s_removed, s_changed, s_unchanged = cls._sync_preview_diff( current_cols=cur_sub_cols, db_cols=db_sub_cols or [], ) preview.sub = GenSyncPreviewSchema( table_name=sn, added=s_added, removed=s_removed, changed=s_changed, unchanged=s_unchanged, ) return preview.model_dump() @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) # 生成代码时按“上级目录”规则矫正最终包名（不落库，仅影响本次生成/预览/下载/写入） gen_table.package_name = await cls._effective_package_name( auth, gen_table.parent_menu_id, gen_table.package_name ) 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