import anyio
from fastapi import APIRouter, Request

from src.config.lic.lic import LicMgr
from src.lic.shcemas import ActivateNetworkRequest, GetAllRequest

router = APIRouter(prefix="/lic/info", tags=["授权"])


def _get_lic_mgr(request: Request) -> LicMgr:
    lic_mgr = getattr(request.app.state, "lic_mgr", None)
    if lic_mgr is None:
        raise RuntimeError("LicMgr not initialized")
    return lic_mgr


@router.get("/get_state", summary="获取授权状态")
async def get_state(request: Request):
    lic_mgr = _get_lic_mgr(request)
    return await anyio.to_thread.run_sync(lic_mgr.get_state)


@router.post("/get_all", summary="获取全部授权信息")
async def get_all(request: Request, req: GetAllRequest):
    lic_mgr = _get_lic_mgr(request)
    return await anyio.to_thread.run_sync(lic_mgr.get_info, req.auth_type)


@router.post("/activate_network", summary="集团网络激活")
async def activate_network(request: Request, req: ActivateNetworkRequest):
    lic_mgr = _get_lic_mgr(request)
    return await anyio.to_thread.run_sync(lic_mgr.activate_network, req.ip, req.port)

from pydantic import BaseModel, ConfigDict, Field


class GetAllRequest(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    auth_type: str = Field(default="no_auth", alias="AuthType")


class ActivateNetworkRequest(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    ip: str = Field(default="", alias="IP")
    port: int = Field(default=8273, alias="Port")

import logging
import os
import pickle
import shutil
import threading
import time
import xml.etree.ElementTree as ET
from datetime import datetime

from tfcommon.utils.translation import gettext

from src.conf import settings

from .bitanswer import BitAnswer
from .bitanswer.BitAnswer import BitException, ErrorCode

logger = logging.getLogger(__name__)


class LicException(Exception):
    pass


def get_expiry_days(end_time):
    return (end_time - datetime.now()).total_seconds() / 86400 if end_time > datetime.now() else 0


class LicMgr:
    # 特征项 (Feature) - 比特云文档中心 http://license.njtf.cn/eee/doc/?t=term&i=feature
    FID0 = "30000"
    FID1 = "30001"

    def __init__(self, url: str = "121.40.58.21", root_path: str | None = None):
        self.url = url
        self.root_path = root_path or str(settings.APPDATA_DIR / "lic_file")

        self.lock = threading.Lock()
        self.config_dict = {}
        self.initialize()

    @property
    def config_path(self) -> str:
        return os.path.join(self.root_path, "config")

    def initialize(self):
        if not os.path.exists(self.root_path):
            os.makedirs(self.root_path, exist_ok=True)
        BitAnswer.BitAnswer().set_root_path(self.root_path)  # 设置lic的路径的

        if os.path.exists(self.config_path):
            with open(self.config_path, "rb") as f:
                self.config_dict = pickle.load(f)
                self.url = self.config_dict.get("url", self.url)
                self.root_path = self.config_dict.get("root_path", self.root_path)
        else:
            with self.lock:
                # "no_auth": "无授权",
                # "trial": "试用",
                # "single": "单机",
                # "group": "集团"
                self.config_dict = {
                    "auth_type": "no_auth",
                    "trial_key": True,  # 有没有试用产品这个按键
                    "no_auth": {},
                    "trial": {"state": False, "expiry_days": None},
                    "single": {"state": False},
                    "group": {"state": False, "ip": None, "port": None},
                    "url": self.url,
                    "root_path": self.root_path,
                }
                self.write_config_dict()

    def write_config_dict(self):
        with open(self.config_path, "wb") as f:
            pickle.dump(self.config_dict, f)

    def check(self):
        if (
            self.config_dict["auth_type"] == "no_auth"
            or (self.config_dict["auth_type"] == "group" and not self.config_dict["group"]["state"])
            or (self.config_dict["auth_type"] == "single" and not self.config_dict["single"]["state"])
        ):
            return False
        elif self.config_dict["auth_type"] == "group":
            _control_type = ["Group, Smart, Float"]
            self.login_network(self.config_dict["group"]["ip"], self.config_dict["group"]["port"])
        else:
            if self.config_dict["auth_type"] == "trial":
                _control_type = ["Local, Demo, Mute"]
            else:
                _control_type = ["Local, Smart, Float", "Local"]
            self.login_local()
        valid_num = self.get_valid_num(_control_type)
        BitAnswer.BitAnswer().logout()
        return valid_num > 0

    def get_valid_num(self, _control_type):
        sn_info_str = BitAnswer.BitAnswer().get_session_info(BitAnswer.SESSION_TYPE.XML_TYPE_SN_INFO.value).decode()
        # 解析XML文件
        root = ET.fromstring(sn_info_str)
        sn_info_list = root.findall("snInfo")
        _sn_info_list = []
        for ele in sn_info_list:
            control_type = ele.find("controlType").text
            if control_type in _control_type:
                _sn_info_list.append(ele)
        local_lic_dict = {}
        for ele in _sn_info_list:
            sn = ele.find("sn").text
            end_date = ele.find("endDate")
            if end_date is None:
                local_lic_dict[sn] = {"end_time": None, "expiry_days": None, "is_expired": 0}
            else:
                end_time = ele.find("endDate").text
                _end_time = datetime.strptime(end_time, "%Y-%m-%d %H:%M:%S")
                is_expired = 0 if _end_time > datetime.now() else 1  # 0是没过期1是过期了
                expiry_days = get_expiry_days(_end_time)
                local_lic_dict[sn] = {
                    "end_time": _end_time.timestamp(),
                    "expiry_days": expiry_days,
                    "is_expired": is_expired,
                }
        return len(local_lic_dict)

    def get_control_type(self, sn):
        info = BitAnswer.BitAnswer().get_info(sn, BitAnswer.LOGIN_MODE.BIT_MODE_AUTO.value).decode()
        # 解析XML文件
        root = ET.fromstring(info)
        sn_info_list = root.findall("snInfo")
        if len(sn_info_list) == 1:
            return sn_info_list[0].find("controlType").text
        return None

    def get_state(self):
        return self.config_dict

    def _get_info(self, auth_type):
        if auth_type == "group":
            if self.config_dict["auth_type"] != "group" or not self.config_dict["group"]["state"]:
                return {}, {}
            _control_type = ["Group, Smart, Float"]
            self.login_network(self.config_dict["group"]["ip"], self.config_dict["group"]["port"])
        else:
            if auth_type == "trial":
                _control_type = ["Local, Demo, Mute"]
            else:
                _control_type = ["Local, Smart, Float", "Local"]
            if not self.config_dict["single"]["state"]:
                return {}, {}
            self.login_local()
        local_lic_dict = {}
        sn_info_b = BitAnswer.BitAnswer().get_session_info(BitAnswer.SESSION_TYPE.XML_TYPE_SN_INFO.value)
        sn_info_str = sn_info_b.decode()
        # 解析XML文件
        root = ET.fromstring(sn_info_str)
        sn_info_list = root.findall("snInfo")
        _sn_info_list = []
        for ele in sn_info_list:
            control_type = ele.find("controlType").text
            if control_type in _control_type:
                _sn_info_list.append(ele)
        for ele in _sn_info_list:
            sn = ele.find("sn").text
            end_date = ele.find("endDate")
            if end_date is None:
                local_lic_dict[sn] = {"end_time": None, "expiry_days": None, "is_expired": 0}
            else:
                end_time = ele.find("endDate").text
                _end_time = datetime.strptime(end_time, "%Y-%m-%d %H:%M:%S")
                is_expired = 0 if _end_time > datetime.now() else 1  # 0是没过期1是过期了
                expiry_days = get_expiry_days(_end_time)
                local_lic_dict[sn] = {
                    "end_time": _end_time.timestamp(),
                    "expiry_days": expiry_days,
                    "is_expired": is_expired,
                }

        feature_info_b = BitAnswer.BitAnswer().get_session_info(BitAnswer.SESSION_TYPE.XML_TYPE_FEATURE_INFO.value)
        feature_info_str = feature_info_b.decode()
        # 解析XML文件
        root = ET.fromstring(feature_info_str)
        features_list = root.findall("features")
        group_highest = {
            "sn": None,
            "end_time": None,
            "expiry_days": None,
            "is_expired": None,
            "name": "DTEmpower",
            "version": None,
            "value": -1,
        }
        for ele in features_list:
            for feature in ele.findall("feature"):
                feature_id = feature.get("id")
                sn = ele.get("sn")
                if auth_type == "group" and sn is None:
                    if feature_id == self.FID0:
                        group_highest["name"] = feature.get("name")
                    if feature_id == self.FID1:
                        value = float(feature.get("value"))
                        end_date = feature.get("endDate")
                        if end_date is None:
                            if value > group_highest["value"]:
                                name = feature.get("name")
                                group_highest["version"] = name
                                group_highest["value"] = value
                                group_highest["expiry_days"] = None
                                group_highest["end_time"] = None
                                group_highest["is_expired"] = 0
                        else:
                            end_time = datetime.strptime(feature.get("endDate"), "%Y-%m-%d %H:%M:%S")
                            if value > group_highest["value"] and end_time > datetime.now():
                                name = feature.get("name")
                                group_highest["version"] = name
                                group_highest["value"] = value
                                expiry_days = get_expiry_days(end_time)
                                group_highest["expiry_days"] = expiry_days
                                group_highest["end_time"] = end_time.timestamp()
                                group_highest["is_expired"] = 0

                if sn in local_lic_dict:
                    if feature_id == self.FID0:
                        local_lic_dict[sn]["name"] = feature.get("name")
                    if feature_id == self.FID1:
                        name = feature.get("name")
                        value = float(feature.get("value"))
                        local_lic_dict[sn]["version"] = name
                        local_lic_dict[sn]["value"] = value
        BitAnswer.BitAnswer().logout()
        return local_lic_dict, group_highest

    def get_info(self, auth_type):
        local_lic_dict, highest = self._get_info(auth_type)
        _list = []
        for k, v in local_lic_dict.items():
            _item = {"sn": k}
            _item.update(v)
            _list.append(_item)
        if auth_type != "group":
            sorted_data = sorted(
                _list,
                key=lambda x: (
                    x["is_expired"],
                    -x["value"],
                    (-float("inf") if x["end_time"] is None else -x["end_time"]),
                ),
            )
            if bool(sorted_data):
                highest = sorted_data[0]
            return {"highest": highest, "version": _list}
        else:
            return {"highest": highest, "state": self.config_dict["group"]}

    def login_local(self):
        BitAnswer.BitAnswer().login(None, None, BitAnswer.LOGIN_MODE.BIT_MODE_AUTO.value)

    def login_network(self, host_name, port):
        BitAnswer.BitAnswer().set_local_service(host_name, port, 10)
        BitAnswer.BitAnswer().login(None, None, BitAnswer.LOGIN_MODE.BIT_MODE_REMOTE.value)

    def activate_trial(self):
        if not self.config_dict["trial"]["state"] and self.config_dict["trial"]["expiry_days"] is not None:
            return self.config_dict
        _control_type = ["Local, Demo, Mute"]
        txt_file = os.path.join(os.path.dirname(__file__), "pre.txt")
        with open(txt_file, "rb") as f:
            update_info = f.read()
        try:
            BitAnswer.BitAnswer().apply_update_info(update_info)
            BitAnswer.BitAnswer().login(None, None, BitAnswer.LOGIN_MODE.BIT_MODE_AUTO.value)
            sn_info_b = BitAnswer.BitAnswer().get_session_info(BitAnswer.SESSION_TYPE.XML_TYPE_SN_INFO.value)
            root = ET.fromstring(sn_info_b.decode())
            sn_info_list = root.findall("snInfo")
            _sn_info_list = []
            for ele in sn_info_list:
                control_type = ele.find("controlType").text
                if control_type in _control_type:
                    _sn_info_list.append(ele)
            if len(_sn_info_list) == 1:
                ele = _sn_info_list[-1]
                end_Date = ele.find("endDate")
                if end_Date is not None:
                    end_time = ele.find("endDate").text
                    _end_time = datetime.strptime(end_time, "%Y-%m-%d %H:%M:%S")
                else:
                    activate_date = ele.find("activateDate").text
                    max_execution_minutes = ele.find("maxExecutionMinutes").text
                    _end_time = datetime.strptime(activate_date, "%Y-%m-%d %H:%M:%S") + datetime.timedelta(
                        seconds=int(max_execution_minutes) * 60
                    )
                with self.lock:
                    self.config_dict["auth_type"] = "trial"
                    self.config_dict["trial"]["state"] = True
                    expiry_days = get_expiry_days(_end_time)
                    self.config_dict["trial"]["expiry_days"] = expiry_days
                    self.write_config_dict()
            BitAnswer.BitAnswer().logout()
        except BitException as e:
            if e.error_code == ErrorCode.AUTH_CODE_EXPIRE:  # 授权码已经过期
                self.config_dict["trial_key"] = False
                self.config_dict["trial"]["state"] = False
                self.config_dict["trial"]["expiry_days"] = 0
                self.write_config_dict()
                return self.config_dict
        return self.config_dict

    def activate_single_online(self, sn):
        BitAnswer.BitAnswer().update_online(self.url, sn)
        with self.lock:
            self.config_dict["auth_type"] = "single"
            self.config_dict["single"]["state"] = True
            self.config_dict["group"]["state"] = False  # 激活单机版网络版断开
            self.config_dict["trial"]["state"] = False
            self.write_config_dict()
        return self.config_dict

    def download_request_info_file(self, sn, tmp_dir):
        request_info = BitAnswer.BitAnswer().get_request_info(sn, BitAnswer.BINDING_TYPE.BINDING_LOCAL.value)
        file_name = "lic.txt"
        if not os.path.exists(tmp_dir.path):
            os.makedirs(tmp_dir.path)
        file_abs_path = os.path.join(tmp_dir.path, file_name)
        with open(file_abs_path, "w") as file:
            file.write(request_info.decode())

        def clean_temp_dir():
            time.sleep(30)
            if os.path.exists(tmp_dir.path):
                shutil.rmtree(tmp_dir.path)

        threading.Thread(target=clean_temp_dir).start()
        return file_abs_path

    def upload_update_info_file(self, update_info_b):
        BitAnswer.BitAnswer().apply_update_info(update_info_b)

    def activate_single_offline(self, update_info_b):
        self.upload_update_info_file(update_info_b)
        with self.lock:
            self.config_dict["auth_type"] = "single"
            self.config_dict["single"]["state"] = True
            self.config_dict["group"]["state"] = False  # 激活单机版网络版断开
            self.config_dict["trial"]["state"] = False
            self.write_config_dict()
        return self.config_dict

    def activate_network(self, host_name, port):
        self.login_network(host_name, port)
        with self.lock:
            self.config_dict["auth_type"] = "group"
            self.config_dict["group"]["state"] = True
            self.config_dict["trial"]["state"] = False
            self.config_dict["single"]["state"] = False
            self.config_dict["group"]["ip"], self.config_dict["group"]["port"] = host_name, port
            self.write_config_dict()
        BitAnswer.BitAnswer().logout()
        return self.config_dict

    def logout(self):
        with self.lock:
            self.config_dict["auth_type"] = "single"
            self.config_dict["group"]["state"] = False
            self.config_dict["single"]["state"] = False  # 激活网络版单机断开
            self.write_config_dict()
        return self.config_dict

    def revoke_online(self, sn):
        control_type = self.get_control_type(sn)
        if control_type != "Local, Smart, Float":  # 不是单机浮动的不允许返还
            raise LicException(gettext("Activation codes are not allowed to be returned"))
        BitAnswer.BitAnswer().revoke(self.url, sn)
        try:
            BitAnswer.BitAnswer().login(None, None, BitAnswer.LOGIN_MODE.BIT_MODE_LOCAL.value)
            valid_num = self.get_valid_num(["Local, Smart, Float", "Local"])
            BitAnswer.BitAnswer().logout()
            if valid_num == 0:
                with self.lock:
                    self.config_dict["single"]["state"] = False
                    self.write_config_dict()
        except Exception as e:
            logger.info(e.args[0])
            if e.args[0] in [276, 291]:
                with self.lock:
                    self.config_dict["single"]["state"] = False
                    self.write_config_dict()
        return self.config_dict

    def revoke_offline(self, sn, tmp_dir):
        control_type = self.get_control_type(sn)
        if control_type != "Local, Smart, Float":  # 不是单机浮动的不允许返还
            raise LicException(gettext("Activation codes are not allowed to be returned"))
        revocation_info = BitAnswer.BitAnswer().revoke(None, sn)
        file_name = "lic.txt"
        if not os.path.exists(tmp_dir.path):
            os.makedirs(tmp_dir.path)
        file_abs_path = os.path.join(tmp_dir.path, file_name)
        with open(file_abs_path, "w") as file:
            file.write(revocation_info.value.decode())

        def clean_temp_dir():
            time.sleep(30)
            if os.path.exists(tmp_dir.path):
                shutil.rmtree(tmp_dir.path)

        threading.Thread(target=clean_temp_dir).start()
        try:
            BitAnswer.BitAnswer().login(None, None, BitAnswer.LOGIN_MODE.BIT_MODE_LOCAL.value)
            valid_num = self.get_valid_num(["Local, Smart, Float", "Local"])
            BitAnswer.BitAnswer().logout()
            if valid_num == 0:
                with self.lock:
                    self.config_dict["single"]["state"] = False
                    self.write_config_dict()
        except Exception as e:
            logger.info(e.args[0])
            if e.args[0] in [276, 291]:
                with self.lock:
                    self.config_dict["single"]["state"] = False
                    self.write_config_dict()
        return file_abs_path

# 必须保留
__all__ = ["BitAnswer"]

import ctypes
import os
import platform
from enum import Enum, IntEnum

from tfcommon.utils.translation import gettext

BASE_DIR = os.path.dirname(__file__)


class ErrorCode(IntEnum):
    AUTH_CODE_EXPIRE = 1793
    LIBRARY_LOADING_FAILED = -1


ERROR_CODE_MESSAGE_MAP: dict[int, str] = {
    ErrorCode.AUTH_CODE_EXPIRE: "Authorization has expired",
    ErrorCode.LIBRARY_LOADING_FAILED: "Library loading failed, check to see if the path is correct",
}


class BitException(Exception):
    def __init__(self, error_code: int):
        self.error_code = error_code

    @property
    def message(self) -> str:
        return gettext(
            ERROR_CODE_MESSAGE_MAP.get(self.error_code, "An error occurred. Please contact the system administrator")
        )


class LOGIN_MODE(Enum):
    BIT_MODE_LOCAL = 0x01  # 登录本地授权，不会尝试寻找集团或云授权服务器。如果本地授权失效也不会尝试自动升级
    BIT_MODE_REMOTE = 0x02  # 登录远程授权服务器，尝试建立云授权或集团授权连接，服务器地址由URL参数或配置文件提供
    BIT_MODE_AUTO = 0x03  # 同时尝试本地授权及远程授权，寻找任何可用的授权方式。可能自动升级失效的本地授权


class SESSION_TYPE(Enum):
    XML_TYPE_SN_INFO = 0x03  # 获取当前会话中授权码的授权信息，以XML结构表示。未作限制的授权项将不会返回
    XML_TYPE_FEATURE_INFO = 0x04  # 获取当前会话中授权码的特征项信息，以XML结构表示
    BIT_SERVER_ADDRESS = 0x101  # 授权服务器IP地址。只对云授权有意义，其它授权方式将返回：127.0.0.1
    BIT_SERVER_TIME = 0x201  # 授权服务器系统时间。如果是本地授权，将返回本机系统时间
    BIT_CONTROL_TYPE = 0x302  # 授权类型。是以逗号分隔的以下几种类型的组合：Group（集团授权），Smart（智能连接）, Float（可浮动）, Demo（演示授权）, USB（U盘授权）, Force（强制认证）
    BIT_VOLUME_NUMBER = 0x303  # 允许使用的机器数量
    BIT_START_DATE = 0x304  # 授权开始日期
    BIT_END_DATE = 0x305  # 授权结束日期
    BIT_EXPIRATION_DAYS = 0x306  # 授权有效期，单位为天。有效期从激活时间算起
    BIT_USAGE_NUMBER = 0x307  # 最大使用次数
    BIT_CONSUMED_USAGE_NUMBER = 0x308  # 对于设定最大使用次数的授权码，返回当前使用次数
    BIT_ACTIVATE_DATE = 0x30A  # 授权码激活时间
    BIT_USER_LIMIT = 0x30B  # 集团授权用户数。对其它方式的授权返回0
    BIT_LAST_UPDATE_DATE = 0x30C  # 最后与服务器连接时间
    BIT_MAX_OFFLINE_MINUTES = 0x30D  # 强制认证最大离线时间（分钟）
    BIT_NEXT_CONNECT_DATE = 0x30E  # 会话信息存储区地址。可以为NULL，此时仅用于获取存储区大小
    BIT_MID_BINARY = 0x1001  # 会话存储区大小


class BINDING_TYPE(Enum):
    BINDING_EXISTING = 0  # 使用当前绑定类型，对已激活授权码进行升级时使用
    BINDING_LOCAL = 1  # 使用本机指纹进行绑定
    BINDING_USB_STORAGE = 2  # 使用USB存储器进行绑定


class BIT_INFO_TYPE(Enum):
    BIT_INFO_SERVER_ADDRESS = 0  # 获取局域网内的集团授权服务器地址列表
    BIT_INFO_SN = 1  # 获取本机已激活的授权码列表
    BIT_INFO_SN_FEATURE = 2  # 获取本机已激活授权码的特征项列表
    BIT_INFO_SN_LICENSE = 3  # 获取本机已激活授权码的授权信息
    BIT_INFO_UPDATE_ERROR = 4  # 获取指定授权码的升级错误详细信息。是否存在详细信息由授权码类型及错误类型决定
    BIT_INFO_CONFIG = 5  # 获取当前产品的配置文件信息
    BIT_INFO_TOKEN_LIST = 7  # 获取当前机器上可用的BIT-ID列表信息
    BIT_INFO_LICENSE_ERROR = 8  #  获取上一次接口调用产生的的详细错误信息(目前仅支持1921错误和网络错误)


class BIT_CUSTOM_TYPE(Enum):
    CUSTOM_CLIENT_ID = 1  # 自定义客户端唯一的机器标识，该自定义标识会发送给服务器并用于今后的机器指纹匹配
    CUSTOM_ACCOUNT_LOCAL = 2  # 自定义客户端用户帐号，用于集团服务标识客户端用户
    CUSTOM_COMMENT = 3  # 自定义客户端登录注释，用于集团服务管理员获取用户登录请求信息


class QUERY_FEATURE_TYPE(Enum):
    BIT_QUERY_DEFAULT = 0x00  # 非阻塞登录，占用required个用户数，如果用户数不够将返回错误
    BIT_QUERY_AVAILABLE = 0x01  # 尽可能的占用 required（不能为 0）个用户数，如果用户数不够将占用剩余的用户数
    BIT_QUERY_CHECK = 0x02  # pCapacity 返回总共占用的用户数,required 必须为 0
    BIT_QUERY_WAIT = 0x03  # 阻塞队列登录，如果用户数不够，将加入队列阻塞等待
    BIT_QUERY_QUEUE = 0x04  # 非阻塞队列登录，如果用户数不够，将加入队列，不会阻塞等待
    BIT_QUERY_TEST = 0x100  # 测试特征项版本是否可用，与DEFAULT、WAIT、QUEUE搭配使用


class TICKET_INFO_TYPE(Enum):
    BIT_TICKET_TYPR_USERS = 0x1  # 获取当前 ticket 占用的用户数


class SET_ATTR_TYPE(Enum):
    ATTR_HB_STOPED_CALLBACK = 0x1  # 心跳回调函数，只会调用一次
    ATTR_WAIT_TIMEOUT = 0x2  # 阻塞队列超时时间，最小 16 秒，最大 24 小时，默认是一直等待
    ATTR_RETRY_COUNT = 0x3  # 客户端连接服务的重试次数，最小 0 次，最大 1024 次，默认是 0 次
    ATTR_RETRY_INTERVAL = 0x4  # 客户端连接服务的重试间隔，最小是 10 秒，最大是 4 小时，默认是 20 秒
    ATTR_CONNECT_TIMEOUT = 0x5  # 网络连接超时时间，最小是 1 秒，最大是 32 秒，默认是 3 秒
    ATTR_CONNECT_RETRY_COUNT = 0x6  # 网络连接重试次数，最小是 0 次，最大是 10 次，默认是 3 秒
    ATTR_HB_RETRY_FAILED_CALLBACK = 0x7  # 心跳回调函数，如果设置了重试连接次数，每次重试连接失败后，都会调用回调函数
    ATTR_HB_INTERVAL = 0x8  # 设置心跳间隔，最小是30s，最大是1/2个timeout时间
    ATTR_HB_AUTO_ENABLE = 0x9
    ATTR_HB_TIMEOUT = 0x10  # 设置回调函数超时时间


class BitAnswer:
    __handle = ctypes.c_void_p(0)

    __m_dll_handle = 0
    __application_data = (ctypes.c_char * 191)(
        0x40,
        0x80,
        0xBC,
        0x00,
        0xD7,
        0x26,
        0x40,
        0xDD,
        0x5F,
        0x43,
        0x0B,
        0x53,
        0xA6,
        0x6D,
        0xF1,
        0x90,
        0xBB,
        0x6A,
        0x13,
        0xCD,
        0xC7,
        0xB7,
        0x7D,
        0xCE,
        0xBC,
        0x73,
        0xAC,
        0xC9,
        0xBC,
        0xE2,
        0x48,
        0x9D,
        0xDB,
        0x31,
        0xF4,
        0xAC,
        0x79,
        0x72,
        0xC5,
        0xAC,
        0x08,
        0xED,
        0x0F,
        0xB5,
        0x43,
        0xF4,
        0xFB,
        0x9B,
        0x4A,
        0xD5,
        0x9F,
        0xBB,
        0xE4,
        0x55,
        0x2C,
        0xEA,
        0x9B,
        0xAC,
        0xF3,
        0xA6,
        0xA6,
        0x3C,
        0x91,
        0xAE,
        0x8D,
        0x0A,
        0x6A,
        0x1C,
        0x5E,
        0x39,
        0x7D,
        0x10,
        0xD4,
        0x78,
        0x78,
        0x61,
        0x9E,
        0x15,
        0xC4,
        0x0C,
        0x8C,
        0x3E,
        0xDD,
        0xE4,
        0xA8,
        0x05,
        0xD1,
        0x38,
        0x7E,
        0x99,
        0xF3,
        0xD9,
        0xBE,
        0x7E,
        0x0A,
        0x23,
        0x5A,
        0x77,
        0x5E,
        0xB8,
        0x5F,
        0x27,
        0x1F,
        0x20,
        0xAF,
        0xBA,
        0x6A,
        0x32,
        0x82,
        0xDA,
        0xFB,
        0xBD,
        0x51,
        0x07,
        0x02,
        0xDA,
        0x9C,
        0x66,
        0xC1,
        0x62,
        0x97,
        0x2E,
        0x64,
        0xD1,
        0xA8,
        0x8F,
        0xC6,
        0x61,
        0xC3,
        0x0E,
        0xCD,
        0xD3,
        0xA5,
        0xA0,
        0xCF,
        0xF9,
        0xBD,
        0xB0,
        0xCD,
        0x8D,
        0xB6,
        0xFB,
        0xC4,
        0x31,
        0x71,
        0x18,
        0x70,
        0x7E,
        0x50,
        0xAD,
        0x66,
        0xE6,
        0xEF,
        0xD8,
        0xD8,
        0xD5,
        0x21,
        0xBA,
        0x07,
        0x4D,
        0x0C,
        0x62,
        0x1E,
        0xD5,
        0x14,
        0xD0,
        0x5A,
        0xD8,
        0x56,
        0x71,
        0x36,
        0xBB,
        0x23,
        0x1E,
        0x19,
        0xCD,
        0x33,
        0x54,
        0x5D,
        0x92,
        0x1D,
        0x22,
        0x52,
        0x0B,
        0x6C,
        0x4F,
        0x80,
        0x36,
        0x82,
        0x8E,
        0xAE,
    )

    def __get_dll_handle(self):
        try:
            if self.__m_dll_handle == 0:
                if platform.system() == "Windows":
                    self.__m_dll_handle = ctypes.windll.LoadLibrary(BASE_DIR + "/00018997_0000000B_x64.dll")
                elif platform.system() == "Linux":
                    self.__m_dll_handle = ctypes.cdll.LoadLibrary(BASE_DIR + "/00018997_0000000B_x64.so")
            return self.__m_dll_handle
        except Exception as e:
            print(e)
            raise BitException(error_code=-1) from e

    def login(self, url, sn, mode):
        if sn is not None:
            sn_arr = ctypes.create_string_buffer(sn.encode())
        else:
            sn_arr = sn
        if url is not None:
            url_arr = ctypes.create_string_buffer(url.encode())
        else:
            url_arr = url
        result = self.__get_dll_handle().Bit_Login(
            url_arr, sn_arr, self.__application_data, ctypes.pointer(self.__handle), mode
        )

        if result != 0:
            raise BitException(error_code=result)

    def login_ex(self, url, sn, feature_id, xml, mode):
        if sn is not None:
            sn_arr = ctypes.create_string_buffer(sn.encode())
        else:
            sn_arr = sn
        if url is not None:
            url_arr = ctypes.create_string_buffer(url.encode())
        else:
            url_arr = url
        result = self.__get_dll_handle().Bit_LoginEx(
            url_arr, sn_arr, feature_id, xml, self.__application_data, ctypes.pointer(self.__handle), mode
        )
        if result != 0:
            raise BitException(error_code=result)

    def logout(self):
        result = self.__get_dll_handle().Bit_Logout(self.__handle)
        if result != 0:
            raise BitException(error_code=result)

    def read_feature(self, feature_id):
        feature_value = ctypes.c_int(0)
        result = self.__get_dll_handle().Bit_ReadFeature(self.__handle, feature_id, ctypes.pointer(feature_value))
        if result != 0:
            raise BitException(error_code=result)
        return feature_value

    def write_feature(self, feature_id, feature_value):
        result = self.__get_dll_handle().Bit_WriteFeature(self.__handle, feature_id, feature_value)
        if result != 0:
            raise BitException(error_code=result)

    def convert_feature(self, feature_id, para1, para2, para3, para4):
        feature_result = ctypes.c_int(0)
        result = self.__get_dll_handle().Bit_ConvertFeature(
            self.__handle, feature_id, para1, para2, para3, para4, ctypes.pointer(feature_result)
        )
        if result != 0:
            raise BitException(error_code=result)
        return feature_result

    def encrypt_feature(self, feature_id, plain_buffer):
        data_buff_size = len(plain_buffer) + 1
        cipher_buffer = (ctypes.c_byte * data_buff_size)(0)
        result = self.__get_dll_handle().Bit_EncryptFeature(
            self.__handle, feature_id, plain_buffer, cipher_buffer, data_buff_size
        )
        if result != 0:
            raise BitException(error_code=result)
        return cipher_buffer

    def decrypt_feature(self, feature_id, cipher_buffer):
        data_buff_size = len(cipher_buffer) + 1
        plain_buffer = (ctypes.c_byte * data_buff_size)(0)
        result = self.__get_dll_handle().Bit_DecryptFeature(
            self.__handle, feature_id, cipher_buffer, plain_buffer, data_buff_size
        )
        if result != 0:
            raise BitException(error_code=result)
        return plain_buffer

    def query_feature(self, feature_id):
        capacity = ctypes.c_int(0)
        result = self.__get_dll_handle().Bit_QueryFeature(self.__handle, feature_id, ctypes.pointer(capacity))
        if result != 0:
            raise BitException(error_code=result)
        return capacity

    def query_feature_ex(self, feature_id, mode, required, version):
        capacity = ctypes.c_int(0)
        if version is not None:
            version_arr = ctypes.create_string_buffer(version.encode())
        else:
            version_arr = version
        result = self.__get_dll_handle().Bit_QueryFeatureEx(
            self.__handle, feature_id, mode, required, ctypes.pointer(capacity), version_arr
        )
        if result != 0:
            raise BitException(error_code=result)
        return capacity

    def query_feature_ex2(self, feature_name, mode, required, version):
        ticket = ctypes.c_void_p(0)
        if feature_name is not None:
            feature_name_arr = ctypes.create_string_buffer(feature_name.encode())
        else:
            feature_name_arr = feature_name
        if version is not None:
            version_arr = ctypes.create_string_buffer(version.encode())
        else:
            version_arr = version
        result = self.__get_dll_handle().Bit_QueryFeatureEx2(
            self.__handle, feature_name_arr, mode, required, version_arr, ctypes.pointer(ticket)
        )
        if result != 0:
            raise BitException(error_code=result)
        return ticket

    def release_feature(self, feature_id):
        capacity = ctypes.c_int(0)
        result = self.__get_dll_handle().Bit_ReleaseFeature(self.__handle, feature_id, ctypes.pointer(capacity))
        if result != 0:
            raise BitException(error_code=result)
        return capacity

    def release_feature_ex(self, feature_id, consumed, version):
        capacity = ctypes.c_int(0)
        if version is not None:
            version_arr = ctypes.create_string_buffer(version.encode())
        else:
            version_arr = version
        result = self.__get_dll_handle().Bit_ReleaseFeatureEx(
            self.__handle, feature_id, consumed, ctypes.pointer(capacity), version_arr
        )
        if result != 0:
            raise BitException(error_code=result)
        return capacity

    def release_feature_ex2(self, ticket):
        result = self.__get_dll_handle().Bit_ReleaseFeatureEx2(ticket, 0)
        if result != 0:
            raise BitException(error_code=result)

    def set_data_item(self, data_item_name, data_item_value):
        if data_item_name is not None:
            data_item_name_arr = ctypes.create_string_buffer(data_item_name.encode())
        else:
            data_item_name_arr = data_item_name
        data_item_value_size = len(data_item_value) + 1
        result = self.__get_dll_handle().Bit_SetDataItem(
            self.__handle, data_item_name_arr, data_item_value, data_item_value_size
        )
        if result != 0:
            raise BitException(error_code=result)

    def remove_data_item(self, data_item_name):
        if data_item_name is not None:
            data_item_name_arr = ctypes.create_string_buffer(data_item_name.encode())
        else:
            data_item_name_arr = data_item_name
        result = self.__get_dll_handle().Bit_RemoveDataItem(self.__handle, data_item_name_arr)
        if result != 0:
            raise BitException(error_code=result)

    def get_data_item(self, data_item_name):
        data_item_value = (ctypes.c_byte * 1024)()
        if data_item_name is not None:
            data_item_name_arr = ctypes.create_string_buffer(data_item_name.encode())
        else:
            data_item_name_arr = data_item_name
        data_item_value_size = ctypes.c_int(1024)
        result = self.__get_dll_handle().Bit_GetDataItem(
            self.__handle, data_item_name_arr, data_item_value, ctypes.pointer(data_item_value_size)
        )
        if result != 0:
            raise BitException(error_code=result)
        return data_item_value

    def get_data_item_num(self):
        num = ctypes.c_int(0)
        result = self.__get_dll_handle().Bit_GetDataItemNum(self.__handle, ctypes.pointer(num))
        if result != 0:
            raise BitException(error_code=result)
        return num

    def get_data_item_name(self, index):
        data_item_name = ctypes.create_string_buffer(128)
        data_item_name_size = ctypes.c_int(len(data_item_name) + 1)
        result = self.__get_dll_handle().Bit_GetDataItemName(
            self.__handle, index, data_item_name, ctypes.pointer(data_item_name_size)
        )
        if result != 0:
            raise BitException(error_code=result)
        return data_item_name.value

    def get_session_info(self, type):
        session_info_size = ctypes.c_int(10240)
        session_info = ctypes.create_string_buffer(session_info_size.value)
        result = self.__get_dll_handle().Bit_GetSessionInfo(
            self.__handle, type, session_info, ctypes.pointer(session_info_size)
        )
        if result != 0:
            raise BitException(error_code=result)
        return session_info.value

    def get_feature_info(self, feature_id):
        feature_info = ctypes.create_string_buffer(10240)
        result = self.__get_dll_handle().Bit_GetFeatureInfo(self.__handle, feature_id, feature_info)
        if result != 0:
            raise BitException(error_code=result)
        return feature_info.value

    def get_request_info(self, sn, type):
        sn_arr = ctypes.create_string_buffer(sn.encode())
        request_info_size = ctypes.c_int(10240)
        request_info = ctypes.create_string_buffer(request_info_size.value)
        result = self.__get_dll_handle().Bit_GetRequestInfo(
            sn_arr, self.__application_data, type, request_info, ctypes.pointer(request_info_size)
        )
        if result != 0:
            raise BitException(error_code=result)
        return request_info.value

    def get_update_info(self, url, sn, request_info):
        if sn is not None:
            sn_arr = ctypes.create_string_buffer(sn.encode())
        else:
            sn_arr = sn
        if url is not None:
            url_arr = ctypes.create_string_buffer(url.encode())
        else:
            url_arr = url
        if request_info is not None:
            request_info_arr = ctypes.create_string_buffer(request_info)
        else:
            request_info_arr = request_info
        update_info_size = ctypes.c_int(10240)
        update_info = ctypes.create_string_buffer(update_info_size.value)
        result = self.__get_dll_handle().Bit_GetUpdateInfo(
            url_arr, sn_arr, self.__application_data, request_info_arr, update_info, ctypes.pointer(update_info_size)
        )
        if result != 0:
            raise BitException(error_code=result)
        return update_info.value

    def apply_update_info(self, update_info):
        receipt_info_size = ctypes.c_int(10240)
        receipt_info = ctypes.create_string_buffer(receipt_info_size.value)
        if update_info is not None:
            update_info_arr = ctypes.create_string_buffer(update_info)
        else:
            update_info_arr = update_info
        result = self.__get_dll_handle().Bit_ApplyUpdateInfo(
            self.__application_data, update_info_arr, receipt_info, ctypes.pointer(receipt_info_size)
        )
        if result != 0:
            raise BitException(error_code=result)
        return receipt_info.value

    def update_online(self, url, sn):
        if sn is not None:
            sn_arr = ctypes.create_string_buffer(sn.encode())
        else:
            sn_arr = sn
        if url is not None:
            url_arr = ctypes.create_string_buffer(url.encode())
        else:
            url_arr = url
        result = self.__get_dll_handle().Bit_UpdateOnline(url_arr, sn_arr, self.__application_data)
        if result != 0:
            raise BitException(error_code=result)

    def revoke(self, url, sn):
        if sn is not None:
            sn_arr = ctypes.create_string_buffer(sn.encode())
        else:
            sn_arr = sn
        if url is not None:
            url_arr = ctypes.create_string_buffer(url.encode())
        else:
            url_arr = url
        revocation_info_size = ctypes.c_int(10240)
        revocation_info = ctypes.create_string_buffer(revocation_info_size.value)
        result = self.__get_dll_handle().Bit_Revoke(
            url_arr, sn_arr, self.__application_data, revocation_info, ctypes.pointer(revocation_info_size)
        )
        if result != 0:
            raise BitException(error_code=result)
        return revocation_info

    def get_info(self, sn, type):
        if sn is not None:
            sn_arr = ctypes.create_string_buffer(sn.encode())
        else:
            sn_arr = sn
        info_size = ctypes.c_int(10240)
        info = ctypes.create_string_buffer(info_size.value)
        result = self.__get_dll_handle().Bit_GetInfo(
            sn_arr, self.__application_data, type, info, ctypes.pointer(info_size)
        )
        if result != 0:
            raise BitException(error_code=result)
        return info.value

    def remove_sn(self, sn):
        if sn is not None:
            sn_arr = ctypes.create_string_buffer(sn.encode())
        else:
            sn_arr = sn
        result = self.__get_dll_handle().Bit_RemoveSn(sn_arr, self.__application_data)
        if result != 0:
            raise BitException(error_code=result)

    def set_custom_info(self, infoId, info_data):
        if info_data is not None:
            info_data_arr = ctypes.create_string_buffer(info_data.encode())
        else:
            info_data_arr = info_data
        info_data_size = ctypes.c_int(len(info_data) + 1)
        result = self.__get_dll_handle().Bit_SetCustomInfo(infoId, info_data_arr, info_data_size)
        if result != 0:
            raise BitException(error_code=result)

    def get_version(self):
        version = ctypes.c_int(128)
        result = self.__get_dll_handle().Bit_GetVersion(ctypes.pointer(version))
        if result != 0:
            raise BitException(error_code=result)
        return version

    def check_out_sn(self, url, feature_id, duration_days):
        if url is not None:
            url_arr = ctypes.create_string_buffer(url.encode())
        else:
            url_arr = url
        result = self.__get_dll_handle().Bit_CheckOutSn(url_arr, feature_id, self.__application_data, duration_days)
        if result != 0:
            raise BitException(error_code=result)

    def check_in(self, url, feature_id):
        if url is not None:
            url_arr = ctypes.create_string_buffer(url.encode())
        else:
            url_arr = url
        result = self.__get_dll_handle().Bit_CheckIn(url_arr, feature_id, self.__application_data)
        if result != 0:
            raise BitException(error_code=result)

    def check_out_features(self, url, feature_list, feature_in_list, duration_days):
        if url is not None:
            url_arr = ctypes.create_string_buffer(url.encode())
        else:
            url_arr = url
        result = self.__get_dll_handle().Bit_CheckOutFeatures(
            url_arr, self.__application_data, feature_list, feature_in_list, duration_days
        )
        if result != 0:
            raise BitException(error_code=result)

    def get_ticket_info(self, ticket, type):
        info_size = ctypes.c_int(10240)
        info = ctypes.create_string_buffer(info_size.value)
        result = self.__get_dll_handle().Bit_GetTicketInfo(ticket, type, info, ctypes.pointer(info_size))
        if result != 0:
            raise BitException(error_code=result)
        return info.value

    def get_feature_info2(self, feature_name, version):
        if feature_name is not None:
            feature_name_arr = ctypes.create_string_buffer(feature_name.encode())
        else:
            feature_name_arr = feature_name
        if version is not None:
            version_arr = ctypes.create_string_buffer(version.encode())
        else:
            version_arr = version
        expired = ctypes.c_int(0)
        result = self.__get_dll_handle().Bit_GetFeatureInfo2(
            self.__handle, feature_name_arr, version_arr, ctypes.pointer(expired)
        )
        if result != 0:
            raise BitException(error_code=result)
        return expired.value

    def set_local_service(self, host_name, port, timeout):
        if host_name is not None:
            host_name_arr = ctypes.create_string_buffer(host_name.encode())
        else:
            host_name_arr = host_name
        result = self.__get_dll_handle().Bit_SetLocalServer(self.__application_data, host_name_arr, port, timeout)
        if result != 0:
            raise BitException(error_code=result)

    def set_root_path(self, path):
        if path is not None:
            path_arr = ctypes.create_string_buffer(path.encode())
        else:
            path_arr = path
        result = self.__get_dll_handle().Bit_SetRootPath(path_arr)
        if result != 0:
            raise BitException(error_code=result)

    def get_product_path(self):
        buf_size = ctypes.c_int(10240)
        buf = ctypes.create_string_buffer(buf_size.value)
        result = self.__get_dll_handle().Bit_GetProductPath(self.__application_data, buf, ctypes.pointer(buf_size))
        if result != 0:
            raise BitException(error_code=result)
        return buf.value

    def heartbeat(self):
        reconnects_num = ctypes.c_int(128)
        result = self.__get_dll_handle().Bit_Heartbeat(self.__handle, ctypes.pointer(reconnects_num))
        if result != 0:
            raise BitException(error_code=result)
        return reconnects_num

    def set_attr(self, type, callback):
        result = self.__get_dll_handle().Bit_SetAttr(self.__handle, type, callback)
        if result != 0:
            raise BitException(error_code=result)

    def get_last_error(self):
        result = self.__get_dll_handle().Bit_GetLastError()
        return result


callback_func = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_void_p)

### 比特云授权使用
当前采用**集团授权模式**：在物理机上部署并运行授权服务，容器内应用通过网络与该授权服务进行交互。

授权服务部署说明参考：

- [AIPOD服务器_许可部署操作说明.pdf](docs/AIPOD服务器_许可部署操作说明.pdf)

注意事项：

- 授权服务中填写的 IP 必须为物理机的 `IPv4` 地址，不要填写容器 IP。

<bitLicense version="5501">
  <version>3</version>
  <sn>WZ7O7NT7Q3VOCUMK</sn>
  <code>EqYZXAxNUpdOVtKAnPrtykDhvNhfdfush9TATXTDcOISILDax8JcGFO1ZneyzZv15llWI1x/9VSproILMCk/CT5mZH12Ll2FJXDt6HATuep8Ynm7HZvnxu7yH7yyBgwHwMm9mMzsaZbzV4x8Ka5fRQL29o4dZ30BoPG19YwGW1dAYE6cn3NzAPxXcka2mhp10wM33NvpTwPNPatFzFWWAzBwuTwY5cQDAxCItDjnRy/Mdz553s491ZFnEyoAZGFcjbFDJU1MQzVFFM4GVw2fcvUD993y3Cvb8X6YDRLU9pcK2Zxr2DM=</code>
  <description>
    <snInfo>
      <product pid="57YHW-OM" name="DTEmpower"/>
      <controlType>PRE</controlType>
      <activateDate>2025-04-08</activateDate>
      <expirationDays>30</expirationDays>
      <operatorType>Download</operatorType>
    </snInfo>
  </description>
</bitLicense>