Forward some script

This commit is contained in:
smallfawn 2026-04-07 19:05:15 +08:00
parent 458dcb885c
commit 37aedb1b2a
3 changed files with 453 additions and 0 deletions

248
daily/BREO.py Normal file
View File

@ -0,0 +1,248 @@
#by:哆啦A梦
#入口:http://mx.qrurl.net/h5/wxa/link?sid=26407uif5Oq
#抓包breoplus.breo.cn的域名下的token多账号换行分割
#账号变量名:BREO
#new Env("BREO")
#cron 8 9,10,11 * * *
import requests
import json
import os
import time
def get_random_one_word():
try:
response = requests.get("https://uapis.cn/api/say")
if response.status_code == 200:
return response.text.strip()
else:
return "无法获取一言"
except Exception as e:
print(f"获取一言时出错: {e}")
return "无法获取一言"
def get_proclamation():
primary_url = "https://github.com/3288588344/toulu/raw/refs/heads/main/tl.txt"
backup_url = "https://tfapi.cn/TL/tl.json"
try:
response = requests.get(primary_url, timeout=10)
if response.status_code == 200:
print("\n" + "=" * 50)
print("📢 公告信息")
print("=" * 35)
print(response.text)
print("=" * 35 + "\n")
print("公告获取成功,开始执行任务...\n")
return
except requests.exceptions.RequestException as e:
print(f"获取公告时发生错误: {e}, 尝试备用链接...")
try:
response = requests.get(backup_url, timeout=10)
if response.status_code == 200:
print("\n" + "=" * 50)
print("📢 公告信息")
print("=" * 35)
print(response.text)
print("=" * 35 + "\n")
print("公告获取成功,开始执行任务...\n")
else:
print(f"⚠️ 获取公告失败,状态码: {response.status_code}")
except requests.exceptions.RequestException as e:
print(f"⚠️ 获取公告时发生错误: {e}, 可能是网络问题或链接无效。")
def post_to_breo(token, content, title):
url = "https://breoplus.breo.cn/breo-app/communityBaseInfo/releasePost"
headers = {
"token": token,
"device-type": "Xiaomi",
"device-version": "10",
"channel": "Breo",
"version_code": "30201",
"version": "3.2.1",
"encrypt": "1",
"Content-Type": "application/json; charset=UTF-8"
}
data = {
"anonymoused": 1,
"content": content,
"expressText": "",
"images": [],
"subTitle": "",
"title": title,
"topicText": ""
}
try:
response = requests.post(url, headers=headers, data=json.dumps(data))
if response.status_code == 200:
result = response.json()
if result.get("success", False):
print("✅ 发帖成功!")
print(f"帖子 ID: {result['result']['id']}")
print(f"帖子标题: {result['result']['title']}")
return result["result"]["id"]
else:
print(f"❌ 发帖失败,错误信息:{result.get('message', '未知错误')}")
return None
else:
print(f"❌ 请求失败,状态码:{response.status_code}")
return None
except Exception as e:
print(f"❌ 请求错误: {e}")
return None
def collect_post(token, post_id):
url = "https://breoplus.breo.cn/breo-app/communityBaseInfo/collect"
headers = {
"token": token,
"device-type": "Xiaomi",
"device-version": "10",
"channel": "Breo",
"version_code": "30201",
"version": "3.2.1",
"encrypt": "1",
"Content-Type": "application/json; charset=UTF-8"
}
data = {
"postId": post_id
}
try:
response = requests.post(url, headers=headers, data=json.dumps(data))
if response.status_code == 200:
result = response.json()
if result.get("success", False):
print("✅ 收藏成功!")
print(f"获得点数: {result['result']['point']}")
print(f"成长值: {result['result']['grow']}")
else:
print(f"❌ 收藏失败,错误信息:{result.get('message', '未知错误')}")
else:
print(f"❌ 请求失败,状态码:{response.status_code}")
except Exception as e:
print(f"❌ 请求错误: {e}")
def comment_post(token, post_id):
for _ in range(2): # 评论2次
comment_content = get_random_one_word() # 使用随机一言作为评论内容
url = "https://breoplus.breo.cn/breo-app/communityBaseInfo/comment"
headers = {
"token": token,
"device-type": "Xiaomi",
"device-version": "10",
"channel": "Breo",
"version_code": "30201",
"version": "3.2.1",
"encrypt": "1",
"Content-Type": "application/json; charset=UTF-8"
}
data = {
"anonymoused": 0,
"commentText": comment_content,
"postId": post_id
}
try:
response = requests.post(url, headers=headers, data=json.dumps(data))
if response.status_code == 200:
result = response.json()
if result.get("success", False):
print("✅ 评论成功!")
print(f"评论内容: {result['result']['rootOutVO']['commentText']}")
print(f"获得点数: {result['result']['point']}")
print(f"成长值: {result['result']['grow']}")
else:
print(f"❌ 评论失败,错误信息:{result.get('message', '未知错误')}")
else:
print(f"❌ 请求失败,状态码:{response.status_code}")
except Exception as e:
print(f"❌ 请求错误: {e}")
time.sleep(1) # 避免频繁请求
def browse_mall(token):
url = "https://breoplus.breo.cn/breo-app/user/po-task-info/mall"
headers = {
"token": token,
"device-type": "Xiaomi",
"device-version": "10",
"channel": "Breo",
"version_code": "30201",
"version": "3.2.1",
"encrypt": "1"
}
try:
response = requests.post(url, headers=headers)
if response.status_code == 200:
result = response.json()
if result.get("success", False):
print("✅ 浏览商城成功!")
print(f"获得点数: {result['result']['point']}")
print(f"成长值: {result['result']['grow']}")
else:
print(f"❌ 浏览商城失败,错误信息:{result.get('message', '未知错误')}")
else:
print(f"❌ 请求失败,状态码:{response.status_code}")
except Exception as e:
print(f"❌ 请求错误: {e}")
def punch_in(token):
url = "https://breoplus.breo.cn/breo-app/user/po-task-info/punch"
headers = {
"Host": "breoplus.breo.cn",
"Connection": "keep-alive",
"Content-Length": "0",
"content-type": "application/json",
"token": token,
"charset": "utf-8",
"Referer": "https://servicewechat.com/wx61457400e4212cec/304/page-frame.html",
"User-Agent": "Mozilla/5.0 (Linux; Android 10; MI 8 Build/QKQ1.190828.002; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/134.0.6998.136 Mobile Safari/537.36 XWEB/1340043 MMWEBSDK/20241202 MMWEBID/3628 MicroMessenger/8.0.56.2800(0x2800385E) WeChat/arm64 Weixin NetType/WIFI Language/zh_CN ABI/arm64 MiniProgramEnv/android",
"Accept-Encoding": "gzip, deflate, br"
}
try:
response = requests.post(url, headers=headers)
if response.status_code == 200:
result = response.json()
if result.get("success", False):
print("✅ 签到成功!")
print(f"获得点数: {result['result']['point']}")
print(f"成长值: {result['result']['grow']}")
else:
print(f"❌ 签到失败,错误信息:{result.get('message', '未知错误')}")
else:
print(f"❌ 请求失败,状态码:{response.status_code}")
except Exception as e:
print(f"❌ 请求错误: {e}")
if __name__ == "__main__":
# 获取公告
#get_proclamation()
# 从环境变量读取 token
tokens = os.getenv("BREO", "").splitlines()
if not tokens:
print("❌ 未检测到 账号信息,退出脚本。")
else:
print("=============== 开始执行任务 ===============")
for i, token in enumerate(tokens, 1):
if token.strip(): # 跳过空行
print(f"\n-------------- 账号 {i} 开始 --------------")
print("🚀 正在签到...")
punch_in(token)
print("\n📝 正在发布帖子...")
post_id = post_to_breo(token, "这是一个自动发布的帖子", "自动化测试")
if post_id:
print("\n⭐ 正在收藏帖子...")
collect_post(token, post_id)
print("\n💬 正在评论帖子...")
comment_post(token, post_id)
else:
print("❌ 发帖失败,跳过后续操作。")
print("\n🛒 正在浏览商城...")
browse_mall(token)
print(f"-------------- 账号 {i} 结束 --------------")
print("\n=============== 所有任务执行完毕 ===============")

102
daily/STOKKE.py Normal file
View File

@ -0,0 +1,102 @@
#抓包www.stokkeshop.cn域名下的Authori-zation值填到环境变量STOKKE中多账号用&分割
#入口:http://mx.qrurl.net/h5/wxa/link?sid=26407QRIfeH
# new Env("stokkes小程序签到")
# cron 1 7 * * *
import os
import re
import requests
from typing import List, Dict, Tuple
# ---------- 通用请求头 ----------
_BASE_HEADERS = {
"Host": "www.stokkeshop.cn",
"Connection": "keep-alive",
"content-type": "application/json",
"charset": "utf-8",
"Referer": "https://servicewechat.com/wxe232c36aaca3dc1a/34/page-frame.html",
"Accept-Encoding": "gzip, deflate, br"
}
# ---------- 获取用户信息 ----------
def _get_user_info(token: str) -> Dict[str, str]:
url = "https://www.stokkeshop.cn/api/front/user"
headers = _BASE_HEADERS.copy()
headers["Authori-zation"] = token
try:
resp = requests.get(url, headers=headers, timeout=10)
resp.raise_for_status()
data = resp.json()
if data.get("code") != 200:
return {}
d = data["data"]
raw_phone = d["phone"]
masked = re.sub(r"(\d{3})\d{4}(\d{4})", r"\1****\2", raw_phone)
return {"nick": d["nickname"], "phone": masked, "integral": str(d["integral"])}
except Exception:
return {}
# ---------- 签到 ----------
def week_sign_all() -> None:
tokens = os.getenv("STOKKE", "").split("&")
if not tokens or tokens == [""]:
print("⚠️ 环境变量 STOKKE 为空,未配置任何 账号信息。")
return
url = "https://www.stokkeshop.cn/api/front/integral-task/finishWeekSign"
payload = {}
for token in tokens:
token = token.strip()
if not token:
continue
user = _get_user_info(token)
if not user:
print("❌ 获取用户信息失败,跳过当前账号。")
print("=" * 45)
continue
headers = _BASE_HEADERS.copy()
headers["Authori-zation"] = token
try:
resp = requests.post(url, json=payload, headers=headers, timeout=10)
resp.raise_for_status()
data = resp.json()
ok = data.get("code") == 200
print(
f"{'' if ok else ''} "
f"{user['nick']}({user['phone']}) 签到{'成功' if ok else '失败'}"
f"当前积分 {user['integral']}"
)
print("=" * 45)
except Exception as e:
print(
f"❗ 网络异常:{user['nick']}({user['phone']}) 签到失败,当前积分 {user['integral']} "
f"({e})"
)
# -------------------- 公告 --------------------
def get_proclamation():
primary_url = "https://github.com/3288588344/toulu/raw/refs/heads/main/tl.txt"
backup_url = "https://tfapi.cn/TL/tl.json"
for url in (primary_url, backup_url):
try:
r = requests.get(url, timeout=10)
if r.status_code == 200:
print("📢 公告信息")
print("=" * 45)
print(r.text)
print("=" * 45 + "\n")
return
except Exception as e:
continue
print("⚠️ 获取公告失败,跳过公告直接执行签到...\n")
# ---------- 示例 ----------
if __name__ == "__main__":
#get_proclamation()
week_sign_all()

103
daily/nwdjg.py Normal file
View File

@ -0,0 +1,103 @@
#入口:https://i.postimg.cc/7YjhgtCH/mmexport1742029346894.jpg
#抓包任意域名下的authorization注意不要开头的Bearer支持多账号
#环境变量名:NWDJG
# new Env("浓五的酒馆")
# cron 8 8 * * *
#by:哆啦A梦
import requests
import os
tokens = os.getenv("NWDJG",'')
if not tokens:
raise ValueError("环境变量 NWDJG 未设置,请确保已正确配置环境变量。")
token_list = tokens.split("&")
headers_template = {
'xweb_xhr': '1',
'content-type': 'application/json',
'sec-fetch-site': 'cross-site',
'sec-fetch-mode': 'cors',
'sec-fetch-dest': 'empty',
'accept-language': 'zh-CN,zh;q=0.9',
}
sign_url = 'https://stdcrm.dtmiller.com/scrm-promotion-service/promotion/sign/today'
sign_params = {
'promotionId': 'PI69cb44d522cc56000aa80bbe',
}
points_url = 'https://stdcrm.dtmiller.com/scrm-promotion-service/mini/wly/user/info'
def get_proclamation():
external_url = "https://github.com/3288588344/toulu/raw/refs/heads/main/tl.txt"
try:
response = requests.get(external_url)
if response.status_code == 200:
print(response.text)
print("公告获取成功,开始执行任务...")
else:
print(f"获取公告失败,状态码: {response.status_code}")
except requests.exceptions.RequestException as e:
print(f"获取公告时发生错误: {e}")
#get_proclamation()
total_accounts = len(token_list)
success_count = 0
fail_count = 0
# 遍历每个账号,依次执行签到和查询积分任务
for index, token in enumerate(token_list, start=1):
print(f"\n正在处理账号 {index}/{total_accounts}...")
headers = headers_template.copy()
headers['authorization'] = f"Bearer {token}"
points_response = requests.get(points_url, headers=headers)
if points_response.status_code == 200:
points_data = points_response.json()
if points_data['code'] == 0:
mobile = points_data['data']['member']['mobile']
account_name = f"{mobile[:3]}*****{mobile[-3:]}"
print(f"账号:{account_name}")
else:
print(f"查询积分失败,{points_data['msg']}")
account_name = "未知"
else:
print(f"查询积分网络请求异常,状态码:{points_response.status_code},响应内容:{points_response.text}")
account_name = "未知"
print("正在尝试签到...")
sign_response = requests.get(sign_url, params=sign_params, headers=headers)
if sign_response.status_code == 200:
sign_data = sign_response.json()
if sign_data['code'] == 0:
print("签到成功!")
print(f"签到天数:{sign_data['data']['signDays']}")
print(f"是否已领取奖励:{sign_data['data']['isReceive']}")
print(f"奖励信息:{sign_data['data']['prize']['goodsName']},积分:{sign_data['data']['prize']['prizePoints']}")
success_count += 1
else:
print(f"签到失败,{sign_data['msg']}")
fail_count += 1
else:
print(f"签到网络请求异常,状态码:{sign_response.status_code},响应内容:{sign_response.text}")
fail_count += 1
print(f"\n任务完成!")
print(f"总账号数量:{total_accounts}")
print(f"签到成功账号:{success_count}")
print(f"签到失败账号:{fail_count}")