""" 模特管理服务 管理模特信息，包括模特编号、模特图片等 """ import json import uuid import os from pathlib import Path from typing import List, Dict, Optional from dataclasses import dataclass, asdict from datetime import datetime from python_core.config import settings from python_core.utils.logger import logger from python_core.utils.jsonrpc import create_response_handler @dataclass class Model: """模特数据结构""" id: str model_number: str # 模特编号 model_image: str # 模特图片路径 created_at: str updated_at: str is_active: bool = True class ModelManager: """模特管理器""" def __init__(self): self.cache_dir = settings.temp_dir / "cache" self.cache_dir.mkdir(parents=True, exist_ok=True) # 模特数据文件 self.models_file = self.cache_dir / "models.json" self.models = self._load_models() # 如果没有数据，创建默认模特 if not self.models: self._create_default_models() def _load_models(self) -> List[Model]: """加载模特数据""" if self.models_file.exists(): try: with open(self.models_file, 'r', encoding='utf-8') as f: data = json.load(f) return [Model(**item) for item in data] except Exception as e: logger.error(f"Failed to load models: {e}") return [] else: return [] def _save_models(self, models: List[Model] = None): """保存模特数据""" if models is None: models = self.models try: data = [asdict(model) for model in models] with open(self.models_file, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) logger.info(f"Models saved to {self.models_file}") except Exception as e: logger.error(f"Failed to save models: {e}") raise def _create_default_models(self): """创建默认模特数据""" default_models = [] now = datetime.now().isoformat() for model_data in default_models: model = Model( id=str(uuid.uuid4()), model_number=model_data["model_number"], model_image=model_data["model_image"], created_at=now, updated_at=now ) self.models.append(model) self._save_models() logger.info("Created default models") def get_all_models(self) -> List[Dict]: """获取所有模特""" return [asdict(model) for model in self.models if model.is_active] def get_model_by_id(self, model_id: str) -> Optional[Dict]: """根据ID获取模特""" for model in self.models: if model.id == model_id and model.is_active: return asdict(model) return None def get_model_by_number(self, model_number: str) -> Optional[Dict]: """根据编号获取模特""" for model in self.models: if model.model_number == model_number and model.is_active: return asdict(model) return None def create_model(self, model_number: str, model_image: str = "") -> Dict: """创建新模特""" # 检查编号是否已存在 if self.get_model_by_number(model_number): raise ValueError(f"Model number {model_number} already exists") now = datetime.now().isoformat() new_model = Model( id=str(uuid.uuid4()), model_number=model_number, model_image=model_image, created_at=now, updated_at=now ) self.models.append(new_model) self._save_models() logger.info(f"Created new model: {model_number}") return asdict(new_model) def update_model(self, model_id: str, model_number: str = None, model_image: str = None) -> Optional[Dict]: """更新模特""" for i, model in enumerate(self.models): if model.id == model_id and model.is_active: # 如果更新编号，检查是否与其他模特冲突 if model_number is not None and model_number != model.model_number: existing = self.get_model_by_number(model_number) if existing and existing['id'] != model_id: raise ValueError(f"Model number {model_number} already exists") model.model_number = model_number if model_image is not None: model.model_image = model_image model.updated_at = datetime.now().isoformat() self.models[i] = model self._save_models() logger.info(f"Updated model: {model_id}") return asdict(model) return None def delete_model(self, model_id: str) -> bool: """删除模特（真删除）""" for i, model in enumerate(self.models): if model.id == model_id: # 真删除：从列表中移除 deleted_model = self.models.pop(i) self._save_models() logger.info(f"Deleted model: {model_id} - {deleted_model.model_number}") return True return False def search_models(self, keyword: str) -> List[Dict]: """搜索模特""" keyword = keyword.lower() results = [] for model in self.models: if (model.is_active and keyword in model.model_number.lower()): results.append(asdict(model)) return results # 全局实例 model_manager = ModelManager() def main(): """命令行接口 - 使用JSON-RPC协议""" import sys import json # 创建响应处理器 rpc = create_response_handler() if len(sys.argv) < 2: rpc.error("INVALID_REQUEST", "No command specified") return command = sys.argv[1] try: if command == "get_all_models": models = model_manager.get_all_models() rpc.success(models) elif command == "get_model_by_id": if len(sys.argv) < 3: rpc.error("INVALID_REQUEST", "Model ID required") return model_id = sys.argv[2] model = model_manager.get_model_by_id(model_id) if model: rpc.success(model) else: rpc.error("NOT_FOUND", "Model not found") elif command == "create_model": if len(sys.argv) < 3: rpc.error("INVALID_REQUEST", "Model number required") return model_number = sys.argv[2] model_image = sys.argv[3] if len(sys.argv) > 3 else "" result = model_manager.create_model(model_number, model_image) rpc.success(result) elif command == "update_model": if len(sys.argv) < 4: rpc.error("INVALID_REQUEST", "Model ID and update data required") return model_id = sys.argv[2] update_data = json.loads(sys.argv[3]) result = model_manager.update_model( model_id, update_data.get('model_number'), update_data.get('model_image') ) if result: rpc.success(result) else: rpc.error("NOT_FOUND", "Model not found") elif command == "delete_model": if len(sys.argv) < 3: rpc.error("INVALID_REQUEST", "Model ID required") return model_id = sys.argv[2] success = model_manager.delete_model(model_id) rpc.success(success) elif command == "search_models": if len(sys.argv) < 3: rpc.error("INVALID_REQUEST", "Search keyword required") return keyword = sys.argv[2] results = model_manager.search_models(keyword) rpc.success(results) else: rpc.error("INVALID_REQUEST", f"Unknown command: {command}") except Exception as e: logger.error(f"Command execution failed: {e}") rpc.error("INTERNAL_ERROR", str(e)) if __name__ == "__main__": main()