Method get_task_list

daily/sfsy.py:645–698

获取任务列表

(self)

            self.http.headers['platform'] = original_platform
    
    def get_task_list(self) -> List[Dict]:
        """获取任务列表"""
        url = 'https://mcs-mimp-web.sf-express.com/mcs-mimp/commonPost/~memberNonactivity~integralTaskStrategyService~queryPointTaskAndSignFromES'
        
        all_tasks = []
        task_codes_seen = set() 
        
        for channel_type in ['1', '2', '3', '4','01','02','03','04']:
            data = {
                'channelType': channel_type,
                'deviceId': self.generate_device_id(),
            }
            
            response = self.http.request(url, data=data)
            
            if response and response.get('success') and response.get('obj'):
                # 只在第一次请求时获取总积分
                if channel_type == '1':
                    self.total_points = response['obj'].get('totalPoint', 0)
                
                tasks = response['obj'].get('taskTitleLevels', [])
                
                # 去重添加任务
                for task in tasks:
                    task_code = task.get('taskCode')
                    task_title = task.get('title', '未知任务')
                    
                    # 尝试提取taskId
                    if 'buttonRedirect' in task:
                        extracted_id = self._extract_task_id_from_url(task['buttonRedirect'])
                        if extracted_id and not task_code:
                            task_code = extracted_id
                            task['taskCode'] = extracted_id
                    
                    # 如果taskCode为空，但能从buttonRedirect中提取到taskId，则使用提取的taskId
                    if not task_code and 'buttonRedirect' in task:
                        extracted_id = self._extract_task_id_from_url(task['buttonRedirect'])
                        if extracted_id:
                            task['taskCode'] = extracted_id
                            task_code = extracted_id
                    
                    # 如果taskCode仍然为空，则跳过
                    if not task_code:
                        continue
                        
                    # 检查是否已存在相同taskCode的任务
                    if task_code not in task_codes_seen:
                        task_codes_seen.add(task_code)
                        all_tasks.append(task)
            else:
                error_msg = response.get('errorMessage', '未知错误') if response else '请求失败'
                self.logger.warning(f'获取 channelType={channel_type} 的任务失败: {error_msg}')
        
        return all_tasks
    
    def execute_task(self) -> bool:
        """执行单个任务"""