scf.py

import httpx
import json
import os
import re
import sentry_sdk
import random
import time
import yaml
import logging

# 配置 Sentry
sentry_sdk.init(
    "https://425d7b4536f94c9fa540fe34dd6609a2@o361988.ingest.sentry.io/6352584",
    traces_sample_rate=1.0,
)

# 配置日志级别
loglevel_env = os.environ.get("MHYY_LOGLEVEL", "INFO").upper()
loglevel = getattr(logging, loglevel_env, logging.INFO)

# 配置日志
logging.basicConfig(
    level=loglevel,
    format="%(asctime)s [%(levelname)s]: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

logger = logging.getLogger()

# 配置数据写入脚本
config_datas = """
accounts:
  - token: your_token_here
    type: 2
    sysver: your_sysver
    deviceid: your_deviceid
    devicename: your_devicename
    devicemodel: your_devicemodel
    appid: 1953439974
    region: cn
"""

class RunError(Exception):
    pass

def handler(*args):
    """
    云函数入口函数。
    """
    # 读取配置
    conf_data = yaml.load(config_datas, Loader=yaml.FullLoader)
    if not conf_data or "accounts" not in conf_data:
        logger.error("请正确配置账户信息后再运行本脚本！")
        return {"statusCode": 1, "message": "配置错误，请检查账户信息。"}

    conf = conf_data["accounts"]
    if not conf:
        logger.error("账户配置为空！")
        return {"statusCode": 1, "message": "账户配置为空，请添加账户信息。"}

    logger.info(f"检测到 {len(conf)} 个账号，正在进行任务……")

    # 获取 SCT 通知配置
    sct_status = os.environ.get("sct")  # https://sct.ftqq.com/
    sct_key = os.environ.get("sct_key")
    sct_url = f"https://sctapi.ftqq.com/{sct_key}.send?title=MHYY-AutoCheckin 自动推送" if sct_key else None

    sct_msg = ""

    try:
        # 随机等待时间以避免被封禁
        debug_mode = os.environ.get("MHYY_DEBUG", "False").upper() == "TRUE"
        if not debug_mode:
            wait_time = random.randint(10, 60)
            logger.info(f"为了避免同一时间签到人数太多导致被官方怀疑，开始休眠 {wait_time} 秒")
            time.sleep(wait_time)

        # 获取最新版本号
        try:
            ver_info = httpx.get(
                "https://hyp-api.mihoyo.com/hyp/hyp-connect/api/getGameBranches?game_ids[]=1Z8W5NHUQb&launcher_id=jGHBHlcOq1",
                timeout=60,
                verify=False,
            ).text
            version = json.loads(ver_info)["data"]["game_branches"][0]["main"]["tag"]
            logger.info(f"从官方API获取到云·原神最新版本号：{version}")
        except Exception as e:
            version = "5.0.0"
            logger.warning(f"获取版本号失败，使用默认版本：{version}")

        # 遍历每个账户进行任务
        for idx, config in enumerate(conf, start=1):
            if not config:
                raise RunError("账户配置为空，请添加账户信息。")

            # 提取账户配置
            token = config.get("token")
            client_type = config.get("type")
            sysver = config.get("sysver")
            deviceid = config.get("deviceid")
            devicename = config.get("devicename")
            devicemodel = config.get("devicemodel")
            appid = config.get("appid")
            region = config.get("region", "cn")

            if not all([token, client_type, sysver, deviceid, devicename, devicemodel, appid]):
                logger.error(f"第 {idx} 个账户配置不完整，请检查配置。")
                sct_msg += f"第 {idx} 个账户配置不完整，请检查配置。\n"
                continue

            # 构建请求头
            headers = {
                "x-rpc-combo_token": token,
                "x-rpc-client_type": str(client_type),
                "x-rpc-app_version": str(version),
                "x-rpc-sys_version": str(sysver),
                "x-rpc-channel": "cyydmihoyo",
                "x-rpc-device_id": deviceid,
                "x-rpc-device_name": devicename,
                "x-rpc-device_model": devicemodel,
                "x-rpc-vendor_id": "1",
                "x-rpc-cg_game_biz": "hk4e_cn",
                "x-rpc-op_biz": "clgm_cn",
                "x-rpc-language": "zh-cn",
                "Host": "api-cloudgame.mihoyo.com",
                "Connection": "Keep-Alive",
                "Accept-Encoding": "gzip",
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36 Edg/121.0.0.0",
            }

            # 处理国际服
            if region == "os":
                headers.update({
                    "x-rpc-channel": "mihoyo",
                    "x-rpc-cg_game_biz": "hk4e_global",
                    "x-rpc-op_biz": "clgm_global",
                    "x-rpc-cg_game_id": "9000254",
                    "x-rpc-app_id": "600493",
                    "User-Agent": "okhttp/4.10.0",
                    "Host": "sg-cg-api.hoyoverse.com",
                })
                NotificationURL = "https://sg-cg-api.hoyoverse.com/hk4e_global/cg/gamer/api/listNotifications?status=NotificationStatusUnread&type=NotificationTypePopup&is_sort=true"
                WalletURL = "https://sg-cg-api.hoyoverse.com/hk4e_global/cg/wallet/wallet/get"
                AnnouncementURL = "https://sg-cg-api.hoyoverse.com/hk4e_global/cg/gamer/api/getAnnouncementInfo"
            else:
                NotificationURL = "https://api-cloudgame.mihoyo.com/hk4e_cg_cn/gamer/api/listNotifications?status=NotificationStatusUnread&type=NotificationTypePopup&is_sort=true"
                WalletURL = "https://api-cloudgame.mihoyo.com/hk4e_cg_cn/wallet/wallet/get"
                AnnouncementURL = "https://api-cloudgame.mihoyo.com/hk4e_cg_cn/gamer/api/getAnnouncementInfo"

            logger.info(f"正在进行第 {idx} 个账号，服务器为{'GLOBAL' if region == 'os' else 'CN'}……")

            try:
                # 获取钱包信息
                wallet_response = httpx.get(WalletURL, headers=headers, timeout=60, verify=False)
                wallet_data = wallet_response.json()
                logger.debug(wallet_data)

                if wallet_data == {
                    "data": None,
                    "message": "登录已失效，请重新登录",
                    "retcode": -100,
                }:
                    logger.error(f"账号 {idx} 登录已过期，请重新登陆！返回为：{wallet_response.text}")
                    sct_msg += f"账号 {idx} 登录已过期，请重新登陆！返回为：{wallet_response.text}\n"
                    continue
                else:
                    free_time = wallet_data['data']['free_time']['free_time']
                    play_card_status = wallet_data['data']['play_card']['short_msg']
                    coin_num = wallet_data['data']['coin']['coin_num']
                    logger.info(
                        f"账号 {idx}: 你当前拥有免费时长 {free_time} 分钟，畅玩卡状态为 {play_card_status}，拥有原点 {coin_num} 点（{int(coin_num)/10}分钟）"
                    )
                    sct_msg += (
                        f"账号 {idx}: 你当前拥有免费时长 {free_time} 分钟，"
                        f"畅玩卡状态为 {play_card_status}，拥有原点 {coin_num} 点（{int(coin_num)/10}分钟）\n"
                    )

                # 获取公告信息
                announcement_response = httpx.get(AnnouncementURL, headers=headers, timeout=60, verify=False)
                announcement_data = announcement_response.json()
                logger.debug(f'获取到公告列表：{announcement_data["data"]}')

                # 获取签到通知
                notification_response = httpx.get(NotificationURL, headers=headers, timeout=60, verify=False)
                notification_data = notification_response.json()
                logger.debug(notification_data)

                # 解析签到状态
                success, Signed, Over = False, False, False
                try:
                    notifications = notification_data["data"]["list"]
                    if not notifications:
                        success = True
                        Signed = True
                        Over = False
                    else:
                        last_msg = json.loads(notifications[-1]["msg"])
                        if last_msg.get("msg") == "每日登录奖励":
                            success = True
                            Signed = False
                            Over = False
                        elif last_msg.get("over_num", 0) > 0:
                            success = True
                            Signed = False
                            Over = True
                        else:
                            success = False
                except (IndexError, json.JSONDecodeError, KeyError) as e:
                    logger.warning(f"解析签到状态时出错: {e}")
                    success = False

                if success:
                    if Signed:
                        logger.info(f"账号 {idx}: 获取签到情况成功！今天是否已经签到过了呢？")
                        sct_msg += f"账号 {idx}: 获取签到情况成功！今天是否已经签到过了呢？\n"
                        logger.debug(f"完整返回体为：{notification_response.text}")
                    elif not Signed and Over:
                        last_msg = json.loads(notifications[-1]["msg"])
                        logger.info(
                            f"账号 {idx}: 获取签到情况成功！当前免费时长已经达到上限！签到情况为{last_msg}"
                        )
                        sct_msg += (
                            f"账号 {idx}: 获取签到情况成功！当前免费时长已经达到上限！签到情况为{last_msg}\n"
                        )
                    else:
                        logger.info(f"账号 {idx}: 已经签到过了！")
                        sct_msg += f"账号 {idx}: 已经签到过了！\n"
                else:
                    logger.info(f"账号 {idx}: 当前没有签到！请稍后再试！")
                    sct_msg += f"账号 {idx}: 当前没有签到！请稍后再试！\n"

                # 发送 SCT 通知
                if sct_url:
                    try:
                        sct_response = httpx.get(sct_url, params={"desp": sct_msg}, timeout=30)
                        if sct_response.status_code == 200:
                            logger.info("SCT 推送完成！")
                        else:
                            logger.warning(f"SCT 无法推送，状态码：{sct_response.status_code}, 响应：{sct_response.text}")
                    except Exception as e:
                        logger.error(f"SCT 推送时出错：{e}")

            except Exception as e:
                logger.error(f"账号 {idx} 执行过程中出错：{str(e)}")
                sct_msg += f"账号 {idx} 执行过程中出错：{str(e)}\n"
                if sct_url:
                    try:
                        httpx.get(sct_url, params={"desp": sct_msg}, timeout=30)
                    except Exception as notify_error:
                        logger.error(f"SCT 推送时出错：{notify_error}")
                continue

        logger.info("所有任务已经执行完毕！")
        return {"statusCode": 0, "message": "所有任务已经执行完毕！", "details": sct_msg}

    except RunError as re:
        logger.error(f"运行错误：{str(re)}")
        if sct_url:
            try:
                httpx.get(sct_url, params={"desp": f"运行错误：{str(re)}"}, timeout=30)
            except Exception as notify_error:
                logger.error(f"SCT 推送时出错：{notify_error}")
        return {"statusCode": 1, "message": f"运行错误：{str(re)}"}

    except Exception as e:
        logger.error(f"未知错误：{str(e)}")
        if sct_url:
            try:
                httpx.get(sct_url, params={"desp": f"未知错误：{str(e)}"}, timeout=30)
            except Exception as notify_error:
                logger.error(f"SCT 推送时出错：{notify_error}")
        return {"statusCode": 1, "message": f"未知错误：{str(e)}"}