MibooGram/modules/pay_log/log.py

import contextlib
from pathlib import Path
from typing import Tuple, Optional, List, Dict

import aiofiles
from genshin import Client, AuthkeyTimeout, InvalidAuthkey
from genshin.models import TransactionKind, BaseTransaction

from modules.pay_log.error import PayLogAuthkeyTimeout, PayLogInvalidAuthkey, PayLogNotFound
from modules.pay_log.models import PayLog as PayLogModel, BaseInfo
from utils.const import PROJECT_ROOT

try:
    import ujson as jsonlib

except ImportError:
    import json as jsonlib

PAY_LOG_PATH = PROJECT_ROOT.joinpath("data", "apihelper", "pay_log")
PAY_LOG_PATH.mkdir(parents=True, exist_ok=True)


class PayLog:
    def __init__(self, pay_log_path: Path = PAY_LOG_PATH):
        self.pay_log_path = pay_log_path

    @staticmethod
    async def load_json(path):
        async with aiofiles.open(path, "r", encoding="utf-8") as f:
            return jsonlib.loads(await f.read())

    @staticmethod
    async def save_json(path, data: PayLogModel):
        async with aiofiles.open(path, "w", encoding="utf-8") as f:
            return await f.write(data.json(ensure_ascii=False, indent=4))

    def get_file_path(
        self,
        user_id: str,
        uid: str,
        bak: bool = False,
    ) -> Path:
        """获取文件路径
        :param user_id: 用户 ID
        :param uid: UID
        :param bak: 是否为备份文件
        :return: 文件路径
        """
        return self.pay_log_path / f"{user_id}-{uid}.json{'.bak' if bak else ''}"

    async def load_history_info(
        self,
        user_id: str,
        uid: str,
        only_status: bool = False,
    ) -> Tuple[Optional[PayLogModel], bool]:
        """读取历史记录数据
        :param user_id: 用户id
        :param uid: 原神uid
        :param only_status: 是否只读取状态
        :return: 抽卡记录数据
        """
        file_path = self.get_file_path(user_id, uid)
        if only_status:
            return None, file_path.exists()
        if not file_path.exists():
            return PayLogModel(info=BaseInfo(uid=uid), list=[]), False
        try:
            return PayLogModel.parse_obj(await self.load_json(file_path)), True
        except jsonlib.JSONDecodeError:
            return PayLogModel(info=BaseInfo(uid=uid), list=[]), False

    async def remove_history_info(
        self,
        user_id: str,
        uid: str,
    ) -> bool:
        """删除历史记录数据
        :param user_id: 用户id
        :param uid: 原神uid
        :return: 是否删除成功
        """
        file_path = self.get_file_path(user_id, uid)
        file_bak_path = self.get_file_path(user_id, uid, bak=True)
        with contextlib.suppress(Exception):
            file_bak_path.unlink(missing_ok=True)
        if file_path.exists():
            try:
                file_path.unlink()
            except PermissionError:
                return False
            return True
        return False

    async def save_pay_log_info(self, user_id: str, uid: str, info: PayLogModel) -> None:
        """保存日志记录数据
        :param user_id: 用户id
        :param uid: 原神uid
        :param info: 记录数据
        """
        save_path = self.pay_log_path / f"{user_id}-{uid}.json"
        save_path_bak = self.pay_log_path / f"{user_id}-{uid}.json.bak"
        # 将旧数据备份一次
        with contextlib.suppress(PermissionError):
            if save_path.exists():
                if save_path_bak.exists():
                    save_path_bak.unlink()
                save_path.rename(save_path.parent / f"{save_path.name}.bak")
        # 写入数据
        await self.save_json(save_path, info)

    async def get_log_data(
        self,
        user_id: int,
        client: Client,
        authkey: str,
    ) -> int:
        """使用 authkey 获取历史记录数据，并合并旧数据
        :param user_id: 用户id
        :param client: genshin client
        :param authkey: authkey
        :return: 更新结果
        """
        new_num = 0
        pay_log, have_old = await self.load_history_info(str(user_id), str(client.uid))
        history_ids = [i.id for i in pay_log.list]
        try:
            async for data in client.transaction_log(TransactionKind.CRYSTAL, authkey=authkey):
                if data.id not in history_ids:
                    pay_log.list.append(data)
                    new_num += 1
        except AuthkeyTimeout as exc:
            raise PayLogAuthkeyTimeout from exc
        except InvalidAuthkey as exc:
            raise PayLogInvalidAuthkey from exc
        if new_num > 0 or have_old:
            pay_log.list.sort(key=lambda x: (x.time, x.id), reverse=True)
            pay_log.info.update_now()
            await self.save_pay_log_info(str(user_id), str(client.uid), pay_log)
        return new_num

    @staticmethod
    async def get_month_data(pay_log: PayLogModel, price_data: List[Dict]) -> Tuple[int, List[Dict]]:
        """获取月份数据
        :param pay_log: 日志数据
        :param price_data: 商品数据
        :return: 月份数据
        """
        all_amount: int = 0
        months: List[int] = []
        month_datas: List[Dict] = []
        last_month: Optional[Dict] = None
        month_data: List[Optional[BaseTransaction]] = []
        for i in pay_log.list:
            if i.amount <= 0:
                continue
            all_amount += i.amount
            if i.time.month not in months:
                months.append(i.time.month)
                if last_month:
                    last_month["amount"] = sum(i.amount for i in month_data)
                    month_data.clear()
                if len(months) <= 6:
                    last_month = {
                        "month": f"{i.time.month}月",
                        "amount": 0,
                    }
                    month_datas.append(last_month)
                else:
                    last_month = None
            for j in price_data:
                if i.amount in j["price"]:
                    j["count"] += 1
                    break
            month_data.append(i)
        if last_month:
            last_month["amount"] = sum(i.amount for i in month_data)
            month_data.clear()
        if not month_datas:
            raise PayLogNotFound
        return all_amount, month_datas

    async def get_analysis(self, user_id: int, client: Client):
        """获取分析数据
        :param user_id: 用户id
        :param client: genshin client
        :return: 分析数据
        """
        pay_log, status = await self.load_history_info(str(user_id), str(client.uid))
        if not status:
            raise PayLogNotFound
        # 单双倍结晶数
        price_data = [
            {
                "price": price,
                "count": 0,
            }
            for price in [[680], [300], [8080, 12960], [3880, 6560], [2240, 3960], [1090, 1960], [330, 600], [60, 120]]
        ]
        price_data_name = ["大月卡", "小月卡", "648", "328", "198", "98", "30", "6"]
        real_price = [68, 30, 648, 328, 198, 98, 30, 6]
        all_amount, month_datas = await PayLog.get_month_data(pay_log, price_data)
        month_data = sorted(month_datas, key=lambda k: k["amount"], reverse=True)
        all_pay = sum((price_data[i]["count"] * real_price[i]) for i in range(len(price_data)))
        datas = [
            {"value": f"￥{all_pay:.0f}", "name": "总消费"},
            {"value": all_amount, "name": "总结晶"},
            {"value": f"{month_data[0]['month']}", "name": "消费最多"},
            {
                "value": f"￥{month_data[0]['amount'] / 10:.0f}",
                "name": f"{month_data[0]['month']}消费",
            },
            *[
                {
                    "value": price_data[i]["count"] if i != 0 else "*",
                    "name": f"{price_data_name[i]}",
                }
                for i in range(len(price_data))
            ],
        ]
        pie_datas = [
            {
                "value": f"{price_data[i]['count'] * real_price[i]:.0f}",
                "name": f"{price_data_name[i]}",
            }
            for i in range(len(price_data))
            if price_data[i]["count"] > 0
        ]
        return {
            "uid": client.uid,
            "datas": datas,
            "bar_data": month_datas,
            "pie_data": pie_datas,
        }