1.购买服务器阿里云:服务器购买地址https://t.aliyun.com/U/Bg6shY若失效,可用地址
阿里云:
服务器购买地址
https://t.aliyun.com/U/Bg6shY若失效,可用地址
https://www.aliyun.com/daily-act/ecs/activity_selection?source=5176.29345612&userCode=49hts92d腾讯云:
https://curl.qcloud.com/wJpWmSfU若失效,可用地址
https://cloud.tencent.com/act/cps/redirect?redirect=2446&cps_key=ad201ee2ef3b771157f72ee5464b1fea&from=console华为云
https://activity.huaweicloud.com/cps.html?fromacct=64b5cf7cc11b4840bb4ed2ea0b2f4468&utm_source=V1g3MDY4NTY=&utm_medium=cps&utm_campaign=2019052.部署教程
3.代码如下
#!/usr/bin/env python3# -*- coding: utf-8 -*-import osimport sysimport timeimport socketimport jsonfrom datetime import datetimeimport requestsfrom requests.adapters import HTTPAdapterfrom urllib3.util.retry import Retry# ========== 配置区 ==========class Config:# 读取信息USERNAME = os.getenv('HXSY_USERNAME')PASSWORD = os.getenv('HXSY_PASSWORD')LOGIN_URL = 'https://www.huaxiashuyu.com/wp-admin/admin-ajax.php'USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'CONNECT_TIMEOUT = 10READ_TIMEOUT = 30MAX_RETRIES = 3PROXY = os.getenv('PROXY')DEBUG = os.getenv('DEBUG', 'false').lower() == 'true'IS_ACTIONS = os.getenv('ACTIONS') == 'true'STATUS_FILE = "status/status_huaxia.json"# ========== 创建会话 ==========def create_session():session = requests.Session()retry_strategy = Retry(total=3,backoff_factor=1,status_forcelist=[429, 500, 502, 503, 504],allowed_methods=["HEAD", "GET", "POST"])adapter = HTTPAdapter(max_retries=retry_strategy)session.mount("http://", adapter)session.mount("https://", adapter)if Config.PROXY:session.proxies = {'http': Config.PROXY, 'https': Config.PROXY}print(f"使用: {Config.PROXY}")return sessionsession = create_session()# ========== 工具函数 ==========def notify(title: str, content: str):timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')message = f"[{timestamp}] {title}\n{content}"print('\n' + '=' * 50)print(message)print('=' * 50 + '\n')# Actions 支持if Config.IS_ACTIONS:summary_file = os.getenv('STEP_SUMMARY')if summary_file:emoji = '' if '成功' in content else '' if '失败' in content else 'ℹ️'with open(summary_file, 'a', encoding='utf-8') as f:f.write(f"## {emoji} {title}\n\n")f.write(f"{content}\n\n")f.write(f"**执行时间**: {timestamp}\n\n")def validate_config():if not Config.USERNAME or not Config.PASSWORD:notify(' 配置错误', '未设置账号或密码!请检查Secrets 配置')sys.exit(1)# 脱敏显示用户名masked_username = Config.USERNAME[:3] + '***' + Config.USERNAME[-3:] if len(Config.USERNAME) > 6 else '***'print(f" 使用账号: {masked_username}")def check_network():try:host = 'www.huaxiashuyu.com'print(f" DNS解析: {host}")ip = socket.gethostbyname(host)print(f" 解析成功: {ip}")sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)sock.settimeout(5)result = sock.connect_ex((host, 443))sock.close()if result == 0:print(f" TCP连接成功\n")return Trueelse:print(f" TCP连接失败\n")return Falseexcept Exception as e:print(f" 网络检测失败: {e}\n")return False# ========== 智能响应判断 ==========def is_response_success(response_data: dict, response_text: str = '') -> bool:if not isinstance(response_data, dict):return Falsestatus = response_data.get('status')if status in [1, '1', 200, '200', 'success']:return Truesuccess = response_data.get('success')if success in [True, 'true', 1, '1']:return Truecode = response_data.get('code')if code in [0, '0', 200, '200']:return Truemsg = response_data.get('msg', '').lower()success_keywords = ['成功', 'success', 'ok', '已签到']if any(keyword in msg for keyword in success_keywords):fail_keywords = ['失败', 'fail', 'error', '错误']if not any(keyword in msg for keyword in fail_keywords):return Trueif '成功' in response_text and '失败' not in response_text:return Truereturn False# ========== 登录函数 ==========def login() -> str:login_data = {'action': 'user_login','username': Config.USERNAME,'password': Config.PASSWORD}headers = {'User-Agent': Config.USER_AGENT,'Content-Type': 'application/x-www-form-urlencoded','Accept': 'application/json, text/javascript, */*; q=0.01','X-Requested-With': 'XMLHttpRequest'}print(f" 正在连接登录接口...")try:response = session.post(Config.LOGIN_URL,data=login_data,headers=headers,timeout=(Config.CONNECT_TIMEOUT, Config.READ_TIMEOUT),verify=True)# 仅在调试模式或本地环境显示详细信息if Config.DEBUG and not Config.IS_ACTIONS:print(f"\n{'='*60}")print(f" HTTP状态码: {response.status_code}")print(f" 响应内容: {response.text[:300]}")print(f" Cookies: {response.cookies.get_dict()}")print(f"{'='*60}\n")try:result = response.json()if Config.DEBUG and not Config.IS_ACTIONS:print(f" 解析后的JSON: {json.dumps(result, ensure_ascii=False, indent=2)}")except json.JSONDecodeError:print(f" JSON解析失败")notify(' 登录错误', '服务器返回格式异常')return Noneif is_response_success(result, response.text):msg = result.get('msg', '登录成功')notify(' 登录成功', msg)cookies = response.cookies.get_dict()if cookies:cookie_str = '; '.join([f"{k}={v}" for k, v in cookies.items()])print(f" 获取到 Cookie ({len(cookie_str)} 字符)")return cookie_strelse:set_cookie = response.headers.get('Set-Cookie', '')if set_cookie:cookie_str = '; '.join([c.split(';')[0] for c in set_cookie.split(', ')])print(f" 从响应头提取 Cookie ({len(cookie_str)} 字符)")return cookie_strelse:return "no-cookie-needed"else:msg = result.get('msg', '未知错误')notify(' 登录失败', msg)return Noneexcept requests.exceptions.Timeout:notify(' 请求超时', '连接超时,请检查网络')return Noneexcept Exception as e:notify(' 登录异常', f'{type(e).__name__}: {str(e)}')return None# ========== 签到函数 ==========def sign_in(cookie: str):headers = {'User-Agent': Config.USER_AGENT,'Content-Type': 'application/x-www-form-urlencoded','Accept': 'application/json, text/javascript, */*; q=0.01','X-Requested-With': 'XMLHttpRequest'}if cookie and cookie != "no-cookie-needed":headers['Cookie'] = cookiedata = {'action': 'user_qiandao'}print(f" 正在执行签到操作...")try:response = session.post(Config.LOGIN_URL,data=data,headers=headers,timeout=(Config.CONNECT_TIMEOUT, Config.READ_TIMEOUT),verify=True)if Config.DEBUG and not Config.IS_ACTIONS:print(f"\n{'='*60}")print(f" 签到响应状态码: {response.status_code}")print(f" 签到响应内容: {response.text[:300]}")print(f"{'='*60}\n")try:result = response.json()if Config.DEBUG and not Config.IS_ACTIONS:print(f"🔍 签到结果: {json.dumps(result, ensure_ascii=False, indent=2)}")except json.JSONDecodeError:print(f" JSON解析失败")notify(' 签到错误', '服务器返回格式异常')returnif is_response_success(result, response.text):msg = result.get('msg', '签到成功')notify(' 签到成功', msg)else:msg = result.get('msg', '签到失败')if '已签' in msg or '重复' in msg:notify(' 今日已签到', msg)else:notify(' 签到异常', msg)except Exception as e:notify('💥 签到异常', f'{type(e).__name__}: {str(e)}')# ========== 主函数 ==========def main():print('\n 花夏数娱自动签到脚本启动...\n')print(f" 执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")print(f" 运行环境: {'Actions' if Config.IS_ACTIONS else '本地环境'}\n")if not check_network():notify(' 网络异常', '无法连接到目标服务器')sys.exit(1)validate_config()try:cookie = login()if cookie:time.sleep(2)sign_in(cookie)print("\n 所有任务执行完成")else:print("\n 登录失败,终止执行")sys.exit(1)except KeyboardInterrupt:print("\n\n 用户中断执行")sys.exit(0)except Exception as e:notify(' 脚本异常', f'{type(e).__name__}: {str(e)}')import tracebackprint(traceback.format_exc())sys.exit(1)if __name__ == '__main__':main()
该脚本为花夏数娱自动签到脚本。
主要作用
在本地,使用账号密码自动登录"花夏数娱"(WordPress 站点),完成每日签到,并把结果输出到控制台。同时内置网络连通性自检、请求重试、"智能判定成功/失败"的容错逻辑。
关键方法
class Config读取环境变量
HXSY_USERNAME/HXSY_PASSWORD、代理、调试开关、是否处于Actions 等;定义接口地址、UA、超时、重试次数等全局配置。
create_session()创建全局
requests.Session;配置
Retry(429/5xx 自动重试、指数退避);如设置了
PROXY,为会话挂代理。notify(title, content)统一的消息输出:控制台打印;
若Actions 中,将结果追加到
${STEP_SUMMARY},并带上表情与时间戳。validate_config()校验账号/密码是否存在;
脱敏打印用户名,防止日志泄露隐私。
check_network()基础连通性检测:DNS 解析
www.huaxiashuyu.com、TCP 443 端口连通测试;失败则提前退出,避免无意义的登录/签到请求。
is_response_success(response_data, response_text='')智能识别成功:兼容不同返回格式字段(
status/success/code/msg等);关键字包含"成功/已签到/success/ok"且未出现失败类关键字时也视为成功;
提高对不稳定/多样化接口返回的容错率。
login() -> str | None调
admin-ajax.php,action=user_login进行登录;解析并判定登录是否成功(用
is_response_success);成功:从
response.cookies或Set-Cookie里提取并组装Cookie字符串返回(若无需 Cookie 返回"no-cookie-needed");失败/异常:做分类通知(超时、JSON 解析失败、服务器格式异常等)。
sign_in(cookie: str)调
admin-ajax.php,action=user_qiandao执行签到;使用登录得到的 Cookie(如需要);
结合
is_response_success判断:成功 / 今日已签 / 异常,分别通知。main()打印启动信息(环境、时间);
先跑
check_network(),再validate_config();login()成功后sign_in();失败则退出;统一异常捕获与可读的报错/堆栈输出。
注意:
本文部分变量已做脱敏处理,仅用于测试和学习研究,禁止用于商业用途,不能保证其合法性,准确性,完整性和有效性,请根据情况自行判断。技术层面需要提供帮助,可以通过打赏的方式进行探讨。
没有评论:
发表评论