mirror/cc892786825-qiandao
1
0
Fork 0
mirror of https://github.com/cc892786825/qiandao.git synced 2025-12-17 07:18:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add files via upload

Browse Source
This commit is contained in:
cc892786825
2025-12-15 23:18:27 +08:00
committed by GitHub
parent afbfd2b023
commit 4f52ef5bef
1 changed files with 453 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

453
ltqycs.py Normal file
View File

@@ -0,0 +1,453 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
专业优化版 — 联通权益超市自动任务脚本
Version: 3.0-Pro
"""
import os
import sys
import time
import json
import logging
import requests
from urllib.parse import urlparse, parse_qs
from datetime import datetime
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# ======================
# 日志格式（带毫秒）
# ======================
class MsFormatter(logging.Formatter):
def formatTime(self, record, datefmt=None):
dt = datetime.fromtimestamp(record.created)
s = dt.strftime("%Y-%m-%d %H:%M:%S.%f")
return s[:-3]
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] %(message)s')
for h in logging.getLogger().handlers:
h.setFormatter(MsFormatter('[%(asctime)s] %(message)s'))
# ======================
# 共享 Session
# ======================
sess = requests.Session()
adapter = HTTPAdapter(max_retries=Retry(total=3, backoff_factor=0.3))
sess.mount("http://", adapter)
sess.mount("https://", adapter)
# ======================
# 统一 UA
# ======================
def ua():
return {
"User-Agent":
"Mozilla/5.0 (Linux; Android 10; Redmi K30 Pro Build/QKQ1.191117.002; wv) "
"AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/100.0.4896.58 "
"Mobile Safari/537.36 unicom{version:android@11.0500}",
"Accept": "*/*",
}
# ======================
# 主类
# ======================
class CUAPI:
def __init__(self, accounts):
self.accounts = accounts
self.GrantPrize = True
# ======================
# ✨ 重构 do_send（专业版 3.0）
# 支持：
# - raw 模式
# - 自动 JSON 解析
# - 自动错误处理
# ======================
def do_send(self, url, method="GET", headers=None,
params=None, data=None, timeout=10,
raw=False, allow_redirects=True):
try:
resp = sess.request(
method=method,
url=url,
headers=headers,
params=params,
json=None if (data and "token_online" in str(data)) else data,
data=data if (data and "token_online" in str(data)) else None,
timeout=timeout,
allow_redirects=allow_redirects
)
except Exception as e:
logging.error(f"请求失败: {e}")
return None
# raw 直接返回响应对象
if raw:
return resp
if resp.status_code == 302:
return resp
try:
return resp.json()
except:
logging.error("响应非 JSON 格式")
return None
# ======================
# 登录 — token_online
# ======================
def login_with_token_online(self, phone, tok, appid):
url = "https://m.client.10010.com/mobileService/onLine.htm"
data = {
"reqtime": str(int(time.time() * 1000)),
"netWay": "Wifi",
"version": "android@11.0000",
"token_online": tok,
"appId": appid,
"deviceModel": "Mi10",
"step": "welcome",
"androidId": "e1d2c3b4a5f6"
}
resp = self.do_send(url, method="POST", headers=ua(), data=data)
if resp and resp.get("ecs_token"):
logging.info(f"{phone} token 登录成功")
return resp["ecs_token"]
logging.error(f"{phone} token 登录失败")
return None
# ======================
# 获取 ticket（核心修复点）
# ======================
def get_ticket(self, ecs_token):
"""
使用联通官方 H5 openPlatLine 跳转链路强制获取 ticket
此链路比 openPlatLineNew 更稳定，token_online 登录也可使用
"""
url = (
"https://m.client.10010.com/mobileService/openPlatform/"
"openPlatLine.htm"
)
headers = {
"User-Agent":
"Mozilla/5.0 (Linux; Android 10; MI 10) "
"AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 "
"Chrome/108.0.5359.128 Mobile Safari/537.36 "
"unicom{version:android@11.0500}",
"X-Requested-With": "com.sinovatech.unicom.ui",
"Origin": "https://img.client.10010.com",
"Referer": "https://img.client.10010.com/",
"Cookie": f"ecs_token={ecs_token}",
}
params = {
"to_url": "https://contact.bol.wo.cn/market",
"reqtime": str(int(time.time() * 1000)),
"version": "android@11.0500"
}
# 强制获取响应，不自动解析
resp = self.do_send(
url, method="GET",
headers=headers,
params=params,
raw=True,
allow_redirects=False
)
if not resp:
logging.error("❌ ticket 请求失败")
return None
# 必须要带 Location 才行
loc = resp.headers.get("Location")
if not loc:
logging.error("❌ 联通拒绝跳转（无 Location）")
return None
qs = parse_qs(urlparse(loc).query)
ticket = qs.get("ticket", [None])[0]
return ticket
# ======================
# 获取 userToken
# ======================
def get_userToken(self, ticket):
url = f"https://backward.bol.wo.cn/prod-api/auth/marketUnicomLogin?ticket={ticket}"
resp = self.do_send(url, method="POST", headers=ua())
return resp.get("data", {}).get("token") if resp else None
# ======================
# 获取任务列表
# ======================
def get_tasks(self, ecs_token, userToken):
url = (
"https://backward.bol.wo.cn/prod-api/promotion/activityTask/"
"getAllActivityTasks?activityId=12"
)
headers = ua()
headers["Authorization"] = f"Bearer {userToken}"
headers["Cookie"] = f"ecs_token={ecs_token}"
resp = self.do_send(url, headers=headers)
if not resp:
return []
return resp.get("data", {}).get("activityTaskUserDetailVOList", [])
# ======================
# 执行单个任务
# ======================
def run_task(self, task, userToken):
name = task.get("name", "")
param = task.get("param1")
target = int(task.get("triggerTime", 1))
done = int(task.get("triggeredTime", 0))
# 跳过购买 / 秒杀任务
if "购买" in name or "秒杀" in name:
logging.info(f"[跳过复杂任务] {name}")
return
if done >= target:
logging.info(f"任务已完成：{name}")
return
# 任务类型判断
if "浏览" in name or "查看" in name:
api = "checkView"
elif "分享" in name:
api = "checkShare"
else:
logging.info(f"无法识别任务类型：{name}")
return
url = f"https://backward.bol.wo.cn/prod-api/promotion/activityTaskShare/{api}?checkKey={param}"
headers = ua()
headers["Authorization"] = f"Bearer {userToken}"
resp = self.do_send(url, method="POST", headers=headers)
if resp and resp.get("code") == 200:
logging.info(f"任务完成：{name}")
else:
logging.error(f"任务失败：{name}")
# ======================
# 检查抽奖池是否放水
# ======================
def check_raffle(self, userToken):
url = (
"https://backward.bol.wo.cn/prod-api/promotion/home/"
"raffleActivity/prizeList?id=12"
)
headers = ua()
headers["Authorization"] = f"Bearer {userToken}"
resp = self.do_send(url, method="POST", headers=headers)
if not resp:
return False
# 判断是否有“月卡”、“月会员”等奖品
prize_list = resp.get("data", [])
has_live = any(("月" in p.get("name", "")) for p in prize_list)
return has_live
# ======================
# 抽奖次数获取 + 循环抽奖
# ======================
def raffle(self, userToken):
url = (
"https://backward.bol.wo.cn/prod-api/promotion/home/"
"raffleActivity/getUserRaffleCount?id=12"
)
headers = ua()
headers["Authorization"] = f"Bearer {userToken}"
resp = self.do_send(url, method="POST", headers=headers)
if not resp:
return
count = resp.get("data", 0)
logging.info(f"当前剩余抽奖次数：{count}")
for _ in range(count):
self.raffle_once(userToken)
time.sleep(1) # 给接口缓冲时间
# ======================
# 执行一次抽奖
# ======================
def raffle_once(self, userToken):
url = (
"https://backward.bol.wo.cn/prod-api/promotion/home/"
"raffleActivity/userRaffle?id=12&channel="
)
headers = ua()
headers["Authorization"] = f"Bearer {userToken}"
resp = self.do_send(url, method="POST", headers=headers)
if not resp:
logging.error("抽奖请求失败")
return
if resp.get("code") != 200:
logging.error("抽奖失败")
return
data = resp.get("data", {})
prize = data.get("prizesName")
msg = data.get("message", "")
logging.info(f"🎁 抽奖结果：{prize or msg}")
# ======================
# 查询待领奖品
# ======================
def get_pending_prizes(self, userToken):
url = "https://backward.bol.wo.cn/prod-api/promotion/home/raffleActivity/getMyPrize"
headers = ua()
headers["Authorization"] = f"Bearer {userToken}"
data = {
"id": 12,
"type": 0,
"page": 1,
"limit": 100
}
resp = self.do_send(url, method="POST", headers=headers, data=data)
if not resp:
return []
return resp.get("data", {}).get("list", [])
# ======================
# 自动领奖
# ======================
def grant_prize(self, userToken, recordId, prizeName):
url = (
"https://backward.bol.wo.cn/prod-api/promotion/home/"
"raffleActivity/grantPrize?activityId=12"
)
headers = ua()
headers["Authorization"] = f"Bearer {userToken}"
headers["Content-Type"] = "application/json"
resp = self.do_send(url, method="POST", headers=headers, data={"recordId": recordId})
if resp and resp.get("code") == 200:
logging.info(f"🎉 奖品领取成功：{prizeName}")
else:
logging.error(f"领奖失败：{prizeName}")
# ======================
# 单账号完整流程
# ======================
def run_account(self, phone, ecs_token=None, token_online=None, appid=None):
logging.info(f"\n===== 开始处理账号：{phone} =====")
# 登录
if ecs_token:
final_token = ecs_token
else:
final_token = self.login_with_token_online(phone, token_online, appid)
if not final_token:
return
# Ticket
ticket = self.get_ticket(final_token)
if not ticket:
logging.error("❌ 获取 ticket 失败")
return
logging.info("✔ ticket 获取成功")
# userToken
userToken = self.get_userToken(ticket)
if not userToken:
logging.error("❌ 获取 userToken 失败")
return
logging.info("✔ userToken 获取成功")
# 任务列表
tasks = self.get_tasks(final_token, userToken)
for t in tasks:
self.run_task(t, userToken)
# 抽奖池检查
logging.info("检查抽奖池放水情况...")
if self.check_raffle(userToken):
logging.info("✔ 抽奖池已放水，开始抽奖")
self.raffle(userToken)
else:
logging.info("❌ 今日未放水，跳过抽奖")
# 查询待领奖品
pending = self.get_pending_prizes(userToken)
if pending:
logging.info(f"发现 {len(pending)} 个待领取奖品，开始领取...")
for item in pending:
recordId = item.get("id")
prizeName = item.get("prizesName")
self.grant_prize(userToken, recordId, prizeName)
else:
logging.info("暂无待领取奖品")
logging.info(f"===== 账号 {phone} 处理完成 =====\n")
# ======================
# 主程序入口
# ======================
def run(self):
for acc in self.accounts:
parts = acc.split("#")
phone = parts[0]
if len(parts) == 2:
self.run_account(phone, ecs_token=parts[1])
elif len(parts) >= 3:
self.run_account(phone, token_online=parts[1], appid=parts[2])
time.sleep(3)
# ======================
# 入口
# ======================
if __name__ == "__main__":
raw = os.getenv("UNICOM_ACCOUNTS", "").strip()
if not raw:
print("❌ 未设置环境变量 UNICOM_ACCOUNTS")
print("示例：")
print(" 手机号#ecs_token")
print(" 手机号#token_online#appid")
sys.exit(1)
accounts = [line for line in raw.splitlines() if line.strip()]
CUAPI(accounts).run()