277 lines
10 KiB
Python
277 lines
10 KiB
Python
# cron: 52 8 * * *
|
||
# 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() |