# cron "0 9 2 * * * # const $ = new Env('恩山签到') # -*- coding: utf-8 -*- import os import re import time import random import requests from DrissionPage import ChromiumPage, ChromiumOptions USER_AGENT = "Mozilla/5.0 (Linux; Android 13; SM-G981B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Mobile Safari/537.36" CACHE_FILE = "enshan_users.txt" # Cookie 缓存文件 def random_wait(): delay = random.randint(0, 200) print(f"🎲 随机延迟启动: 将在 {delay} 秒后开始执行任务...") time.sleep(delay) print("⏰ 倒计时结束，任务开始！") def force_kill_chrome(): print("🧹 正在清理残留的浏览器进程...") try: os.system("pkill -f chromium") os.system("pkill -f chrome") time.sleep(2) except: pass def push_pushplus(token, content): if not token: return url = "https://www.pushplus.plus/send" data = {"token": token, "title": "恩山签到结果", "content": content} try: requests.post(url, json=data) print("📨 PushPlus 通知已发送") except Exception as e: print(f"❌ 推送失败: {e}") def get_cookies_safe(page): try: ret = page.run_cdp('Network.getCookies') cookies_list = ret.get('cookies', []) return "; ".join([f"{item['name']}={item['value']}" for item in cookies_list]) except Exception as e: print(f"❌ 获取 Cookie 异常: {e}") return "" def extract_regex(pattern, text, default="0"): try: match = re.search(pattern, text) return match.group(1).strip() if match else default except: return default def load_cache(): """从本地缓存文件读取 {uid: cookie} 字典""" cache = {} if not os.path.exists(CACHE_FILE): return cache with open(CACHE_FILE, 'r', encoding='utf-8') as f: for line in f: line = line.strip() if not line or line.startswith('#'): continue if '#' in line: parts = line.split('#', 1) uid = parts[0].strip() cookie = parts[1].strip() if uid and cookie: cache[uid] = cookie return cache def save_cache(cache): """将 {uid: cookie} 字典写入缓存文件""" with open(CACHE_FILE, 'w', encoding='utf-8') as f: for uid, cookie in cache.items(): f.write(f"{uid}#{cookie}\n") print(f"💾 Cookie 缓存已更新至 {CACHE_FILE}") def parse_env_users(): """从环境变量 ENS_CK 解析用户列表，返回 {uid: cookie} 字典""" ens_ck = os.environ.get('ENS_CK') if not ens_ck: return {} users = {} for line in ens_ck.strip().splitlines(): line = line.strip() if not line or line.startswith('#'): continue if '#' in line: parts = line.split('#', 1) uid = parts[0].strip() cookie = parts[1].strip() if uid and cookie: users[uid] = cookie return users def sign_for_user(uid, cookie, push_token): """执行单用户签到，返回 (成功否, 消息, 新cookie)""" print(f"\n========== 开始处理用户 UID: {uid} ==========") co = ChromiumOptions() co.set_argument('--no-sandbox') co.set_argument('--disable-gpu') co.set_argument('--disable-dev-shm-usage') co.set_argument('--headless=new') co.set_argument('--window-size=375,812') if os.path.exists("/usr/bin/chromium-browser"): co.set_paths(browser_path="/usr/bin/chromium-browser") elif os.path.exists("/usr/bin/chromium"): co.set_paths(browser_path="/usr/bin/chromium") co.set_user_agent(user_agent=USER_AGENT) page = None for attempt in range(2): try: force_kill_chrome() page = ChromiumPage(co) break except Exception as e: print(f"⚠️ 浏览器启动失败 (第 {attempt+1} 次尝试): {e}") time.sleep(3) if not page: error_msg = "浏览器连续启动失败" print(f"❌ {error_msg}") push_pushplus(push_token, f"用户 {uid} 签到失败：{error_msg}") return False, error_msg, cookie try: print("1. 访问主页确立作用域...") page.get('https://www.right.com.cn/forum/forum.php?mobile=2') try: page.set.cookies(cookie) except: pass print("2. 刷新页面并过盾...") page.refresh() time.sleep(5) title = page.title if "安全" in title or "验证" in title: print("🛡️ 检测到防火墙拦截，正在等待自动跳转...") time.sleep(15) print("3. 正在获取签到信息...") check_url = "https://www.right.com.cn/forum/erling_qd-sign_in_m.html" page.get(check_url) time.sleep(3) is_signed = False html = page.html formhash = extract_regex(r"var FORMHASH = '([0-9a-zA-Z]+)'", html, "") if not formhash: formhash = extract_regex(r'name="formhash" value="([0-9a-zA-Z]+)"', html, "") if not formhash: formhash = extract_regex(r'formhash=([0-9a-zA-Z]+)', html, "") if not formhash: try: body_text = page.ele('tag:body').text if "登录" in body_text or "Login" in body_text: error_msg = "Cookie 已失效，变为游客状态" print(f"❌ {error_msg}") push_pushplus(push_token, f"用户 {uid} 签到失败：{error_msg}，请更新环境变量 ENS_CK 中的 Cookie。") return False, error_msg, cookie except: pass try: body_text = page.ele('tag:body').text if "连续签到" in body_text and "立即签到" not in body_text: is_signed = True print("ℹ️ 状态: 今天已经签到过了。") except: pass if not formhash and not is_signed: error_msg = "无法提取 formhash (可能页面结构改变)" print(f"❌ {error_msg}") push_pushplus(push_token, f"用户 {uid} 签到失败：{error_msg}") return False, error_msg, cookie if formhash: print(f"🔑 获取 Formhash 成功: {formhash}") sign_success = False sign_msg = "已签到" if not is_signed: sign_api = "https://www.right.com.cn/forum/plugin.php?id=erling_qd:action&action=sign" print("🚀 正在发送签到请求...") js_code = f""" return fetch("{sign_api}", {{ method: "POST", headers: {{ "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", "X-Requested-With": "XMLHttpRequest" }}, body: "formhash={formhash}" }}).then(response => response.json()); """ try: result = page.run_js(js_code) print(f"📥 签到接口返回: {result}") if result and (result.get('success') or "已经签到" in str(result)): sign_success = True sign_msg = result.get('message', '签到成功') else: sign_msg = result.get('message', '未知错误') if result else "接口无响应" except Exception as js_err: print(f"❌ JS 执行异常: {js_err}") sign_success = False sign_msg = "JS执行失败或WAF拦截" else: sign_success = True if sign_success: print("4. 正在获取最终积分数据...") page.get(check_url) time.sleep(2) sign_html = page.html today_points = extract_regex(r'erqd-current-point[^>]*>(\d+)', sign_html, "未知") if today_points == "未知": today_points = extract_regex(r'今日积分.*?(\d+)', sign_html, "未知") continuous_days = extract_regex(r'erqd-continuous-days[^>]*>(\d+)', sign_html, "未知") if continuous_days == "未知": continuous_days = extract_regex(r'连续签到.*?(\d+)', sign_html, "未知") total_days = extract_regex(r'erqd-total-days[^>]*>(\d+)', sign_html, "未知") if total_days == "未知": total_days = extract_regex(r'总签到天数.*?(\d+)', sign_html, "未知") print("🔄 正在刷新积分缓存...") credit_log_url = "https://www.right.com.cn/forum/home.php?mod=spacecp&ac=credit&op=log&mobile=2" page.get(credit_log_url) time.sleep(2) profile_url = f"https://www.right.com.cn/forum/home.php?mod=space&uid={uid}&do=profile&mycenter=1&mobile=2" print(f"📥 正在抓取个人资料页 (UID: {uid})...") page.get(profile_url) total_points = "未知" contribution = "未知" enshan_coin = "未知" try: time.sleep(5) all_lis = page.eles('tag:li') for li in all_lis: clean_text = li.text.replace(" ", "").replace("\n", "").replace("\r", "") if not clean_text: continue if ("积分" in clean_text and "今日" not in clean_text) or "Points" in clean_text: match_cn = re.search(r'(\d+)积分', clean_text) match_en = re.search(r'(\d+)Points', clean_text) if match_cn: total_points = match_cn.group(1) elif match_en: total_points = match_en.group(1) if "贡献" in clean_text or "Contributions" in clean_text: match_cn = re.search(r'(\d+)分贡献', clean_text) match_en = re.search(r'(\d+)pointsContributions', clean_text) if match_cn: contribution = match_cn.group(1) elif match_en: contribution = match_en.group(1) if "恩山币" in clean_text or "EnshanCoin" in clean_text: match_cn = re.search(r'(\d+)币恩山币', clean_text) match_en = re.search(r'(\d+)coinsEnshanCoin', clean_text) if match_cn: enshan_coin = match_cn.group(1) elif match_en: enshan_coin = match_en.group(1) print(f"📊 抓取结果: 积分={total_points}, 贡献={contribution}, 币={enshan_coin}") except Exception as e: print(f"❌ 数据解析异常: {e}") new_cookie = get_cookies_safe(page) if not new_cookie: new_cookie = cookie notify_content = ( f"✅ 用户 {uid} 签到成功！🎊
" f"📊 积分统计如下：
" f"===========
" f"今日积分：{today_points}
" f"连续签到：{continuous_days} 天
" f"总签到天数：{total_days} 天
" f"总积分：{total_points}
" f"贡献分：{contribution} 分
" f"恩山币：{enshan_coin} 币" ) print("=== 推送内容预览 ===") print(notify_content.replace("
", "\n")) push_pushplus(push_token, notify_content) return True, "签到成功", new_cookie else: error_msg = f"签到失败：{sign_msg}" print(f"❌ {error_msg}") push_pushplus(push_token, f"用户 {uid} 签到失败：{sign_msg}") return False, error_msg, cookie except Exception as e: import traceback traceback.print_exc() error_msg = f"运行异常: {str(e)}" push_pushplus(push_token, f"用户 {uid} 签到异常：{error_msg}") return False, error_msg, cookie finally: try: if page: page.quit() except: pass force_kill_chrome() def main(): random_wait() # 1. 从环境变量读取账号列表（权威来源） env_users = parse_env_users() if not env_users: print("❌ 环境变量 ENS_CK 未设置或为空，无法继续。") return # 2. 加载本地 Cookie 缓存 cache = load_cache() # 3. 合并：优先使用缓存，若缓存不存在或环境变量中的 Cookie 与缓存不同，则使用环境变量（视为手动更新） # 我们以 env_users 为基准，遍历每个 UID，决定使用的 Cookie final_users = {} # {uid: cookie_to_use} for uid, env_cookie in env_users.items(): if uid in cache: cached_cookie = cache[uid] # 如果环境变量中的 cookie 与缓存不同，则使用环境变量（用户可能手动更新） if env_cookie != cached_cookie: print(f"ℹ️ 用户 {uid} 的环境变量 Cookie 与缓存不同，将使用环境变量（视为手动更新）。") final_users[uid] = env_cookie else: final_users[uid] = cached_cookie else: final_users[uid] = env_cookie # 新用户，首次使用环境变量 print(f"📋 共加载 {len(final_users)} 个用户，开始逐个签到...") push_token = os.environ.get('PUSHPLUS_TOKEN', '') new_cache = {} # 用于存储处理后的最新 Cookie for idx, (uid, cookie) in enumerate(final_users.items(), 1): print(f"\n>>> 正在处理第 {idx}/{len(final_users)} 个用户 (UID: {uid})") ok, msg, new_cookie = sign_for_user(uid, cookie, push_token) # 无论签到是否成功，只要有新 cookie（可能是旧 cookie 或更新后的），都保存 # 如果签到成功，new_cookie 是新的；如果失败，new_cookie 是原 cookie（或原样） new_cache[uid] = new_cookie if idx < len(final_users): time.sleep(5) # 4. 将更新后的所有 Cookie 写回缓存文件 save_cache(new_cache) # 汇总信息 success_count = sum(1 for u in final_users if u in new_cache and new_cache[u] != '') print(f"\n📊 完成，共 {len(final_users)} 个用户，成功 {success_count} 个。") if __name__ == "__main__": main()