import json import time import base64 import re import os import csv import random import string from datetime import datetime from playwright.sync_api import sync_playwright # ===WWW.MAFLYA.COM=== 配置 ===TG几频道@flymaf======= BASE_URL = "https://wula001.com" REGISTER_API = f"{BASE_URL}/api/v1/passport/auth/register" LOGIN_API = f"{BASE_URL}/api/v1/passport/auth/login" SUBSCRIBE_API = f"{BASE_URL}/api/v1/user/getSubscribe" SUBSCRIBE_BASE = "https://windowsv1.com/s/?token=" OUTPUT_DIR = "nodes_wula" CSV_FILE = "accounts.csv" JSON_FILE = "wula_data.json" AUTO_SUMMARY_FILE = "wulaauto24.txt" MANUAL_SUMMARY_FILE = "wulabyhand.txt" os.makedirs(OUTPUT_DIR, exist_ok=True) # 关键：订阅下载必须用代理软件 UA，不能用浏览器 UA PROXY_UA_LIST = [ "clash-verge/v2.0", "ClashMetaForAndroid/2.11", "FlClash/0.8.70", "Shadowrocket/2.1.30", "Stash/2.6.0", "Surge/5.8.0", "Quantumult X/1.0.30", ] ALL_ACCOUNTS = [] ALL_NODES_ACCUMULATOR = [] def log_step(msg): print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True) def random_delay(min_sec=1.0, max_sec=3.0): time.sleep(random.uniform(min_sec, max_sec)) def generate_email(): ts = hex(int(time.time() * 1000))[-6:] prefix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=10)) + ts return f"{prefix}@gmail.com" def generate_password(length=12): if length < 8: length = 8 return ''.join(random.choices(string.ascii_letters + string.digits, k=length)) def browser_post(page, url, data=None, json_data=None, headers=None): if json_data is not None: body = json.dumps(json_data) ct = 'application/json' else: body = '&'.join([f"{k}={v}" for k, v in (data or {}).items()]) ct = 'application/x-www-form-urlencoded' result = page.evaluate(""" async ({url, body, ct, headers}) => { const resp = await fetch(url, { method: 'POST', headers: Object.assign({'Content-Type': ct}, headers || {}), body: body }); const text = await resp.text(); return {status: resp.status, body: text}; } """, {"url": url, "body": body, "ct": ct, "headers": headers or {}}) return result def register_wula(email, password): """用 Playwright 打开首页 -> cap.solve() -> 浏览器内 fetch 注册登录拿订阅 URL""" with sync_playwright() as p: browser = p.chromium.launch(headless=True) context = browser.new_context( user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/153.0.0.0 Safari/537.36", viewport={"width": 1280, "height": 800}, ignore_https_errors=True ) page = context.new_page() log_step("打开乌拉云首页...") page.goto(BASE_URL + "/", wait_until="domcontentloaded", timeout=60000) page.wait_for_timeout(5000) log_step("开始求解验证码...") token = None for attempt in range(1, 4): try: token = page.evaluate(""" async () => { let widget = document.querySelector('cap-widget'); if (widget && typeof widget.solve === 'function') { const r = await widget.solve(); return r && r.token ? r.token : null; } if (typeof Cap === 'function') { const cap = new Cap({ apiEndpoint: 'https://wula001.com/fd0fb2130e/' }); const r = await cap.solve(); return r && r.token ? r.token : null; } return null; } """) if token: break except Exception as e: log_step(f"验证码尝试 {attempt} 失败: {e}") page.wait_for_timeout(3000) if not token: browser.close() raise Exception("获取验证码 token 失败") log_step(f"验证码 token 成功: {token[:30]}...") reg_headers = {"Referer": BASE_URL + "/", "Origin": BASE_URL, "Accept": "application/json, text/plain, */*"} log_step("注册...") reg = browser_post(page, REGISTER_API, data={ "email": email, "password": password, "invite_code": "", "email_code": "", "recaptcha_data": token }, headers=reg_headers) log_step(f"注册响应: {reg['status']}") if reg['status'] != 200: browser.close() raise Exception(f"注册失败: {reg['status']} - {reg['body']}") log_step("登录...") login = browser_post(page, LOGIN_API, json_data={"email": email, "password": password}, headers=reg_headers) if login['status'] != 200: browser.close() raise Exception(f"登录失败: {login['status']} - {login['body']}") login_json = json.loads(login['body']) auth_data = login_json.get("data", {}).get("auth_data") if not auth_data: browser.close() raise Exception(f"登录响应缺少 auth_data: {login_json}") log_step("获取订阅链接...") sub = page.evaluate(""" async ({url, auth}) => { const resp = await fetch(url, { headers: { 'Authorization': auth } }); const text = await resp.text(); return {status: resp.status, body: text}; } """, {"url": SUBSCRIBE_API, "auth": auth_data}) if sub['status'] != 200: browser.close() raise Exception(f"获取订阅失败: {sub['status']}") sub_json = json.loads(sub['body']) subscribe_url = sub_json.get("data", {}).get("subscribe_url") if not subscribe_url: browser.close() raise Exception(f"订阅链接为空: {sub_json}") log_step(f"订阅地址: {subscribe_url}") # **关键：用 page.request 下载订阅（走浏览器网络栈，无跨域限制）** # 但是 page.request 仍会用浏览器 UA，所以我们强制覆盖 User-Agent 为代理软件 UA content = None for ua in PROXY_UA_LIST: try: log_step(f"下载订阅 (UA={ua})...") req = page.request.get(subscribe_url, headers={"User-Agent": ua, "Accept": "*/*"}, timeout=40000) if req.ok: body = req.text() if len(body) > 100 and ('vless://' in body or 'anytls://' in body or 'proxies:' in body): content = body log_step(f"订阅下载成功 (UA={ua}), 长度={len(body)}") break else: log_step(f" UA={ua} 返回内容过短或格式不对，长度={len(body)}") else: log_step(f" UA={ua} 状态={req.status}") except Exception as e: log_step(f" 下载异常: {e}") browser.close() if not content: raise Exception("订阅下载失败（所有代理软件 UA 均失败）") return content, subscribe_url, auth_data, email def parse_nodes_from_yaml(yaml_text): log_step("解析 YAML 节点...") try: import yaml obj = yaml.safe_load(yaml_text) proxies = obj.get("proxies", []) if isinstance(obj, dict) else [] nodes = {"vless": [], "anytls": [], "hysteria2": []} for p in proxies: if not isinstance(p, dict): continue name = str(p.get("name", "")) bad = ["剩余流量", "套餐到期", "官网", "公告", "过期", "流量", "时间"] if any(x in name for x in bad): continue ptype = p.get("type") server = p.get("server") port = p.get("port") pwd = p.get("password") or p.get("uuid") if not server or not port: continue if ptype == "vless" and p.get("uuid"): link = f"vless://{p['uuid']}@{server}:{port}?type=tcp&security=none#{name.replace(' ', '%20')}" if link not in nodes["vless"]: nodes["vless"].append(link) elif ptype == "anytls" and pwd: link = f"anytls://{pwd}@{server}:{port}#{name.replace(' ', '%20')}" if link not in nodes["anytls"]: nodes["anytls"].append(link) elif ptype == "hysteria2" and pwd: link = f"hysteria2://{pwd}@{server}:{port}#{name.replace(' ', '%20')}" if link not in nodes["hysteria2"]: nodes["hysteria2"].append(link) log_step(f"解析完成: vless={len(nodes['vless'])}, anytls={len(nodes['anytls'])}, hysteria2={len(nodes['hysteria2'])}") return nodes except Exception as e: log_step(f"YAML 解析失败: {e}") return {"vless": [], "anytls": [], "hysteria2": []} def extract_vless_links(text): links = re.findall(r'(vless://[^\n]+)', text) return list(dict.fromkeys(links)) def save_nodes_to_files(email, nodes): log_step("保存节点文件...") if nodes["vless"]: with open(os.path.join(OUTPUT_DIR, f"nodes_vless_{email}.txt"), "w", encoding="utf-8") as f: f.write("\n".join(nodes["vless"])) log_step(f"保存 {len(nodes['vless'])} 个 vless") if nodes["anytls"]: with open(os.path.join(OUTPUT_DIR, f"nodes_anytls_{email}.txt"), "w", encoding="utf-8") as f: f.write("\n".join(nodes["anytls"])) log_step(f"保存 {len(nodes['anytls'])} 个 anytls") if nodes["hysteria2"]: with open(os.path.join(OUTPUT_DIR, f"nodes_hysteria2_{email}.txt"), "w", encoding="utf-8") as f: f.write("\n".join(nodes["hysteria2"])) log_step(f"保存 {len(nodes['hysteria2'])} 个 hysteria2") all_nodes = nodes["vless"] + nodes["anytls"] + nodes["hysteria2"] if all_nodes: with open(os.path.join(OUTPUT_DIR, f"nodes_all_{email}.txt"), "w", encoding="utf-8") as f: f.write("\n".join(all_nodes)) ALL_NODES_ACCUMULATOR.extend(all_nodes) b64 = base64.b64encode("\n".join(all_nodes).encode("utf-8")).decode("utf-8") with open(os.path.join(OUTPUT_DIR, f"nodes_all_{email}_base64.txt"), "w", encoding="utf-8") as f: f.write(b64) log_step(f"保存合并文件 {len(all_nodes)} 个节点") def write_summary_file(): if not ALL_NODES_ACCUMULATOR: log_step("无节点可写汇总") return ev = os.getenv("GITHUB_EVENT_NAME", "") fname = AUTO_SUMMARY_FILE if ev == "schedule" else MANUAL_SUMMARY_FILE uniq = list(dict.fromkeys(ALL_NODES_ACCUMULATOR)) with open(fname, "w", encoding="utf-8") as f: f.write("\n".join(uniq)) log_step(f"汇总写入 {fname}: {len(uniq)} 个节点") def process_one(index, total, password=None): log_step(f"处理第 {index}/{total} 个账号") email = generate_email() if password and len(password) >= 8: pwd = password else: pwd = generate_password(12) log_step(f"邮箱: {email}") try: random_delay(1, 3) content, subscribe_url, auth_data, email = register_wula(email, pwd) with open(os.path.join(OUTPUT_DIR, f"subscription_{email}.txt"), "w", encoding="utf-8") as f: f.write(content) nodes = parse_nodes_from_yaml(content) if not nodes["vless"]: vless = extract_vless_links(content) if vless: nodes["vless"] = vless save_nodes_to_files(email, nodes) all_links = nodes["vless"] + nodes["anytls"] + nodes["hysteria2"] ALL_ACCOUNTS.append({ "email": email, "password": pwd, "subscribe_url": subscribe_url, "nodes": all_links, "registered_at": datetime.now().isoformat() }) file_exists = os.path.isfile(CSV_FILE) with open(CSV_FILE, "a", newline="", encoding="utf-8") as f: w = csv.writer(f) if not file_exists: w.writerow(["email", "password", "subscribe_url", "registered_at"]) w.writerow([email, pwd, subscribe_url, datetime.now().isoformat()]) log_step(f"账号已写入 {CSV_FILE}") except Exception as e: log_step(f"处理失败: {e}") def main(): log_step("========== 开始运行 ==========") if not os.path.exists(CSV_FILE): with open(CSV_FILE, "w", newline="", encoding="utf-8") as f: csv.writer(f).writerow(["email", "password", "subscribe_url", "registered_at"]) total = int(os.getenv("REGISTER_COUNT", "1")) fixed_pwd = os.getenv("REGISTER_PASSWORD", "").strip() or None log_step(f"注册数量={total}, 固定密码={'设置' if fixed_pwd else '随机'}") for i in range(1, total + 1): process_one(i, total, fixed_pwd) if ALL_ACCOUNTS: with open(JSON_FILE, "w", encoding="utf-8") as f: json.dump(ALL_ACCOUNTS, f, ensure_ascii=False, indent=2) log_step(f"已保存 {len(ALL_ACCOUNTS)} 个账号到 {JSON_FILE}") write_summary_file() log_step("========== 全部完成 ==========") if __name__ == "__main__": main()