aszerW
3e6d9d1fdb
.env 配置命名统一： - 群1: DINGTALK_WEBHOOK_1 + DINGTALK_SECRET_1 - 群2: DINGTALK_WEBHOOK_2 + DINGTALK_SECRET_2 settings.py 读取逻辑简化： - 从 i=1 开始读取编号配置 - 移除不带编号的默认配置处理
84 lines
1.8 KiB
Python
84 lines
1.8 KiB
Python
"""
|
|
配置管理模块
|
|
|
|
从环境变量读取配置信息
|
|
"""
|
|
|
|
import os
|
|
from typing import Dict, List, Optional
|
|
|
|
# 默认基准
|
|
DEFAULT_BENCHMARK_CODE = "000300.SH"
|
|
DEFAULT_BENCHMARK_NAME = "沪深300指数"
|
|
|
|
# 默认代码名称映射
|
|
DEFAULT_CODE_NAME_MAP = {
|
|
"000300.SH": "沪深300",
|
|
"000905.SH": "中证500",
|
|
"000852.SH": "中证1000",
|
|
}
|
|
|
|
|
|
def get_dingtalk_config() -> Dict[str, str]:
|
|
"""
|
|
获取钉钉机器人配置（第一个群）
|
|
|
|
Returns:
|
|
dict: 包含 webhook 和 secret
|
|
"""
|
|
webhook = os.getenv("DINGTALK_WEBHOOK", "")
|
|
secret = os.getenv("DINGTALK_SECRET", "")
|
|
|
|
return {
|
|
"webhook": webhook,
|
|
"secret": secret,
|
|
}
|
|
|
|
|
|
def get_all_dingtalk_configs() -> List[Dict[str, str]]:
|
|
"""
|
|
获取所有钉钉机器人配置（支持多群）
|
|
|
|
环境变量格式:
|
|
群1: DINGTALK_WEBHOOK_1 + DINGTALK_SECRET_1
|
|
群2: DINGTALK_WEBHOOK_2 + DINGTALK_SECRET_2
|
|
群3: DINGTALK_WEBHOOK_3 + DINGTALK_SECRET_3
|
|
...
|
|
|
|
Returns:
|
|
list: 配置列表，每项包含 webhook 和 secret
|
|
"""
|
|
configs = []
|
|
|
|
# 从 i=1 开始读取编号配置
|
|
i = 1
|
|
while True:
|
|
webhook = os.getenv(f"DINGTALK_WEBHOOK_{i}", "")
|
|
secret = os.getenv(f"DINGTALK_SECRET_{i}", "")
|
|
|
|
if not webhook:
|
|
break
|
|
|
|
configs.append({
|
|
"webhook": webhook,
|
|
"secret": secret,
|
|
})
|
|
i += 1
|
|
|
|
return configs
|
|
|
|
|
|
def get_db_config() -> Dict[str, str]:
|
|
"""
|
|
获取数据库配置
|
|
|
|
Returns:
|
|
dict: 数据库配置
|
|
"""
|
|
return {
|
|
"host": os.getenv("DB_HOST", "localhost"),
|
|
"port": os.getenv("DB_PORT", "5432"),
|
|
"user": os.getenv("DB_USER", ""),
|
|
"password": os.getenv("DB_PASSWORD", ""),
|
|
"database": os.getenv("DB_NAME", ""),
|
|
} |