wehub-resource-sync
245 lines
9.5 KiB
Python
245 lines
9.5 KiB
Python
import os
|
|
from dotenv import load_dotenv
|
|
from urllib.parse import quote_plus
|
|
from typing import List, Optional
|
|
from pathlib import Path
|
|
|
|
# 加载环境变量
|
|
load_dotenv()
|
|
|
|
# 导入embedding配置
|
|
try:
|
|
from config.embedding_config import EmbeddingConfig
|
|
_embedding_config = None # 延迟加载
|
|
except ImportError:
|
|
_embedding_config = None
|
|
|
|
|
|
class Settings:
|
|
"""应用配置类"""
|
|
|
|
# 应用基础配置
|
|
APP_NAME: str = os.getenv("APP_NAME", "KUMI Service")
|
|
DEBUG: bool = os.getenv("DEBUG", "False").lower() == "true"
|
|
HOST: str = os.getenv("HOST", "0.0.0.0")
|
|
PORT: int = int(os.getenv("PORT", "8000"))
|
|
|
|
ADMIN_USER_NAME: str = os.getenv("ADMIN_USER_NAME", "admin")
|
|
ADMIN_PASSWORD: str = os.getenv("ADMIN_PASSWORD", "KUMI-admin-123456")
|
|
|
|
test_host: str = os.getenv("TEST_HOST", "127.0.0.1")
|
|
test_port: int = int(os.getenv("TEST_PORT", "8001"))
|
|
debug: bool = os.getenv("DEBUG", "False").lower() == "true"
|
|
|
|
# 资源目录配置
|
|
RESOURCE_BASE_DIR: str = os.getenv("RESOURCE_BASE_DIR", "resources")
|
|
DDL_EXPORT_DIR: str = os.getenv("DDL_EXPORT_DIR", "ddl_exports")
|
|
|
|
# 服务控制 - 控制哪些API服务启用
|
|
ENABLE_KNOWLEDGE_API: bool = os.getenv("ENABLE_KNOWLEDGE_API", "true").lower() == "true"
|
|
ENABLE_DOCUMENT_CONVERSION_API: bool = os.getenv("ENABLE_DOCUMENT_CONVERSION_API", "true").lower() == "true"
|
|
|
|
# MarkItDown配置
|
|
MARKITDOWN_ENABLE_PLUGINS: bool = os.getenv("MARKITDOWN_ENABLE_PLUGINS", "false").lower() == "true"
|
|
MARKITDOWN_MAX_FILE_SIZE: int = int(os.getenv("MARKITDOWN_MAX_FILE_SIZE", str(50 * 1024 * 1024))) # 50MB
|
|
|
|
# 知识库上传文件大小限制
|
|
KNOWLEDGE_UPLOAD_MAX_FILE_SIZE: int = int(os.getenv("KNOWLEDGE_UPLOAD_MAX_FILE_SIZE", str(100 * 1024 * 1024))) # 100MB
|
|
|
|
# 支持的文件扩展名（从环境变量读取，用逗号分隔）
|
|
_allowed_extensions = os.getenv("MARKITDOWN_ALLOWED_EXTENSIONS",
|
|
".pdf,.docx,.pptx,.xlsx,.xls,.txt,.md,.html,.csv,.wav,.mp3")
|
|
MARKITDOWN_ALLOWED_EXTENSIONS: List[str] = [ext.strip() for ext in _allowed_extensions.split(",")]
|
|
|
|
# Chroma向量数据库配置
|
|
vector_db_type: str = os.getenv('VECTOR_DB_TYPE', 'chroma')
|
|
|
|
# Milvus Configuration
|
|
milvus_host: str = os.getenv('MILVUS_HOST', 'localhost')
|
|
milvus_port: int = int(os.getenv('MILVUS_PORT', '19530'))
|
|
milvus_user: str = os.getenv('MILVUS_USER', 'root')
|
|
milvus_password: str = os.getenv('MILVUS_PASSWORD', 'Milvus')
|
|
|
|
# ChromaDB Configuration
|
|
chroma_host: str = os.getenv('CHROMA_HOST', 'localhost')
|
|
chroma_port: int = int(os.getenv('CHROMA_PORT', '8000'))
|
|
|
|
# Embedding配置文件路径
|
|
_EMBEDDING_CONFIG_PATH: str = os.getenv('EMBEDDING_CONFIG_PATH', 'config/embedding_providers.yaml')
|
|
|
|
VERIFY_TOKEN: str = os.getenv('VERIFY_TOKEN', 'NOTPOSSIBLEADMINp0ss')
|
|
|
|
# PATH
|
|
PROJECT_ROOT: str = str(Path(__file__).parent.parent.absolute())
|
|
|
|
# 如果需要自定义项目根目录，可以通过环境变量覆盖
|
|
_custom_root = os.getenv('PROJECT_ROOT')
|
|
if _custom_root:
|
|
PROJECT_ROOT = str(Path(_custom_root).absolute())
|
|
|
|
# 相对于项目根目录的路径配置
|
|
_CSV_TEST_RELATIVE_PATH: str = os.getenv('CSV_TEST_RELATIVE_PATH', 'dataset/dataset_csv_test')
|
|
_YAML_EVALUATE_RELATIVE_PATH: str = os.getenv('YAML_EVALUATE_RELATIVE_PATH', 'dataset/yaml_eval')
|
|
_YAML_EVALUATE_TEMPLATES_RELATIVE_PATH: str = os.getenv('YAML_EVALUATE_TEMPLATES_RELATIVE_PATH',
|
|
'dataset/yaml_eval_templates')
|
|
|
|
_WORKFLOWS_CONFIG_RELATIVE_PATH: str = os.getenv('WORKFLOWS_CONFIG_RELATIVE_PATH', 'dataset/workflows.json')
|
|
_TABLE_FOLDER_RELATIVE_PATH: str = os.getenv('TABLE_FOLDER_RELATIVE_PATH', 'dataset/dataset_csv_test')
|
|
_TEMP_RESULTS_RELATIVE_PATH: str = os.getenv('TEMP_RESULTS_RELATIVE_PATH', 'output/temp_results')
|
|
_EVALUATION_RESULTS_RELATIVE_PATH: str = os.getenv('EVALUATION_RESULTS_RELATIVE_PATH', 'output/evaluation_results')
|
|
_YAML_RULES_RELATIVE_PATH: str = os.getenv('YAML_RULES_RELATIVE_PATH', 'dataset/yaml_eval')
|
|
|
|
# LLM配置
|
|
OPENAI_API_KEY: Optional[str] = os.getenv("OPENAI_API_KEY")
|
|
OPENAI_BASE_URL: Optional[str] = os.getenv("OPENAI_BASE_URL")
|
|
CLAUDE_API_KEY: Optional[str] = os.getenv("CLAUDE_API_KEY")
|
|
DEFAULT_LLM_PROVIDER: str = os.getenv("DEFAULT_LLM_PROVIDER", "openai")
|
|
DEFAULT_MODEL: str = os.getenv("DEFAULT_MODEL", "gpt-4.1-mini")
|
|
DEFAULT_TEMPERATURE: float = float(os.getenv("DEFAULT_TEMPERATURE", "0.4"))
|
|
DEFAULT_MAX_TOKENS: int = int(os.getenv("DEFAULT_MAX_TOKENS", "1000"))
|
|
|
|
API_API_KEY: Optional[str] = os.getenv("API_API_KEY")
|
|
|
|
# KUMI API URL（用于 UMAP 可视化服务调用主服务）
|
|
KUMI_API_URL: str = os.getenv("KUMI_API_URL", "http://localhost:8000")
|
|
|
|
# UMAP 缓存目录（Embedding Atlas 兼容格式）
|
|
_UMAP_CACHE_DIR: Optional[str] = os.getenv("UMAP_CACHE_DIR")
|
|
|
|
# 日志配置
|
|
LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")
|
|
LOG_FILE: Optional[str] = os.getenv("LOG_FILE")
|
|
|
|
# 生成绝对路径
|
|
@property
|
|
def CSV_TEST_PATH(self) -> str:
|
|
return str(Path(self.PROJECT_ROOT) / self._CSV_TEST_RELATIVE_PATH)
|
|
|
|
@property
|
|
def YAML_EVALUATE_PATH(self) -> str:
|
|
return str(Path(self.PROJECT_ROOT) / self._YAML_EVALUATE_RELATIVE_PATH)
|
|
|
|
@property
|
|
def YAML_EVALUATE_TEMPLATES_PATH(self) -> str:
|
|
return str(Path(self.PROJECT_ROOT) / self._YAML_EVALUATE_TEMPLATES_RELATIVE_PATH)
|
|
|
|
@property
|
|
def WORKFLOWS_CONFIG_PATH(self) -> str:
|
|
return str(Path(self.PROJECT_ROOT) / self._WORKFLOWS_CONFIG_RELATIVE_PATH)
|
|
|
|
@property
|
|
def TABLE_FOLDER_PATH(self) -> str:
|
|
return str(Path(self.PROJECT_ROOT) / self._TABLE_FOLDER_RELATIVE_PATH)
|
|
|
|
@property
|
|
def TEMP_RESULTS_PATH(self) -> str:
|
|
return str(Path(self.PROJECT_ROOT) / self._TEMP_RESULTS_RELATIVE_PATH)
|
|
|
|
@property
|
|
def EVALUATION_RESULTS_PATH(self) -> str:
|
|
return str(Path(self.PROJECT_ROOT) / self._EVALUATION_RESULTS_RELATIVE_PATH)
|
|
|
|
@property
|
|
def YAML_RULES_PATH(self) -> str:
|
|
return str(Path(self.PROJECT_ROOT) / self._YAML_RULES_RELATIVE_PATH)
|
|
|
|
def get_absolute_path(self, relative_path: str) -> str:
|
|
"""
|
|
根据相对路径获取绝对路径
|
|
|
|
Args:
|
|
relative_path: 相对于项目根目录的路径
|
|
|
|
Returns:
|
|
str: 绝对路径
|
|
"""
|
|
return str(Path(self.PROJECT_ROOT) / relative_path)
|
|
|
|
def create_directories(self):
|
|
"""创建所有配置的目录"""
|
|
directories = [
|
|
self.CSV_TEST_PATH,
|
|
self.YAML_EVALUATE_PATH,
|
|
self.YAML_EVALUATE_TEMPLATES_PATH,
|
|
self.TABLE_FOLDER_PATH,
|
|
self.TEMP_RESULTS_PATH,
|
|
self.EVALUATION_RESULTS_PATH,
|
|
self.YAML_RULES_PATH,
|
|
# 配置文件的目录
|
|
str(Path(self.WORKFLOWS_CONFIG_PATH).parent)
|
|
]
|
|
|
|
for directory in directories:
|
|
Path(directory).mkdir(parents=True, exist_ok=True)
|
|
|
|
def print_config(self):
|
|
"""打印当前配置信息"""
|
|
print("⚙️ 当前路径配置:")
|
|
print(f" 项目根目录: {self.PROJECT_ROOT}")
|
|
print(f" 工作流配置: {self.WORKFLOWS_CONFIG_PATH}")
|
|
print(f" 表格文件夹: {self.TABLE_FOLDER_PATH}")
|
|
print(f" 临时结果: {self.TEMP_RESULTS_PATH}")
|
|
print(f" 评测结果: {self.EVALUATION_RESULTS_PATH}")
|
|
print(f" YAML规则: {self.YAML_RULES_PATH}")
|
|
|
|
|
|
@property
|
|
def chroma_url(self) -> str:
|
|
"""构建Chroma连接URL"""
|
|
return f"http://{self.CHROMA_HOST}:{self.CHROMA_PORT}"
|
|
|
|
@property
|
|
def resource_dir(self) -> str:
|
|
"""获取资源目录完整路径"""
|
|
return os.path.abspath(self.RESOURCE_BASE_DIR)
|
|
|
|
@property
|
|
def ddl_export_path(self) -> str:
|
|
"""获取DDL导出目录完整路径"""
|
|
return os.path.join(self.resource_dir, self.DDL_EXPORT_DIR)
|
|
|
|
@property
|
|
def EMBEDDING_CONFIG_PATH(self) -> str:
|
|
"""获取embedding配置文件完整路径"""
|
|
return str(Path(self.PROJECT_ROOT) / self._EMBEDDING_CONFIG_PATH)
|
|
|
|
@property
|
|
def UMAP_CACHE_DIR(self) -> str:
|
|
"""
|
|
获取 UMAP 缓存目录（Embedding Atlas 兼容格式）
|
|
|
|
默认值按平台:
|
|
- Windows: %LOCALAPPDATA%\\embedding_atlas\\projections\\
|
|
- Linux: ~/.cache/embedding_atlas/projections/
|
|
- macOS: ~/Library/Caches/embedding_atlas/projections/
|
|
"""
|
|
if self._UMAP_CACHE_DIR:
|
|
return self._UMAP_CACHE_DIR
|
|
|
|
import sys
|
|
if sys.platform == 'win32':
|
|
local_app_data = os.environ.get('LOCALAPPDATA', '')
|
|
if local_app_data:
|
|
return str(Path(local_app_data) / 'embedding_atlas' / 'projections')
|
|
return str(Path.home() / 'AppData' / 'Local' / 'embedding_atlas' / 'projections')
|
|
elif sys.platform == 'darwin':
|
|
return str(Path.home() / 'Library' / 'Caches' / 'embedding_atlas' / 'projections')
|
|
else:
|
|
return str(Path.home() / '.cache' / 'embedding_atlas' / 'projections')
|
|
|
|
def get_embedding_config(self) -> 'EmbeddingConfig':
|
|
"""
|
|
获取embedding配置实例(单例模式)
|
|
|
|
Returns:
|
|
EmbeddingConfig实例
|
|
"""
|
|
global _embedding_config
|
|
if _embedding_config is None:
|
|
from config.embedding_config import EmbeddingConfig
|
|
_embedding_config = EmbeddingConfig(self.EMBEDDING_CONFIG_PATH)
|
|
return _embedding_config
|
|
|
|
|
|
# 创建全局配置实例
|
|
settings = Settings() |