qnloft-spider/PT/pt_operation.py

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}")