yuedu/叮叮联盟_api_1.6.py
2024-11-14 01:23:08 +08:00

493 lines
19 KiB
Python

# --------------------------------注释区--------------------------------
# 入口:http://nmub7w.jlnrj.cn?mid=1786918&r=OKgDUXIa&e=1730731520
#
# 需抓取数据:
# * 填写穿透出来的api接口
#
# 抓取请求头中的user-agent填入yuanshen_useragent 无论多少个号都只填一个即可!!!!
#
# 变量名:yuanshen_api
#
# --------------------------------祈求区--------------------------------
# _ooOoo_
# o8888888o
# 88" . "88
# (| -_- |)
# O\ = /O
# ____/`---'\____
# . ' \\| |// `.
# / \\||| : |||// \
# / _||||| -:- |||||- \
# | | \\\ - /// | |
# | \_| ''\---/'' | |
# \ .-\__ `-` ___/-. /
# ___`. .' /--.--\ `. . __
# ."" '< `.___\_<|>_/___.' >'"".
# | | : `- \`.;`\ _ /`;.`/ - ` : | |
# \ \ `-. \_ __\ /__ _/ .-` / /
# ======`-.____`-.___\_____/___.-`____.-'======
# `=---='
#
# .............................................
# 佛祖保佑 永无BUG
# 佛祖镇楼 BUG辟邪
# --------------------------------代码区--------------------------------
import requests
import time
import os
import json
import hashlib
import random
import logging
from urllib.parse import urlparse, parse_qs,quote,unquote
import sys
import string
from base64 import b64encode
import uuid
from datetime import datetime,timedelta
import re
logging.basicConfig(level=logging.INFO, format='[%(levelname)s] ===> %(message)s')
code = "叮叮联盟_api版"
ver = "1.6"
envname = "yuanshen_api"
debug = False #debug模式 开启即从脚本内部获取环境变量
debugcookie = "" #debug模式cookie
is_bulletin = False #公告开关
is_toulu = False #偷撸公告开关
is_with_sleep = False #是否开启随机延时
def get_sm():
global bizlist
try:
url = "https://api.huaji.asia/read_api/biz_ddlm.txt"
r = requests.get(url)
if r.status_code == 200:
logging.info(f"联网获取检测文章列表成功")
bizlist = r.text.split(",")
else:
logging.error("联网获取检测文章列表失败!")
bizlist = []
except:
logging.error("联网获取检测文章列表失败!")
bizlist = []
class env():
"""
env模块,获取cookie并转成列表,统计时间,提示脚本开始结束,检测一些配置,try except异常处理
:param args: 传递给env的参数
:param kwargs: 传递给env的关键字参数
:return: Null
Powered by huaji
"""
def __init__(self, *args, **kwargs):
self.cookie = None
self.env_ver = '1.5' #版本号
self.split_chars = ['@', '&', '\n'] #分隔符
self.identifiers = ['Powered By Huaji', 'QQ Group:901898186', 'yuanshen'] #标识符
def check_file(self):
file_path = __file__
try:
with open(file_path, 'r', encoding='utf-8') as file:
content = file.read()
results = {}
# 批量检查每个标识符
for identifier in self.identifiers:
if identifier in content:
results[identifier] = True
else:
results[identifier] = False
logging.error(f"文件可能被恶意篡改,请勿修改文件内容")
all_identifiers_present = all(results.values())
if not all_identifiers_present:
logging.error(f"文件可能被恶意篡改,请勿修改文件内容")
self.force_exit()
except Exception as e:
print(f"读取文件时发生错误: {e}")
self.force_exit()
def split_cookies(self):
"""根据多个分隔符分割cookie"""
for sep in self.split_chars:
if sep in self.cookie:
return self.cookie.split(sep)
return [self.cookie]
def scmain(self):
get_sm()
apiurl = 'http://' + self.cookies[0]
r = requests.get(apiurl + '/getallwx').json()
for i, cookie in enumerate(r, 1):
print(f"--------开始第{i}个账号--------")
main = yuanshen(cookie,apiurl)
main.main()
print(f"--------第{i}个账号执行完毕--------")
def force_exit(self,code=0):
exit()
print("Warning: 篡改你妈")
os._exit(code) # 强制退出程序
sys.exit(code) # 正常退出
import ctypes
while True:
ctypes.string_at(114514)
ctypes.string_at(1919810)
ctypes.string_at(666666)
print("Warning: 篡改你妈")
def run(self):
if not os.getenv(envname) and not debug:
logging.warning(f"请先设置环境变量[{envname}]")
self.force_exit()
self.cookie = os.getenv(envname, "")
if debug:
self.cookie = debugcookie
if is_bulletin:
try:
print(requests.get("https://gitee.com/HuaJiB/yuanshen34/raw/master/pubilc.txt").text, "\n\n\n")
except:
logging.error("网络异常,链接公告服务器失败(gitee),请检查网络")
self.force_exit()
if is_toulu:
try:
txt = '''
此为滑稽的偷撸本本 如你不在滑稽的小群却意外通过某种渠道获得了该脚本
请联系QQ3487934983 提供证据后 你将代替泄露人员获得该群位置
=======================================================
'''
print(txt*5)
except:
self.force_exit()
if is_with_sleep:
random_time = random.randint(10,60)
logging.info(f"随机延时[{random_time}]秒")
time.sleep(random_time)
self.cookies = self.split_cookies()
account_count = len(self.cookies)
logging.info(f"一共获取到{account_count}个账号")
print(f"=========🔔开始执行[{code}][{ver}]=========\n")
start_time = time.time()
if debug:
self.scmain()
else:
try:
self.scmain()
except Exception as e:
logging.error(f"脚本执行出错: {e}")
end_time = time.time()
execution_time = end_time - start_time
print(f"\n============🔔脚本[{code}]执行结束============")
print(f"本次脚本总运行时间: [{execution_time:.2f}] 秒")
self.force_exit()
def main(self):
self.check_file()
self.run()
def random_str(self,charset="all", length=8, to_upper=False,to_lower=False):
if charset == "all": #包含大小写字母和数字
chars = string.ascii_letters + string.digits
elif charset == "letters": #大小写字母
chars = string.ascii_letters
elif charset == "digits":
chars = string.digits
elif charset == "lowercase": #小写字母
chars = string.ascii_lowercase
elif charset == "uuid":
return str(uuid.uuid4())
elif charset == "uuid_str":
return str(uuid.uuid4()).replace("-", "")
elif charset != '' and charset is not None:
chars = charset
random_string = ''.join(random.choice(chars) for _ in range(length))
if to_upper:
return random_string.upper()
elif to_lower:
return random_string.lower()
else:
return random_string
def aes_encrypt(self,data):
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from binascii import hexlify, unhexlify
key = self.key.encode('utf-8')
iv = self.iv.encode('utf-8')
cipher = AES.new(key, AES.MODE_CBC, iv)
padded_plaintext = pad(data.encode('utf-8'), AES.block_size)
ciphertext = cipher.encrypt(padded_plaintext)
encrypted_base64 = b64encode(ciphertext).decode('utf-8')
return encrypted_base64
class yuanshen:
def __init__(self,cookie,apiurl) -> None:
self.apiurl = apiurl
self.Wxid = cookie['Wxid']
self.bz = cookie['wxname']
logging.info(f'[{self.bz}]开始运行')
def login(self):
headers = {
'User-Agent': ua,
# 'Accept-Encoding': 'gzip, deflate, br, zstd',
'Content-Type': 'application/json',
'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="126", "Android WebView";v="126"',
'sec-ch-ua-mobile': '?1',
'Authorization': 'Bearer',
'F-RTYFGHVBN': '0cb5cc12b50dddc84c3066433e88194f',
'sec-ch-ua-platform': '"Android"',
'Origin': 'http://bsdspbkbbt.r8it3o.cn',
'X-Requested-With': 'com.tencent.mm',
'Sec-Fetch-Site': 'cross-site',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Dest': 'empty',
'Referer': 'http://bsdspbkbbt.r8it3o.cn/',
'Accept-Language': 'zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7',
}
url = 'https://api.jiudingliliang.com/login/get-auth-url'
data = {"route":"","mid":"1786918"}
r = requests.post(url,json=data,headers=headers).json()
if r['code'] == 10000:
url = r['result']['url']
data = {'Wxid':self.Wxid,'url':url}
url = requests.post(self.apiurl+'/loginbyweb',json=data).json()['url']
#print(url)
h = {
'User-Agent': ua,
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/wxpic,image/tpg,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
# 'Accept-Encoding': 'gzip, deflate, br, zstd',
'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="126", "Android WebView";v="126"',
'sec-ch-ua-mobile': '?1',
'sec-ch-ua-platform': '"Android"',
'upgrade-insecure-requests': '1',
'x-requested-with': 'com.tencent.mm',
'sec-fetch-site': 'none',
'sec-fetch-mode': 'navigate',
'sec-fetch-user': '?1',
'sec-fetch-dest': 'document',
'accept-language': 'zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7',
'priority': 'u=0, i',
}
r = requests.get(url,headers=h,allow_redirects=False)
#print(r.headers)
token = r.headers['Location'].split('token=')[1]
self.h = {
'User-Agent': ua,
# 'Accept-Encoding': 'gzip, deflate, br, zstd',
'Content-Type': 'application/json',
'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="126", "Android WebView";v="126"',
'sec-ch-ua-mobile': '?1',
'Authorization': f'Bearer {token}',
'X-NIUBILITY-Y': '',
'F-RTYFGHVBN': env().random_str(length=32,to_lower=True),
'sec-ch-ua-platform': '"Android"',
'Origin': 'http://rdecfgw.cecua.cn',
'X-Requested-With': 'com.tencent.mm',
'Sec-Fetch-Site': 'cross-site',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Dest': 'empty',
'Referer': 'http://rdecfgw.cecua.cn/',
'Accept-Language': 'zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7',
}
r = requests.post('https://api.jiudingliliang.com/user/info',json={},headers=self.h).json()
if r['code'] == 10000:
logging.info(f'登录成功')
self.h['X-NIUBILITY-Y'] = r["result"]["uok"]
logging.info(f'登录成功')
else:
logging.info(f'获取阅读入口失败[{r}]')
def get_readtime(self,url):
h = {
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
'accept-language': 'zh-CN,zh;q=0.9',
'cache-control': 'no-cache',
# 'cookie': 'RK=x93V1EQnnk; ptcz=70105284a73b47db6aa5f48a05337a8cd84e98de7bd26d75975059126646db21; pac_uid=0_ZkHRNfy6PMhrH; ua_id=SyF8Nc7qjhUZ6RH8AAAAAHO013VXlTF21GY7ajzLLgs=; _clck=19reoir|1|fpg|0; wxuin=27186344812218; eas_sid=x1Y7v2y759E6j1a811G6q158d0; qq_domain_video_guid_verify=08d525e64ffa56c3; _qimei_uuid42=18a1b130e12100363fb639c628b466967c0e1ac0aa; pgv_pvid=6915337851; _qimei_fingerprint=3c42962857167cc55475bb65b2331f0b; _qimei_h38=fded42463fb639c628b4669602000006018a1b; o_cookie=3487934983; _qimei_q32=ec50b72a0f5a646accacf4393f1271f2; _qimei_q36=9e46147a2ab8bee5904ab74e30001741890c; rewardsn=; wxtokenkey=777',
'pragma': 'no-cache',
'priority': 'u=0, i',
'sec-ch-ua': '"Chromium";v="130", "Google Chrome";v="130", "Not?A_Brand";v="99"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'none',
'sec-fetch-user': '?1',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36',
}
r = requests.get(url,headers=h).text
match = re.search(r"var createTime = '(.+?)';", r)
if match:
date_string = (match.group(1))
date = datetime.strptime(date_string, "%Y-%m-%d %H:%M")
half_year_ago = datetime.now() - timedelta(days=30)
if date > half_year_ago:
return False,date_string
else:
return True,date_string
def getread(self):
url = 'https://api.jiudingliliang.com/task/read-batch'
data = {"u": 1}
r = requests.post(url,json=data,headers=self.h).json()
if r['code'] == 10000:
logging.info(f'获取阅读入口成功')
print(r)
#http://agifh.liliangda.cn?route=/pages/read/read&bn=2409211329003132289067&token=p5wP0pTb9eMHHJjf1fy133qF8tiaE0_k7jJzJRBtBXc&k=15G8EuNiZKx4b9W9SjgmmVJL60Tm0e_SNCQtW5IVdFdpYVsSHoL_UHd9btHBIunG
j = urlparse(r["result"]["qrcode"]);
self.domain, p = j.netloc, parse_qs(j.query);
self.k = p.get('k')[0] if p.get('k') else ''
self.readh = {
"Host": "api.jiudingliliang.com",
"Connection": "keep-alive",
"Content-Length": "28",
"sec-ch-ua": "\"Not/A)Brand\";v=\"8\", \"Chromium\";v=\"126\", \"Android WebView\";v=\"126\"",
"sec-ch-ua-mobile": "?1",
"User-Agent":ua,
"Authorization": "Bearer",
"Content-Type": "application/json",
"X-NIUBILITY-Y": "",
"F-RTYFGHVBN":self.h['F-RTYFGHVBN'],
"sec-ch-ua-platform": "\"Android\"",
"Accept": "*/*",
"Origin": f"http://{self.domain}",
"X-Requested-With": "com.tencent.mm",
"Sec-Fetch-Site": "cross-site",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Dest": "empty",
"Referer": f"http://{self.domain}/",
"Accept-Encoding": "gzip, deflate, br, zstd",
"Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7"
}
return True
else:
logging.error(f'获取阅读入口失败[{r}]')
return False
def tuisong(self):
# 发送消息到wxpusher
url = f"{self.apiurl}/zdgjc"
data = {"url":self.read_url}
r = requests.post(url,json=data).json()
logging.info(f"遇到检测文章:推送结果[{r}]")
def read(self):
#print(self.k)
data = {"k":self.k}
url = 'https://api.jiudingliliang.com/task/read-article'
r = requests.post(url,json=data,headers=self.readh).json()
if r['code'] == 10000:
self.read_url = r["result"]["articleUrl"]
va, date_string = self.get_readtime(self.read_url)
logging.info(f'获取文章成功[{self.read_url}],发布时间[{date_string}]')
j = urlparse(self.read_url.replace('\\',''))
biz = parse_qs(j.query).get('__biz', [''])[0] if '__biz' in parse_qs(j.query) else ''
if 'biz' not in self.read_url or biz in bizlist or va:
self.tuisong()
time.sleep(random.randint(20,25))
else:
time.sleep(random.randint(8,10))
k = r["result"]["k"]
else:
logging.error(f'获取文章失败[{r}]')
return False
url = 'https://api.jiudingliliang.com/task/read-article-completed'
data = {'k':k}
r = requests.post(url,json=data,headers=self.readh).json()
if r['code'] == 10000:
now = r["result"]["readingNumber"]
all = r["result"]["totalReading"]
logging.info(f'第[{now}/{all}]篇文章阅读成功')
if now == all:
logging.info("本轮阅读完成")
return False
self.k = r["result"]["k"]
return True
else:
logging.error(f'阅读失败[{r}]')
def userinfo(self):
url = 'https://api.jiudingliliang.com/user/info'
r = requests.post(url,json={},headers=self.h).json()
if r['code'] == 10000:
name = r["result"]["nickname"]
m = int(float(r["result"]["balanceCash"]))
logging.info(f'当前账号[{name}]余额[{m}]')
url = 'https://api.jiudingliliang.com/user/withdraw'
r = requests.post(url,headers=self.h,json={}).json()
# if r['code'] == 10000:
# lessmoney = r["result"]["transfer"][0]["m"]
# if m >= lessmoney:
# url = 'https://api.jiudingliliang.com/user/transfer'
# data = {'amount':lessmoney}
# r = requests.post(url,json=data,headers=self.h).json()
# if r['code'] == 10000:
# logging.info(f'提现成功')
# else:
# logging.error(f'提现失败[{r}]')
# else:
# logging.info(f'余额不足,跳过提现')
else:
logging.error(f'获取用户信息失败[{r}]')
def main(self):
self.login()
if self.getread():
print('='*30)
while self.read():
time.sleep(random.randint(0,1))
else:
logging.error('获取阅读入口失败')
self.userinfo()
if __name__ == '__main__':
ua = os.getenv("yuanshen_useragent")
#ua = 'Mozilla/5.0 (Linux; Android 14; 23113RKC6C Build/UKQ1.230804.001; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/126.0.6478.188 Mobile Safari/537.36 XWEB/1260183 MMWEBSDK/20240301 MMWEBID/98 MicroMessenger/8.0.48.2580(0x28003035) WeChat/arm64 Weixin NetType/WIFI Language/zh_CN ABI/arm64'
if not ua:
logging.info("不填ua玩你妹妹\n"*999)
exit()
env().main()