import os import requests 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 from app.models import Setting class BookService: def __init__(self): self.setting = Setting.query.filter_by(name="book_download").first().to_dict() self.setting["targetDate"] = timestamp_to_datetime(datetime_to_timestamp(self.setting["targetDate"]) - 24 * 60 * 60) app_dir = os.path.dirname(os.path.dirname(__file__)) self.download_path = os.path.join(os.path.dirname(app_dir), "download") # 下载目录 self.today_timestamp = get_today_timestamp() self.one_date_loading = False self.headers = { "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; " } def legitimate_naming(self, name): # 合法命名 name = name.replace(":", "：") name = name.replace("<", "《") name = name.replace(">", "》") name = name.replace("/", " ") name = name.replace("\\", " ") name = name.replace("?", "？") return name def create_folder(self, target_dir, name): # 创建文件夹 real_dir = os.path.join(target_dir, name) folder = os.path.exists(real_dir) if not folder: os.makedirs(real_dir) else: print(f"文件夹：{name} 已存在") def isDownloadTarget(self, soup): # 是否是可以下载的文件 return soup.headers.get('Content-Type') in ['"application/octet-stream"', "application/octet-stream"] def post_page(self, name, url): # 帖子页面 response = requests.get(f"{self.setting["baseUrl"]}{url}", headers=self.headers) print(f"详情页面状态码：{str(response.status_code)}") soup = BeautifulSoup(response.text, 'lxml') resource_boxs = soup.select('ignore_js_op') for i in resource_boxs: download_dir_path = os.path.join(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) except RetryError as e: print('重试三次也不成功') @retry(stop=stop_after_attempt(3)) def download_file(self, file_url, dir_name, file_name): # 下载文件 file_res = requests.get(file_url, headers=self.headers, stream=True, allow_redirects=False) if file_res.status_code in (301, 302, 307, 308): # 非200状态码，重新请求 location = file_res.headers["location"] print(f'跳转页面了：{location}') file_res = requests.get(location, headers={"referer": location}, stream=True) print(f"下载状态码：{file_res.status_code} {file_url}") print(dir_name + "/" + file_name) yield f"data: {self.setting["targetDate"]} {file_name}\n\n" if self.isDownloadTarget(file_res): with open(f"download/{self.setting["targetDate"]}/{dir_name}/{file_name}", 'wb') as file2: for chunk in file_res.iter_content(chunk_size=1024): if chunk: file2.write(chunk) 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 = requests.get(url, headers=self.headers) # 请求 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.setting["totalPage"] = new_total_page # 更新总页数 Setting.query.filter_by(name="book_download").update({"totalPage": self.setting["totalPage"]}) db.session.commit() 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.setting["targetDate"] = post_date Setting.query.filter_by(name="book_download").update({"targetDate": self.setting["targetDate"]}) db.session.commit() 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.setting["targetCountdownPage"] = min(self.setting["targetCountdownPage"] + 1, self.setting["totalPage"] - 1) 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 def get_post_date(self, element): if element.select_one(".by em span span"): if element.select_one(".by em span span").get("title"): return element.select_one(".by em span span").get("title") else: return element.select_one(".by em span span").string else: return element.select_one(".by em span").string def book_download(self): 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"