opt/tools/upload_log.py

import time
from dataclasses import dataclass

from net.task_api import task_api

PULL_ACCOUNT_OK = 4101099
PULL_ACCOUNT_fail = 4101001
START_SIMULATOR_OK = 4301099
START_GAME_DEFAULT_OK = 4501099
LOGING_GAME_OLD_OK = 4606099
MAIN_DEFAULT_OK = 4701099
FEE_XMY_OK = 4801099
START_SIMULATOR_FAIL = 4301001
START_GAME_DEFAULT_FAIL = 4501001
LOGING_GAME_OLD_FAIL = 4606001
MAIN_DEFAULT_FAIL = 4701099
FEE_XMY_FAIL = 4801099

@dataclass
class UploadLog:
    simulator_ip: str  # 模拟器ip
    simulator_mac: str  # 模拟器mac
    pc_code: str  # 电脑编号
    pc_ip: str  # 电脑IP
    pc_mac: str  # 电脑mac
    device_id: str  # 手机_aid
    account: str  # 游戏账号
    account_type: int  # 游戏_账号类型
    pwd: str  # 游戏账号密码
    game_id: int  # 游戏编号
    coding: int  # 错误码
    log_uuid: str
    operator: str  # 负责人
    remarks: str  # 备注
    task_type: int  # 任务类型新增0活跃1
    script_type: int  # 脚本类型
    simulator_code: str  # 模拟器编号
    device_manufacturer: str  # 手机_厂商
    device_model: str  # 手机_型号
    device_imei: str  # 手机_imei
    device_sdk: str  # 手机_sdk
    device_mac: str  # 手机_mac
    device_number: str  # 手机_号码
    script_device_id: str  # 脚本中上传
    err: str  # 0或不传表示没有异常，其他表示异常，脚本端自定义
    simulator_ip_city: str  # 手机_IP_城市

    # 定义一个方法，将对象转换为字典
    def to_dict(self):
        return {
            'simulator_ip': self.simulator_ip,
            'simulator_mac': self.simulator_mac,
            'pc_code': self.pc_code,
            'pc_ip': self.pc_ip,
            'pc_mac': self.pc_mac,
            'device_id': self.device_id,
            'account': self.account,
            'account_type': self.account_type,
            'pwd': self.pwd,
            'game_id': self.game_id,
            'coding': self.coding,
            'log_uuid': self.log_uuid,
            'operator': self.operator,
            'remarks': self.remarks,
            'task_type': self.task_type,
            'script_type': self.script_type,
            'simulator_code': self.simulator_code,
            'device_manufacturer': self.device_manufacturer,
            'device_model': self.device_model,
            'device_imei': self.device_imei,
            'device_sdk': self.device_sdk,
            'device_mac': self.device_mac,
            'device_number': self.device_number,
            'script_device_id': self.script_device_id,
            'err': self.err,
            'simulator_ip_city': self.simulator_ip_city
        }


def log_init(uuid: str):
    return UploadLog(
        simulator_ip='',
        simulator_mac='',
        pc_code='',
        pc_ip='',
        pc_mac='',
        device_id='',
        account='',
        account_type=1,
        pwd='',
        game_id=5001,
        coding=PULL_ACCOUNT_OK,
        log_uuid=uuid,
        operator='',
        remarks='',
        task_type=1,
        script_type=1,
        simulator_code='',
        device_manufacturer='',
        device_model='',
        device_imei='',
        device_sdk='',
        device_mac='',
        device_number='',
        script_device_id='',
        err='',
        simulator_ip_city='',
    )


def log_uuid(account: str) -> str:
    timestamp = int(time.time() * 1000)
    timestamp_str = str(timestamp)
    result = timestamp_str + "_" + account
    return result




def _get_account_type(account_type: str):
    # 0:QQ 1:小绵羊 2：微信 3:魅族
    if account_type == "qq":
        return 0
    elif account_type == "小绵羊":
        return 1
    elif account_type == "微信":
        return 2
    elif account_type == "魅族":
        return 3
    else:
        return 1


class LogInfo:

    def __init__(self, uuid: str):
        # uuid = log_uuid(account)
        self.upload_log = log_init(uuid)

    def set_account_info(self, game_id: int, account_type: str, pwd: str, account: str, task_type: int):
        try:
            self.upload_log.game_id = game_id
            self.upload_log.account_type = _get_account_type(account_type)
            self.upload_log.pwd = pwd
            self.upload_log.account = account
            self.upload_log.task_type = task_type
            return True, ""
        except Exception as e:
            return False, str(e)

    def set_device_info(self, device_id: str, device_manufacturer: str, device_model: str, device_imei: str,
                        device_sdk: str, device_mac: str, device_number: str, script_device_id: str):
        try:
            self.upload_log.device_id = device_id
            self.upload_log.device_manufacturer = device_manufacturer
            self.upload_log.device_model = device_model
            self.upload_log.device_imei = device_imei
            self.upload_log.device_sdk = device_sdk
            self.upload_log.device_mac = device_mac
            self.upload_log.device_number = device_number
            self.upload_log.script_device_id = script_device_id
            return True, ""
        except Exception as e:
            return False, str(e)

    def set_pc_info(self, pc_code: str, pc_ip: str, pc_mac: str, operator: str):
        try:
            self.upload_log.pc_code = pc_code
            self.upload_log.pc_ip = pc_ip
            self.upload_log.pc_mac = pc_mac
            self.upload_log.operator = operator
            return True, ""
        except Exception as e:
            return False, str(e)

    def set_simulator_info(self, simulator_ip: str, simulator_mac: str, simulator_code: str, simulator_ip_city: str):
        self.upload_log.simulator_ip = simulator_ip
        self.upload_log.simulator_mac = simulator_mac
        self.upload_log.simulator_code = simulator_code
        self.upload_log.simulator_ip_city = simulator_ip_city

    def set_coding(self, coding: int, remarks: str):
        self.upload_log.coding = coding
        self.upload_log.remarks = remarks

    def set_err(self, err: str):
        self.upload_log.err = err

    def get_upload_log(self):
        return self.upload_log

    def upload_longin_log(self, status: int):
        if status == 1:
            self.upload_log.coding = LOGING_GAME_OLD_OK
        else:
            self.upload_log.coding = LOGING_GAME_OLD_FAIL
        task_api.upload_log_to_server(self.upload_log)

    def upload_main_log(self, status: int):
        if status == 1:
            self.upload_log.coding = MAIN_DEFAULT_OK
        else:
            self.upload_log.coding = MAIN_DEFAULT_FAIL
        task_api.upload_log_to_server(self.upload_log)

    def upload_fee_log(self, status: int):
        if status == 1:
            self.upload_log.coding = FEE_XMY_OK
        else:
            self.upload_log.coding = FEE_XMY_FAIL
        task_api.upload_log_to_server(self.upload_log)

    def upload_start_simulator_log(self, status: int):
        if status == 1:
            self.upload_log.coding = START_SIMULATOR_OK
        else:
            self.upload_log.coding = START_SIMULATOR_FAIL
        ret = task_api.upload_log_to_server(self.upload_log)
        return ret

    def upload_start_game_log(self, status: int):
        if status == 1:
            self.upload_log.coding = START_GAME_DEFAULT_OK
        else:
            self.upload_log.coding = START_GAME_DEFAULT_FAIL
        task_api.upload_log_to_server(self.upload_log)

    def upload_pull_account_log(self, status: int):
        if status == 1:
            self.upload_log.coding = PULL_ACCOUNT_OK
        else:
            self.upload_log.coding = PULL_ACCOUNT_fail
        task_api.upload_log_to_server(self.upload_log)