172 lines
6.6 KiB
Python
172 lines
6.6 KiB
Python
"""任务三配置管理"""
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Dict, List
|
||
|
||
# 添加项目根目录到Python路径
|
||
project_root = Path(__file__).parent.parent
|
||
sys.path.insert(0, str(project_root))
|
||
|
||
from src.config import ConfigManager
|
||
|
||
|
||
class Task3ConfigManager(ConfigManager):
|
||
"""任务三配置管理器"""
|
||
|
||
GROUP_SECTION = "GroupPush"
|
||
DEFAULT_GROUP_COUNT = 3
|
||
|
||
def __init__(self):
|
||
project_root = Path(__file__).parent.parent
|
||
config_path = project_root / "config3" / "config.ini"
|
||
super().__init__(config_path)
|
||
|
||
def get_workspace_id(self) -> str:
|
||
"""获取TAPD工作空间ID"""
|
||
if not self.config.has_section("TAPD"):
|
||
raise ValueError("配置文件缺少[TAPD]节")
|
||
if not self.config.has_option("TAPD", "workspace_id"):
|
||
raise ValueError("配置文件[TAPD]节缺少workspace_id配置项")
|
||
workspace_id = self.config.get("TAPD", "workspace_id").strip()
|
||
if not workspace_id:
|
||
raise ValueError("workspace_id配置项不能为空")
|
||
return workspace_id
|
||
|
||
def get_push_time(self) -> str:
|
||
"""获取推送时间"""
|
||
if not self.config.has_section("Schedule"):
|
||
return "09:30"
|
||
return self.config.get("Schedule", "push_time", fallback="09:30").strip()
|
||
|
||
def get_skip_weekend(self) -> bool:
|
||
"""是否跳过周末"""
|
||
if not self.config.has_section("Schedule"):
|
||
return True
|
||
return self.config.getboolean("Schedule", "skip_weekend", fallback=True)
|
||
|
||
def get_smartsheet_config(self) -> Dict[str, str]:
|
||
"""获取智能表格配置（仅docid）"""
|
||
if not self.config.has_section("Smartsheet"):
|
||
raise ValueError("配置文件缺少[Smartsheet]节")
|
||
if not self.config.has_option("Smartsheet", "docid"):
|
||
raise ValueError("配置文件[Smartsheet]节缺少docid配置项")
|
||
|
||
docid = self.config.get("Smartsheet", "docid").strip()
|
||
if not docid:
|
||
raise ValueError("docid配置项不能为空")
|
||
|
||
return {"docid": docid}
|
||
|
||
def get_group_push_configs(self) -> List[Dict[str, str]]:
|
||
"""读取多群配置（按顺序一一对应）"""
|
||
section = self.GROUP_SECTION
|
||
if not self.config.has_section(section):
|
||
raise ValueError("配置文件缺少[GroupPush]节")
|
||
|
||
group_count = self.config.getint(section, "group_count", fallback=self.DEFAULT_GROUP_COUNT)
|
||
if group_count <= 0:
|
||
raise ValueError("group_count必须为正整数")
|
||
|
||
group_configs: List[Dict[str, str]] = []
|
||
for idx in range(1, group_count + 1):
|
||
name_key = f"group{idx}_name"
|
||
title_key = f"group{idx}_sheet_title"
|
||
webhook_key = f"group{idx}_webhook_url"
|
||
|
||
if not self.config.has_option(section, name_key):
|
||
raise ValueError(f"[GroupPush]缺少{name_key}配置项")
|
||
if not self.config.has_option(section, title_key):
|
||
raise ValueError(f"[GroupPush]缺少{title_key}配置项")
|
||
if not self.config.has_option(section, webhook_key):
|
||
raise ValueError(f"[GroupPush]缺少{webhook_key}配置项")
|
||
|
||
group_name = self.config.get(section, name_key).strip()
|
||
sheet_title = self.config.get(section, title_key).strip()
|
||
webhook_url = self.config.get(section, webhook_key).strip()
|
||
|
||
if not group_name or not sheet_title or not webhook_url:
|
||
raise ValueError(f"[GroupPush]第{idx}组配置不能为空")
|
||
|
||
group_configs.append(
|
||
{
|
||
"group_name": group_name,
|
||
"sheet_title": sheet_title,
|
||
"webhook_url": webhook_url,
|
||
}
|
||
)
|
||
|
||
return group_configs
|
||
|
||
def get_group_team_configs(self, logger=None) -> List[Dict]:
|
||
"""加载多组成员配置并解析子表ID"""
|
||
from src.token_manager import TokenManager
|
||
from src.smartsheet import SmartSheetAPI
|
||
|
||
smartsheet_config = self.get_smartsheet_config()
|
||
group_push_configs = self.get_group_push_configs()
|
||
docid = smartsheet_config["docid"]
|
||
|
||
token_manager = TokenManager(logger=logger)
|
||
access_token = token_manager.get_token()
|
||
|
||
api = SmartSheetAPI(access_token, docid)
|
||
sheet_list = api.get_sheet_list()
|
||
title_to_sheet_id = self._build_title_to_sheet_id(sheet_list)
|
||
|
||
group_team_configs: List[Dict] = []
|
||
for group_config in group_push_configs:
|
||
sheet_title = group_config["sheet_title"]
|
||
sheet_id = title_to_sheet_id.get(sheet_title)
|
||
if not sheet_id:
|
||
raise ValueError(f"未找到成员配置子表: {sheet_title}")
|
||
|
||
records_result = api.get_records(sheet_id)
|
||
records = records_result.get("records", [])
|
||
|
||
member_list: List[str] = []
|
||
user_mapping: Dict[str, str] = {}
|
||
|
||
for record in records:
|
||
tapd_name = api.get_field_value_by_title(record, "TAPD用户名")
|
||
wework_id = api.get_field_value_by_title(record, "企微UserID")
|
||
enabled = api.get_field_value_by_title(record, "是否启用")
|
||
|
||
tapd_name = str(tapd_name).strip() if tapd_name else ""
|
||
wework_id = str(wework_id).strip() if wework_id else ""
|
||
|
||
if not tapd_name or not self._is_enabled(enabled):
|
||
continue
|
||
|
||
if tapd_name not in member_list:
|
||
member_list.append(tapd_name)
|
||
if wework_id:
|
||
user_mapping[tapd_name] = wework_id
|
||
|
||
group_team_configs.append(
|
||
{
|
||
"group_name": group_config["group_name"],
|
||
"sheet_title": sheet_title,
|
||
"sheet_id": sheet_id,
|
||
"webhook_url": group_config["webhook_url"],
|
||
"member_list": member_list,
|
||
"user_mapping": user_mapping,
|
||
}
|
||
)
|
||
|
||
return group_team_configs
|
||
|
||
def _build_title_to_sheet_id(self, sheet_list: List[Dict]) -> Dict[str, str]:
|
||
"""构建子表标题到sheet_id映射"""
|
||
mapping: Dict[str, str] = {}
|
||
for sheet in sheet_list:
|
||
title = str(sheet.get("title", "")).strip()
|
||
sheet_id = str(sheet.get("sheet_id", "")).strip()
|
||
if title and sheet_id:
|
||
mapping[title] = sheet_id
|
||
return mapping
|
||
|
||
def _is_enabled(self, value) -> bool:
|
||
"""判断成员是否启用"""
|
||
normalized = str(value).strip().lower()
|
||
return normalized in {"是", "true", "1", "yes", "y"}
|