256 lines
7.6 KiB
Python
256 lines
7.6 KiB
Python
import os
|
||
import json
|
||
import time
|
||
import datetime
|
||
import requests
|
||
import schedule
|
||
import pymysql.cursors
|
||
|
||
HEADERS = {}
|
||
TOKEN = ""
|
||
if os.path.exists('./accounts.json'):
|
||
with open('./accounts.json', 'r', encoding="utf-8") as f:
|
||
account_json = json.loads(f.read())
|
||
else:
|
||
account_json = {
|
||
"ECS": {
|
||
"ip": "localhost", # 自己输入
|
||
"user": "user",
|
||
"password": "password",
|
||
"db": "user",
|
||
}
|
||
}
|
||
|
||
db_config = {
|
||
'host': account_json["ECS"]["ip"],
|
||
'user': account_json["ECS"]["user"],
|
||
'password': account_json["ECS"]["password"],
|
||
'db': account_json["ECS"]["db"],
|
||
'charset': 'utf8mb4',
|
||
'cursorclass': pymysql.cursors.DictCursor
|
||
}
|
||
def input_headers() -> dict:
|
||
"""
|
||
输入curl命令，返回headers
|
||
:return: headers
|
||
"""
|
||
headers = {}
|
||
print("请输入请求的bash，按下Ctrl+Z或者输入quit()并回车结束输入：")
|
||
curl_command = []
|
||
while True:
|
||
try:
|
||
line = input()
|
||
if line == "quit()":
|
||
break
|
||
curl_command.append(line)
|
||
except EOFError:
|
||
break
|
||
curl_command = ''.join(curl_command)
|
||
parts = curl_command.split('-H ')
|
||
for i in range(len(parts)):
|
||
if "curl" in parts[i]:
|
||
pass
|
||
else:
|
||
tmp = parts[i].replace(" ", "").replace("\n", "")
|
||
name = tmp.split(":")[0][1:]
|
||
value = tmp.split(":")[1][:-1].replace("\\", "").replace("'", "")
|
||
headers[name] = value
|
||
headers["origin"] = "https://linux.do"
|
||
headers["sec-ch-ua"] = '"Chromium";v="122", "Not(A:Brand";v="24", "Google Chrome";v="122"'
|
||
headers["sec-ch-ua-platform"] = '"Windows"'
|
||
csrf_token = ""
|
||
for key in headers.keys():
|
||
if key.lower() == "x-csrf-token":
|
||
if headers[key] == "undefined":
|
||
print()
|
||
print("尚未登录！！")
|
||
headers = input_headers()
|
||
else:
|
||
csrf_token = headers[key]
|
||
break
|
||
return {"headers": headers, "csrf_token": csrf_token}
|
||
|
||
|
||
def try_cookies(headers):
|
||
response = requests.get('https://linux.do/u/dengdai/invited/pending', headers=headers)
|
||
if response.status_code == 404:
|
||
return False
|
||
else:
|
||
return True
|
||
|
||
|
||
def gen_new_invite(_headers: dict) -> dict:
|
||
"""
|
||
生成新的邀请
|
||
:param _headers: 请求头，必须包含 x-csrf-token
|
||
:return: response
|
||
"""
|
||
now = datetime.datetime.now()
|
||
data = {
|
||
'max_redemptions_allowed': '10',
|
||
'expires_at': f'{now.year + 1}-{now.month:02d}-{now.day} 00:00+08:00',
|
||
}
|
||
response = requests.post('https://linux.do/invites', headers=_headers, data=data)
|
||
return response.json()
|
||
|
||
|
||
def change_invite(_headers: dict, id_: int) -> dict:
|
||
"""
|
||
邀请链接信息更改
|
||
:param _headers: 请求头，必须包含 x-csrf-token
|
||
:param id_: 邀请链接id
|
||
:return:
|
||
"""
|
||
now = datetime.datetime.now()
|
||
data = {
|
||
'max_redemptions_allowed': '10',
|
||
'expires_at': f'{now.year + 1}-{now.month:02d}-{now.day} 00:00+08:00',
|
||
}
|
||
|
||
response = requests.put(f'https://linux.do/invites/{id_}', headers=_headers, data=data)
|
||
return response.json()
|
||
|
||
|
||
def select_mysql(_headers: dict) -> dict:
|
||
"""
|
||
从mysql获取邀请链接，并判断是否可用
|
||
:param _headers:
|
||
:return:
|
||
"""
|
||
try:
|
||
connection = pymysql.connect(**db_config)
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
sql = "SELECT * FROM links"
|
||
cursor.execute(sql)
|
||
result = cursor.fetchall()
|
||
for res in result:
|
||
print(test_invite(res["link"], _headers))
|
||
|
||
except pymysql.MySQLError as e:
|
||
print("MySQL连接错误！", e)
|
||
return {"status": 502, "message": "MySQL连接错误！"}
|
||
finally:
|
||
connection.close()
|
||
return {"status": 200, "message": f"链接测试完毕！"}
|
||
except Exception as e:
|
||
print("连接失败！\n", e)
|
||
return {"status": 403, "message": "MySQL连接失败！"}
|
||
|
||
|
||
def insert_mysql(invite_link: str) -> dict:
|
||
"""
|
||
邀请链接插入sql
|
||
:param invite_link: 邀请链接
|
||
:return: 成功与否
|
||
"""
|
||
try:
|
||
connection = pymysql.connect(**db_config)
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
sql = "INSERT INTO `links` (link) values (%s)"
|
||
cursor.execute(sql, invite_link)
|
||
connection.commit()
|
||
except pymysql.MySQLError as e:
|
||
print("MySQL连接错误！", e)
|
||
return {"status": 502, "message": "MySQL连接错误！"}
|
||
finally:
|
||
connection.close()
|
||
return {"status": 200, "message": f"链接 {invite_link} 已上传！"}
|
||
except Exception as e:
|
||
print("连接失败！\n", e)
|
||
return {"status": 403, "message": "MySQL连接失败！"}
|
||
|
||
|
||
def del_mysql(invite_link: str) -> dict:
|
||
"""
|
||
删除失效链接
|
||
:param invite_link: 邀请链接
|
||
:return: 成功与否
|
||
"""
|
||
try:
|
||
connection = pymysql.connect(**db_config)
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
sql = "DELETE FROM links where link = %s"
|
||
cursor.execute(sql, invite_link)
|
||
connection.commit()
|
||
except pymysql.MySQLError as e:
|
||
print("MySQL连接错误！", e)
|
||
return {"status": 502, "message": "MySQL连接错误！"}
|
||
finally:
|
||
connection.close()
|
||
return {"status": 200, "message": f"链接 {invite_link} 已删除！"}
|
||
except Exception as e:
|
||
print("连接失败！\n", e)
|
||
return {"status": 403, "message": "MySQL连接失败！"}
|
||
|
||
|
||
def test_invite(invite_link: str, _headers: dict) -> dict:
|
||
"""
|
||
邀请链接测试
|
||
:param invite_link: 邀请链接
|
||
:param _headers: 请求头，必须包含 x-csrf-token
|
||
:return:
|
||
"""
|
||
response = requests.get(invite_link)
|
||
if "invite_info" in response.text:
|
||
return {"status": 200, "message": f"链接 {invite_link} 可用！"}
|
||
else:
|
||
return delete_invite(invite_link, _headers)
|
||
|
||
|
||
def delete_invite(invite_link: str, _headers: dict) -> dict:
|
||
"""
|
||
删除邀请
|
||
:param invite_link: 邀请链接
|
||
:param _headers: 请求头，必须包含 x-csrf-token
|
||
:return:
|
||
"""
|
||
# 删除邀请
|
||
print(del_mysql(invite_link))
|
||
# 获取邀请
|
||
invite_new = gen_new_invite(_headers)
|
||
# 插入邀请
|
||
print(insert_mysql(invite_new["link"]))
|
||
return {"status": 200, "message": f"邀请码：{invite_link} 已删除，并增添新的 {invite_new['link']}！"}
|
||
|
||
|
||
def main():
|
||
global HEADERS, TOKEN
|
||
print("开始测试！")
|
||
if try_cookies(HEADERS):
|
||
print(select_mysql(HEADERS))
|
||
else:
|
||
# requests.post(f'https://iyuu.cn/IYUU20053Tcb524cc74f1cbdc348e75a25124525282bd78bd6.send?text=Input Headers!')
|
||
print("请重新输入headers！")
|
||
headers = input_headers()
|
||
HEADERS = headers["headers"]
|
||
TOKEN = headers_["csrf_token"]
|
||
main()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
headers_ = input_headers()
|
||
HEADERS = headers_["headers"]
|
||
TOKEN = headers_["csrf_token"]
|
||
main()
|
||
# 整点运行
|
||
schedule.every().hour.at(":00").do(main)
|
||
while True:
|
||
schedule.run_pending()
|
||
time.sleep(1)
|
||
# ok https://linux.do/invites/8bRLymaDhL
|
||
# 过期 https://linux.do/invites/wwPLXVwYgT
|
||
# 失效 https://linux.do/invites/xggGwUazzE
|
||
|
||
"""
|
||
无限邀请1：
|
||
`https://linuxdo-invites.speedcow.top/`
|
||
无限邀请2：
|
||
`https://invite.fc88.xyz/`
|
||
无限邀请3：
|
||
`https://l.briqt.dev/`
|
||
参见：https://linux.do/t/topic/36752?u=dengdai
|
||
"""
|