group_manage/ruoyi_generator/service/__init__.py

# -*- coding: utf-8 -*-
# @Author  : YY
# @FileName: __init__.py

import json
from typing import List, Optional, Tuple
from datetime import datetime

from ruoyi_common.utils import DateUtil, StringUtil
from ruoyi_generator.domain.entity import GenTable, GenTableColumn
from ruoyi_generator.mapper import gen_table_mapper, gen_table_column_mapper
from ruoyi_generator.util import GenUtils, to_underscore
from ruoyi_generator.config import GeneratorConfig


class GenTableService:
    def select_gen_table_list(self, gen_table: GenTable) -> Tuple[List[GenTable], int]:
        """
        查询代码生成表列表

        Args:
            gen_table (GenTable): 代码生成表对象

        Returns:
            Tuple[List[GenTable], int]: 代码生成表列表和总数
        """
        # 查询列表
        gen_tables = gen_table_mapper.select_list(gen_table)
        # 查询总数
        # 注意：这里需要根据实际需求实现总数查询逻辑
        total = len(gen_tables)
        return gen_tables, total

    def select_db_table_list(self, gen_table: GenTable) -> Tuple[List[GenTable], int]:
        """
        查询数据库表列表

        Args:
            gen_table (GenTable): 代码生成表对象

        Returns:
            Tuple[List[GenTable], int]: 数据库表列表和总数
        """
        # 查询列表
        gen_tables = gen_table_mapper.select_db_list(gen_table)
        # 查询总数
        # 注意：这里需要根据实际需求实现总数查询逻辑
        total = len(gen_tables)
        return gen_tables, total

    def select_gen_table_by_id(self, table_id: int) -> Optional[GenTable]:
        """
        根据ID查询代码生成表

        Args:
            table_id (int): 表ID

        Returns:
            Optional[GenTable]: 代码生成表对象
        """
        gen_table = gen_table_mapper.select_by_id(table_id)
        if gen_table:
            gen_table.columns = gen_table_column_mapper.select_list_by_table_id(table_id)
        return gen_table

    def select_gen_table_by_name(self, table_name: str) -> Optional[GenTable]:
        """
        根据表名查询代码生成表

        Args:
            table_name (str): 表名

        Returns:
            Optional[GenTable]: 代码生成表对象
        """
        return gen_table_mapper.select_by_table_name(table_name)

    def delete_gen_table_by_id(self, table_id: int):
        """
        根据ID删除代码生成表

        Args:
            table_id (int): 表ID
        """
        # 先删除字段信息
        gen_table_column_mapper.delete_by_table_id(table_id)
        # 再删除表信息
        gen_table_mapper.delete_by_id(table_id)

    def delete_gen_table_by_ids(self, table_ids: List[int]):
        """
        批量删除代码生成表

        Args:
            table_ids (List[int]): 表ID列表
        """
        for table_id in table_ids:
            # 先删除字段信息
            gen_table_column_mapper.delete_by_table_id(table_id)
            # 再删除表信息
            gen_table_mapper.delete_by_id(table_id)

    def import_gen_table(self, table_names: List[str]) -> int:
        """
        导入代码生成表

        Args:
            table_names (List[str]): 表名列表

        Returns:
            int: 导入的表数量
        """
        success_count = 0
        for table_name in table_names:
            # 检查表是否已存在
            if gen_table_mapper.exists_table(table_name):
                continue

            # 创建GenTable对象
            table = GenTable()

            # 设置默认值
            table.table_name = table_name
            clean_table_name = GenUtils.remove_table_prefix(
                table_name) if GeneratorConfig.auto_remove_pre else table_name
            table.class_name = to_underscore(clean_table_name)
            table.business_name = GenUtils.get_business_name(clean_table_name)
            table.package_name = GeneratorConfig.package_name
            table.module_name = StringUtil.substring_before(clean_table_name, "_") if hasattr(StringUtil,
                                                                                              'substring_before') and "_" in clean_table_name else clean_table_name

            # 获取表注释
            try:
                result = gen_table_mapper.select_db_table_comment_by_name(table_name)
                table.table_comment = result if result else table_name
            except:
                table.table_comment = table_name

            table.function_name = table.table_comment
            table.function_author = GeneratorConfig.author
            table.create_by = "admin"
            table.create_time = datetime.now()

            # 保存表信息
            table_id = gen_table_mapper.insert(table)
            if table_id > 0:
                # 保存列信息
                columns = gen_table_mapper.select_db_table_columns_by_name(table_name)
                for column in columns:
                    column.table_id = table_id
                    column.create_by = "admin"
                    column.create_time = datetime.now()
                    gen_table_column_mapper.insert(column)
                success_count += 1
        return success_count

    def update_gen_table(self, gen_table: GenTable):
        """
        更新代码生成表

        Args:
            gen_table (GenTable): 代码生成表对象
        """
        print("更新代码生成表:", gen_table.columns)
        # 获取列信息
        columns = gen_table.columns
        if columns:
            # 处理列信息
            for column in columns:
                print("处理列信息:", column)
                # 确保column是GenTableColumn对象而不是dict
                column_obj = None
                if isinstance(column, dict):
                    # 直接使用字典数据创建对象，避免手动字段映射
                    # 前端传过来的已经是正确的字符串格式，直接使用
                    print("原始列数据:", column)

                    # 使用 model_validate 方法处理别名字段
                    column_obj = GenTableColumn.model_validate(column)
                    print("处理后列对象:", column_obj.model_dump())
                else:
                    column_obj = column

                # 设置表ID
                column_obj.table_id = gen_table.table_id
                column_obj.update_time = datetime.now()

                # 如果有column_id则更新，否则插入
                if hasattr(column_obj, 'column_id') and column_obj.column_id:
                    # 更新列信息
                    gen_table_column_mapper.update(column_obj)
                else:
                    # 插入列信息
                    column_obj.create_by = "admin"
                    column_obj.create_time = datetime.now()
                    gen_table_column_mapper.insert(column_obj)
        # 更新表信息
        gen_table.update_time = datetime.now()
        gen_table_mapper.update(gen_table)

    def synch_db(self, table_name: str):
        """
        同步数据库表结构

        Args:
            table_name (str): 表名
        """
        try:
            # 查询表信息
            gen_table = gen_table_mapper.select_by_table_name(table_name)
            if not gen_table:
                raise Exception(f"表{table_name}不存在")

            # 查询数据库表列信息
            db_columns = gen_table_mapper.select_db_table_columns_by_name(table_name)
            # 查询代码生成表列信息
            gen_columns = gen_table_column_mapper.select_list_by_table_id(gen_table.table_id)

            # 处理新增和更新的列
            for db_column in db_columns:
                exist_column = None
                for gen_column in gen_columns:
                    if db_column.column_name == gen_column.column_name:
                        exist_column = gen_column
                        break

                if exist_column:
                    # 更新列信息
                    exist_column.column_comment = db_column.column_comment
                    exist_column.column_type = db_column.column_type
                    exist_column.java_type = db_column.java_type
                    exist_column.java_field = db_column.java_field
                    exist_column.is_pk = db_column.is_pk
                    exist_column.is_increment = db_column.is_increment
                    exist_column.is_required = db_column.is_required
                    exist_column.update_by = "admin"
                    exist_column.update_time = datetime.now()
                    gen_table_column_mapper.update(exist_column)
                else:
                    # 新增列信息
                    db_column.table_id = gen_table.table_id
                    db_column.create_by = "admin"
                    db_column.create_time = datetime.now()
                    gen_table_column_mapper.insert(db_column)

            # 处理删除的列
            for gen_column in gen_columns:
                exist_column = False
                for db_column in db_columns:
                    if gen_column.column_name == db_column.column_name:
                        exist_column = True
                        break

                if not exist_column:
                    # 删除列信息
                    gen_table_column_mapper.delete_by_id(gen_column.column_id)
            
            return True
        except Exception as e:
            print(f"同步数据库失败: {e}")
            raise e

    def generator_code(self, table_name: str) -> bytes:
        """
        生成代码

        Args:
            table_name (str): 表名

        Returns:
            bytes: 生成的代码文件
        """
        # 查询表信息
        gen_table = gen_table_mapper.select_by_table_name(table_name)
        if not gen_table:
            raise Exception(f"表{table_name}不存在")

        # 查询列信息
        gen_table.columns = gen_table_column_mapper.select_list_by_table_id(gen_table.table_id)

        # 生成代码
        return GenUtils.generator_code(gen_table).getvalue()

    def batch_generator_code(self, table_names: List[str]) -> bytes:
        """
        批量生成代码

        Args:
            table_names (List[str]): 表名列表

        Returns:
            bytes: 生成的代码文件
        """
        gen_tables = []
        for table_name in table_names:
            # 查询表信息
            gen_table = gen_table_mapper.select_by_table_name(table_name)
            if gen_table:
                # 查询列信息
                gen_table.columns = gen_table_column_mapper.select_list_by_table_id(gen_table.table_id)
                gen_tables.append(gen_table)

        # 生成代码
        return GenUtils.batch_generator_code(gen_tables).getvalue()

    def preview_code(self, table_id: int) -> dict:
        """
        预览代码

        Args:
            table_id (int): 表ID

        Returns:
            dict: 预览代码
        """
        # 查询表信息
        gen_table = gen_table_mapper.select_by_id(table_id)
        if not gen_table:
            raise Exception(f"表ID{table_id}不存在")

        # 查询列信息
        gen_table.columns = gen_table_column_mapper.select_list_by_table_id(table_id)

        # 预览代码
        return GenUtils.preview_code(gen_table)

    def select_db_table_comment_by_name(self, table_name: str) -> Optional[str]:
        """
        根据表名查询数据库表注释

        Args:
            table_name (str): 表名

        Returns:
            Optional[str]: 表注释
        """
        try:
            result = gen_table_mapper.session.execute(
                text(
                    "SELECT table_comment FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = :table_name"),
                {"table_name": table_name}
            ).fetchone()
            return result[0] if result else None
        except Exception:
            return None


# 实例化Service
gen_table_service = GenTableService()