opt/net/task_api.py

import importlib.util
import json
import os
import requests

from dacite import from_dict
from model.custom_struct import PcConfig, AccountInfo, WindowInfo, UploadLog
from tools.log import logger


class TaskApi:
    def __init__(self):
        self.pc_name = os.getenv('COMPUTERNAME') or os.uname().nodename

        # 配置内容及路径
        config_content = """# 接口配置\n#http://dataset.skfzs.top/api\nconfig_url='http://dcf.loadfa.com/api'\n# http://dataset.skfzs.top:8001/api\nlog_url='http://assist.qiming321.cn:8888'"""
        config_path = os.path.join(os.getcwd(), 'config', 'config.py')

        if not os.path.exists(config_path):
            # 文件不存在，创建并写入内容
            os.makedirs(os.path.dirname(config_path), exist_ok=True)  # 确保目录存在
            with open(config_path, 'w', encoding='utf-8') as file:
                file.write(config_content)

        # 动态导入配置文件
        config_spec = importlib.util.spec_from_file_location("config", config_path)
        config_module = importlib.util.module_from_spec(config_spec)
        config_spec.loader.exec_module(config_module)

        try:
            self.config_url = config_module.config_url
            self.log_url = config_module.log_url
        except AttributeError as e:
            raise ImportError("配置文件中缺少必要的变量 config_url 或 log_url") from e

        self.task_url = 'http://xjf.lianyou.fun:8099'

    def get_device_info(self) -> tuple[bool,PcConfig|str]:

        url = f'{self.config_url}/deviceComputer/uploadAndFindCCVersion?pcNum={self.pc_name}'
        try:
            result = requests.get(url, timeout=10)
            if result.status_code == 200:
                json_obj = result.json()
                if json_obj['code'] == 0:
                    obj = json_obj['data']['config']
                    pc_config = PcConfig(
                        pc_name=json_obj['data']['pcNum'],
                        officer=json_obj['data']['directorName'],
                        use_wuyouip=obj['使用无忧ip'],
                        only_new=obj['只做新增'],
                        only_retained=obj['只做留存'],
                        timed_start=0,
                        check_md5_on_start=obj['启动时md5检测'],
                        start_dkw_on_start=obj['多开王自启动'],
                        check_script_update=obj['检查脚本更新'],
                        check_image_update=obj['检查镜像更新'],
                        window_nums=int(obj['窗口数量'])
                    )
                    return True, pc_config
            return False, result.text
        except Exception as e:
            return False,str(e)

    def get_window_list(self) -> list[WindowInfo]:
        url = f'{self.config_url}/window/getWindowPublicV2?computerID={self.pc_name}'
        try:
            result = requests.get(url, timeout=10)
            if result.status_code == 200:
                json_obj = result.json()
                if json_obj['code'] == 0:
                    window_list = []
                    for obj in json_obj['data']['list']:
                        window_info = WindowInfo(
                            window_id=obj['windowID'],
                            game_list=obj['configGame']
                        )
                        window_list.append(window_info)
                    return window_list
            return []
        except Exception as e:
            logger.exception(f'get_window_list error: {e}')
            return []

    def heartbeat(self,operator) -> tuple[bool, str]:

        url = f'http://assist.qiming321.cn:8888/loging/computerHeartbeat'
        data = {
            'pc_code': self.pc_name,
            'operator': operator
        }

        try:
            result = requests.post(url, data= json.dumps(data), timeout=10)

            if result.status_code == 200 :
                if result.json()['code'] == 0:
                    return True,result.json()['msg']
            return False,result.json()['msg']
        except Exception as e:
            logger.exception(f'heartbeat error: {e}')
            return False, str(e)

    def get_account(self, task_id: str,new,retained) -> tuple[bool, AccountInfo] | tuple[bool, str]:
        if new == '1':
            url = f'{self.task_url}/v1/task/get_account?game_id={task_id}&new=1'
        elif retained == '1':
            url = f'{self.task_url}/v1/task/get_account?game_id={task_id}&new=2'
        else:
            url = f'{self.task_url}/v1/task/get_account?game_id={task_id}'
        try:
            result = requests.get(url, timeout=10)
            if result.status_code == 200:
                if isinstance(result.json(), dict):
                    json_obj = result.json()
                    return True , from_dict(AccountInfo, json_obj)
            return False , result.text
        except Exception as e:
            return False , str(e)

    def up_log_profession(self, account, game_id, action, action_result) -> str:
        url = f'{self.task_url}/v1/device/setAccountLog'
        data = {
            'account': account,
            'game_id': game_id,
            'action': action,
            'action_result': action_result,
        }
        try:
            result = requests.get(url, params=data, timeout=10)
            return result.text
        except Exception as e:
            logger.exception(f'up_log_profession error: {e}')
            return str(e)

    def upload_log_to_server(self, params: UploadLog) -> (bool, str):
        json_data = json.dumps(params.to_dict())
        url = f'{self.log_url}/loging/setLog'

        # 发送POST请求
        try:
            response = requests.post(url, data=json_data, headers={'Content-Type': 'application/json'}, timeout=10)
            return response.text
        except Exception as e:
            return str(e)


    def notify(self,desc:str):
        url = f'{self.config_url}/loging/supConErr'
        data = {
            'pc_code': self.pc_name,
            'describe': desc
        }
        try:
            result = requests.post(url, data=data, timeout=10)
            return result.text
        except Exception as e:
            return str(e)


task_api = TaskApi()