任务执行器
| 425 | |
| 426 | # ==================== 任务执行器 ==================== |
| 427 | class TaskExecutor: |
| 428 | """任务执行器""" |
| 429 | |
| 430 | def __init__( |
| 431 | self, |
| 432 | http_client: SFHttpClient, |
| 433 | logger: Logger, |
| 434 | config: Config, |
| 435 | user_id: str |
| 436 | ): |
| 437 | self.http = http_client |
| 438 | self.logger = logger |
| 439 | self.config = config |
| 440 | self.user_id = user_id |
| 441 | self.total_points = 0 |
| 442 | |
| 443 | # 任务相关属性 |
| 444 | self.taskId = "" |
| 445 | self.taskCode = "" |
| 446 | self.strategyId = "" |
| 447 | self.title = "" |
| 448 | |
| 449 | @staticmethod |
| 450 | def generate_device_id(characters: str = 'abcdef0123456789') -> str: |
| 451 | """生成设备ID""" |
| 452 | result = '' |
| 453 | for char in 'xxxxxxxx-xxxx-xxxx': |
| 454 | if char == 'x': |
| 455 | result += random.choice(characters) |
| 456 | else: |
| 457 | result += char |
| 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 |