211 lines
11 KiB
Python
211 lines
11 KiB
Python
import os
|
||
|
||
import requests
|
||
from pathlib import Path
|
||
from bs4 import BeautifulSoup
|
||
from app.extensions import db
|
||
from tenacity import retry, stop_after_attempt, RetryError
|
||
from app.utils import datetime_to_timestamp, timestamp_to_datetime, get_today_timestamp, find_project_root
|
||
from app.models import Setting
|
||
|
||
|
||
class BookService:
|
||
ALLOWED_CONTENT_TYPES = ['"application/octet-stream"', "application/octet-stream", "text/plain",
|
||
"application/zip"]
|
||
|
||
def __init__(self):
|
||
setting_record = Setting.query.filter_by(name="book_download").first()
|
||
if not setting_record:
|
||
raise RuntimeError("Setting 'book_download' not found in database")
|
||
self.setting = setting_record.to_dict()
|
||
|
||
# 目标日期减一天
|
||
target_ts = datetime_to_timestamp(self.setting["targetDate"]) - 24 * 60 * 60
|
||
self.setting["targetDate"] = timestamp_to_datetime(target_ts)
|
||
|
||
app_dir = find_project_root()
|
||
self.download_path = app_dir / "download" # 下载目录
|
||
print(self.download_path)
|
||
self.today_timestamp = get_today_timestamp()
|
||
self.one_date_loading = False
|
||
self.session = requests.session()
|
||
self.session.headers.update({
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36",
|
||
"cookie": "cPNj_2132_saltkey=CeJLs8Ed; "
|
||
"cPNj_2132_auth=44bdzFfx4TYjBMhuKf4AZdi2JM%2BXZ0Okxd0JEl1Hmwbz%2Fr3WiX0sLmATmolWkLlRgorFz%2BULizz5o6G%2FJ3YGWYPF0xw; "
|
||
})
|
||
self.session.timeout = 30
|
||
|
||
# ------------------------------------------------------------
|
||
# 数据库辅助方法
|
||
# ------------------------------------------------------------
|
||
def _update_setting(self, **kwargs):
|
||
Setting.query.filter_by(name="book_download").update(kwargs)
|
||
db.session.commit()
|
||
for key, value in kwargs.items():
|
||
self.setting[key] = value
|
||
|
||
# ------------------------------------------------------------
|
||
# 文件与目录处理
|
||
# ------------------------------------------------------------
|
||
@staticmethod
|
||
def legitimate_naming(name: str) -> str:
|
||
"""将文件名中的非法字符替换为合法字符"""
|
||
name = name.rstrip(" .")
|
||
replacements = {":": "：", "<": "《", ">": "》", "/": " ", "\\": " ", "?": "？"}
|
||
for old, new in replacements.items():
|
||
name = name.replace(old, new)
|
||
return name
|
||
|
||
@staticmethod
|
||
def create_folder(parent_dir: Path, folder_name: str) -> Path: # 创建文件夹
|
||
"""创建文件夹（如果不存在）并返回路径"""
|
||
target = parent_dir / folder_name
|
||
target.mkdir(parents=True, exist_ok=True)
|
||
return target
|
||
|
||
def post_page(self, name, url): # 帖子页面
|
||
full_url = f"{self.setting['baseUrl']}{url}"
|
||
try:
|
||
response = self.session.get(full_url)
|
||
response.raise_for_status()
|
||
except requests.RequestException as e:
|
||
yield f"event: error\ndata: 请求帖子详情失败 {url}: {str(e)}\n\n"
|
||
return
|
||
print(f"详情页面 {str(response.status_code)} {full_url}")
|
||
|
||
soup = BeautifulSoup(response.text, 'lxml')
|
||
resource_boxs = soup.select('ignore_js_op')
|
||
|
||
for i in resource_boxs:
|
||
download_dir_path = self.download_path / self.setting["targetDate"]
|
||
self.create_folder(download_dir_path, name)
|
||
try:
|
||
for string in i.select_one(".tip.tip_4").stripped_strings:
|
||
if "阅读权限: " in repr(string):
|
||
privilege_level = int(
|
||
repr(string).strip("'").strip('阅读权限: '))
|
||
if privilege_level > self.setting["privilegeLevel"]:
|
||
print("下载失败，权限等级不够")
|
||
return
|
||
yield from self.download_file(f"{self.setting['baseUrl']}{i.select_one('a')['href']}", name, i.select_one('a').string)
|
||
# yield from self.download_file(f"{self.setting["baseUrl"]}{i.select_one(
|
||
# 'a')['href']}", name, self.legitimate_naming(i.select_one('a').string))
|
||
except RetryError as e:
|
||
print('重试三次也不成功')
|
||
|
||
@retry(stop=stop_after_attempt(3))
|
||
def download_file(self, file_url: str, dir_name: str, file_name: str): # 下载文件
|
||
try:
|
||
response = self.session.get(file_url, stream=True, allow_redirects=False)
|
||
if response.status_code in (301, 302, 307, 308): # 非200状态码，重新请求
|
||
file_url = response.headers["location"]
|
||
print(f'跳转页面了：{file_url}')
|
||
response = self.session.get(file_url, headers={"referer": file_url}, stream=True)
|
||
print(f"下载状态码：{response.status_code} {file_url}")
|
||
print(dir_name + "/" + file_name)
|
||
content_type = response.headers.get('Content-Type', '')
|
||
if content_type not in self.ALLOWED_CONTENT_TYPES:
|
||
yield f"event: mes_error\ndata: {self.setting['targetDate']} {file_name}\n\n"
|
||
print("不是可下载文件")
|
||
return
|
||
target_dir = self.download_path / self.setting["targetDate"] / dir_name / file_name
|
||
print(target_dir)
|
||
with open(target_dir, 'wb') as f:
|
||
for chunk in response.iter_content(chunk_size=1024):
|
||
if chunk:
|
||
f.write(chunk)
|
||
yield f"data: {self.setting['targetDate']} {file_name}\n\n"
|
||
except requests.RequestException as e:
|
||
yield f"event: error\ndata: {self.setting['targetDate']} {file_name} 网络错误\n\n"
|
||
raise # 触发 tenacity 重试
|
||
|
||
def section_page(self, page): # 版块页面
|
||
print(f"当前页面 {page}")
|
||
target_date_timestamp = datetime_to_timestamp(self.setting["targetDate"]) # 目标时间戳
|
||
if target_date_timestamp >= self.today_timestamp: # 判断时间（是否大于当前时间戳）
|
||
# print("不允许下载当天的")
|
||
yield f"data: 不允许下载当天的\n\n"
|
||
return None
|
||
# yield f"data: 当前页面 {page}\n\n"
|
||
url = f"{self.setting['baseUrl']}forum.php?mod={self.setting['mod']}&fid={self.setting['fid']}&page={page}" # 版块页面地址
|
||
response = self.session.get(url) # 请求
|
||
response.raise_for_status()
|
||
soup = BeautifulSoup(response.text, 'lxml') # 解析
|
||
total_page_element = soup.select_one("a.bm_h") # 获取总页数元素
|
||
if total_page_element is not None: # 总页数元素存在 则更新总页数
|
||
new_total_page = int(total_page_element.get("totalpage")) # 获取总页数
|
||
if self.setting["totalPage"] != new_total_page: # 总页数有更新
|
||
self._update_setting(totalPage=new_total_page)
|
||
yield from self.section_page(new_total_page - self.setting["targetCountdownPage"]) # 重新跳转指点版块页面
|
||
return None
|
||
|
||
posts = soup.select("table#threadlisttableid tbody[id^=normalthread]") # 获取帖子列表
|
||
posts.reverse() # 帖子列表倒序
|
||
posts_min_timestamp = datetime_to_timestamp(self.get_post_date(posts[0])) # 帖子列表最小时间戳
|
||
posts_max_timestamp = datetime_to_timestamp(self.get_post_date(posts[-1])) # 帖子列表最大时间戳
|
||
|
||
# 本页最早的帖子时间戳小于等于目标时间戳 且 倒计时页数大于 0 则跳转到上一页继续查找
|
||
if posts_min_timestamp >= target_date_timestamp and not self.one_date_loading and self.setting[
|
||
"targetCountdownPage"] > 0:
|
||
print("跳转到上一页")
|
||
# yield f"data: 跳转到上一页\n\n"
|
||
self.setting["targetCountdownPage"] = max(self.setting["targetCountdownPage"] - 1, 0)
|
||
Setting.query.filter_by(name="book_download").update(
|
||
{"targetCountdownPage": self.setting["targetCountdownPage"]})
|
||
db.session.commit()
|
||
yield from self.section_page(self.setting["totalPage"] - self.setting["targetCountdownPage"])
|
||
return None
|
||
else:
|
||
print("不需要跳转上一页")
|
||
self.one_date_loading = True
|
||
# yield f"data: 不需要跳转上一页\n\n"
|
||
for post in posts:
|
||
post_element = post.select_one(".s.xst")
|
||
post_date = self.get_post_date(post) # 获取帖子时间
|
||
post_date_timestamp = datetime_to_timestamp(post_date) # 帖子时间戳
|
||
if post_date_timestamp == target_date_timestamp: # 判断是否是目标时间
|
||
print(f"目标时间是{post_date}")
|
||
# yield f"data: 目标时间是{post_date}\n\n"
|
||
yield from self.post_page(self.legitimate_naming(post_element.string), post_element['href']) # 跳转帖子页面
|
||
if post_date_timestamp > target_date_timestamp: # 帖子时间大于目标时间并且没有下一天时间
|
||
self._update_setting(targetDate=post_date)
|
||
print(f"下一天是 {post_date}")
|
||
# yield f"data: 下一天是 {post_date}\n\n"
|
||
self.one_date_loading = False
|
||
yield from self.section_page(self.setting["totalPage"] - self.setting["targetCountdownPage"])
|
||
return None
|
||
# 判断是否需要跳转到下一页
|
||
if posts_max_timestamp <= target_date_timestamp and self.setting["targetCountdownPage"] < self.setting[
|
||
"totalPage"] - 1:
|
||
print("跳转到下一页")
|
||
# yield f"data: 跳转到下一页\n\n"
|
||
self._update_setting(
|
||
targetCountdownPage=min(self.setting["targetCountdownPage"] + 1, self.setting["totalPage"] - 1))
|
||
yield from self.section_page(self.setting["totalPage"] - self.setting["targetCountdownPage"])
|
||
return None
|
||
|
||
# ------------------------------------------------------------
|
||
# 获取帖子发布时间
|
||
# ------------------------------------------------------------
|
||
@staticmethod
|
||
def get_post_date(element):
|
||
span_span = element.select_one(".by em span span")
|
||
if span_span:
|
||
if span_span.get("title"):
|
||
return span_span.get("title")
|
||
else:
|
||
return span_span.string
|
||
else:
|
||
return element.select_one(".by em span").string
|
||
|
||
# ------------------------------------------------------------
|
||
# 下载
|
||
# ------------------------------------------------------------
|
||
def book_download(self):
|
||
print(os.getenv('aa'))
|
||
self.create_folder(self.download_path, self.setting["targetDate"]) # 创建文件夹
|
||
target_page = self.setting["totalPage"] - self.setting["targetCountdownPage"] # 目标页
|
||
yield from self.section_page(target_page) # 前往目标页
|
||
yield f"event: close\ndata: 已完成\n\n"
|