154 lines
4.9 KiB
Python
154 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
项目配置管理系统
|
||
负责管理项目ID到数据目录的映射，以及项目访问权限控制
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
from typing import Dict, Optional, List
|
||
from dataclasses import dataclass, asdict
|
||
|
||
|
||
@dataclass
|
||
class ProjectConfig:
|
||
"""项目配置数据类"""
|
||
project_id: str
|
||
data_dir: str
|
||
name: str
|
||
description: str = ""
|
||
allowed_file_types: List[str] = None
|
||
max_file_size_mb: int = 100
|
||
is_active: bool = True
|
||
|
||
def __post_init__(self):
|
||
if self.allowed_file_types is None:
|
||
self.allowed_file_types = [".json", ".txt", ".csv", ".pdf"]
|
||
|
||
|
||
class ProjectManager:
|
||
"""项目管理器"""
|
||
|
||
def __init__(self, config_file: str = "./projects/project_registry.json"):
|
||
self.config_file = config_file
|
||
self.projects: Dict[str, ProjectConfig] = {}
|
||
self._ensure_config_dir()
|
||
self._load_projects()
|
||
|
||
def _ensure_config_dir(self):
|
||
"""确保配置目录存在"""
|
||
config_dir = os.path.dirname(self.config_file)
|
||
if not os.path.exists(config_dir):
|
||
os.makedirs(config_dir, exist_ok=True)
|
||
|
||
def _load_projects(self):
|
||
"""从配置文件加载项目"""
|
||
if os.path.exists(self.config_file):
|
||
try:
|
||
with open(self.config_file, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
for project_data in data.get('projects', []):
|
||
config = ProjectConfig(**project_data)
|
||
self.projects[config.project_id] = config
|
||
except Exception as e:
|
||
print(f"加载项目配置失败: {e}")
|
||
self._create_default_config()
|
||
else:
|
||
self._create_default_config()
|
||
|
||
def _create_default_config(self):
|
||
"""创建默认配置"""
|
||
default_project = ProjectConfig(
|
||
project_id="default",
|
||
data_dir="./data",
|
||
name="默认项目",
|
||
description="默认数据项目"
|
||
)
|
||
self.projects["default"] = default_project
|
||
self._save_projects()
|
||
|
||
def _save_projects(self):
|
||
"""保存项目配置到文件"""
|
||
data = {
|
||
"projects": [asdict(project) for project in self.projects.values()]
|
||
}
|
||
try:
|
||
with open(self.config_file, 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
print(f"保存项目配置失败: {e}")
|
||
|
||
def get_project(self, project_id: str) -> Optional[ProjectConfig]:
|
||
"""获取项目配置"""
|
||
return self.projects.get(project_id)
|
||
|
||
def add_project(self, config: ProjectConfig) -> bool:
|
||
"""添加项目"""
|
||
if config.project_id in self.projects:
|
||
return False
|
||
|
||
# 确保数据目录存在
|
||
if not os.path.isabs(config.data_dir):
|
||
config.data_dir = os.path.abspath(config.data_dir)
|
||
|
||
os.makedirs(config.data_dir, exist_ok=True)
|
||
|
||
self.projects[config.project_id] = config
|
||
self._save_projects()
|
||
return True
|
||
|
||
def update_project(self, project_id: str, **kwargs) -> bool:
|
||
"""更新项目配置"""
|
||
if project_id not in self.projects:
|
||
return False
|
||
|
||
project = self.projects[project_id]
|
||
for key, value in kwargs.items():
|
||
if hasattr(project, key):
|
||
setattr(project, key, value)
|
||
|
||
self._save_projects()
|
||
return True
|
||
|
||
def delete_project(self, project_id: str) -> bool:
|
||
"""删除项目"""
|
||
if project_id not in self.projects:
|
||
return False
|
||
|
||
del self.projects[project_id]
|
||
self._save_projects()
|
||
return True
|
||
|
||
def list_projects(self) -> List[ProjectConfig]:
|
||
"""列出所有项目"""
|
||
return list(self.projects.values())
|
||
|
||
def get_project_dir(self, project_id: str) -> str:
|
||
"""获取项目数据目录"""
|
||
project = self.get_project(project_id)
|
||
if project:
|
||
return project.data_dir
|
||
|
||
# 如果项目不存在，创建默认目录结构
|
||
default_dir = f"./projects/{project_id}/data"
|
||
os.makedirs(default_dir, exist_ok=True)
|
||
|
||
# 自动创建新项目配置
|
||
new_project = ProjectConfig(
|
||
project_id=project_id,
|
||
data_dir=default_dir,
|
||
name=f"项目 {project_id}",
|
||
description=f"自动创建的项目 {project_id}"
|
||
)
|
||
self.add_project(new_project)
|
||
|
||
return default_dir
|
||
|
||
def validate_project_access(self, project_id: str) -> bool:
|
||
"""验证项目访问权限"""
|
||
project = self.get_project(project_id)
|
||
return project and project.is_active
|
||
|
||
|
||
# 全局项目管理器实例
|
||
project_manager = ProjectManager() |