# coding: utf-8

import functools
import json
import os
import time
import uuid
from dataclasses import dataclass
from datetime import date, datetime, timedelta

import airflow.configuration
import pendulum
import requests
import sqlparse
from airflow.models import Connection, Variable

local_tz = pendulum.timezone('Asia/Shanghai')


def dag_start_date(hourly=False, start_hour=0):
    if hourly:
        start_date = datetime.now(tz=local_tz) - timedelta(hours=1)
        return datetime(start_date.year, start_date.month, start_date.day, start_date.hour, tzinfo=local_tz)
    else:
        start_date = datetime.now(tz=local_tz) - timedelta(days=1)
        return datetime(start_date.year, start_date.month, start_date.day, start_hour, tzinfo=local_tz)


def deployment_env():
    return Variable.get('deployment_env', default_var='PROD')


def get_splitted_sql(sql_file):
    file_path = os.path.join(airflow.configuration.get_airflow_home(), sql_file)
    with open(file_path, encoding='UTF-8') as f:
        content = f.read()
        splitted = sqlparse.split(content)
        splitted = [sql for sql in splitted if sql]
        return splitted


def get_file_content(file):
    file_path = os.path.join(airflow.configuration.get_airflow_home(), file)
    with open(file_path, encoding='UTF-8') as f:
        return f.read()


def today():
    now = datetime.now()
    return datetime(year=now.year, month=now.month, day=now.day)


def lru_ttl_cache(ttl_seconds, maxsize=32, typed=False):
    """Least-recently-used cache decorator with time-based cache invalidation."""

    def wrapper(func):
        @functools.lru_cache(maxsize=maxsize, typed=typed)
        def _real_call(*args, __time_salt, **kwargs):
            return func(*args, **kwargs)

        @functools.wraps(func)
        def _wrapped(*args, **kwargs):
            return _real_call(*args, __time_salt=int(time.time() / ttl_seconds), **kwargs)

        return _wrapped

    return wrapper


@dataclass
class Quote:
    price_date: date  # 日期
    price_high: float | None  # 最高价
    price_low: float | None  # 最低价
    price_open: float | None  # 开盘价
    price_close: float | None  # 收盘价
    volume: float | None  # 成交量
    turnover: float | None  # 成交额


# 每日行情
def daily_quote(ticker: str, days: int = 20) -> list[Quote]:
    fosun_quote_conn = Connection.get_connection_from_secrets(conn_id='fosun_quote')
    api_url = f'{fosun_quote_conn.schema}://{fosun_quote_conn.host}/market/v1/kline'

    request_id = uuid.uuid4().hex
    headers = {
        "Content-type": "application/json",
        "X-request-id": request_id,
        "X-lang": "zhTc",
    }
    req_data = {
        "jsonrpc": "2.0",
        "params": {
            "code": ticker,
            "ktype": "day",
            "time": 0,
            "num": days,
            "right": "forward",
            "suspension": 1,
        },
        "method": "getKline",
        "id": request_id,
    }
    req_data = json.dumps(req_data)

    resp = requests.post(url=api_url, headers=headers, data=req_data)
    resp.raise_for_status()
    resp_result = resp.json()["result"]

    ndigits_raw = resp_result.get("power", 3)
    try:
        ndigits = int(ndigits_raw)
    except ValueError:
        if ticker.startswith("us"):
            ndigits = 4
        elif ticker.startswith("hk"):
            ndigits = 3
        else:
            ndigits = 2
    price_factor = 10.0 ** ndigits

    resp_data = resp_result["data"]
    result = [Quote(
        price_date=date(
            int(d["time"]) // 10000,
            (int(d["time"]) % 10000) // 100,
            int(d["time"]) % 100,
        ),
        price_high=d["high"] / price_factor if "high" in d else None,
        price_low=d["low"] / price_factor if "low" in d else None,
        price_open=d["open"] / price_factor if "open" in d else None,
        price_close=d["close"] / price_factor if "close" in d else None,
        volume=d["vol"] if "vol" in d else None,
        turnover=d["turnover"] / price_factor if "turnover" in d else None,
    ) for d in resp_data]

    return result


# 方便阅读的金额格式
def format_money(amount: float) -> str:
    if amount >= 100000000:
        amount /= 100000000
        return f'{amount:,.1f}亿'
    elif amount >= 10000:
        amount /= 10000
        return f'{amount:,.1f}万'
    else:
        return f'{amount:,.1f}'
