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())
|
