PamGram/modules/apihelper/client/components/calendar.py

import re
from datetime import datetime, timedelta
from typing import List, Tuple, Optional, Dict, Union, TYPE_CHECKING

from httpx import AsyncClient
from metadata.shortname import roleToId
from modules.apihelper.client.components.remote import Remote
from modules.apihelper.models.genshin.calendar import Date, FinalAct, ActEnum, ActDetail, ActTime, BirthChar
from modules.wiki.character import Character


if TYPE_CHECKING:
    from core.dependence.assets import AssetsService


class Calendar:
    """活动日历"""

    ANNOUNCEMENT_LIST = "https://hkrpg-api.mihoyo.com/common/hkrpg_cn/announcement/api/getAnnList"
    ANNOUNCEMENT_CONTENT = "https://hkrpg-api-static.mihoyo.com/common/hkrpg_cn/announcement/api/getAnnContent"
    ANNOUNCEMENT_PARAMS = {
        "game": "hkrpg",
        "game_biz": "hkrpg_cn",
        "lang": "zh-cn",
        "bundle_id": "hkrpg_cn",
        "platform": "pc",
        "region": "prod_gf_cn",
        "level": "55",
        "uid": "100000000",
    }
    IGNORE_IDS = [
        194,  # 有奖问卷调查开启！
        183,  # 官方社群一览
        171,  # 米游社《崩坏：星穹铁道》专属工具一览
        187,  # 《崩坏：星穹铁道》防沉迷系统公告
        185,  # 《崩坏：星穹铁道》公平运营声明
    ]
    IGNORE_RE = re.compile(r"(内容专题页|版本更新说明|调研|防沉迷|米游社|专项意见|游戏优化及已知问题说明|问卷调查|版本更新通知|更新时间说明|预下载功能|周边限时|周边上新|角色演示)")
    FULL_TIME_RE = re.compile(r"(魔神任务)")

    def __init__(self):
        self.client = AsyncClient()

    @staticmethod
    def get_now_hour() -> datetime:
        """获取当前时间"""
        return datetime.now().replace(minute=0, second=0, microsecond=0)

    async def parse_official_content_date(self) -> Dict[str, ActTime]:
        """解析官方内容时间"""
        time_map = {}
        req = await self.client.get(self.ANNOUNCEMENT_CONTENT, params=self.ANNOUNCEMENT_PARAMS)
        if req.status_code != 200:
            return time_map
        detail_data = req.json()
        for data in detail_data.get("data", {}).get("list", []):
            ann_id = data.get("ann_id", 0)
            title = data.get("title", "")
            content = data.get("content", "")
            if ann_id in self.IGNORE_IDS or self.IGNORE_RE.findall(title):
                continue
            content = re.sub(r'(<|&lt;)[\w "%:;=\-\\/\\(\\),\\.]+(>|&gt;)', "", content)
            try:
                if reg_ret := re.search(r"(?:活动时间|祈愿介绍|任务开放时间|冒险....包|折扣时间)\s*〓([^〓]+)(〓|$)", content):
                    if time_ret := re.search(r"(?:活动时间)?(?:〓|\s)*([0-9\\/\\: ~]{6,})", reg_ret[1]):
                        start_time, end_time = time_ret[1].split("~")
                        start_time = start_time.replace("/", "-").strip()
                        end_time = end_time.replace("/", "-").strip()
                        time_map[str(ann_id)] = ActTime(
                            title=title,
                            start=start_time,
                            end=end_time,
                        )
            except (IndexError, ValueError):
                continue
        return time_map

    async def req_cal_data(self) -> Tuple[List[List[ActDetail]], Dict[str, ActTime]]:
        """请求日历数据"""
        list_data = await self.client.get(self.ANNOUNCEMENT_LIST, params=self.ANNOUNCEMENT_PARAMS)
        list_data = list_data.json()

        new_list_data = [[], []]
        for idx, data in enumerate(list_data.get("data", {}).get("list", [])):
            for item in data.get("list", []):
                new_list_data[idx].append(ActDetail(**item))
        time_map = {}
        time_map.update(await self.parse_official_content_date())
        req = await self.client.get(self.MIAO_API)
        if req.status_code == 200:
            miao_data = req.json()
            time_map.update({key: ActTime(**value) for key, value in miao_data.get("data", {}).items()})
        remote_data = await Remote.get_remote_calendar()
        if remote_data:
            time_map.update({key: ActTime(**value) for key, value in remote_data.get("data", {}).items()})
        return new_list_data, time_map

    @staticmethod
    def date_to_weekday(date_: datetime) -> str:
        """日期转换为星期"""
        time = ["一", "二", "三", "四", "五", "六", "日"]
        return time[date_.weekday()]

    async def get_date_list(self) -> Tuple[List[Date], datetime, datetime, timedelta, float]:
        """获取日历数据"""
        data_list: List[Date] = []
        today = self.get_now_hour()
        temp = today - timedelta(days=7)
        month = 0
        date, week, is_today = [], [], []
        start_date, end_date = None, None
        for i in range(13):
            temp += timedelta(days=1)
            m, d, w = temp.month, temp.day, self.date_to_weekday(temp)
            if month == 0:
                start_date = temp
                month = m
            if month != m and len(date) > 0:
                data_list.append(Date(month=month, date=date, week=week, is_today=is_today))
                date, week, is_today = [], [], []
                month = m
            date.append(d)
            week.append(w)
            is_today.append(temp == today)
            if i == 12:
                data_list.append(Date(month=month, date=date, week=week, is_today=is_today))
                end_date = temp
        start_time = start_date.replace(hour=0, minute=0, second=0, microsecond=0)
        end_time = end_date.replace(hour=23, minute=59, second=59, microsecond=999999)
        total_range: timedelta = end_time - start_time
        now_left: float = (self.get_now_hour() - start_time) / total_range * 100
        return (
            data_list,
            start_time,
            end_time,
            total_range,
            now_left,
        )

    @staticmethod
    def human_read(d: timedelta) -> str:
        """将日期转换为人类可读"""
        hour = d.seconds // 3600
        minute = d.seconds // 60 % 60
        if minute >= 59:
            hour += 1
        text = ""
        if d.days:
            text += f"{d.days}天"
        if hour:
            text += f"{hour}小时"
        return text

    @staticmethod
    def count_width(
        act: FinalAct,
        detail: Optional[ActTime],
        ds: ActDetail,
        start_time: datetime,
        end_time: datetime,
        total_range: timedelta,
    ) -> Tuple[datetime, datetime]:
        """计算宽度"""

        def get_date(d1: str, d2: str) -> datetime:
            if d1 and len(d1) > 6:
                return datetime.strptime(d1, "%Y-%m-%d %H:%M:%S")
            return datetime.strptime(d2, "%Y-%m-%d %H:%M:%S")

        s_date = get_date(detail and detail.start, ds.start_time)
        e_date = get_date(detail and detail.end, ds.end_time)
        s_time = max(s_date, start_time)
        e_time = min(e_date, end_time)

        s_range = s_time - start_time
        e_range = e_time - start_time

        act.left = s_range / total_range * 100
        act.width = e_range / total_range * 100 - act.left
        act.duration = (e_time - s_time).total_seconds()
        act.start = s_date.strftime("%m-%d %H:%M")
        act.end = e_date.strftime("%m-%d %H:%M")
        return s_date, e_date

    def parse_label(self, act: FinalAct, is_act: bool, s_date: datetime, e_date: datetime) -> None:
        """解析活动标签"""
        now = self.get_now_hour()
        label = ""
        if self.FULL_TIME_RE.findall(act.title) or e_date - s_date > timedelta(days=365):
            label = f"{s_date.strftime('%m-%d %H:%M')} 后永久有效" if s_date < now else "永久有效"
        elif s_date < now < e_date:
            label = f'{e_date.strftime("%m-%d %H:%M")} ({self.human_read(e_date - now)}后结束)'
            if act.width > (38 if is_act else 55):
                label = f"{s_date.strftime('%m-%d %H:%M')} ~ {label}"
        elif s_date > now:
            label = f'{s_date.strftime("%m-%d %H:%M")} ({self.human_read(s_date - now)}后开始)'
        elif is_act:
            label = f"{s_date.strftime('%m-%d %H:%M')} ~ {e_date.strftime('%m-%d %H:%M')}"
        act.label = label

    @staticmethod
    async def parse_type(act: FinalAct, assets: "AssetsService") -> None:
        """解析活动类型"""
        if "神铸赋形" in act.title:
            act.type = ActEnum.weapon
            act.title = re.sub(r"(单手剑|双手剑|长柄武器|弓|法器|·)", "", act.title)
            act.sort = 2
        elif "祈愿" in act.title:
            act.type = ActEnum.character
            if reg_ret := re.search(r"·(.*)\(", act.title):
                char_name = reg_ret[1]
                char = assets.avatar(roleToId(char_name))
                act.banner = (await assets.namecard(char.id).navbar()).as_uri()
                act.face = (await char.icon()).as_uri()
                act.sort = 1
        elif "纪行" in act.title:
            act.type = ActEnum.no_display
        elif act.title == "深渊":
            act.type = ActEnum.abyss

    async def get_list(
        self,
        ds: ActDetail,
        start_time: datetime,
        end_time: datetime,
        total_range: timedelta,
        time_map: Dict[str, ActTime],
        is_act: bool,
        assets: "AssetsService",
    ) -> Optional[FinalAct]:
        """获取活动列表"""
        act = FinalAct(
            id=ds.ann_id,
            type=ActEnum.activity if is_act else ActEnum.normal,
            title=ds.title,
            banner=ds.banner if is_act else "",
            sort=5 if is_act else 10,
            icon=ds.tag_icon,
        )
        detail: Optional[ActTime] = time_map.get(str(act.id))

        if act.id in self.IGNORE_IDS or self.IGNORE_RE.findall(act.title) or (detail and not detail.display):
            return None
        await self.parse_type(act, assets)
        s_date, e_date = self.count_width(act, detail, ds, start_time, end_time, total_range)
        self.parse_label(act, is_act, s_date, e_date)

        if s_date <= end_time and e_date >= start_time:
            act.mergeStatus = 1 if act.type in {ActEnum.activity, ActEnum.normal} else 0
            return act

    @staticmethod
    def get_abyss_cal(start_time: datetime, end_time: datetime) -> List[List[Union[datetime, str]]]:
        """获取深渊日历"""
        last = datetime.now().replace(day=1) - timedelta(days=2)
        last_month = last.month
        curr = datetime.now()
        curr_month = curr.month
        next_date = last + timedelta(days=40)
        next_month = next_date.month

        def start(date: datetime, up: bool = False):
            return date.replace(day=1 if up else 16, hour=4, minute=0, second=0, microsecond=0)

        def end(date: datetime, up: bool = False):
            return date.replace(day=1 if up else 16, hour=3, minute=59, second=59, microsecond=999999)

        check = [
            [start(last, False), end(last, True), f"{last_month}月下半"],
            [start(curr, True), end(curr, False), f"{curr_month}月上半"],
            [start(curr, False), end(next_date, True), f"{curr_month}月下半"],
            [start(next_date, True), end(next_date, False), f"{next_month}月上半"],
        ]
        ret = []
        for ds in check:
            s, e, _ = ds
            if (s <= start_time <= e) or (s <= end_time <= e):
                ret.append(ds)
        return ret

    async def get_birthday_char(
        self, date_list: List[Date], assets: "AssetsService"
    ) -> Tuple[int, Dict[str, Dict[str, List[BirthChar]]]]:
        """获取生日角色"""
        birthday_list = await self.async_gen_birthday_list()
        birthday_char_line = 0
        birthday_chars = {}
        for date in date_list:
            birthday_chars[str(date.month)] = {}
            for d in date.date:
                key = f"{date.month}_{d}"
                if char := birthday_list.get(key):
                    birthday_char_line = max(len(char), birthday_char_line)
                    birthday_chars[str(date.month)][str(d)] = []
                    for c in char:
                        character = await Character.get_by_name(c)
                        birthday_chars[str(date.month)][str(d)].append(
                            BirthChar(
                                name=c,
                                star=character.rarity,
                                icon=(await assets.avatar(roleToId(c)).icon()).as_uri(),
                            )
                        )
        return birthday_char_line, birthday_chars

    @staticmethod
    def get_merge_next(target: List[FinalAct], li: FinalAct) -> Optional[FinalAct]:
        """获取下一个可以合并的活动"""
        return next(
            (li2 for li2 in target if (li2.mergeStatus == 1) and (li.left + li.width <= li2.left)),
            None,
        )

    def merge_list(self, target: List[FinalAct]) -> Tuple[List[List[FinalAct]], int, int]:
        """将两个活动合并为一行"""
        char_count = 0
        char_old = 0
        ret: List[List[FinalAct]] = []
        for idx, li in enumerate(target):
            if li.type == ActEnum.character:
                char_count += 1
                if li.left == 0:
                    char_old += 1
                li.idx = char_count
            if li.mergeStatus == 1:
                if li2 := self.get_merge_next(target[idx + 1 :], li):
                    li.mergeStatus = 2
                    li2.mergeStatus = 2
                    ret.append([li, li2])
            if li.mergeStatus != 2:
                li.mergeStatus = 2
                ret.append([li])
        return ret, char_count, char_old

    async def get_photo_data(self, assets: "AssetsService") -> Dict:
        """获取数据"""
        now = self.get_now_hour()
        list_data, time_map = await self.req_cal_data()
        (
            date_list,
            start_time,
            end_time,
            total_range,
            now_left,
        ) = await self.get_date_list()
        birthday_char_line, birthday_chars = await self.get_birthday_char(date_list, assets)
        target: List[FinalAct] = []
        abyss: List[FinalAct] = []

        for ds in list_data[1]:
            if act := await self.get_list(ds, start_time, end_time, total_range, time_map, True, assets):
                target.append(act)
        for ds in list_data[0]:
            if act := await self.get_list(ds, start_time, end_time, total_range, time_map, False, assets):
                target.append(act)
        abyss_cal = self.get_abyss_cal(start_time, end_time)
        for t in abyss_cal:
            ds = ActDetail(
                title=f"「深境螺旋」· {t[2]}",
                start_time=t[0].strftime("%Y-%m-%d %H:%M:%S"),
                end_time=t[1].strftime("%Y-%m-%d %H:%M:%S"),
            )
            if act := await self.get_list(ds, start_time, end_time, total_range, {}, True, assets):
                abyss.append(act)
        target.sort(key=lambda x: (x.sort, x.start, x.duration))
        target, char_count, char_old = self.merge_list(target)
        return {
            "date_list": date_list,
            "now_left": now_left,
            "list": target,
            "abyss": abyss,
            "char_mode": f"char-{char_count}-{char_old}",
            "now_time": now.strftime("%Y-%m-%d %H 时"),
            "birthday_char_line": birthday_char_line,
            "birthday_chars": birthday_chars,
        }