diff --git a/WX_Applet/Applet_JYHS_KBJY.py b/WX_Applet/Applet_JYHS_KBJY.py new file mode 100644 index 0000000..8fedc3e --- /dev/null +++ b/WX_Applet/Applet_JYHS_KBJY.py @@ -0,0 +1,522 @@ +# cron: 52 10 * * * +# new Env("回收蛙旧衣服回收") +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +脚本名称：快宝旧衣签到抽奖答题（新小程序专用版） +创建时间：2026-06-2 +功能：自动从养鸡场获取账号 → 登录 → 签到 → 抽奖 → 答题 → 自动提现 +环境变量： + wx_cloud : 养鸡场地址（必填） + wx_token : 养鸡场 Authorization（必填） + SINGLE_TEST_WXID : 可选，只处理指定 wxid + PUSHPLUS_TOKEN : 可选，推送结果到微信 +""" + +import base64 +import json +import os +import sys +import time +import uuid +from dataclasses import dataclass +from urllib.parse import quote, urljoin, urlparse +from typing import Optional, List, Dict, Any + +try: + import httpx + import requests +except ImportError: + print("错误: 需要安装 httpx 和 requests") + sys.exit(1) + +try: + from SendNotify import capture_output +except ImportError: + print("[警告] 通知模块 SendNotify.py 未找到，将跳过通知推送。") + def capture_output(title: str = "脚本运行结果"): + def decorator(func): + return func + return decorator + +# ================= 免责声明 ================= +DISCLAIMER = """ +╔══════════════════════════════════════════════╗ +║ 本脚本仅供技术学习与编程研究使用。 ║ +║ 严禁用于任何商业用途或违反目标平台 ║ +║ 服务条款的行为。使用者须自行承担 ║ +║ 因使用本脚本所产生的一切法律风险 ║ +║ 及后果，开发者不承担任何责任。 ║ +║ 请遵守相关法律法规及平台规定， ║ +║ 并在24小时内删除本脚本及所获数据。 ║ +╚══════════════════════════════════════════════╝ +""" + +# ---------- 新小程序专用配置 ---------- +APPID = "wxbe2f25800165b6f8" +CHANNEL_ID = "5658" +LOGIN_PATH = "/api/kuaibao/login/wx_login" +COMMON_REFERER = "https://servicewechat.com/wxbe2f25800165b6f8/24/page-frame.html" + +# ---------- 养鸡场客户端 ---------- +class JycClient: + def __init__(self, server_url: str, token: str): + self.server_url = server_url.rstrip('/') + self.token = token + self.client = httpx.Client(timeout=30.0) + + def get_all_accounts(self) -> List[Dict[str, str]]: + url = f"{self.server_url}/prod-api/wechat/wechat/list" + params = {"pageNum": 1, "pageSize": 1000} + headers = {"Authorization": self.token} + try: + resp = self.client.get(url, headers=headers, params=params) + resp.raise_for_status() + data = resp.json() + if data.get("rows"): + accounts = [] + for row in data["rows"]: + wxid = row.get("wxId") or row.get("wxid") + if wxid: + nick = row.get("nickName", wxid) + accounts.append({"wxid": wxid, "nickName": nick}) + print(f"成功获取 {len(accounts)} 个账号") + return accounts + else: + print(f"获取账号列表失败: {data}") + return [] + except Exception as e: + print(f"拉取账号列表异常: {e}") + return [] + + def get_wechat_code(self, wxid: str) -> Optional[str]: + url = f"{self.server_url}/prod-api/wechat/api/getMiniProgramCode" + headers = {"Authorization": self.token, "Content-Type": "application/json"} + payload = {"wxid": wxid, "appid": APPID} + try: + resp = self.client.post(url, json=payload, headers=headers) + resp.raise_for_status() + result = resp.json() + if result.get("code") == 200: + code = result.get("data", {}).get("code") + if code: + return code + else: + print(f"返回的 data 中没有 code: {result}") + else: + print(f"获取 {wxid} code 失败: {result.get('msg')}") + return None + except Exception as e: + print(f"获取code异常: {e}") + return None + + def close(self): + self.client.close() + + +# ---------- 铛铛一下业务类 ---------- +DEFAULT_BASE_URL = "https://vues.dd1x.cn" +COMMON_HEADERS = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36 MicroMessenger/7.0.20.1781(0x6700143B) NetType/WIFI MiniProgramEnv/Windows WindowsWechat/WMPF WindowsWechat(0x63090a13) UnifiedPCWindowsWechat(0xf2541022) XWEB/16467", + "Content-Type": "application/json", + "Accept": "*/*", + "Referer": COMMON_REFERER, + "Accept-Language": "zh-CN,zh;q=0.9", +} + +@dataclass +class AccountConfig: + token: str + base_url: str + raw: str + +def parse_account_line(line: str) -> Optional[AccountConfig]: + parts = [part.strip() for part in line.split("#") if part.strip()] + if not parts: + return None + token = parts[0] + base_url = DEFAULT_BASE_URL + for part in parts[1:]: + if part.lower().startswith("base_url="): + base_url = part.split("=", 1)[1].strip() + parsed = urlparse(base_url) + if not parsed.scheme or not parsed.netloc: + return None + return AccountConfig(token=token, base_url=f"{parsed.scheme}://{parsed.netloc}", raw=line) + +def decode_openid_from_jwt(token: str) -> str: + try: + parts = token.split(".") + if len(parts) < 2: + return "" + payload = parts[1].replace("-", "+").replace("_", "/") + payload += "=" * (-len(payload) % 4) + data = json.loads(base64.b64decode(payload).decode("utf-8")) + return data.get("openid") or data.get("openId") or "" + except Exception: + return "" + +def assert_ok(resp: dict) -> None: + if resp.get("code") == 0: + return + raise RuntimeError(str(resp.get("msg") or resp.get("message") or "请求失败")) + +def call_api(acc: AccountConfig, method: str, path: str, body: dict | list | None = None) -> dict: + url = urljoin(acc.base_url, path) + headers = {**COMMON_HEADERS, "token": acc.token} + if method.upper() == "GET": + response = requests.get(url, headers=headers, timeout=30) + else: + response = requests.post(url, headers=headers, json=body or {}, timeout=30) + text = response.text + try: + return response.json() + except Exception as exc: + return {"code": -1, "msg": f"JSON解析失败: {exc}; body={text[:500]}{'...' if len(text) > 500 else ''}"} + +def api_get(acc: AccountConfig, path: str) -> dict: + return call_api(acc, "GET", path) + +def api_post(acc: AccountConfig, path: str, body: dict | list | None = None) -> dict: + return call_api(acc, "POST", path, body) + +def send_tracking(open_id: str, path: str, action: str, page_query_obj: dict | None = None, random_args: dict | None = None) -> None: + payload = { + "type": "1", + "platform": "weapp", + "appLaunch": { + "path": "pages/index/index", + "query": {}, + "scene": 1256, + "referrerInfo": {}, + "apiCategory": "default", + }, + "pageQueryObj": page_query_obj or {}, + "appHeader": { + "platformVersion": "4.1.0.34", + "resolution": "978*519", + "pixelRatio": 1.25, + "os": "windows", + "fontSizeSetting": 15, + "deviceModel": "microsoft", + "deviceBrand": "microsoft", + "deviceManufacturer": "microsoft", + "deviceManuid": "microsoft", + "deviceName": "microsoft", + "osVersion": "Windows 10 x64", + "language": "zh_CN", + "access": "wifi", + }, + "path": path, + "uuid": str(uuid.uuid4()), + "randomArgs": random_args or {}, + "appid": APPID, + "channelId": CHANNEL_ID, + "openId": open_id, + "action": action, + } + try: + requests.post("https://data.dd1x.cn/api/test_api", headers=COMMON_HEADERS, json=payload, timeout=15) + except Exception: + pass + +def xcx_point(acc: AccountConfig, process_id: str, note: str, page_name: str) -> None: + if not process_id: + return + try: + api_get(acc, f"/front/xcxPoint?processId={process_id}&processNote={quote(note)}&channel=154&pageName={quote(page_name)}") + except Exception: + pass + +def build_answer_payload(data) -> list[dict]: + if not isinstance(data, list): + return [] + payload = [] + for item in data: + try: + questions_id = int(item.get("id")) + answer_id = int(item.get("correctAnswerId")) + except Exception: + continue + payload.append({"answerId": answer_id, "questionsId": questions_id}) + return payload + +def get_token_by_code(code: str) -> Optional[str]: + url = f"{DEFAULT_BASE_URL}{LOGIN_PATH}" + params = {"code": code, "channelId": CHANNEL_ID} + try: + resp = requests.get(url, params=params, timeout=10) + if resp.status_code != 200: + print(f"登录请求失败: HTTP {resp.status_code}") + return None + data = resp.json() + if data.get("code") == 0: + token = data.get("data", {}).get("token") + if token: + print(f"登录成功，获取到 token") + return token + else: + print(f"登录响应中未找到 token 字段: {data}") + else: + print(f"登录失败: {data.get('msg')}") + return None + except Exception as e: + print(f"登录请求异常: {e}") + return None + +def run_for_account(acc: AccountConfig) -> Dict[str, Any]: + result = {"nickname": "未知", "success": False, "message": "", "money_before": 0, "money_after": 0} + open_id = decode_openid_from_jwt(acc.token) + print("正在初始化会话...") + access_res = api_get(acc, "/front/accessXcx?channelId=154&processId=") + process_id = str(access_res.get("data") or "") + if process_id: + print(f"会话初始化成功: {process_id}") + api_get(acc, f"/front/accessXcx?channelId=154&processId={process_id}") + else: + print("警告: 未获取到 processId，部分任务可能失效") + + send_tracking(open_id, "pages/index/index", "page_show") + send_tracking(open_id, "pages/index/index", "page_click", random_args={"event_name": "进入小程序"}) + xcx_point(acc, process_id, "进入小程序", "首页") + + user_info = api_get(acc, "/ali/getUserInfo") + assert_ok(user_info) + nick_name = str(user_info.get("data", {}).get("nickName") or "未知") + result["nickname"] = nick_name + print(f"账号【{nick_name}】Token有效") + + send_tracking(open_id, "pages/index/index", "page_show") + member_info = api_get(acc, "/api/v2/get_member_info") + assert_ok(member_info) + money_before = float(member_info.get("data", {}).get("money") or 0) + result["money_before"] = money_before + print(f"当前余额{money_before}元") + + sign = api_get(acc, "/api/v2/sign_join") + if sign.get("code") == 0: + print(f"签到成功，获得【{sign.get('data', {}).get('name', '未知奖励')}】") + send_tracking(open_id, "pages/index/index", "page_click", random_args={"event_name": "首页-立即签到"}) + xcx_point(acc, process_id, "首页-立即签到", "首页") + else: + msg = str(sign.get("msg") or sign.get("message") or "签到失败") + if "签" in msg and ("过" in msg or "已经" in msg): + print("今天已经签到过，继续执行抽奖/答题任务") + else: + raise RuntimeError(msg) + + send_tracking(open_id, "pages/activity/turntable/turntable", "page_show") + lottery_info = api_get(acc, "/front/activity/get_lottery_info?id=13&channelId=154") + assert_ok(lottery_info) + times = max(int(lottery_info.get("data", {}).get("member_count") or 0), 0) + print(f"今日有{times}次抽奖机会") + + for _ in range(times): + send_tracking(open_id, "pages/activity/turntable/turntable", "page_click", random_args={"event_name": "抽奖页-立即抽奖"}) + xcx_point(acc, process_id, "抽奖页-立即抽奖", "抽奖页") + result_data = api_get(acc, "/front/activity/get_lottery_result?id=13") + assert_ok(result_data) + record_id = result_data.get("data", {}).get("record_id") + print(f"获得奖励{result_data.get('data', {}).get('prizeName', '未知奖励')}") + if record_id is not None: + assert_ok(api_get(acc, f"/front/activity/update_lottery_result?id={quote(str(record_id))}")) + + print("开始获取今日题目...") + send_tracking(open_id, "pages/find_page/index", "page_show") + send_tracking(open_id, "pages/index/index", "page_click", random_args={"event_name": "底部导航-发现"}) + send_tracking(open_id, "pages/find_page/index", "page_click", random_args={"event_name": "回收问答-立即参与"}) + send_tracking(open_id, "pages/find_page/answerQues/index", "page_show") + + questions = api_get(acc, "/api/questions/get_questions") + assert_ok(questions) + answer_payload = build_answer_payload(questions.get("data")) + if answer_payload: + send_tracking(open_id, "pages/find_page/answerSelectQues/index", "page_show") + judge = api_post(acc, "/api/questions/judge", answer_payload) + assert_ok(judge) + if judge.get("data") == 2: + print("今日已经答过题了") + else: + print("答题提交完成") + send_tracking(open_id, "pages/find_page/answerSelectQues/index", "page_click", random_args={"event_name": "提现说明-立即提现"}) + xcx_point(acc, process_id, "提现说明-立即提现", "回答问题页") + + send_tracking(open_id, "pages/mine/mine", "page_show") + send_tracking(open_id, "pages/index/index", "page_click", random_args={"event_name": "底部导航-我的"}) + + member_info = api_get(acc, "/api/v2/get_member_info") + assert_ok(member_info) + current_money = float(member_info.get("data", {}).get("money") or 0) + result["money_after"] = current_money + print(f"任务完毕，当前余额{current_money}元") + + if current_money >= 0.3: + print("余额满足提现要求，准备提现...") + send_tracking(open_id, "pages/mine/mine", "page_click", random_args={"event_name": "设置-我的钱包"}) + xcx_point(acc, process_id, "中心首页-我的钱包", "我的") + send_tracking(open_id, "pages/mine/withdrawal/index", "page_show", page_query_obj={"channelId": "154"}) + xcx_point(acc, process_id, "进入钱包", "提现") + send_tracking(open_id, "pages/mine/withdrawal/index", "page_click", random_args={"event_name": "钱包-提现"}) + xcx_point(acc, process_id, "钱包-提现", "提现") + + withdrawal_list = api_get(acc, "/api/h/get_withdrawal_trade_list") + if isinstance(withdrawal_list, list): + trade_list = withdrawal_list + else: + trade_list = withdrawal_list.get("data") if isinstance(withdrawal_list.get("data"), list) else [] + + available = [item for item in trade_list if not item.get("disabled") and float(item.get("money") or 0) >= 0.3] + if available: + total_money = f"{sum(float(item.get('money') or 0) for item in available):.2f}" + print(f"检测到可提现订单 {len(available)} 个，合计 {total_money} 元") + withdraw_res = api_post( + acc, + "/api/h/withdrawal", + {"totalMoney": total_money, "type": 1, "withdrawalDetailPojoList": available}, + ) + if withdraw_res.get("code") == 1: + print(f"提现成功: {withdraw_res.get('msg') or '确定'}") + send_tracking(open_id, "pages/mine/mine", "page_click", random_args={"event_name": "全选-提现成功"}) + xcx_point(acc, process_id, "全选-提现成功", "提现") + else: + print(f"提现失败: {withdraw_res.get('msg') or '未知错误'}") + else: + print("没有满足提现金额(>=0.3元)的订单") + + result["success"] = True + result["message"] = "签到抽奖答题完成" + return result + + +def push_to_pushplus(token: str, title: str, content: str) -> bool: + if not token: + return False + url = "http://www.pushplus.plus/send" + data = {"token": token, "title": title, "content": content, "template": "txt"} + try: + resp = httpx.post(url, json=data, timeout=10) + if resp.status_code == 200: + result = resp.json() + if result.get("code") == 200: + print("推送成功") + return True + else: + print(f"推送失败: {result.get('msg')}") + else: + print(f"推送失败: HTTP {resp.status_code}") + except Exception as e: + print(f"推送异常: {e}") + return False + + +def load_accounts_from_env() -> List[AccountConfig]: + lines = [line.strip() for line in os.getenv("dd1x", "").splitlines() if line.strip()] + accounts = [] + for line in lines: + acc = parse_account_line(line) + if acc: + accounts.append(acc) + else: + print(f"手动配置格式错误，跳过: {line}") + return accounts + + +def load_accounts_from_jyc(jyc: JycClient, single_wxid: str) -> List[AccountConfig]: + if single_wxid: + accounts_info = [{"wxid": single_wxid, "nickName": single_wxid}] + print(f"单号测试模式: {single_wxid}") + else: + accounts_info = jyc.get_all_accounts() + if not accounts_info: + print("未获取到任何在线账号") + return [] + accounts = [] + for info in accounts_info: + wxid = info["wxid"] + nick = info.get("nickName", wxid) + print(f"正在为 {nick} ({wxid}) 获取登录 code...") + code = jyc.get_wechat_code(wxid) + if not code: + print(f"获取 {nick} 的 code 失败，跳过") + continue + token = get_token_by_code(code) + if not token: + print(f"登录 {nick} 失败，无法获取 token，跳过") + continue + accounts.append(AccountConfig(token=token, base_url=DEFAULT_BASE_URL, raw=f"{token}#{DEFAULT_BASE_URL}")) + return accounts + + +@capture_output("铛铛一下签到抽奖答题运行结果") +def main(): + # 打印免责声明 + print(DISCLAIMER) + + jyc_server = os.getenv("wx_cloud") + jyc_token = os.getenv("wx_token") + single_test = os.getenv("SINGLE_TEST_WXID") + pushplus_token = os.getenv("PUSHPLUS_TOKEN") + + if jyc_server and jyc_token: + jyc = JycClient(jyc_server, jyc_token) + try: + accounts = load_accounts_from_jyc(jyc, single_test) + finally: + jyc.close() + if accounts: + print(f"从养鸡场获取到 {len(accounts)} 个账号") + else: + print("养鸡场未获取到有效账号，尝试使用环境变量 dd1x 手动配置") + accounts = load_accounts_from_env() + else: + print("未配置养鸡场，使用环境变量 dd1x 手动配置") + accounts = load_accounts_from_env() + + if not accounts: + print("未找到任何可用账号，退出") + return + + print(f"共处理 {len(accounts)} 个账号") + results = [] + for idx, acc in enumerate(accounts, 1): + print(f"\n{'='*50}") + print(f"处理第 {idx}/{len(accounts)} 个账号") + print(f"{'='*50}") + try: + res = run_for_account(acc) + results.append(res) + except Exception as e: + print(f"处理失败: {e}") + results.append({ + "nickname": "未知", + "success": False, + "message": str(e), + "money_before": 0, + "money_after": 0 + }) + if idx < len(accounts): + print("等待3秒后处理下一个账号...") + time.sleep(3) + + summary_lines = [] + success_cnt = 0 + for r in results: + status = "✅" if r.get("success") else "❌" + nickname = r.get("nickname", "未知") + msg = r.get("message", "") + line = f"{status} {nickname}: {msg}" + if r.get("money_before", 0) != 0 or r.get("money_after", 0) != 0: + line += f"，余额 {r['money_before']} → {r['money_after']} 元" + summary_lines.append(line) + if r.get("success"): + success_cnt += 1 + summary = "\n".join(summary_lines) + title = f"铛铛一下签到结果 - 成功 {success_cnt}/{len(results)}" + print("\n" + summary) + if pushplus_token: + push_to_pushplus(pushplus_token, title, summary) + else: + print("未设置 PUSHPLUS_TOKEN，跳过推送") + +if __name__ == "__main__": + main() \ No newline at end of file