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
| import asyncio
import aiohttp
import functools
import time
import hashlib
import json
from typing import AsyncGenerator, Dict, List
from dataclasses import dataclass
def async_cache(expire_seconds=60):
"""异步缓存装饰器。"""
cache = {}
def decorator(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
key_parts = (func.__name__, args, tuple(sorted(kwargs.items())))
cache_key = hashlib.md5(str(key_parts).encode()).hexdigest()
if cache_key in cache:
data, timestamp = cache[cache_key]
if time.time() - timestamp < expire_seconds:
print(f"[缓存命中] {func.__name__}")
return data
result = await func(*args, **kwargs)
cache[cache_key] = (result, time.time())
return result
wrapper.cache = cache
return wrapper
return decorator
def async_retry(max_attempts=3, delay=1, exceptions=(Exception,)):
"""异步重试装饰器。"""
def decorator(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(1, max_attempts + 1):
try:
return await func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_attempts:
await asyncio.sleep(delay)
print(f"[重试 {attempt}/{max_attempts}] {func.__name__}: {e}")
raise last_exception
return wrapper
return decorator
def async_timer(func):
"""异步计时装饰器。"""
@functools.wraps(func)
async def wrapper(*args, **kwargs):
start = time.time()
result = await func(*args, **kwargs)
elapsed = time.time() - start
print(f"[计时] {func.__name__} 耗时: {elapsed:.3f}s")
return result
return wrapper
@dataclass
class Article:
title: str
url: str
content: str
published_at: str
source: str
class AsyncNewsCrawler:
"""异步新闻爬虫。"""
def __init__(self, max_concurrent=10):
self.max_concurrent = max_concurrent
self.semaphore = asyncio.Semaphore(max_concurrent)
self.session = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
@async_cache(expire_seconds=300)
@async_retry(max_attempts=3, delay=1, exceptions=(aiohttp.ClientError,))
@async_timer
async def fetch_page(self, url: str) -> str:
"""获取页面内容（带缓存、重试、计时）。"""
async with self.semaphore:
async with self.session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
return await resp.text()
async def parse_articles(self, html: str, source: str) -> List[Article]:
"""解析文章列表（简化版，实际应使用 BeautifulSoup）。"""
import re
articles = []
title_pattern = re.compile(r'<h[12][^>]*>([^<]+)</h[12]>')
titles = title_pattern.findall(html)
for i, title in enumerate(titles[:10]):
articles.append(Article(
title=title.strip(),
url=f"{source}/article/{i}",
content=f"内容 {i}",
published_at="2026-01-01",
source=source
))
return articles
async def crawl_source(self, base_url: str) -> List[Article]:
"""爬取单个来源。"""
html = await self.fetch_page(base_url)
articles = await self.parse_articles(html, base_url)
print(f"[{base_url}] 获取 {len(articles)} 篇文章")
return articles
async def article_pipeline(articles: List[Article]) -> AsyncGenerator[Dict, None]:
"""异步生成器：处理文章数据流。"""
for article in articles:
article.title = article.title.strip()
article.content = article.content.replace('\n', ' ')
yield {
'title': article.title,
'url': article.url,
'content': article.content[:200],
'published_at': article.published_at,
'source': article.source
}
async def filter_keywords(data_stream: AsyncGenerator, keywords: List[str]):
"""过滤生成器：只保留包含关键词的文章。"""
async for article in data_stream:
if any(kw.lower() in article['title'].lower() for kw in keywords):
yield article
async def save_to_json(data_stream: AsyncGenerator, output_file: str):
"""保存生成器：写入 JSON 文件。"""
articles = []
async for article in data_stream:
articles.append(article)
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(articles, f, ensure_ascii=False, indent=2)
print(f"[保存] 写入 {len(articles)} 篇文章到 {output_file}")
async def main():
sources = [
"https://news.ycombinator.com",
"https://www.reddit.com/r/programming",
]
all_articles = []
async with AsyncNewsCrawler(max_concurrent=5) as crawler:
tasks = [crawler.crawl_source(url) for url in sources]
results = await asyncio.gather(*tasks, return_exceptions=True)
for result in results:
if isinstance(result, Exception):
print(f"爬取失败: {result}")
else:
all_articles.extend(result)
print(f"\n总共获取 {len(all_articles)} 篇文章")
pipeline = article_pipeline(all_articles)
filtered = filter_keywords(pipeline, keywords=['Python', 'AI', 'Web'])
await save_to_json(filtered, 'articles.json')
print("\n=== 性能统计 ===")
print("爬虫缓存命中率:", sum(1 for _ in range(0)))
print("总耗时:", time.time() - start_time)
if __name__ == '__main__':
start_time = time.time()
asyncio.run(main())
|
