307 lines
11 KiB
Python
307 lines
11 KiB
Python
"""
|
|
项目管理服务
|
|
管理项目信息，包括项目名、本地目录、分类文件夹初始化、商品信息等
|
|
"""
|
|
|
|
import json
|
|
import uuid
|
|
import os
|
|
import shutil
|
|
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
|
|
from python_core.services.resource_category_manager import resource_category_manager
|
|
|
|
|
|
@dataclass
|
|
class Project:
|
|
"""项目数据结构"""
|
|
id: str
|
|
name: str # 项目名
|
|
local_directory: str # 本地目录路径
|
|
product_name: str # 商品名
|
|
product_image: str # 商品图片路径
|
|
created_at: str
|
|
updated_at: str
|
|
is_active: bool = True
|
|
|
|
|
|
class ProjectManager:
|
|
"""项目管理器"""
|
|
|
|
def __init__(self):
|
|
self.cache_dir = settings.temp_dir / "cache"
|
|
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 项目数据文件
|
|
self.projects_file = self.cache_dir / "projects.json"
|
|
self.projects = self._load_projects()
|
|
|
|
def _load_projects(self) -> List[Project]:
|
|
"""加载项目数据"""
|
|
if self.projects_file.exists():
|
|
try:
|
|
with open(self.projects_file, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
return [Project(**item) for item in data]
|
|
except Exception as e:
|
|
logger.error(f"Failed to load projects: {e}")
|
|
return []
|
|
else:
|
|
return []
|
|
|
|
def _save_projects(self, projects: List[Project] = None):
|
|
"""保存项目数据"""
|
|
if projects is None:
|
|
projects = self.projects
|
|
|
|
try:
|
|
data = [asdict(project) for project in projects]
|
|
with open(self.projects_file, 'w', encoding='utf-8') as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
logger.info(f"Projects saved to {self.projects_file}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to save projects: {e}")
|
|
raise
|
|
|
|
def _initialize_category_folders(self, project_directory: str) -> bool:
|
|
"""为项目初始化分类文件夹"""
|
|
try:
|
|
project_path = Path(project_directory)
|
|
if not project_path.exists():
|
|
project_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 获取所有启用的分类
|
|
active_categories = resource_category_manager.get_all_categories()
|
|
|
|
for category in active_categories:
|
|
if category.get('is_active', True): # 只为启用的分类创建文件夹
|
|
category_folder = project_path / category['title']
|
|
category_folder.mkdir(exist_ok=True)
|
|
logger.info(f"Created category folder: {category_folder}")
|
|
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to initialize category folders: {e}")
|
|
return False
|
|
|
|
def get_all_projects(self) -> List[Dict]:
|
|
"""获取所有项目"""
|
|
return [asdict(project) for project in self.projects if project.is_active]
|
|
|
|
def get_project_by_id(self, project_id: str) -> Optional[Dict]:
|
|
"""根据ID获取项目"""
|
|
for project in self.projects:
|
|
if project.id == project_id and project.is_active:
|
|
return asdict(project)
|
|
return None
|
|
|
|
def create_project(self, name: str, local_directory: str, product_name: str = "",
|
|
product_image: str = "") -> Dict:
|
|
"""创建新项目"""
|
|
now = datetime.now().isoformat()
|
|
|
|
# 验证目录路径
|
|
if not os.path.isabs(local_directory):
|
|
raise ValueError("Local directory must be an absolute path")
|
|
|
|
new_project = Project(
|
|
id=str(uuid.uuid4()),
|
|
name=name,
|
|
local_directory=local_directory,
|
|
product_name=product_name,
|
|
product_image=product_image,
|
|
created_at=now,
|
|
updated_at=now
|
|
)
|
|
|
|
# 初始化分类文件夹
|
|
if not self._initialize_category_folders(local_directory):
|
|
raise RuntimeError("Failed to initialize category folders")
|
|
|
|
self.projects.append(new_project)
|
|
self._save_projects()
|
|
|
|
logger.info(f"Created new project: {name}")
|
|
return asdict(new_project)
|
|
|
|
def update_project(self, project_id: str, name: str = None, local_directory: str = None,
|
|
product_name: str = None, product_image: str = None) -> Optional[Dict]:
|
|
"""更新项目"""
|
|
for i, project in enumerate(self.projects):
|
|
if project.id == project_id and project.is_active:
|
|
if name is not None:
|
|
project.name = name
|
|
if local_directory is not None:
|
|
# 验证新目录路径
|
|
if not os.path.isabs(local_directory):
|
|
raise ValueError("Local directory must be an absolute path")
|
|
project.local_directory = local_directory
|
|
if product_name is not None:
|
|
project.product_name = product_name
|
|
if product_image is not None:
|
|
project.product_image = product_image
|
|
project.updated_at = datetime.now().isoformat()
|
|
|
|
self.projects[i] = project
|
|
self._save_projects()
|
|
|
|
logger.info(f"Updated project: {project_id}")
|
|
return asdict(project)
|
|
return None
|
|
|
|
def delete_project(self, project_id: str) -> bool:
|
|
"""删除项目（软删除）"""
|
|
for i, project in enumerate(self.projects):
|
|
if project.id == project_id and project.is_active:
|
|
project.is_active = False
|
|
project.updated_at = datetime.now().isoformat()
|
|
|
|
self.projects[i] = project
|
|
self._save_projects()
|
|
|
|
logger.info(f"Deleted project: {project_id}")
|
|
return True
|
|
return False
|
|
|
|
def search_projects(self, keyword: str) -> List[Dict]:
|
|
"""搜索项目"""
|
|
keyword = keyword.lower()
|
|
results = []
|
|
|
|
for project in self.projects:
|
|
if (project.is_active and
|
|
(keyword in project.name.lower() or
|
|
keyword in project.product_name.lower())):
|
|
results.append(asdict(project))
|
|
|
|
return results
|
|
|
|
def open_project_directory(self, project_id: str) -> bool:
|
|
"""打开项目目录"""
|
|
project = self.get_project_by_id(project_id)
|
|
if project and os.path.exists(project['local_directory']):
|
|
try:
|
|
import subprocess
|
|
import platform
|
|
|
|
directory = project['local_directory']
|
|
system = platform.system()
|
|
|
|
if system == "Windows":
|
|
subprocess.run(['explorer', directory])
|
|
elif system == "Darwin": # macOS
|
|
subprocess.run(['open', directory])
|
|
else: # Linux
|
|
subprocess.run(['xdg-open', directory])
|
|
|
|
logger.info(f"Opened project directory: {directory}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to open directory: {e}")
|
|
return False
|
|
return False
|
|
|
|
|
|
# 全局实例
|
|
project_manager = ProjectManager()
|
|
|
|
|
|
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_projects":
|
|
projects = project_manager.get_all_projects()
|
|
rpc.success(projects)
|
|
|
|
elif command == "get_project_by_id":
|
|
if len(sys.argv) < 3:
|
|
rpc.error("INVALID_REQUEST", "Project ID required")
|
|
return
|
|
project_id = sys.argv[2]
|
|
project = project_manager.get_project_by_id(project_id)
|
|
if project:
|
|
rpc.success(project)
|
|
else:
|
|
rpc.error("NOT_FOUND", "Project not found")
|
|
|
|
elif command == "create_project":
|
|
if len(sys.argv) < 4:
|
|
rpc.error("INVALID_REQUEST", "Name and directory required")
|
|
return
|
|
name = sys.argv[2]
|
|
local_directory = sys.argv[3]
|
|
product_name = sys.argv[4] if len(sys.argv) > 4 else ""
|
|
product_image = sys.argv[5] if len(sys.argv) > 5 else ""
|
|
result = project_manager.create_project(name, local_directory, product_name, product_image)
|
|
rpc.success(result)
|
|
|
|
elif command == "update_project":
|
|
if len(sys.argv) < 4:
|
|
rpc.error("INVALID_REQUEST", "Project ID and update data required")
|
|
return
|
|
project_id = sys.argv[2]
|
|
update_data = json.loads(sys.argv[3])
|
|
result = project_manager.update_project(
|
|
project_id,
|
|
update_data.get('name'),
|
|
update_data.get('local_directory'),
|
|
update_data.get('product_name'),
|
|
update_data.get('product_image')
|
|
)
|
|
if result:
|
|
rpc.success(result)
|
|
else:
|
|
rpc.error("NOT_FOUND", "Project not found")
|
|
|
|
elif command == "delete_project":
|
|
if len(sys.argv) < 3:
|
|
rpc.error("INVALID_REQUEST", "Project ID required")
|
|
return
|
|
project_id = sys.argv[2]
|
|
success = project_manager.delete_project(project_id)
|
|
rpc.success(success)
|
|
|
|
elif command == "search_projects":
|
|
if len(sys.argv) < 3:
|
|
rpc.error("INVALID_REQUEST", "Search keyword required")
|
|
return
|
|
keyword = sys.argv[2]
|
|
results = project_manager.search_projects(keyword)
|
|
rpc.success(results)
|
|
|
|
elif command == "open_project_directory":
|
|
if len(sys.argv) < 3:
|
|
rpc.error("INVALID_REQUEST", "Project ID required")
|
|
return
|
|
project_id = sys.argv[2]
|
|
success = project_manager.open_project_directory(project_id)
|
|
rpc.success(success)
|
|
|
|
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() |