import httpx import json import os import re import sentry_sdk import random import time import yaml import logging if os.environ.get("ZZZCLOUD_LOGLEVEL", "").upper() == "DEBUG": loglevel = logging.DEBUG elif os.environ.get("ZZZCLOUD_LOGLEVEL", "").upper() == "WARNING": loglevel = logging.WARNING elif os.environ.get("ZZZCLOUD_LOGLEVEL", "").upper() == "ERROR": loglevel = logging.ERROR else: loglevel = logging.INFO # 设置日志配置 logging.basicConfig( level=loglevel, format="%(asctime)s [%(levelname)s]: %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logger = logging.getLogger() def ReadConf(variable_name, default_value=None): # Try to get the variable from the environment env_value = os.environ.get(variable_name) if env_value: config_data = yaml.load(env_value, Loader=yaml.FullLoader) return config_data # If not found in environment, try to read from config.yml try: with open("config.yml", "r", encoding="utf-8") as config_file: config_data = yaml.load(config_file, Loader=yaml.FullLoader) return config_data except FileNotFoundError: return default_value sentry_sdk.init( "https://379c1d210fc6ea113172c5444297a646@o361988.ingest.us.sentry.io/4508426724966400", traces_sample_rate=1.0, ) conf = ReadConf("ZZZCLOUD_CONFIG")["accounts"] if not conf: logger.error("请正确配置环境变量或者config.yml后再运行本脚本！") os._exit(0) logger.info(f"检测到 {len(conf)} 个账号，正在进行任务……") # Options 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=ZZZCloud-AutoCheckin 自动推送" sct_msg = "" class RunError(Exception): pass if __name__ == "__main__": if not os.environ.get("ZZZCLOUD_DEBUG", False): wait_time = random.randint(10, 60) # Random Sleep to Avoid Ban 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[]=x6znKlJ0xK&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 = "1.3.0" logger.warning(f"获取版本号失败，使用默认版本：{version}") for config in conf: # 各种API的URL LoginURL = "https://cg-nap-api.mihoyo.com/nap_cn/cg/gamer/api/login" NotificationURL = "https://cg-nap-api.mihoyo.com/nap_cn/cg/gamer/api/listNotifications?status=NotificationStatusUnread&type=NotificationTypePopup&is_sort=true" WalletURL = "https://cg-nap-api.mihoyo.com/nap_cn/cg/wallet/wallet/get" AnnouncementURL = ( "https://cg-nap-api.mihoyo.com/nap_cn/cg/gamer/api/getAnnouncementInfo" ) if config == "": # Verify config raise RunError( f"请在Settings->Secrets->Actions页面中新建名为config的变量，并将你的配置填入后再运行！" ) else: token = config["token"] client_type = config["type"] sysver = config["sysver"] deviceid = config["deviceid"] devicename = config["devicename"] devicemodel = config["devicemodel"] appid = config["appid"] 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": "oosnsdyu_C", "x-rpc-device_id": deviceid, "x-rpc-device_name": devicename, "x-rpc-device_model": devicemodel, "x-rpc-vendor_id": "2", "x-rpc-cg_game_biz": "nap_cn", "x-rpc-op_biz": "clgm_nap-cn", "x-rpc-language": "zh-cn", "Host": "cg-nap-api.mihoyo.com", "Connection": "Keep-Alive", "Accept-Encoding": "gzip", "User-Agent": "okhttp/4.10.0", } bbsid = re.findall(r"oi=[0-9]+", token)[0].replace("oi=", "") logger.info( f"正在进行第 {conf.index(config) + 1} 个账号……" ) try: wallet = httpx.get(WalletURL, headers=headers, timeout=60, verify=False) logger.debug(wallet.text) if json.loads(wallet.text) == { "data": None, "message": "登录已失效，请重新登录", "retcode": -100, }: logger.error(f"当前登录已过期，请重新登陆！返回为：{wallet.text}") sct_msg += f"当前登录已过期，请重新登陆！返回为：{wallet.text}" else: logger.info( f"你当前拥有免费时长 {json.loads(wallet.text)['data']['free_time']['free_time']} 分钟，畅玩卡状态为 {json.loads(wallet.text)['data']['play_card']['short_msg']}，拥有邦邦点 {json.loads(wallet.text)['data']['coin']['coin_num']} 点（{int(json.loads(wallet.text)['data']['coin']['coin_num'])/10}分钟）" ) sct_msg += f"你当前拥有免费时长 {json.loads(wallet.text)['data']['free_time']['free_time']} 分钟，畅玩卡状态为 {json.loads(wallet.text)['data']['play_card']['short_msg']}，拥有邦邦点 {json.loads(wallet.text)['data']['coin']['coin_num']} 点（{int(json.loads(wallet.text)['data']['coin']['coin_num'])/10}分钟）" announcement = httpx.get( AnnouncementURL, headers=headers, timeout=60, verify=False ) logger.debug(f'获取到公告列表：{json.loads(announcement.text)["data"]}') res = httpx.get( NotificationURL, headers=headers, timeout=60, verify=False ) success, Signed = False, False logger.debug(res.text) try: if list(json.loads(res.text)["data"]["list"]) == []: success = True Signed = True Over = False elif ( json.loads(json.loads(res.text)["data"]["list"][-1]["msg"])[ "msg" ] == "每日登录奖励" ): success = True Signed = False Over = False elif ( json.loads(json.loads(res.text)["data"]["list"][-1]["msg"])[ "over_num" ] > 0 ): success = True Signed = False Over = True else: success = False except IndexError: success = False if success: if Signed: logger.info(f"获取签到情况成功！今天是否已经签到过了呢？") sct_msg += f"获取签到情况成功！今天是否已经签到过了呢？" logger.debug(f"完整返回体为：{res.text}") elif not Signed and Over: logger.info( f'获取签到情况成功！当前免费时长已经达到上限！签到情况为{json.loads(res.text)["data"]["list"][0]["msg"]}' ) sct_msg += f"获取签到情况成功！当前免费时长已经达到上限！签到情况为{json.loads(res.text)['data']['list'][0]['msg']}" else: logger.info(f"已经签到过了！") sct_msg += f"已经签到过了！" else: logger.info(f"当前没有签到！请稍后再试！") sct_msg += f"当前没有签到！请稍后再试！" if sct_key: httpx.get(sct_url, params={"desp": sct_msg}) except Exception as e: logger.error(f"执行过程中出错：{str(e)}") sct_msg += f"执行过程中出错：{str(e)}" if sct_key: httpx.get(sct_url, params={"desp": sct_msg}) raise RunError(str(e)) logger.info("所有任务已经执行完毕！")