添加 WX_Applet/Applet_JYHS_TNJY.py

WX_Applet/Applet_JYHS_TNJY.py

@@ -0,0 +1,277 @@
+# cron: 52 7 * * *
+# new Env("旧衣回收_天牛旧衣")
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+天牛旧衣回收全自动化脚本 - 修复登录异常版
+"""
+
+import os
+import sys
+import json
+import time
+import logging
+import requests
+from datetime import datetime
+from typing import Optional, Dict, Any, List, Tuple
+
+# ==================== 配置区 ====================
+WX_CLOUD = os.getenv('wx_cloud', '')
+AUTH_TOKEN = os.getenv('wx_token', '')
+TIAN_NIU_APPID = "wx887c2f947bffa76e"
+TIAN_NIU_NEW_URL = "https://tianniunew.fzjingzhou.com"
+TIAN_NIU_OLD_URL = "https://tianniu.fzjingzhou.com"
+LOG_DIR = "/opt/data/logs/tianniu"
+
+SINGLE_TEST_WXID = os.getenv('SINGLE_TEST_WXID', '')
+WX_IDS_ENV = os.getenv('WX_IDS', '')
+
+# 全局变量
+logger: logging.Logger = None
+log_file_path: str = ""
+
+# ==================== 获取 wxid 列表 ====================
+def fetch_all_wxids_from_yjc() -> List[str]:
+    """从养鸡场获取全部可用的 wxid 列表"""
+    if not WX_CLOUD or not AUTH_TOKEN:
+        print("❌ 养鸡场配置缺失，无法获取全部账号")
+        return []
+    url = f"{WX_CLOUD}/prod-api/wechat/wechat/list?pageNum=1&pageSize=1000"
+    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
+    try:
+        resp = requests.get(url, headers=headers, timeout=30)
+        if resp.status_code != 200:
+            print(f"❌ 获取账号列表失败: HTTP {resp.status_code}")
+            return []
+        data = resp.json()
+        rows = data.get('rows', [])
+        if not rows:
+            print("⚠️ 养鸡场返回账号列表为空")
+            return []
+        wxids = []
+        for item in rows:
+            wxid = item.get('wxId') or item.get('wxid')
+            if wxid:
+                wxids.append(wxid)
+        print(f"✅ 成功获取 {len(wxids)} 个账号")
+        return wxids
+    except Exception as e:
+        print(f"❌ 请求异常: {e}")
+        return []
+
+def get_wxid_list() -> List[str]:
+    """获取待处理的 wxid 列表"""
+    if SINGLE_TEST_WXID:
+        return [SINGLE_TEST_WXID]
+    if WX_IDS_ENV:
+        return [wxid.strip() for wxid in WX_IDS_ENV.split(',') if wxid.strip()]
+    print("📡 未指定 wxid，正在从养鸡场获取全部账号...")
+    all_wxids = fetch_all_wxids_from_yjc()
+    if not all_wxids:
+        print("❌ 从养鸡场获取全部账号失败")
+        return []
+    return all_wxids
+
+# ==================== 初始化日志 ====================
+def init_logger(wxid: str) -> None:
+    global logger, log_file_path
+    os.makedirs(LOG_DIR, exist_ok=True)
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    safe_wxid = wxid.replace('@', '_').replace('/', '_')
+    log_file_path = os.path.join(LOG_DIR, f"tianniu_{safe_wxid}_{timestamp}.log")
+    
+    logger = logging.getLogger(f"tianniu_{wxid}")
+    logger.setLevel(logging.INFO)
+    if logger.handlers:
+        logger.handlers.clear()
+    
+    fh = logging.FileHandler(log_file_path, encoding="utf-8")
+    fh.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"))
+    logger.addHandler(fh)
+    
+    ch = logging.StreamHandler(sys.stdout)
+    ch.setFormatter(logging.Formatter("%(message)s"))
+    logger.addHandler(ch)
+    
+    logger.info("=" * 50)
+    logger.info(f"          天牛旧衣回收脚本 - 账号: {wxid}")
+    logger.info("=" * 50)
+    logger.info(f"时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+    logger.info("=" * 50)
+
+def check_config() -> bool:
+    if not WX_CLOUD:
+        logger.error("❌ 环境变量 wx_cloud 未设置")
+        return False
+    if not AUTH_TOKEN:
+        logger.error("❌ 环境变量 wx_token 未设置")
+        return False
+    logger.info(f"✅ 养鸡场地址: {WX_CLOUD}")
+    return True
+
+# ==================== HTTP 请求 ====================
+def http_post(url: str, data=None, json_data=None, headers=None, timeout=30) -> Optional[Dict]:
+    try:
+        resp = requests.post(url, data=data, json=json_data, headers=headers, timeout=timeout)
+        if resp.status_code != 200:
+            logger.error(f"HTTP {resp.status_code}: {url}")
+            return None
+        # 确保响应是 JSON
+        try:
+            return resp.json()
+        except Exception as e:
+            logger.error(f"JSON解析失败: {e}, 响应预览: {resp.text[:200]}")
+            return None
+    except Exception as e:
+        logger.error(f"请求失败 {url}: {str(e)}")
+        return None
+
+# ==================== 业务步骤 ====================
+def get_miniprogram_code(wxid: str) -> Optional[str]:
+    logger.info("\n步骤1/4：获取小程序code...")
+    start = time.time()
+    url = f"{WX_CLOUD}/prod-api/wechat/api/getMiniProgramCode"
+    payload = {"wxid": wxid, "appid": TIAN_NIU_APPID}
+    headers = {"Authorization": f"Bearer {AUTH_TOKEN}", "Content-Type": "application/json"}
+    result = http_post(url, json_data=payload, headers=headers, timeout=30)
+    elapsed = time.time() - start
+    if result and isinstance(result, dict) and result.get("code") == 200:
+        code = result.get("data", {}).get("code")
+        if code:
+            logger.info(f"✅ 获取成功（耗时：{elapsed:.2f}秒）")
+            return code
+    logger.error("❌ 获取失败")
+    return None
+
+def login_tianniu(code: str) -> Optional[Tuple[str, Dict]]:
+    logger.info("\n步骤2/4：登录天牛...")
+    start = time.time()
+    url = f"{TIAN_NIU_NEW_URL}/api/login/getWxMiniProgramSessionKey"
+    payload = {"code": code, "appid": TIAN_NIU_APPID}
+    headers = {"Content-Type": "application/json"}
+    result = http_post(url, json_data=payload, headers=headers, timeout=30)
+    elapsed = time.time() - start
+    # 增加类型和字段检查
+    if result and isinstance(result, dict) and result.get("code") == 1000:
+        data = result.get("data")
+        if isinstance(data, dict) and data.get("token"):
+            token = data["token"]
+            logger.info(f"✅ 登录成功（耗时：{elapsed:.2f}秒）")
+            info = data.get("personInfo", {})
+            user_info = {
+                "mobile": info.get("mobile", "未知"),
+                "exchange": info.get("exchange", 0),
+                "sign_in_num": info.get("sign_in_num", 0)
+            }
+            logger.info(f"   手机号：{user_info['mobile']}  可兑换：{user_info['exchange']}元  已签到：{user_info['sign_in_num']}天")
+            return token, user_info
+    logger.error(f"❌ 登录失败，响应类型: {type(result)} 内容: {result}")
+    return None
+
+def do_sign_in(token: str) -> None:
+    logger.info("\n步骤3/4：执行签到...")
+    start = time.time()
+    headers = {"content-type": "application/x-www-form-urlencoded", "platform": "MP-WEIXIN"}
+    data = {"token": token}
+    result = None
+    try:
+        resp = requests.post(f"{TIAN_NIU_NEW_URL}/api/Person/sign", data=data, headers=headers, timeout=30)
+        if resp.status_code == 200:
+            result = resp.json()
+    except:
+        pass
+    if not result:
+        try:
+            resp = requests.post(f"{TIAN_NIU_OLD_URL}/api/Person/sign", data=data, headers=headers, timeout=30)
+            if resp.status_code == 200:
+                result = resp.json()
+        except:
+            pass
+    elapsed = time.time() - start
+    if result:
+        code = result.get("code", 0)
+        msg = result.get("msg", "")
+        if code == 1000:
+            logger.info(f"✅ 签到成功（耗时：{elapsed:.2f}秒）")
+        elif code == 1001 and "已签到" in msg:
+            logger.info(f"📅 今日已签到（耗时：{elapsed:.2f}秒）")
+        else:
+            logger.info(f"⚠️ 签到失败：{msg}")
+    else:
+        logger.info("❌ 签到请求失败")
+
+def check_withdraw(token: str, exchange: float) -> None:
+    logger.info("\n步骤4/4：提现检查...")
+    if exchange >= 2.0:
+        logger.info(f"✅ 余额满足提现条件：¥{exchange:.2f}，执行提现...")
+        url = f"{TIAN_NIU_OLD_URL}/api/cash/scoreWithdraw"
+        headers = {"content-type": "application/x-www-form-urlencoded", "platform": "MP-WEIXIN"}
+        data = {"type": "wx_account", "score": 20, "token": token}
+        try:
+            resp = requests.post(url, data=data, headers=headers, timeout=30)
+            if resp.status_code == 200:
+                result = resp.json()
+                if result.get("code") == 1000 or "成功" in str(result):
+                    logger.info("✅ 提现成功")
+                else:
+                    logger.warning("⚠️ 提现失败")
+            else:
+                logger.warning("⚠️ 提现请求异常")
+        except Exception as e:
+            logger.error(f"提现异常: {e}")
+    else:
+        logger.info(f"📊 余额不足：¥{exchange:.2f} < ¥2，不执行提现")
+
+def process_one_wxid(wxid: str) -> bool:
+    """处理单个 wxid，捕获所有异常不中断主流程"""
+    try:
+        init_logger(wxid)
+        if not check_config():
+            return False
+        code = get_miniprogram_code(wxid)
+        if not code:
+            return False
+        login_result = login_tianniu(code)
+        if not login_result:
+            return False
+        token, user_info = login_result
+        do_sign_in(token)
+        check_withdraw(token, user_info.get("exchange", 0))
+        # 保存最新日志副本
+        try:
+            latest = f"/opt/data/tianniu_latest_{wxid}.log"
+            os.makedirs(os.path.dirname(latest), exist_ok=True)
+            with open(log_file_path, "r", encoding="utf-8") as src, open(latest, "w", encoding="utf-8") as dst:
+                dst.write(src.read())
+        except:
+            pass
+        logger.info(f"\n✅ 账号 {wxid} 处理完毕")
+        return True
+    except Exception as e:
+        # 异常捕获，避免脚本崩溃
+        print(f"❌ 账号 {wxid} 处理异常: {e}")
+        return False
+
+# ==================== 主函数 ====================
+def main():
+    wxid_list = get_wxid_list()
+    if not wxid_list:
+        print("❌ 没有可处理的 wxid，请检查环境变量或养鸡场接口")
+        sys.exit(1)
+    total = len(wxid_list)
+    success_count = 0
+    for idx, wxid in enumerate(wxid_list, 1):
+        print(f"\n>>> 正在处理第 {idx}/{total} 个账号: {wxid}")
+        if process_one_wxid(wxid):
+            success_count += 1
+        # 重置日志句柄，避免多个账号干扰
+        for handler in logging.root.handlers[:]:
+            logging.root.removeHandler(handler)
+        if idx < total:
+            time.sleep(2)
+    print(f"\n{'='*50}")
+    print(f"批量处理完成：成功 {success_count}/{total} 个账号")
+    print(f"{'='*50}")
+
+if __name__ == "__main__":
+    main()