import json import logging import sys from loguru import logger import time import requests import toml from bs4 import BeautifulSoup from main_notify import pushme, pushplus_bot ''' 0 0 15 * * ? ''' class PtOperation: def __init__(self): logger.add("../log/PtOperation_{time:YYYY-MM-DD}.log", rotation="1 day", level="INFO") logger.add(sys.stderr, level="INFO") self.toml_file = 'PT/pt_config.toml' self.headers = { '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,zh-CN;q=0.9', 'cache-control': 'max-age=0', 'sec-ch-ua': '"Google Chrome";v="119", "Chromium";v="119", "Not?A_Brand";v="24"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"macOS"', 'sec-fetch-dest': 'document', 'sec-fetch-mode': 'navigate', 'sec-fetch-site': 'same-origin', 'sec-fetch-user': '?1', 'upgrade-insecure-requests': '1', 'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36', } async def attendances(self, section_data): """ 签到 :param cookie: :param url: :return: """ section_name, url, cookie, attendance_uri = section_data.get('name'), \ section_data.get('url'), \ section_data.get('cookie'), \ section_data.get('attendance_uri') res_text, log_txt = "", "" self.headers["cookie"] = cookie # cookie不为空时候，可以签到 if cookie is not None and len(cookie.strip()) > 0: # 获取网站自定义的签到链接 attendance_uri = section_data.get('attendance_uri') if attendance_uri is not None and len(attendance_uri.strip()) > 0: url = url + attendance_uri else: url = url + "/attendance.php" self.headers["referer"] = url for i in range(5): try: response_result = requests.get(url, headers=self.headers) # print(response_result.status_code) # print(self.headers) # print("=" * 20) # print(response_result.text) if response_result.status_code == 200 or response_result.text in "签到成功": res_text = '签到成功！' break else: res_text = f"签到失败，网站请求结果 {response_result.status_code}" break except Exception: time.sleep(2) res_text = f"{i + 1} 次出现错误，请关注！！！" finally: log_txt = f"开始对 [{section_name}] 进行签到操作! 请求连接为：[{url}]，签到结果：[{res_text}]" logger.info(log_txt) else: # log_txt = f"开始对 [{section_name}] 进行签到操作! 请求连接为：[{url}]，签到结果：[{res_text}]" # logger.error(log_txt) pushme("签到错误提示：" + section_name, log_txt) pushplus_bot("签到错误提示：" + section_name, log_txt) self.headers["cookie"] = "" async def signup(self, section_data): """ 是否开注册 :return: """ section_name, url = section_data.get('name'), section_data.get('url') request_url = url + "/signup.php" text = f"网站名：{section_name}, 网址：{request_url}" logger.info(f"开始 -->> {text}") html = self.request_get(request_url, text) if len(html) == 0: return try: # 打印结果 if "再次输入密码" in html and "submit" in html: remind_me = f"现在 {text} 正在开放注册，请前往~！" pushme("开注提醒" + section_name, remind_me) pushplus_bot("开注提醒" + section_name, remind_me) logger.info(remind_me) elif "对不起" in html or ("关闭" in html and "自由注册" in html) or "维护" in html: remind_me = f"现在 {text} 关闭注册，请知晓~！" logger.info(remind_me) else: soup = BeautifulSoup(html, 'html.parser') text_content = soup.find(class_='text').get_text() print(text_content) except Exception: logger.error(f"{text} , 页面无法解析，请知晓！！！") logger.info("=" * 100) def request_get(self, url, text): self.headers["Referer"] = url for _ in range(5): try: response = requests.get(url, headers=self.headers, timeout=5 * 60) response.encoding = "utf-8" if response.status_code == 403 or response.text in "Just a moment": return self.flaresolverr_get(url, text) elif response.status_code == 200: return response.text else: logger.error(f"{text} , 出现错误，code码是：{response.status_code}！！！") return "" except Exception: time.sleep(2) else: logger.error(f"{text} , 5次出现错误，无法访问！！！") self.headers["Referer"] = "" return "" def flaresolverr_get(self, url, text): # 5 * 1000 * 60 代表 5分钟 for _ in range(5): try: flaresolverr_url = "http://152.136.50.100:7024/v1" payload = json.dumps({ "cmd": "request.get", "url": url, "maxTimeout": 5 * 1000 * 60 }) headers = { 'Content-Type': 'application/json' } response = requests.post(flaresolverr_url, headers=headers, data=payload) res = json.loads(response.text) if res['status'] == 'ok' and res['solution']['status'] == 200: logging.info(f"最终耗时：{(res['endTimestamp'] - res['startTimestamp']) / 1000 / 60:.2f} 分钟") return res['solution']['response'] elif res['status'] == 'error': logger.error(f"{text} , 访问返回 {res['message']} ！！！") return "" except Exception: time.sleep(2) else: logger.error(f"{text} , 5次出现错误，无法访问！！！") return "" def opt(self): try: with open(self.toml_file, 'r', encoding='utf-8') as file: config_data = toml.load(file) # 迭代每个 section for section_name, section_data in config_data.items(): url, cookie, flag = section_data.get('url'), section_data.get('cookie'), section_data.get('flag') if flag != 1: print(f"Processing section: {section_name} --- {section_data.get('url')}") # 签到 self.attendances(section_data) # 检测是否可以注册 self.signup(section_data) except FileNotFoundError: print(f"Error: The file '{self.toml_file}' was not found.") except toml.TomlDecodeError as e: print(f"Error decoding TOML: {e}")