新增计划字段及映射同步
This commit is contained in:
parent
f91dadffd4
commit
ecf9ccbc0b
@ -31,6 +31,7 @@ FIELD_TAPD_STATUS = "TAPD状态" # 工具回写
|
||||
FIELD_OWNER = "处理人" # 工具回写
|
||||
FIELD_BEGIN_DATE = "TAPD预计开始日期" # 工具回写
|
||||
FIELD_DUE_DATE = "TAPD预计完成日期" # 工具回写
|
||||
FIELD_PLAN = "计划" # 工具回写，TAPD计划字段
|
||||
FIELD_SYNC_STATUS = "同步状态" # 工具回写，标记同步结果
|
||||
|
||||
# 必要字段列表
|
||||
@ -40,6 +41,7 @@ REQUIRED_FIELDS = [
|
||||
FIELD_OWNER,
|
||||
FIELD_BEGIN_DATE,
|
||||
FIELD_DUE_DATE,
|
||||
FIELD_PLAN,
|
||||
FIELD_SYNC_STATUS,
|
||||
]
|
||||
|
||||
@ -163,7 +165,8 @@ class SmartSheetSync:
|
||||
|
||||
def build_update_record(self, record_id: str, status: str = None,
|
||||
owner: str = None, begin_date: str = None,
|
||||
due_date: str = None, sync_status: str = None) -> Dict:
|
||||
due_date: str = None, plan: str = None,
|
||||
sync_status: str = None) -> Dict:
|
||||
"""
|
||||
构造更新记录的数据结构
|
||||
|
||||
@ -173,6 +176,7 @@ class SmartSheetSync:
|
||||
owner: 处理人
|
||||
begin_date: 预计开始日期
|
||||
due_date: 预计完成日期
|
||||
plan: 计划（中文名称）
|
||||
sync_status: 同步状态（"成功" 或 "失败"）
|
||||
|
||||
Returns:
|
||||
@ -194,6 +198,9 @@ class SmartSheetSync:
|
||||
if due_date is not None and due_date != "":
|
||||
values[FIELD_DUE_DATE] = [{"type": "text", "text": due_date}]
|
||||
|
||||
if plan is not None and plan != "":
|
||||
values[FIELD_PLAN] = [{"type": "text", "text": plan}]
|
||||
|
||||
if sync_status is not None and sync_status != "":
|
||||
values[FIELD_SYNC_STATUS] = [{"type": "text", "text": sync_status}]
|
||||
|
||||
@ -303,6 +310,7 @@ class SmartSheetSync:
|
||||
FIELD_OWNER: self.api.get_field_value_by_title(record, FIELD_OWNER),
|
||||
FIELD_BEGIN_DATE: self.api.get_field_value_by_title(record, FIELD_BEGIN_DATE),
|
||||
FIELD_DUE_DATE: self.api.get_field_value_by_title(record, FIELD_DUE_DATE),
|
||||
FIELD_PLAN: self.api.get_field_value_by_title(record, FIELD_PLAN),
|
||||
}
|
||||
|
||||
|
||||
|
||||
@ -62,6 +62,15 @@ class SyncService:
|
||||
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" 正在获取计划字段映射...")
|
||||
try:
|
||||
self.plan_mapping = self.tapd_api.get_plan_mapping()
|
||||
print(f" ✓ 计划字段映射获取完成，共 {len(self.plan_mapping)} 个选项")
|
||||
except Exception as e:
|
||||
print(f" ⚠ 计划字段映射获取失败: {e}，将使用空映射")
|
||||
self.plan_mapping = {}
|
||||
|
||||
print(f" ✓ 同步服务初始化完成")
|
||||
|
||||
def sync_once(self) -> Dict[str, Any]:
|
||||
@ -278,11 +287,15 @@ class SyncService:
|
||||
begin_date = story_info.get('begin', '')
|
||||
due_date = story_info.get('due', '')
|
||||
|
||||
# 提取计划字段并转换为中文名称
|
||||
plan_id = story_info.get('custom_plan_field_1', '')
|
||||
plan_name = self.tapd_api.map_plan_id_to_name(plan_id)
|
||||
|
||||
# 获取当前字段值，判断是否需要更新
|
||||
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
|
||||
current_values, status, owner, begin_date, due_date, plan_name
|
||||
)
|
||||
|
||||
# 构造更新记录（包含业务字段 + 同步状态=成功）
|
||||
@ -293,6 +306,7 @@ class SyncService:
|
||||
owner=owner,
|
||||
begin_date=begin_date,
|
||||
due_date=due_date,
|
||||
plan=plan_name,
|
||||
sync_status="成功"
|
||||
)
|
||||
record_result["update_record"] = update_record
|
||||
@ -306,7 +320,8 @@ class SyncService:
|
||||
|
||||
def _check_needs_update(self, current_values: Dict,
|
||||
status: str, owner: str,
|
||||
begin_date: str, due_date: str) -> bool:
|
||||
begin_date: str, due_date: str,
|
||||
plan: str = "") -> bool:
|
||||
"""
|
||||
检查是否需要更新记录
|
||||
|
||||
@ -316,6 +331,7 @@ class SyncService:
|
||||
owner: 新处理人
|
||||
begin_date: 新开始日期
|
||||
due_date: 新结束日期
|
||||
plan: 新计划
|
||||
|
||||
Returns:
|
||||
bool: 是否需要更新
|
||||
@ -339,6 +355,7 @@ class SyncService:
|
||||
current_owner = extract_text(current_values.get('处理人'))
|
||||
current_begin = extract_text(current_values.get('TAPD预计开始日期'))
|
||||
current_due = extract_text(current_values.get('TAPD预计完成日期'))
|
||||
current_plan = extract_text(current_values.get('计划'))
|
||||
|
||||
# 比较是否有变化
|
||||
if current_status != status:
|
||||
@ -349,6 +366,8 @@ class SyncService:
|
||||
return True
|
||||
if current_due != due_date:
|
||||
return True
|
||||
if current_plan != plan:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
@ -50,6 +50,9 @@ class TAPDStoryApi:
|
||||
# TAPD API基础URL（与任务一相同）
|
||||
BASE_URL = "https://tapd-api.bilibili.co/tapd"
|
||||
|
||||
# 计划字段名称
|
||||
PLAN_FIELD_NAME = "custom_plan_field_1"
|
||||
|
||||
def __init__(self, workspace_id: str, test_mode: bool = False):
|
||||
"""
|
||||
初始化TAPD Story API
|
||||
@ -82,6 +85,9 @@ class TAPDStoryApi:
|
||||
# 初始化任务二专用的日志记录器
|
||||
self.logger = get_task2_logger()
|
||||
|
||||
# 计划字段映射缓存（ID -> 中文名称）
|
||||
self._plan_mapping = None
|
||||
|
||||
print(f" ✓ TAPD Story API初始化完成 (workspace_id: {workspace_id})")
|
||||
if test_mode:
|
||||
print(f" ⚠ 测试模式已启用：将显示所有API调用的详细信息")
|
||||
@ -249,6 +255,67 @@ class TAPDStoryApi:
|
||||
"""
|
||||
return f"https://www.tapd.cn/{self.workspace_id}/prong/stories/view/{story_id}"
|
||||
|
||||
def get_story_fields_info(self) -> Dict:
|
||||
"""
|
||||
获取需求所有字段及候选值
|
||||
|
||||
Returns:
|
||||
Dict: 字段信息，包含各字段的名称、选项等
|
||||
|
||||
Raises:
|
||||
RuntimeError: 获取失败时抛出
|
||||
"""
|
||||
params = {
|
||||
'workspace_id': self.workspace_id
|
||||
}
|
||||
|
||||
result = self._make_request("stories/get_fields_info", params=params)
|
||||
return result.get('data', {})
|
||||
|
||||
def get_plan_mapping(self) -> Dict[str, str]:
|
||||
"""
|
||||
获取计划字段的ID到中文名称映射
|
||||
|
||||
Returns:
|
||||
Dict[str, str]: 计划ID到中文名称的映射
|
||||
例如: {"1158335167001034196": "M1版本", ...}
|
||||
"""
|
||||
# 获取字段信息
|
||||
fields_info = self.get_story_fields_info()
|
||||
|
||||
# 提取计划字段的options
|
||||
plan_field = fields_info.get(self.PLAN_FIELD_NAME, {})
|
||||
options = plan_field.get('options', {})
|
||||
|
||||
# 缓存映射
|
||||
self._plan_mapping = options
|
||||
|
||||
if self.test_mode:
|
||||
print(f"\n【测试模式】计划字段映射:")
|
||||
for plan_id, plan_name in options.items():
|
||||
print(f" {plan_id} -> {plan_name}")
|
||||
|
||||
return options
|
||||
|
||||
def map_plan_id_to_name(self, plan_id: str) -> str:
|
||||
"""
|
||||
将计划ID转换为中文名称
|
||||
|
||||
Args:
|
||||
plan_id: 计划ID（如 "1158335167001034196"）
|
||||
|
||||
Returns:
|
||||
str: 中文名称（如 "M1版本"），未找到则返回空字符串
|
||||
"""
|
||||
if not plan_id or plan_id == "0":
|
||||
return ""
|
||||
|
||||
# 如果映射未初始化，先获取
|
||||
if self._plan_mapping is None:
|
||||
self.get_plan_mapping()
|
||||
|
||||
return self._plan_mapping.get(plan_id, "")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("=== TAPD Story API 测试 ===\n")
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user