1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
| import asyncio
import os
import re
import time
import feedparser
import aiohttp
import aiosqlite
import yaml
import html
import argparse
import logging
import random
from bs4 import BeautifulSoup
from urllib.parse import urlparse, urljoin
from tqdm.asyncio import tqdm as aio_tqdm
from http.cookies import SimpleCookie
# ==============================================================================
# 日志设置
# ==============================================================================
logger = logging.getLogger("下载器")
def setup_loggers(console_level=logging.INFO, error_file='downloader_errors.log'):
"""配置控制台和文件日志记录器。"""
if logger.hasHandlers():
logger.handlers.clear()
logger.setLevel(logging.DEBUG)
# 1. 控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(console_level)
console_formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - [%(name)s] - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
console_handler.setFormatter(console_formatter)
logger.addHandler(console_handler)
# 2. 错误文件处理器
try:
file_handler = logging.FileHandler(error_file, 'a', encoding='utf-8')
file_handler.setLevel(logging.WARNING)
file_formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(file_formatter)
logger.addHandler(file_handler)
logger.info(f"错误日志将记录到: {os.path.abspath(error_file)}")
except Exception as e:
logger.error(f"无法创建错误日志文件 {error_file}: {e}")
# ==============================================================================
# 配置加载与正则表达式预编译
# ==============================================================================
CONFIG = {}
REGEX = {
"image_file_ext": re.compile(r'\.(jpg|jpeg|png|webp|gif)$', re.I),
"blogspot_thumb": re.compile(r'/s\d+(-[a-z])?/', re.I),
"thumbnail_resize": re.compile(r'-\d+x\d+(\.\w+)$'),
}
def load_config(path='config.yaml'):
"""从 YAML 文件加载配置。"""
global CONFIG
try:
with open(path, 'r', encoding='utf-8') as f:
CONFIG = yaml.safe_load(f)
setup_loggers(error_file=CONFIG.get('error_log_file', 'downloader_errors.log'))
logger.info("成功从 %s 加载配置。", path)
except FileNotFoundError:
print(f"致命错误: 未找到 {path} 文件，请创建它。")
exit(1)
except Exception as e:
print(f"致命错误: 加载 {path} 出错: {e}")
exit(1)
# ==============================================================================
# 数据库与Cookie (异步)
# ==============================================================================
class DatabaseManager:
"""异步管理用于存储下载历史的 SQLite 数据库 (基于文章标题全局去重)。"""
def __init__(self, db_path):
self._db_path = db_path
self._conn = None
async def connect(self):
try:
self._conn = await aiosqlite.connect(self._db_path)
await self._conn.execute("PRAGMA journal_mode=WAL;")
await self._conn.execute("""
CREATE TABLE IF NOT EXISTS download_history (
post_title TEXT PRIMARY KEY,
download_date TEXT NOT NULL,
source_feed TEXT
)
""")
await self._conn.commit()
logger.info("数据库连接成功 (全局去重模式): %s", self._db_path)
except Exception as e:
logger.error("数据库连接失败: %s", e)
raise
async def is_downloaded(self, post_title):
"""检查一个文章标题是否已经被下载过。"""
async with self._conn.execute(
"SELECT 1 FROM download_history WHERE post_title = ?",
(post_title,)
) as cursor:
return await cursor.fetchone() is not None
async def add_entry(self, post_title, feed_title):
"""将一个文章标题添加到下载历史中。"""
now = time.strftime('%Y-%m-%d %H:%M:%S')
try:
await self._conn.execute(
"INSERT OR IGNORE INTO download_history (post_title, download_date, source_feed) VALUES (?, ?, ?)",
(post_title, now, feed_title)
)
await self._conn.commit()
except Exception as e:
logger.error("添加记录 '%s' 到数据库失败: %s", post_title, e)
async def close(self):
if self._conn:
await self._conn.close()
logger.info("数据库连接已关闭。")
class NetscapeCookieJar(aiohttp.CookieJar):
"""一个可以从 Netscape 格式的 cookies.txt 文件加载 cookie 的 CookieJar。"""
def __init__(self, cookie_file=None):
super().__init__()
if cookie_file and os.path.exists(cookie_file):
self.load_from_file(cookie_file)
def load_from_file(self, cookie_file):
try:
with open(cookie_file, 'r') as f:
for line in f:
if line.strip().startswith('#') or not line.strip():
continue
try:
domain, _, path, secure, expires, name, value = line.strip().split('\t')
cookie = SimpleCookie()
cookie[name] = value
cookie[name]['path'] = path
cookie[name]['domain'] = domain
cookie[name]['expires'] = int(float(expires))
cookie[name]['secure'] = secure.upper() == 'TRUE'
self.update_cookies(cookie)
except ValueError:
logger.warning(f"无法解析Cookie行: {line.strip()}")
logger.info(f"成功从 {cookie_file} 加载 Cookies。")
except Exception as e:
logger.error(f"加载Cookies文件 {cookie_file} 失败: {e}")
# ==============================================================================
# 工具函数
# ==============================================================================
def sanitize_folder_name(name):
sanitized = html.unescape(name).strip()
invalid_chars = r'[\\/:*?"<>|]'
sanitized = re.sub(invalid_chars, '_', sanitized)
return sanitized or "未命名"
def get_full_image_url(url):
if "bp.blogspot.com" in url or "googleusercontent.com" in url:
return REGEX["blogspot_thumb"].sub('/s0/', url)
if REGEX["thumbnail_resize"].search(url):
return REGEX["thumbnail_resize"].sub(r'\1', url)
return url
# ==============================================================================
# 网络与下载核心 (异步)
# ==============================================================================
async def fetch(session, url, log_prefix=""):
for attempt in range(CONFIG.get('max_retries', 3) + 1):
try:
async with session.get(url, timeout=CONFIG.get('request_timeout', 45), headers=CONFIG.get('request_headers', {})) as response:
response.raise_for_status()
return await response.text()
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
if attempt < CONFIG.get('max_retries', 3):
delay = CONFIG.get('retry_delay', 2) * (2 ** attempt)
logger.warning("%s第 %d/%d 次抓取 %s 失败。将在 %.1f 秒后重试...", log_prefix, attempt + 1, CONFIG.get('max_retries', 3) + 1, url, delay)
await asyncio.sleep(delay)
else:
logger.error("%s抓取 %s 在 %d 次重试后仍然失败: %s", log_prefix, url, CONFIG.get('max_retries', 3) + 1, e)
return None
async def download_image(session, img_url, filepath, referer_url, log_prefix=""):
"""异步下载单张图片，并在失败时记录详细错误。"""
full_img_url = get_full_image_url(img_url)
headers = CONFIG.get('image_headers', {}).copy()
headers['Referer'] = referer_url
last_exception = None
for attempt in range(CONFIG.get('max_retries', 3) + 1):
try:
async with session.get(full_img_url, timeout=CONFIG.get('request_timeout', 45), headers=headers) as response:
if response.status == 404 and full_img_url != img_url:
logger.warning("%s高清图URL %s 返回404，尝试原始URL: %s", log_prefix, full_img_url, img_url)
return await download_image(session, img_url, filepath, referer_url, log_prefix)
response.raise_for_status()
content = await response.read()
os.makedirs(os.path.dirname(filepath), exist_ok=True)
with open(filepath, 'wb') as f:
f.write(content)
return True, os.path.basename(filepath)
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
last_exception = e
if attempt < CONFIG.get('max_retries', 3):
delay = CONFIG.get('retry_delay', 5) * (2 ** attempt)
await asyncio.sleep(delay)
else:
# 最终失败时记录详细错误
error_message = f"重试 {CONFIG.get('max_retries', 3) + 1} 次后失败。URL: {img_url}"
logger.error("%s%s, 底层错误: %s", log_prefix, error_message, last_exception, exc_info=False)
return False, str(last_exception)
return False, "已达最大重试次数"
async def fetch_paginated_content(session, url, log_prefix=""):
"""异步获取一个分页文章的所有页面HTML内容。"""
full_html = ""
current_url = url
processed_urls = set()
while current_url and current_url not in processed_urls:
logger.debug("%s正在抓取分页内容: %s", log_prefix, current_url)
processed_urls.add(current_url)
html_content = await fetch(session, current_url, log_prefix)
if not html_content:
break
soup = BeautifulSoup(html_content, 'html.parser')
content_area = soup.select_one('div.entry-content, div.post-content, article.post, div#content')
if content_area:
full_html += str(content_area)
else:
full_html += html_content
next_page_link = soup.select_one('a.next.page-numbers, a[rel=next], .pagination-next a, a:-soup-contains("Next Page")')
if next_page_link and next_page_link.get('href'):
current_url = urljoin(current_url, next_page_link['href'])
await asyncio.sleep(random.uniform(0.5, 2.0)) # 抓取分页时也增加随机延迟
else:
current_url = None
return full_html
async def extract_images_from_page(session, article_url, log_prefix=""):
"""(更精确版) 从一个可能分页的文章中提取所有图片URL。"""
full_html_content = await fetch_paginated_content(session, article_url, log_prefix)
if not full_html_content:
return []
soup = BeautifulSoup(full_html_content, 'html.parser')
image_urls = set()
content_area = soup.select_one('div.entry-content, div.post-content, article.post, div#content')
search_area = content_area if content_area else soup
for img in search_area.find_all('img'):
src = img.get('data-src') or img.get('src')
if src:
full_url = urljoin(article_url, src.strip())
if REGEX["image_file_ext"].search(full_url.split('?')[0]):
image_urls.add(full_url)
return list(image_urls)
# ==============================================================================
# 主要处理逻辑 (异步)
# ==============================================================================
async def process_entry(session, db, entry, feed_title):
"""处理单个 RSS 条目，并记录下载失败 (基于文章标题全局去重)。"""
entry_id = getattr(entry, 'id', entry.link)
article_url = entry.link
log_prefix = f"[{feed_title[:15]}] "
# 1. 首先获取并清理标题
post_title = sanitize_folder_name(entry.title)
# 2. 使用清理后的标题进行全局重复检查
if not article_url or await db.is_downloaded(post_title):
if not article_url:
return 0, 0
logger.info("%s文章 '%s' 已在全局历史中存在，跳过。", log_prefix, post_title)
return 0, 0
# 如果没跳过，说明是新文章，继续处理
logger.info("%s处理新文章: '%s'", log_prefix, post_title)
image_urls = []
html_description = getattr(entry, 'description', '')
if html_description:
soup = BeautifulSoup(html_description, 'html.parser')
gallery_div = soup.select_one('div.gallery, div#gallery')
if gallery_div:
links = gallery_div.select('a')
for link in links:
href = link.get('href')
if href and REGEX["image_file_ext"].search(href.split('?')[0]):
image_urls.append(urljoin(article_url, href.strip()))
if not image_urls:
for img in soup.find_all('img'):
src = img.get('src') or img.get('data-src')
if src:
image_urls.append(urljoin(article_url, src.strip()))
if not image_urls and CONFIG.get('allow_fallback_to_source_site', False):
logger.warning("%s在RSS源中未找到图片，将访问源网站: %s", log_prefix, article_url)
image_urls = await extract_images_from_page(session, article_url, log_prefix)
elif image_urls:
unique_urls = list(dict.fromkeys(image_urls))
image_urls = unique_urls
logger.info("%s已成功从RSS源描述中提取 %d 个图片地址。", log_prefix, len(image_urls))
if not image_urls:
logger.warning("%s文章 '%s' 未能找到任何图片。", log_prefix, post_title)
# 将标题添加到数据库
await db.add_entry(post_title, feed_title)
return 1, 0
post_folder = os.path.join(CONFIG['base_folder'], sanitize_folder_name(feed_title), post_title)
extended_log_prefix = f"[{feed_title}][{post_title}] "
tasks = []
for i, img_url in enumerate(image_urls):
ext = os.path.splitext(urlparse(img_url).path)[1]
if not REGEX["image_file_ext"].search(ext):
ext = ".jpg"
filename = f"{i+1:03d}{ext}"
filepath = os.path.join(post_folder, filename)
if not os.path.exists(filepath):
task = asyncio.create_task(download_image(session, img_url, filepath, article_url, extended_log_prefix))
tasks.append(task)
else:
await asyncio.sleep(0.05)
if not tasks:
logger.info("%s所有 %d 张图片均已存在，跳过下载。", extended_log_prefix, len(image_urls))
# 将标题添加到数据库
await db.add_entry(post_title, feed_title)
return 1, 0
# 使用 asyncio.as_completed 来逐个处理下载任务，并加入随机延时
success_count = 0
pbar = aio_tqdm(total=len(tasks), desc=f"{log_prefix}下载 '{post_title[:20]}…'", unit="张", leave=False)
for task in asyncio.as_completed(tasks):
result, _ = await task
if result:
success_count += 1
pbar.update(1)
# 关键优化：在每次下载后都加入一个随机的短暂延时
await asyncio.sleep(random.uniform(0.5, 1.5))
pbar.close()
failed_count = len(tasks) - success_count
if failed_count == 0:
logger.info("%s成功为 '%s' 下载 %d 张新图片。", log_prefix, post_title, success_count)
else:
logger.error("%s'%s' 下载完成，%d 张成功, %d 张失败。详情请查看错误日志文件。", log_prefix, post_title, success_count, failed_count)
# 将标题添加到数据库
await db.add_entry(post_title, feed_title)
return 1, success_count
async def process_feed(session, db, feed_info):
"""
处理单个 RSS 订阅源。
优先尝试正常解析，如果因内容非XML而失败，则自动使用浏览器User-Agent重试。
"""
feed_title, feed_url = feed_info
log_prefix = f"[{feed_title}] "
logger.info("%s开始处理...", log_prefix)
try:
# --- 步骤 1: 首次尝试，不带任何伪装 ---
feed_data = await asyncio.to_thread(feedparser.parse, feed_url)
# --- 步骤 2: 检查首次尝试是否因“反爬虫”失败 ---
if feed_data.bozo and isinstance(feed_data.bozo_exception, feedparser.NonXMLContentType):
logger.warning("%s首次尝试失败，服务器返回了HTML页面。正在切换伪装模式重试...", log_prefix)
# --- 步骤 3: 切换为伪装模式进行重试 ---
agent = CONFIG.get('request_headers', {}).get('User-Agent', 'Mozilla/5.0')
feed_data = await asyncio.to_thread(feedparser.parse, feed_url, agent=agent)
# --- 步骤 4: 检查最终结果 ---
if feed_data.bozo:
logger.warning("%s订阅源可能格式错误或无法访问: %s. 错误: %s", log_prefix, feed_url, feed_data.bozo_exception)
# 后续逻辑保持不变
entry_tasks = [process_entry(session, db, entry, feed_title) for entry in feed_data.entries]
results = await asyncio.gather(*entry_tasks)
total_new_entries = sum(r[0] for r in results)
total_new_images = sum(r[1] for r in results)
logger.info("%s处理完毕。发现 %d 篇新文章，下载了 %d 张新图片。", log_prefix, total_new_entries, total_new_images)
return total_new_entries, total_new_images
except Exception as e:
logger.error("%s处理时发生意外错误: %s", log_prefix, e, exc_info=True)
return 0, 0
async def main():
"""异步主函数，运行下载器。"""
load_config()
os.makedirs(CONFIG['base_folder'], exist_ok=True)
db_path = os.path.join(CONFIG['base_folder'], CONFIG['db_file'])
db = DatabaseManager(db_path)
await db.connect()
try:
# 如果配置了cookie文件则加载，否则忽略
cookie_file = CONFIG.get('cookie_file')
if cookie_file and not os.path.exists(cookie_file):
logger.warning(f"配置了cookie文件 '{cookie_file}' 但文件不存在，将不加载Cookie。")
cookie_jar = aiohttp.CookieJar()
else:
cookie_jar = NetscapeCookieJar(cookie_file)
connector = aiohttp.TCPConnector(limit_per_host=CONFIG.get('max_concurrent_downloads', 8))
async with aiohttp.ClientSession(connector=connector, cookie_jar=cookie_jar) as session:
with open(CONFIG['opml_file'], 'r', encoding='utf-8') as f:
opml_data = f.read()
opml_soup = BeautifulSoup(opml_data, 'xml')
all_feeds = [
(outline.get('text', '未命名源'), outline.get('xmlUrl'))
for outline in opml_soup.find_all('outline') if outline.get('xmlUrl')
]
feeds_to_process = [f for f in all_feeds if f[0] not in CONFIG.get('skip_feeds', [])]
logger.info("在 %s 中发现 %d 个订阅源。将处理 %d 个。", CONFIG['opml_file'], len(all_feeds), len(feeds_to_process))
stats = {'new_entries': 0, 'new_images': 0}
semaphore = asyncio.Semaphore(CONFIG.get('max_concurrent_feeds', 4))
async def run_with_semaphore(feed_info):
async with semaphore:
return await process_feed(session, db, feed_info)
feed_tasks = [run_with_semaphore(feed) for feed in feeds_to_process]
for future in aio_tqdm.as_completed(feed_tasks, total=len(feed_tasks), desc="处理订阅源", unit="个"):
new_entries, new_images = await future
stats['new_entries'] += new_entries
stats['new_images'] += new_images
logger.info("=" * 60)
logger.info("所有订阅源处理完毕。")
logger.info("总计处理新文章: %d 篇", stats['new_entries'])
logger.info("总计下载新图片: %d 张", stats['new_images'])
logger.info("数据库位于: %s", os.path.abspath(db_path))
logger.info("=" * 60)
except FileNotFoundError as e:
logger.error("未找到关键文件: %s", e)
except Exception as e:
logger.critical("主程序发生严重错误: %s", e, exc_info=True)
finally:
await db.close()
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description='异步 RSS 图片下载器 (最终优化版 - 模拟人类行为)',
formatter_class=argparse.RawTextHelpFormatter
)
parser.add_argument('--daemon', action='store_true', help='以守护进程模式运行，按固定间隔重复执行。')
parser.add_argument(
'--interval',
type=int,
default=7200,
help='守护进程模式下的检查间隔（秒）。\n默认值: 7200 (2小时)。'
)
args = parser.parse_args()
if not args.daemon:
try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info("用户中断了程序。正在关闭...")
else:
logger.info(f"守护进程模式已启动，运行间隔为 {args.interval} 秒 ({args.interval / 3600:.1f} 小时)。")
while True:
start_time = time.time()
try:
logger.info("开始新一轮的检查与下载...")
asyncio.run(main())
except KeyboardInterrupt:
logger.info("守护进程被用户中断。正在退出...")
break
except Exception as e:
logger.critical(f"守护进程在执行任务时发生严重错误: {e}", exc_info=True)
end_time = time.time()
elapsed = end_time - start_time
sleep_time = max(60, args.interval - elapsed)
next_run_time_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time() + sleep_time))
logger.info(f"本轮任务执行完毕，耗时 {elapsed:.1f} 秒。将在 {sleep_time:.1f} 秒后开始下一轮（预计时间: {next_run_time_str}）。")
time.sleep(sleep_time)
|
