from pathlib import Path import random import os import re import time from datetime import datetime import requests import asyncio from bs4 import BeautifulSoup from docx import Document from docx.shared import Inches from docx.enum.text import WD_ALIGN_PARAGRAPH from PIL import Image from io import BytesIO from playwright.async_api import async_playwright, TimeoutError as PlaywrightTimeoutError from auto_media_publisher.config.conf_base import PLAYWRIGHT_STEALTH_JS_PATH,DOCX_DIR_DOWNLOADED,WEIXIN_GZH_WATERMARK_HEIGHT,DAJIALA_URL_PREFIX,DAJIALA_CATEGORY_DICT,DEFAULT_USER_AGENT,DAJIALA_MAX_CONCURRENT_PAGE from auto_media_publisher.utils.utils_database import dajiala_get_pending_downloads,dajiala_update_download_status from auto_media_publisher.utils.logger import get_logger from auto_media_publisher.utils.utils_network import get_free_valid_proxies class DajialaDownloader(): def __init__(self): super().__init__() self.base_url = DAJIALA_URL_PREFIX self.category_dict = DAJIALA_CATEGORY_DICT self.logger = get_logger(self.__class__.__name__,"admin") self.valid_proxies = [] self.proxy_index = 0 self.failed_proxies = set() async def refresh_proxies(self): self.valid_proxies = await get_free_valid_proxies() if self.valid_proxies: self.logger.info(f"[Proxy] 获取并验证有效代理 {len(self.valid_proxies)} 个") def get_next_proxy(self): if self.valid_proxies: while self.proxy_index < len(self.valid_proxies): proxy = self.valid_proxies[self.proxy_index] self.proxy_index += 1 if proxy not in self.failed_proxies: return proxy return None async def init_browser_with_proxy(self, playwright, proxy_raw): try: proxy = {"server": f"http://{proxy_raw}"} browser = await playwright.chromium.launch(headless=False, channel= "chrome",args=["--blink-settings=imagesEnabled=false"]) context = await browser.new_context(proxy=proxy,user_agent=DEFAULT_USER_AGENT,viewport={"width": 1920, "height": 1080}) if os.path.exists(PLAYWRIGHT_STEALTH_JS_PATH): await context.add_init_script(path=Path(PLAYWRIGHT_STEALTH_JS_PATH)) page = await context.new_page() return browser, page except Exception as e: self.logger.error(f"[Worker] 浏览器初始化失败 proxy={proxy_raw}：{e}") async def worker(self, playwright, proxy_raw, url_queue: asyncio.Queue): try: browser, page = await self.init_browser_with_proxy(playwright, proxy_raw) self.logger.info(f"[Worker] 浏览器启动完成，代理: {proxy_raw}") except Exception as e: self.logger.error(f"[Worker] 浏览器初始化失败 proxy={proxy_raw}：{e}") return while True: try: row = await url_queue.get() except asyncio.CancelledError: break if row is None: # 队列结束信号 break url = row["url"] self.logger.info(f"[Worker] 开始处理 URL: {url}") docx_path_name = self.get_docx_path_name(row) try: success = await self.download_single_url(page, url, docx_path_name) if success: dajiala_update_download_status(url, "done") self.logger.info(f"[Worker] 下载成功: {url}") else: dajiala_update_download_status(url, "fail") self.failed_proxies.add(proxy_raw) self.logger.warning(f"[Worker] 下载失败: {url}") await asyncio.sleep(random.uniform(15, 30)) # 秒 except Exception as e: self.failed_proxies.add(proxy_raw) self.logger.error(f"[Worker] 异常下载: {url} - {e}") url_queue.task_done() await browser.close() self.logger.info(f"[Worker] 浏览器关闭，代理: {proxy_raw}") def get_docx_path_name(self, row): try: datepart = row["updatetime"][:10] title_part = re.sub(r'[\\/:*?"<>|]', '', row["title"]) return os.path.join(DOCX_DIR_DOWNLOADED, str(row["type"]) + "_" + row["type_name"], f"{datepart}_{title_part}.docx") except: return os.path.join(DOCX_DIR_DOWNLOADED, "0_未知", f"{datetime.now().isoformat()}.docx") async def download_single_url(self,page, url,docx_path_name): try: await page.goto(url, timeout=30000) await page.wait_for_timeout(30000) # 等待页面加载完成 # 检查是否有验证框 if await page.query_selector("#js_verify"): self.logger.info(f"[download] 检测到验证框，尝试点击验证: {url}") try: await page.click("#js_verify") await page.wait_for_timeout(5000) # 再等 5 秒 except Exception as ve: self.logger.warning(f"[download] 点击验证失败: {ve}") html_content = await page.content() soup = BeautifulSoup(html_content, "html.parser") div = soup.find("div", id="img-content") if not div: self.logger.warning(f"[download] 没找到 #img-content: {url}") return False doc = Document() for element in div.find_all(["h1", "p", "img", "audio", "video"]): if element.name in ["h1", "p"]: paragraph = doc.add_paragraph() run = paragraph.add_run(element.get_text()) # 字体颜色处理 if "style" in element.attrs: style = element.attrs["style"].lower() if "color" in style: color = style.split("color:")[1].split(";")[0].strip() rgb = self.get_rgb_color(color) if rgb: run.font.color.rgb = rgb elif element.name == "img": img_url = element.get("data-src") or element.get("src") if img_url: self.insert_image(doc, img_url) elif element.name in ["audio", "video"]: media_url = element.get("src") if media_url: doc.add_paragraph(f"媒体资源: {media_url}") os.makedirs(os.path.dirname(docx_path_name), exist_ok=True) doc.save(docx_path_name) self.logger.info(f"[download] 保存完成: {docx_path_name}") return True except Exception as e: self.logger.error(f"[download] 下载失败 {url}: {e}") return False def insert_image(self, doc, img_url): try: time.sleep(1) headers = {"User-Agent": DEFAULT_USER_AGENT} response = requests.get(img_url, timeout=50,headers=headers) if response.status_code == 200: image = Image.open(BytesIO(response.content)) width, height = image.size cropped = image.crop((0, 0, width, height - WEIXIN_GZH_WATERMARK_HEIGHT)) buffer = BytesIO() cropped.save(buffer, format=image.format or "PNG") buffer.seek(0) paragraph = doc.add_paragraph() run = paragraph.add_run() run.add_picture(buffer, width=Inches(6)) paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER except Exception as e: self.logger.warning(f"[image] 下载或插图失败 {img_url}: {e}") def get_rgb_color(self, color_str): """ 解析 HTML 颜色字符串并转换为 RGB """ from docx.shared import RGBColor import re match = re.search(r'#([0-9a-fA-F]{6})', color_str) if match: hex_color = match.group(1) r, g, b = int(hex_color[0:2], 16), int(hex_color[2:4], 16), int(hex_color[4:6], 16) return RGBColor(r, g, b) return None async def _async_run(self): await self.refresh_proxies() urls = dajiala_get_pending_downloads() self.logger.info(f"[DownloaderDajiala] 待下载 URL 数量: {len(urls)}") url_queue = asyncio.Queue() for row in urls: await url_queue.put(row) if not self.valid_proxies: return async with async_playwright() as playwright: workers = [] for proxy_raw in self.valid_proxies[:DAJIALA_MAX_CONCURRENT_PAGE]: workers.append(asyncio.create_task(self.worker(playwright, proxy_raw, url_queue))) # 等待队列处理完成 await url_queue.join() # 结束所有 worker for _ in workers: await url_queue.put(None) # 结束信号 await asyncio.gather(*workers) def run(self): asyncio.run(self._async_run())