feat: add cocker-compose yaml

python_core/cli/cli.py

@@ -12,6 +12,7 @@ from python_core.cli.commands import scene_detect
 from python_core.cli.commands.template import template_app
 from python_core.cli.commands.resource_category import resource_category_app
 from python_core.cli.commands.auth import auth_app
+from python_core.cli.commands.model import model_app
 
 app = typer.Typer(
     name="mixvideo",
@@ -35,6 +36,7 @@ app.add_typer(scene_detect, name="scene")
 app.add_typer(template_app, name="template")
 app.add_typer(resource_category_app, name="resource-category")
 app.add_typer(auth_app, name="auth")
+app.add_typer(model_app, name="model")
 
 @app.command()
 def init():

python_core/cli/commands/model.py

@@ -0,0 +1,307 @@
+"""
+模特管理 CLI 命令
+"""
+
+from pathlib import Path
+from typing import Optional, List
+from dataclasses import asdict
+from datetime import datetime
+import typer
+from python_core.utils.jsonrpc_enhanced import create_response_handler, create_progress_reporter
+from python_core.database.model_postgres import model_table
+from python_core.utils.logger import logger
+from uuid import uuid4
+
+# 创建模特应用
+model_app = typer.Typer(help="模特管理命令")
+
+
+@model_app.command("create")
+def create_model(
+    model_number: str = typer.Argument(..., help="模特编号"),
+    model_image: str = typer.Argument(..., help="模特图片路径"),
+    user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
+    is_cloud: bool = typer.Option(False, "--cloud", help="是否为云端模特"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
+    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
+):
+    """创建模特"""
+    response = create_response_handler()
+    try:
+        # 创建模特
+        model = model_table.create_model(
+            model_number=model_number,
+            model_image=model_image,
+            user_id=user_id or "default",
+            is_cloud=is_cloud
+        )
+        
+        if model:
+            response.success({
+                'model': asdict(model),
+                'message': f'模特创建成功: {model.model_number}'
+            })
+        else:
+            response.error(-32603, "创建模特失败")
+            
+    except ValueError as e:
+        response.error(-32602, str(e))
+    except Exception as e:
+        logger.error(f"创建模特失败: {e}")
+        response.error(-32603, f"创建模特失败: {str(e)}")
+
+
+@model_app.command("list")
+def list_models(
+    user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
+    include_cloud: bool = typer.Option(True, "--include-cloud", help="包含云端模特"),
+    include_inactive: bool = typer.Option(False, "--include-inactive", help="包含已禁用模特"),
+    limit: int = typer.Option(100, "--limit", "-l", help="显示数量限制"),
+    offset: int = typer.Option(0, "--offset", help="偏移量"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
+    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
+):
+    """获取模特列表"""
+    response = create_response_handler()
+    try:
+        # 获取模特列表
+        models = model_table.get_all_models(
+            user_id=user_id or "default",
+            include_cloud=include_cloud,
+            include_inactive=include_inactive,
+            limit=limit,
+            offset=offset
+        )
+        
+        # 获取总数
+        total_count = model_table.get_model_count(
+            user_id=user_id or "default",
+            include_cloud=include_cloud,
+            include_inactive=include_inactive
+        )
+        
+        response.success({
+            'models': [asdict(model) for model in models],
+            'total_count': total_count,
+            'limit': limit,
+            'offset': offset,
+            'message': f'获取到 {len(models)} 个模特'
+        })
+        
+    except Exception as e:
+        logger.error(f"获取模特列表失败: {e}")
+        response.error(-32603, f"获取模特列表失败: {str(e)}")
+
+
+@model_app.command("get")
+def get_model(
+    model_id: str = typer.Argument(..., help="模特ID"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
+    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
+):
+    """获取模特详情"""
+    response = create_response_handler()
+    try:
+        model = model_table.get_model_by_id(model_id)
+        
+        if model:
+            response.success({
+                'model': asdict(model),
+                'message': f'获取模特成功: {model.model_number}'
+            })
+        else:
+            response.error(-32604, f"模特不存在: {model_id}")
+            
+    except Exception as e:
+        logger.error(f"获取模特失败: {e}")
+        response.error(-32603, f"获取模特失败: {str(e)}")
+
+
+@model_app.command("get-by-number")
+def get_model_by_number(
+    model_number: str = typer.Argument(..., help="模特编号"),
+    user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
+    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
+):
+    """根据编号获取模特"""
+    response = create_response_handler()
+    try:
+        model = model_table.get_model_by_number(
+            model_number=model_number,
+            user_id=user_id or "default"
+        )
+        
+        if model:
+            response.success({
+                'model': asdict(model),
+                'message': f'获取模特成功: {model.model_number}'
+            })
+        else:
+            response.error(-32604, f"模特不存在: {model_number}")
+            
+    except Exception as e:
+        logger.error(f"获取模特失败: {e}")
+        response.error(-32603, f"获取模特失败: {str(e)}")
+
+
+@model_app.command("update")
+def update_model(
+    model_id: str = typer.Argument(..., help="模特ID"),
+    model_number: Optional[str] = typer.Option(None, "--model-number", help="模特编号"),
+    model_image: Optional[str] = typer.Option(None, "--model-image", help="模特图片路径"),
+    is_active: Optional[bool] = typer.Option(None, "--active", help="是否激活"),
+    is_cloud: Optional[bool] = typer.Option(None, "--cloud", help="是否为云端模特"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
+    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
+):
+    """更新模特"""
+    response = create_response_handler()
+    try:
+        # 构建更新数据
+        updates = {}
+        if model_number is not None:
+            updates['model_number'] = model_number
+        if model_image is not None:
+            updates['model_image'] = model_image
+        if is_active is not None:
+            updates['is_active'] = is_active
+        if is_cloud is not None:
+            updates['is_cloud'] = is_cloud
+        
+        if not updates:
+            response.error(-32602, "没有提供更新字段")
+            return
+        
+        success = model_table.update_model(model_id, updates)
+        
+        if success:
+            response.success({
+                'model_id': model_id,
+                'updates': updates,
+                'message': '模特更新成功'
+            })
+        else:
+            response.error(-32604, f"模特不存在: {model_id}")
+            
+    except ValueError as e:
+        response.error(-32602, str(e))
+    except Exception as e:
+        logger.error(f"更新模特失败: {e}")
+        response.error(-32603, f"更新模特失败: {str(e)}")
+
+
+@model_app.command("delete")
+def delete_model(
+    model_id: str = typer.Argument(..., help="模特ID"),
+    hard_delete: bool = typer.Option(False, "--hard", help="硬删除（永久删除）"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
+    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
+):
+    """删除模特"""
+    response = create_response_handler()
+    try:
+        success = model_table.delete_model(model_id, hard_delete=hard_delete)
+        
+        if success:
+            action = "删除" if hard_delete else "禁用"
+            response.success({
+                'model_id': model_id,
+                'hard_delete': hard_delete,
+                'message': f'模特{action}成功'
+            })
+        else:
+            response.error(-32604, f"模特不存在: {model_id}")
+            
+    except Exception as e:
+        logger.error(f"删除模特失败: {e}")
+        response.error(-32603, f"删除模特失败: {str(e)}")
+
+
+@model_app.command("search")
+def search_models(
+    query: str = typer.Argument(..., help="搜索关键词"),
+    user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
+    include_cloud: bool = typer.Option(True, "--include-cloud", help="包含云端模特"),
+    limit: int = typer.Option(50, "--limit", "-l", help="显示数量限制"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
+    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
+):
+    """搜索模特"""
+    response = create_response_handler()
+    try:
+        models = model_table.search_models(
+            query=query,
+            user_id=user_id or "default",
+            include_cloud=include_cloud,
+            limit=limit
+        )
+        
+        response.success({
+            'models': [asdict(model) for model in models],
+            'query': query,
+            'count': len(models),
+            'message': f'搜索到 {len(models)} 个模特'
+        })
+        
+    except Exception as e:
+        logger.error(f"搜索模特失败: {e}")
+        response.error(-32603, f"搜索模特失败: {str(e)}")
+
+
+@model_app.command("toggle")
+def toggle_model_status(
+    model_id: str = typer.Argument(..., help="模特ID"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
+    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
+):
+    """切换模特状态"""
+    response = create_response_handler()
+    try:
+        success = model_table.toggle_model_status(model_id)
+        
+        if success:
+            response.success({
+                'model_id': model_id,
+                'message': '模特状态切换成功'
+            })
+        else:
+            response.error(-32604, f"模特不存在: {model_id}")
+            
+    except Exception as e:
+        logger.error(f"切换模特状态失败: {e}")
+        response.error(-32603, f"切换模特状态失败: {str(e)}")
+
+
+@model_app.command("count")
+def get_model_count(
+    user_id: Optional[str] = typer.Option(None, "--user-id", help="用户ID"),
+    include_cloud: bool = typer.Option(True, "--include-cloud", help="包含云端模特"),
+    include_inactive: bool = typer.Option(False, "--include-inactive", help="包含已禁用模特"),
+    verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出"),
+    json_output: bool = typer.Option(True, "--json", help="JSON格式输出")
+):
+    """获取模特数量"""
+    response = create_response_handler()
+    try:
+        count = model_table.get_model_count(
+            user_id=user_id or "default",
+            include_cloud=include_cloud,
+            include_inactive=include_inactive
+        )
+        
+        response.success({
+            'count': count,
+            'user_id': user_id or "default",
+            'include_cloud': include_cloud,
+            'include_inactive': include_inactive,
+            'message': f'共有 {count} 个模特'
+        })
+        
+    except Exception as e:
+        logger.error(f"获取模特数量失败: {e}")
+        response.error(-32603, f"获取模特数量失败: {str(e)}")
+
+
+if __name__ == "__main__":
+    model_app()

python_core/config.py

@@ -18,7 +18,7 @@ except ImportError:
 project_root = Path(__file__).parent.parent
 class Settings(BaseSettings):
     """Application settings with environment variable support."""
-    db: str = "postgres://192.168.0.126:5433/mixvideo"
+    db: str = "postgres://mixvideo_user:mixvideo_password@192.168.0.126:5433/mixvideo"
     # Application Info
     app_name: str = "MixVideo V2"
     app_version: str = "2.0.0"

python_core/database/model.py

@@ -1,8 +0,0 @@
-# 模特表
-
-from python_core.kv import kv
-from .db import Db
-class ModelDb(Db):
-    def __init__(self):
-        self.key = "model"
-        pass

python_core/database/model_postgres.py

@@ -0,0 +1,398 @@
+"""
+模特表 PostgreSQL 实现
+"""
+
+import uuid
+from datetime import datetime
+from typing import List, Optional, Dict, Any
+import psycopg2
+from psycopg2.extras import RealDictCursor
+from python_core.database.types import Model
+from python_core.database.db_postgres import DatabasePostgres
+from python_core.utils.logger import logger
+
+
+class ModelTablePostgres(DatabasePostgres):
+    """模特表 PostgreSQL 实现"""
+    
+    def __init__(self):
+        super().__init__()
+        self._init_model_table()
+    
+    def _init_model_table(self):
+        """初始化模特表"""
+        try:
+            with self.get_connection() as conn:
+                with conn.cursor() as cursor:
+                    # 创建模特表
+                    cursor.execute("""
+                        CREATE TABLE IF NOT EXISTS models (
+                            id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+                            model_number VARCHAR(100) NOT NULL,
+                            model_image VARCHAR(500) NOT NULL,
+                            is_active BOOLEAN DEFAULT true,
+                            is_cloud BOOLEAN DEFAULT false,
+                            user_id VARCHAR(100) NOT NULL DEFAULT 'default',
+                            created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
+                            updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
+                            UNIQUE(model_number, user_id)
+                        )
+                    """)
+                    
+                    # 创建索引
+                    cursor.execute("""
+                        CREATE INDEX IF NOT EXISTS idx_models_user_id ON models(user_id)
+                    """)
+                    cursor.execute("""
+                        CREATE INDEX IF NOT EXISTS idx_models_model_number ON models(model_number)
+                    """)
+                    cursor.execute("""
+                        CREATE INDEX IF NOT EXISTS idx_models_is_active ON models(is_active)
+                    """)
+                    cursor.execute("""
+                        CREATE INDEX IF NOT EXISTS idx_models_is_cloud ON models(is_cloud)
+                    """)
+                    cursor.execute("""
+                        CREATE INDEX IF NOT EXISTS idx_models_created_at ON models(created_at)
+                    """)
+                    
+                    # 创建更新时间触发器
+                    cursor.execute("""
+                        CREATE OR REPLACE FUNCTION update_models_updated_at()
+                        RETURNS TRIGGER AS $$
+                        BEGIN
+                            NEW.updated_at = CURRENT_TIMESTAMP;
+                            RETURN NEW;
+                        END;
+                        $$ language 'plpgsql';
+                    """)
+                    
+                    cursor.execute("""
+                        DROP TRIGGER IF EXISTS update_models_updated_at_trigger ON models;
+                        CREATE TRIGGER update_models_updated_at_trigger
+                            BEFORE UPDATE ON models
+                            FOR EACH ROW
+                            EXECUTE FUNCTION update_models_updated_at();
+                    """)
+                    
+                    conn.commit()
+                    logger.info("模特表初始化成功")
+                    
+        except Exception as e:
+            logger.error(f"初始化模特表失败: {e}")
+            raise
+    
+    def _row_to_model(self, row: Dict[str, Any]) -> Model:
+        """将数据库行转换为 Model 对象"""
+        return Model(
+            id=str(row['id']),
+            model_number=row['model_number'],
+            model_image=row['model_image'],
+            is_active=row['is_active'],
+            is_cloud=row['is_cloud'],
+            user_id=row['user_id'],
+            created_at=row['created_at'].isoformat() if row['created_at'] else '',
+            updated_at=row['updated_at'].isoformat() if row['updated_at'] else ''
+        )
+    
+    def create_model(self, model_number: str, model_image: str, user_id: str = "default", 
+                    is_cloud: bool = False) -> Optional[Model]:
+        """创建模特"""
+        try:
+            with self.get_connection() as conn:
+                with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+                    cursor.execute("""
+                        INSERT INTO models (model_number, model_image, user_id, is_cloud)
+                        VALUES (%s, %s, %s, %s)
+                        RETURNING *
+                    """, (model_number, model_image, user_id, is_cloud))
+                    
+                    row = cursor.fetchone()
+                    conn.commit()
+                    
+                    if row:
+                        model = self._row_to_model(row)
+                        logger.info(f"创建模特成功: {model.model_number}")
+                        return model
+                    
+        except psycopg2.IntegrityError as e:
+            if "unique" in str(e).lower():
+                logger.warning(f"模特编号已存在: {model_number}")
+                raise ValueError(f"模特编号 '{model_number}' 已存在")
+            else:
+                logger.error(f"创建模特失败: {e}")
+                raise
+        except Exception as e:
+            logger.error(f"创建模特失败: {e}")
+            raise
+        
+        return None
+    
+    def get_model_by_id(self, model_id: str) -> Optional[Model]:
+        """根据ID获取模特"""
+        try:
+            with self.get_connection() as conn:
+                with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+                    cursor.execute("""
+                        SELECT * FROM models WHERE id = %s
+                    """, (model_id,))
+                    
+                    row = cursor.fetchone()
+                    if row:
+                        return self._row_to_model(row)
+                    
+        except Exception as e:
+            logger.error(f"获取模特失败: {e}")
+            raise
+        
+        return None
+    
+    def get_model_by_number(self, model_number: str, user_id: str = "default") -> Optional[Model]:
+        """根据模特编号获取模特"""
+        try:
+            with self.get_connection() as conn:
+                with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+                    cursor.execute("""
+                        SELECT * FROM models 
+                        WHERE model_number = %s AND user_id = %s
+                    """, (model_number, user_id))
+                    
+                    row = cursor.fetchone()
+                    if row:
+                        return self._row_to_model(row)
+                    
+        except Exception as e:
+            logger.error(f"获取模特失败: {e}")
+            raise
+        
+        return None
+    
+    def get_all_models(self, user_id: str = "default", include_cloud: bool = True, 
+                      include_inactive: bool = False, limit: int = 100, 
+                      offset: int = 0) -> List[Model]:
+        """获取所有模特"""
+        try:
+            with self.get_connection() as conn:
+                with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+                    # 构建查询条件
+                    conditions = []
+                    params = []
+                    
+                    if include_cloud:
+                        conditions.append("(user_id = %s OR is_cloud = true)")
+                        params.append(user_id)
+                    else:
+                        conditions.append("user_id = %s")
+                        params.append(user_id)
+                    
+                    if not include_inactive:
+                        conditions.append("is_active = true")
+                    
+                    where_clause = " AND ".join(conditions) if conditions else "1=1"
+                    
+                    cursor.execute(f"""
+                        SELECT * FROM models 
+                        WHERE {where_clause}
+                        ORDER BY created_at DESC
+                        LIMIT %s OFFSET %s
+                    """, params + [limit, offset])
+                    
+                    rows = cursor.fetchall()
+                    return [self._row_to_model(row) for row in rows]
+                    
+        except Exception as e:
+            logger.error(f"获取模特列表失败: {e}")
+            raise
+        
+        return []
+    
+    def update_model(self, model_id: str, updates: Dict[str, Any]) -> bool:
+        """更新模特"""
+        try:
+            if not updates:
+                return True
+            
+            # 允许更新的字段
+            allowed_fields = {
+                'model_number', 'model_image', 'is_active', 'is_cloud'
+            }
+            
+            # 过滤允许的字段
+            filtered_updates = {k: v for k, v in updates.items() if k in allowed_fields}
+            
+            if not filtered_updates:
+                logger.warning("没有有效的更新字段")
+                return False
+            
+            with self.get_connection() as conn:
+                with conn.cursor() as cursor:
+                    # 构建更新语句
+                    set_clauses = []
+                    params = []
+                    
+                    for field, value in filtered_updates.items():
+                        set_clauses.append(f"{field} = %s")
+                        params.append(value)
+                    
+                    params.append(model_id)
+                    
+                    cursor.execute(f"""
+                        UPDATE models 
+                        SET {', '.join(set_clauses)}
+                        WHERE id = %s
+                    """, params)
+                    
+                    affected_rows = cursor.rowcount
+                    conn.commit()
+                    
+                    if affected_rows > 0:
+                        logger.info(f"更新模特成功: {model_id}")
+                        return True
+                    else:
+                        logger.warning(f"模特不存在: {model_id}")
+                        return False
+                    
+        except psycopg2.IntegrityError as e:
+            if "unique" in str(e).lower():
+                logger.warning(f"模特编号已存在")
+                raise ValueError("模特编号已存在")
+            else:
+                logger.error(f"更新模特失败: {e}")
+                raise
+        except Exception as e:
+            logger.error(f"更新模特失败: {e}")
+            raise
+    
+    def delete_model(self, model_id: str, hard_delete: bool = False) -> bool:
+        """删除模特"""
+        try:
+            with self.get_connection() as conn:
+                with conn.cursor() as cursor:
+                    if hard_delete:
+                        # 硬删除
+                        cursor.execute("""
+                            DELETE FROM models WHERE id = %s
+                        """, (model_id,))
+                    else:
+                        # 软删除
+                        cursor.execute("""
+                            UPDATE models SET is_active = false WHERE id = %s
+                        """, (model_id,))
+                    
+                    affected_rows = cursor.rowcount
+                    conn.commit()
+                    
+                    if affected_rows > 0:
+                        action = "删除" if hard_delete else "禁用"
+                        logger.info(f"{action}模特成功: {model_id}")
+                        return True
+                    else:
+                        logger.warning(f"模特不存在: {model_id}")
+                        return False
+                    
+        except Exception as e:
+            logger.error(f"删除模特失败: {e}")
+            raise
+    
+    def search_models(self, query: str, user_id: str = "default", 
+                     include_cloud: bool = True, limit: int = 50) -> List[Model]:
+        """搜索模特"""
+        try:
+            with self.get_connection() as conn:
+                with conn.cursor(cursor_factory=RealDictCursor) as cursor:
+                    # 构建查询条件
+                    conditions = ["is_active = true"]
+                    params = [f"%{query}%"]
+                    
+                    if include_cloud:
+                        conditions.append("(user_id = %s OR is_cloud = true)")
+                        params.append(user_id)
+                    else:
+                        conditions.append("user_id = %s")
+                        params.append(user_id)
+                    
+                    params.append(limit)
+                    
+                    where_clause = " AND ".join(conditions)
+                    
+                    cursor.execute(f"""
+                        SELECT * FROM models 
+                        WHERE model_number ILIKE %s AND {where_clause}
+                        ORDER BY 
+                            CASE WHEN model_number ILIKE %s THEN 1 ELSE 2 END,
+                            created_at DESC
+                        LIMIT %s
+                    """, [f"%{query}%"] + params)
+                    
+                    rows = cursor.fetchall()
+                    return [self._row_to_model(row) for row in rows]
+                    
+        except Exception as e:
+            logger.error(f"搜索模特失败: {e}")
+            raise
+        
+        return []
+    
+    def get_model_count(self, user_id: str = "default", include_cloud: bool = True, 
+                       include_inactive: bool = False) -> int:
+        """获取模特数量"""
+        try:
+            with self.get_connection() as conn:
+                with conn.cursor() as cursor:
+                    # 构建查询条件
+                    conditions = []
+                    params = []
+                    
+                    if include_cloud:
+                        conditions.append("(user_id = %s OR is_cloud = true)")
+                        params.append(user_id)
+                    else:
+                        conditions.append("user_id = %s")
+                        params.append(user_id)
+                    
+                    if not include_inactive:
+                        conditions.append("is_active = true")
+                    
+                    where_clause = " AND ".join(conditions) if conditions else "1=1"
+                    
+                    cursor.execute(f"""
+                        SELECT COUNT(*) FROM models WHERE {where_clause}
+                    """, params)
+                    
+                    result = cursor.fetchone()
+                    return result[0] if result else 0
+                    
+        except Exception as e:
+            logger.error(f"获取模特数量失败: {e}")
+            raise
+        
+        return 0
+    
+    def toggle_model_status(self, model_id: str) -> bool:
+        """切换模特状态"""
+        try:
+            with self.get_connection() as conn:
+                with conn.cursor() as cursor:
+                    cursor.execute("""
+                        UPDATE models 
+                        SET is_active = NOT is_active 
+                        WHERE id = %s
+                    """, (model_id,))
+                    
+                    affected_rows = cursor.rowcount
+                    conn.commit()
+                    
+                    if affected_rows > 0:
+                        logger.info(f"切换模特状态成功: {model_id}")
+                        return True
+                    else:
+                        logger.warning(f"模特不存在: {model_id}")
+                        return False
+                    
+        except Exception as e:
+            logger.error(f"切换模特状态失败: {e}")
+            raise
+
+
+# 创建全局实例
+model_table = ModelTablePostgres()

python_core/database/types.py

@@ -1,6 +1,19 @@
 from dataclasses import dataclass, field
 from typing import Dict, List, Any, Optional
 
+@dataclass
+class Model:
+    """模特数据结构"""
+    id: str
+    model_number: str  # 模特编号
+    model_image: str   # 模特图片路径
+    created_at: str
+    updated_at: str
+    is_active: bool = True
+    # is_cloud false 代表本地资源 那么只加载自己的 如果是 true 就是公用的
+    is_cloud: bool = False
+    # 创建用户
+    user_id: str = ""
 
 @dataclass
 class ResourceCategory: