"""
Authorization计算工具

提供基于AK/SK的BCE认证签名计算功能，用于生成API请求的Authorization头。
"""

import datetime
import hashlib
import hmac
import urllib.parse

from requests.auth import AuthBase

# mypy: allow-untyped-defs


def get_canonical_time(timestamp=0):
    """Get cannonical time.

    Args:
        timestamp: Unix时间戳，如果为0则使用当前时间

    Returns:
        规范化的时间字符串，格式如 "2025-12-13T03:19:46Z"
    """
    if timestamp == 0:
        utctime = datetime.datetime.now(datetime.timezone.utc)
    else:
        utctime = datetime.datetime.fromtimestamp(timestamp, tz=datetime.timezone.utc)
    return "%04d-%02d-%02dT%02d:%02d:%02dZ" % (  # noqa: UP031
        utctime.year,
        utctime.month,
        utctime.day,
        utctime.hour,
        utctime.minute,
        utctime.second,
    )


def gen_authorization(request, ak, sk, timestamp=None, expire_period=1800):
    """generate authorization string
    if not specify timestamp, then use current time;

    Args:
        request: 请求字典，包含 'uri', 'params', 'method', 'headers'
        ak: Access Key
        sk: Secret Key
        timestamp: 时间戳（可选）
        expire_period: 过期时间（秒），默认为1800

    Returns:
        授权字符串
    """
    signedheaders = []
    if "headers" in request:
        signedheaders = list(key.lower() for key in request["headers"].keys() if key != "")
        signedheaders.sort()
    authorization = build_authorization(ak, signedheaders, expire_period, timestamp)
    signingkey = _calc_signingkey(authorization, sk)
    signature = _calc_signature(signingkey, request, signedheaders)
    authorization["signature"] = signature
    return serialize_authorization(authorization)


def serialize_authorization(auth):
    """serialize Authorization object to authorization string

    Args:
        auth: 授权对象字典

    Returns:
        序列化后的授权字符串
    """
    val = "/".join(
        (
            auth["version"],
            auth["access"],
            auth["timestamp"],
            auth["period"],
            ";".join(auth["signedheaders"]),
            auth["signature"],
        )
    )
    return get_utf8_value(val)


def build_authorization(accesskey, signedheaders, period=1800, timestamp=None):
    """build Authorization object

    Args:
        accesskey: Access Key
        signedheaders: 需要签名的请求头列表
        period: 过期时间（秒），默认为1800
        timestamp: 时间戳（可选）

    Returns:
        授权对象字典
    """
    auth = {}
    auth["version"] = "bce-auth-v1"
    auth["access"] = accesskey
    if not timestamp:
        auth["timestamp"] = get_canonical_time()
    else:
        auth["timestamp"] = timestamp
    auth["period"] = str(period)
    auth["signedheaders"] = signedheaders
    return auth


def _calc_signingkey(auth, sk):
    """Get a signing key

    Args:
        auth: 授权对象字典
        sk: Secret Key

    Returns:
        签名密钥（hex字符串）
    """
    string_to_sign = "/".join((auth["version"], auth["access"], auth["timestamp"], auth["period"]))

    signingkey = hmac.new(bytes(sk, "utf-8"), bytes(string_to_sign, "utf-8"), hashlib.sha256).hexdigest()
    return signingkey


def get_utf8_value(value):
    """Get the UTF8-encoded version of a value.

    Args:
        value: 输入值（可以是字符串或bytes）

    Returns:
        UTF-8编码的字符串
    """
    if isinstance(value, bytes):
        return value.decode("utf-8")
    return value.encode("utf-8").decode("utf-8")


def normalized_uri(uri):
    """Construct a normalized(except slash '/') uri
    eg. /json-api/v1/example/ ==> /json-api/v1/example/

    Args:
        uri: URI路径

    Returns:
        规范化后的URI
    """
    return urllib.parse.quote(get_utf8_value(uri), safe="-_.~/")


def normalized(msg):
    """Construct a normalized uri

    Args:
        msg: 要规范化的字符串

    Returns:
        规范化后的字符串
    """
    return urllib.parse.quote(get_utf8_value(msg), safe="-_.~")


def canonical_qs(params):
    """Construct a sorted, correctly encoded query string

    Args:
        params: 参数字典

    Returns:
        规范化后的查询字符串
    """
    keys = list(params)
    keys.sort()
    pairs = []
    for key in keys:
        if key == "authorization":
            continue
        val = normalized(params[key])
        pairs.append(urllib.parse.quote(key, safe="") + "=" + val)
    qs = "&".join(pairs)
    return qs


def canonical_header_str(headers, signedheaders=None):
    """calculate canonicalized header string

    Args:
        headers: 请求头字典
        signedheaders: 需要签名的请求头列表（可选）

    Returns:
        规范化后的请求头字符串
    """
    headers_norm_lower = dict()
    for k, v in headers.items():
        key_norm_lower = normalized(k.lower())
        value_norm_lower = normalized(v.strip())
        headers_norm_lower[key_norm_lower] = value_norm_lower
    keys = list(headers_norm_lower)
    keys.sort()
    if "host" not in keys:
        raise ValueError("host header is required")
    header_list = []
    default_signed = ("host", "content-length", "content-type", "content-md5")
    if signedheaders:
        for key in signedheaders:
            key = normalized(key.lower())
            if key not in keys:
                raise ValueError
            if headers_norm_lower[key]:
                header_list.append(key + ":" + headers_norm_lower[key])
    else:
        for key in keys:
            if key.startswith("x-bce-") or key in default_signed:
                header_list.append(key + ":" + headers_norm_lower[key])
    return "\n".join(header_list)


def _calc_signature(key, request, signedheaders):
    """Generate BCE signature string.

    Args:
        key: 签名密钥（hex字符串）
        request: 请求字典，包含 'uri', 'params', 'method', 'headers'
        signedheaders: 需要签名的请求头列表

    Returns:
        签名（hex字符串）
    """
    # Create canonical request
    params = {}
    headers = {}
    # print request
    if "params" in request:
        params = request["params"]
    if "headers" in request:
        headers = request["headers"]
    cr = "\n".join(
        (
            request["method"].upper(),
            normalized_uri(request["uri"]),
            canonical_qs(params),
            canonical_header_str(headers, signedheaders),
        )
    )
    signature = hmac.new(bytes(key, "utf-8"), bytes(cr, "utf-8"), hashlib.sha256).hexdigest()
    return signature


class BceAuth(AuthBase):
    """BCE 认证类，用于requests库的认证支持。"""

    def __init__(self, ak, sk):
        self.ak = ak
        self.sk = sk

    def __call__(self, r):
        # 解析URL和参数
        parsed = urllib.parse.urlparse(r.url)

        domain = parsed.netloc
        uri = parsed.path

        params = {}
        if parsed.query:
            for item in parsed.query.split("&"):
                if "=" in item:
                    k, v = item.split("=", 1)
                    params[k] = v
                else:
                    params[item] = ""

        # 保证 Host header
        r.headers.setdefault("Host", domain)

        bce_request = {
            "uri": uri,
            "params": params,
            "method": r.method.upper(),
            "headers": {
                "Host": domain,
            },
        }

        # 添加 x-bce-security-token
        # session_token = Config.get_session_token()
        # if session_token:
        #     r.headers["x-bce-security-token"] = session_token
        #     bce_request["headers"]["x-bce-security-token"] = session_token

        auth = gen_authorization(bce_request, self.ak, self.sk)
        r.headers["Authorization"] = get_utf8_value(auth)

        return r
