从URL中提取taskId
(self, url: str)
| 458 | return result |
| 459 | |
| 460 | def _extract_task_id_from_url(self, url: str) -> str: |
| 461 | """从URL中提取taskId""" |
| 462 | try: |
| 463 | from urllib.parse import parse_qs, urlparse, unquote |
| 464 | import json |
| 465 | |
| 466 | # 处理_ug_view_param参数 |
| 467 | parsed = urlparse(url) |
| 468 | params = parse_qs(parsed.query) |
| 469 | |
| 470 | if '_ug_view_param' in params: |
| 471 | ug_params = json.loads(unquote(params['_ug_view_param'][0])) |
| 472 | if 'taskId' in ug_params: |
| 473 | return str(ug_params['taskId']) # 确保返回字符串类型 |
| 474 | |
| 475 | # 如果URL是JSON格式的,尝试解析 |
| 476 | if url.startswith('com.sf-express://'): |
| 477 | json_str = url.split('_ug_view_param=')[1] |
| 478 | ug_params = json.loads(unquote(json_str)) |
| 479 | if 'taskId' in ug_params: |
| 480 | return str(ug_params['taskId']) # 确保返回字符串类型 |
| 481 | |
| 482 | except Exception as e: |
| 483 | self.logger.warning(f'从URL提取taskId失败: {e}') |
| 484 | |
| 485 | return '' |
| 486 | |
| 487 | def _set_task_attrs(self, task: Dict) -> None: |
| 488 | """设置任务属性""" |
no test coverage detected