mirror of https://github.com/yuxian158/check.git
张洛
3 years ago
5 changed files with 117 additions and 6 deletions
@ -0,0 +1,33 @@
@@ -0,0 +1,33 @@
|
||||
import requests |
||||
import os |
||||
|
||||
|
||||
def message2telegram(tg_api_host, tg_proxy, tg_bot_token, tg_user_id, content): |
||||
print("Telegram 推送开始") |
||||
send_data = {"chat_id": tg_user_id, "text": content, "disable_web_page_preview": "true"} |
||||
if tg_api_host: |
||||
url = f"https://{tg_api_host}/bot{tg_bot_token}/sendMessage" |
||||
else: |
||||
url = f"https://api.telegram.org/bot{tg_bot_token}/sendMessage" |
||||
if tg_proxy: |
||||
proxies = { |
||||
"http": tg_proxy, |
||||
"https": tg_proxy, |
||||
} |
||||
else: |
||||
proxies = None |
||||
if requests.post(url=url, data=send_data, proxies=proxies) == 200: |
||||
print("推送成功") |
||||
return 1 |
||||
else: |
||||
print("推送失败") |
||||
return 0 |
||||
|
||||
|
||||
def send(content): |
||||
tg_api_host = os.environ.get('TG_API_HOST') |
||||
tg_proxy = '' |
||||
tg_bot_token = os.environ.get('TG_BOT_TOKEN') |
||||
tg_user_id = os.environ.get('TG_USER_ID') |
||||
message2telegram(tg_api_host=tg_api_host, tg_proxy=tg_proxy, tg_bot_token=tg_bot_token, tg_user_id=tg_user_id, |
||||
content=content) |
@ -0,0 +1,25 @@
@@ -0,0 +1,25 @@
|
||||
const whitelist = ["/bot"]; |
||||
const tg_host = "api.telegram.org"; |
||||
addEventListener('fetch', event =>{ |
||||
event.respondWith(handleRequest(event.request)) |
||||
}) |
||||
function validate(path) { |
||||
for (var i = 0; i < whitelist.length; i++) { |
||||
if (path.startsWith(whitelist[i])) return true; |
||||
} |
||||
return false; |
||||
} |
||||
async function handleRequest(request) { |
||||
var u = new URL(request.url); |
||||
u.host = tg_host; |
||||
if (!validate(u.pathname)) return new Response('Unauthorized', { |
||||
status: 403 |
||||
}); |
||||
var req = new Request(u, { |
||||
method: request.method, |
||||
headers: request.headers, |
||||
body: request.body |
||||
}); |
||||
const result = await fetch(req); |
||||
return result; |
||||
} |
@ -0,0 +1,44 @@
@@ -0,0 +1,44 @@
|
||||
# -*- coding: utf-8 -*- |
||||
import json |
||||
import os |
||||
import time |
||||
from urllib import parse |
||||
|
||||
import requests |
||||
|
||||
|
||||
class WZYDCheckIn: |
||||
def __init__(self, check_item): |
||||
self.check_item = check_item |
||||
|
||||
@staticmethod |
||||
def sign(data): |
||||
response = requests.post(url="https://ssl.kohsocialapp.qq.com:10001/play/h5sign", data=data).json() |
||||
try: |
||||
if response["result"] == 0: |
||||
msg = "签到成功" |
||||
else: |
||||
msg = response["returnMsg"] |
||||
except: |
||||
msg = "请求失败,请检查接口" |
||||
return msg |
||||
|
||||
def main(self): |
||||
wzyd_data = self.check_item[0] |
||||
data = {k: v[0] for k, v in parse.parse_qs(wzyd_data).items()} |
||||
try: |
||||
user_id = data.get("userId", "") |
||||
except Exception as e: |
||||
print(f"获取用户信息失败: {e}") |
||||
user_id = "未获取到用户信息" |
||||
sign_msg = self.sign(data=data) |
||||
msg = f"帐号信息: {user_id}\n签到信息: {sign_msg}" |
||||
return msg |
||||
|
||||
|
||||
if __name__ == "__main__": |
||||
if 'wzry' in os.environ: |
||||
print('王者营地签到开始') |
||||
text = WZYDCheckIn(check_item=os.environ).main() |
||||
localtime = time.asctime( time.localtime(time.time()) ) |
||||
|
Loading…
Reference in new issue