363 lines
12 KiB
Python
363 lines
12 KiB
Python
"""
|
||
任务二同步服务模块
|
||
整合链接解析、TAPD查询、表格回写，实现完整的同步流程
|
||
|
||
功能：
|
||
1. 单次同步流程
|
||
2. 执行统计
|
||
3. 错误处理
|
||
"""
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Dict, List, Any, Optional
|
||
from datetime import datetime
|
||
|
||
# 将项目根目录添加到 Python 路径
|
||
project_root = Path(__file__).parent.parent
|
||
sys.path.insert(0, str(project_root))
|
||
|
||
from src.token_manager import TokenManager
|
||
from src2.config import Task2ConfigManager
|
||
from src2.logger import get_task2_logger
|
||
from src2.tapd_api import TAPDStoryApi
|
||
from src2.smartsheet_sync import SmartSheetSync, REQUIRED_FIELDS
|
||
|
||
|
||
class SyncService:
|
||
"""TAPD状态同步服务"""
|
||
|
||
def __init__(self, config_manager: Task2ConfigManager = None,
|
||
access_token: str = None, test_mode: bool = False):
|
||
"""
|
||
初始化同步服务
|
||
|
||
Args:
|
||
config_manager: 配置管理器，如果为None则自动创建
|
||
access_token: 企业微信access_token，如果为None则自动获取
|
||
test_mode: 是否启用测试模式
|
||
"""
|
||
self.test_mode = test_mode
|
||
self.logger = get_task2_logger()
|
||
|
||
# 初始化配置管理器
|
||
if config_manager is None:
|
||
self.config_manager = Task2ConfigManager()
|
||
else:
|
||
self.config_manager = config_manager
|
||
|
||
# 获取配置
|
||
self.config = self.config_manager.get_all_config()
|
||
self.workspace_id = self.config['tapd']['workspace_id']
|
||
self.docid = self.config['smartsheet']['docid']
|
||
|
||
# 获取access_token
|
||
if access_token is None:
|
||
token_manager = TokenManager()
|
||
self.access_token = token_manager.get_token()
|
||
else:
|
||
self.access_token = access_token
|
||
|
||
# 初始化API模块
|
||
self.tapd_api = TAPDStoryApi(self.workspace_id, test_mode=test_mode)
|
||
self.smartsheet = SmartSheetSync(self.access_token, self.docid, test_mode=test_mode)
|
||
|
||
print(f" ✓ 同步服务初始化完成")
|
||
|
||
def sync_once(self) -> Dict[str, Any]:
|
||
"""
|
||
执行一次完整的同步流程
|
||
|
||
Returns:
|
||
Dict: 同步结果统计
|
||
"""
|
||
result = {
|
||
"success": False,
|
||
"start_time": datetime.now().isoformat(),
|
||
"end_time": None,
|
||
"sheets_processed": 0,
|
||
"sheets_skipped": 0,
|
||
"total_records": 0,
|
||
"records_with_link": 0,
|
||
"records_synced": 0,
|
||
"records_updated": 0,
|
||
"records_failed": 0,
|
||
"error_message": None,
|
||
"sheet_results": []
|
||
}
|
||
|
||
try:
|
||
# 1. 获取所有子表
|
||
print("\n正在获取子表列表...")
|
||
sheets = self.smartsheet.api.get_sheet_list()
|
||
print(f" ✓ 找到 {len(sheets)} 个子表")
|
||
|
||
# 2. 处理每个子表
|
||
for sheet in sheets:
|
||
sheet_id = sheet.get('sheet_id', '')
|
||
sheet_title = sheet.get('title', '未命名')
|
||
|
||
sheet_result = self._process_sheet(sheet_id, sheet_title)
|
||
result["sheet_results"].append(sheet_result)
|
||
|
||
if sheet_result["skipped"]:
|
||
result["sheets_skipped"] += 1
|
||
else:
|
||
result["sheets_processed"] += 1
|
||
result["total_records"] += sheet_result["total_records"]
|
||
result["records_with_link"] += sheet_result["records_with_link"]
|
||
result["records_synced"] += sheet_result["records_synced"]
|
||
result["records_updated"] += sheet_result["records_updated"]
|
||
result["records_failed"] += sheet_result["records_failed"]
|
||
|
||
result["success"] = True
|
||
|
||
except Exception as e:
|
||
result["error_message"] = str(e)
|
||
print(f"\n✗ 同步失败: {e}")
|
||
|
||
if self.test_mode:
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
result["end_time"] = datetime.now().isoformat()
|
||
return result
|
||
|
||
def _process_sheet(self, sheet_id: str, sheet_title: str) -> Dict[str, Any]:
|
||
"""
|
||
处理单个子表的同步
|
||
|
||
Args:
|
||
sheet_id: 子表ID
|
||
sheet_title: 子表标题
|
||
|
||
Returns:
|
||
Dict: 子表处理结果
|
||
"""
|
||
print(f"\n{'='*60}")
|
||
print(f"处理子表: {sheet_title}")
|
||
print(f"{'='*60}")
|
||
|
||
sheet_result = {
|
||
"sheet_id": sheet_id,
|
||
"sheet_title": sheet_title,
|
||
"skipped": False,
|
||
"skip_reason": None,
|
||
"total_records": 0,
|
||
"records_with_link": 0,
|
||
"records_synced": 0,
|
||
"records_updated": 0,
|
||
"records_failed": 0,
|
||
"details": []
|
||
}
|
||
|
||
try:
|
||
# 1. 获取字段信息并检查必要字段
|
||
fields = self.smartsheet.api.get_fields(sheet_id)
|
||
all_present, missing_fields, field_mapping = self.smartsheet.check_required_fields(fields)
|
||
|
||
if not all_present:
|
||
sheet_result["skipped"] = True
|
||
sheet_result["skip_reason"] = f"缺少必要字段: {', '.join(missing_fields)}"
|
||
print(f" ⚠ 跳过此子表: {sheet_result['skip_reason']}")
|
||
return sheet_result
|
||
|
||
# 2. 获取包含TAPD链接的记录
|
||
records_with_link = self.smartsheet.get_records_with_tapd_link(sheet_id)
|
||
sheet_result["records_with_link"] = len(records_with_link)
|
||
|
||
if not records_with_link:
|
||
print(f" ℹ 没有包含TAPD链接的记录")
|
||
return sheet_result
|
||
|
||
# 3. 处理每条记录
|
||
success_records = [] # 成功同步的记录（包含业务字段 + 同步状态）
|
||
failed_record_ids = [] # 失败的记录ID列表
|
||
|
||
for record_info in records_with_link:
|
||
record_result = self._process_record(record_info)
|
||
sheet_result["details"].append(record_result)
|
||
|
||
if record_result["success"]:
|
||
sheet_result["records_synced"] += 1
|
||
|
||
if record_result["update_record"]:
|
||
success_records.append(record_result["update_record"])
|
||
sheet_result["records_updated"] += 1
|
||
else:
|
||
sheet_result["records_failed"] += 1
|
||
failed_record_ids.append(record_info["record_id"])
|
||
|
||
# 4. 批量回写成功记录（业务字段 + 同步状态=成功）
|
||
if success_records:
|
||
print(f"\n正在回写 {len(success_records)} 条成功记录...")
|
||
try:
|
||
self.smartsheet.batch_update_records(sheet_id, success_records)
|
||
print(f" ✓ 成功记录回写完成")
|
||
except Exception as e:
|
||
print(f" ✗ 成功记录回写失败: {e}")
|
||
|
||
# 5. 批量回写失败记录（只写入同步状态=失败）
|
||
if failed_record_ids:
|
||
print(f"\n正在回写 {len(failed_record_ids)} 条失败记录的状态...")
|
||
try:
|
||
failed_updates = [
|
||
self.smartsheet.build_update_record(
|
||
record_id=record_id,
|
||
sync_status="失败"
|
||
)
|
||
for record_id in failed_record_ids
|
||
]
|
||
self.smartsheet.batch_update_records(sheet_id, failed_updates)
|
||
print(f" ✓ 失败记录状态回写完成")
|
||
except Exception as e:
|
||
print(f" ✗ 失败记录状态回写失败: {e}")
|
||
|
||
sheet_result["total_records"] = len(records_with_link)
|
||
|
||
except Exception as e:
|
||
sheet_result["skipped"] = True
|
||
sheet_result["skip_reason"] = f"处理异常: {str(e)}"
|
||
print(f" ✗ 处理子表异常: {e}")
|
||
|
||
return sheet_result
|
||
|
||
def _process_record(self, record_info: Dict) -> Dict[str, Any]:
|
||
"""
|
||
处理单条记录的同步
|
||
|
||
Args:
|
||
record_info: 记录信息（包含解析结果）
|
||
|
||
Returns:
|
||
Dict: 记录处理结果
|
||
"""
|
||
record_result = {
|
||
"record_id": record_info["record_id"],
|
||
"tapd_link": record_info["tapd_link"],
|
||
"success": False,
|
||
"update_record": None,
|
||
"error_message": None,
|
||
"story_info": None
|
||
}
|
||
|
||
# 检查链接解析结果
|
||
if not record_info["parse_success"]:
|
||
record_result["error_message"] = record_info.get("parse_error", "链接解析失败")
|
||
return record_result
|
||
|
||
story_id = record_info["story_id"]
|
||
|
||
try:
|
||
# 查询TAPD获取需求信息
|
||
story_info = self.tapd_api.get_story(story_id)
|
||
record_result["story_info"] = story_info
|
||
|
||
# 提取需要同步的字段
|
||
status = story_info.get('status', '')
|
||
owner = story_info.get('owner', '')
|
||
begin_date = story_info.get('begin', '')
|
||
due_date = story_info.get('due', '')
|
||
|
||
# 获取当前字段值，判断是否需要更新
|
||
current_values = self.smartsheet.get_current_field_values(record_info["record"])
|
||
|
||
needs_update = self._check_needs_update(
|
||
current_values, status, owner, begin_date, due_date
|
||
)
|
||
|
||
# 构造更新记录（包含业务字段 + 同步状态=成功）
|
||
# 即使业务字段没有变化，也要写入同步状态
|
||
update_record = self.smartsheet.build_update_record(
|
||
record_id=record_info["record_id"],
|
||
status=status,
|
||
owner=owner,
|
||
begin_date=begin_date,
|
||
due_date=due_date,
|
||
sync_status="成功"
|
||
)
|
||
record_result["update_record"] = update_record
|
||
|
||
record_result["success"] = True
|
||
|
||
except Exception as e:
|
||
record_result["error_message"] = str(e)
|
||
|
||
return record_result
|
||
|
||
def _check_needs_update(self, current_values: Dict,
|
||
status: str, owner: str,
|
||
begin_date: str, due_date: str) -> bool:
|
||
"""
|
||
检查是否需要更新记录
|
||
|
||
Args:
|
||
current_values: 当前字段值
|
||
status: 新状态
|
||
owner: 新处理人
|
||
begin_date: 新开始日期
|
||
due_date: 新结束日期
|
||
|
||
Returns:
|
||
bool: 是否需要更新
|
||
"""
|
||
# 提取当前值的文本内容
|
||
def extract_text(value):
|
||
if value is None:
|
||
return ""
|
||
if isinstance(value, str):
|
||
return value
|
||
if isinstance(value, list) and len(value) > 0:
|
||
first = value[0]
|
||
if isinstance(first, dict):
|
||
return first.get('text', '')
|
||
return str(first)
|
||
if isinstance(value, dict):
|
||
return value.get('text', '')
|
||
return str(value)
|
||
|
||
current_status = extract_text(current_values.get('TAPD状态'))
|
||
current_owner = extract_text(current_values.get('处理人'))
|
||
current_begin = extract_text(current_values.get('TAPD预计开始日期'))
|
||
current_due = extract_text(current_values.get('TAPD预计完成日期'))
|
||
|
||
# 比较是否有变化
|
||
if current_status != status:
|
||
return True
|
||
if current_owner != owner:
|
||
return True
|
||
if current_begin != begin_date:
|
||
return True
|
||
if current_due != due_date:
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def run_once(config_manager: Task2ConfigManager = None,
|
||
access_token: str = None,
|
||
test_mode: bool = False) -> Dict[str, Any]:
|
||
"""
|
||
执行一次同步流程（供调度器调用）
|
||
|
||
Args:
|
||
config_manager: 配置管理器
|
||
access_token: access_token
|
||
test_mode: 测试模式
|
||
|
||
Returns:
|
||
Dict: 同步结果
|
||
"""
|
||
service = SyncService(
|
||
config_manager=config_manager,
|
||
access_token=access_token,
|
||
test_mode=test_mode
|
||
)
|
||
return service.sync_once()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print("=== 任务二同步服务测试 ===\n")
|
||
print("请使用 main.py 或 scheduler.py 运行同步服务")
|
||
print("或使用 test_phase4.py 进行测试")
|