MibooGram/modules/pay_log/log.py

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

import aiofiles
from simnet import GenshinClient, Region
from simnet.errors import AuthkeyTimeout, InvalidAuthkey
from simnet.models.genshin.transaction import TransactionKind, BaseTransaction
from simnet.utils.player import recognize_genshin_server

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
from utils.uid import mask_number

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, by_alias=True))

    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)

    @staticmethod
    def get_game_client(player_id: int) -> GenshinClient:
        if recognize_genshin_server(player_id) in ["cn_gf01", "cn_qd01"]:
            return GenshinClient(player_id=player_id, region=Region.CHINESE, lang="zh-cn")
        return GenshinClient(player_id=player_id, region=Region.OVERSEAS, lang="zh-cn")

    async def get_log_data(
        self,
        user_id: int,
        player_id: int,
        authkey: str,
    ) -> int:
        """使用 authkey 获取历史记录数据，并合并旧数据
        :param user_id: 用户id
        :param player_id: 游戏id
        :param authkey: authkey
        :return: 更新结果
        """
        new_num = 0
        pay_log, have_old = await self.load_history_info(str(user_id), str(player_id))
        history_ids = [i.id for i in pay_log.list]
        client = self.get_game_client(player_id)
        try:
            transaction_log = await client.transaction_log(authkey=authkey, kind=TransactionKind.CRYSTAL.value)
            for data in transaction_log:
                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
        finally:
            await client.shutdown()
        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.player_id), 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, player_id: int):
        """获取分析数据
        :param user_id: 用户id
        :param player_id: 玩家id
        :return: 分析数据
        """
        pay_log, status = await self.load_history_info(str(user_id), str(player_id))
        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": mask_number(player_id),
            "datas": datas,
            "bar_data": month_datas,
            "pie_data": pie_datas,
        }