yumoqing
297a97974d
- parser_config field in hotspot_source for per-source extraction rules - Pluggable: json_path/css/regex/rss/builtin - 11 builtin platform parsers (weibo/zhihu/baidu/toutiao/bilibili/douyin/36kr/github/hn/v2ex/rsshub)
571 lines
22 KiB
Plaintext
571 lines
22 KiB
Plaintext
"""
|
|
抓取引擎 — 可配置提取规则 + 内置平台解析器
|
|
|
|
每条 hotspot_source 可带 parser_config JSON:
|
|
方式 A — 内置解析器: {"parser": "builtin", "builtin_name": "weibo"}
|
|
方式 B — JSONPath: {"parser": "json_path", "item_path": "$.data.list[*]", "field_map": {...}}
|
|
方式 C — CSS选择器: {"parser": "css", "item_selector": ".hot-item", "field_map": {...}}
|
|
方式 D — Regex: {"parser": "regex", "item_regex": "...", "field_map": {...}}
|
|
方式 E — RSS: {"parser": "rss"}
|
|
|
|
field_map: {"title":"...", "url":"...", "heat_score":"...", "summary":"...", "category":"...",
|
|
"tags":"...", "engagement_count":"...", "comment_count":"...", "share_count":"..."}
|
|
"""
|
|
import json, re, time
|
|
import xml.etree.ElementTree as ET
|
|
from datetime import datetime
|
|
from appPublic.uniqueID import getID
|
|
from sqlor.dbpools import DBPools
|
|
|
|
def now_str():
|
|
return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
# ============================================================
|
|
# 内置平台解析器 — 每个返回 (url, headers, extractor函数)
|
|
# ============================================================
|
|
|
|
BUILTIN_PARSERS = {}
|
|
|
|
def _reg(name, url, parser_fn, headers=None):
|
|
BUILTIN_PARSERS[name] = {'url': url, 'parser': parser_fn, 'headers': headers or {}}
|
|
|
|
# -- 微博热搜 --
|
|
_reg('weibo_hot',
|
|
'https://weibo.com/ajax/side/hotSearch',
|
|
lambda data: [
|
|
{'title': item.get('word', ''), 'url': f'https://s.weibo.com/weibo?q={item.get("word","")}',
|
|
'heat_score': item.get('raw_hot', item.get('num', 0)),
|
|
'summary': item.get('word_scheme', item.get('word', ''))}
|
|
for item in data.get('data', {}).get('realtime', [])[:50] if isinstance(item, dict)
|
|
],
|
|
{'User-Agent': 'Mozilla/5.0', 'X-Requested-With': 'XMLHttpRequest'})
|
|
|
|
# -- 知乎热榜 --
|
|
_reg('zhihu_hot',
|
|
'https://www.zhihu.com/api/v3/feed/topstory/hot-lists/total?limit=50',
|
|
lambda data: [
|
|
{'title': item.get('target', {}).get('title', ''),
|
|
'url': f'https://www.zhihu.com/question/{item.get("target",{}).get("id","")}',
|
|
'heat_score': int(item.get('detail_text', '0').replace('万', '0000').replace('亿', '00000000').replace('热度', '') or 0),
|
|
'summary': item.get('target', {}).get('excerpt', '')[:500]}
|
|
for item in data.get('data', []) if isinstance(item, dict)
|
|
],
|
|
{'User-Agent': 'Mozilla/5.0'})
|
|
|
|
# -- 百度热搜 --
|
|
_reg('baidu_hot',
|
|
'https://top.baidu.com/board?tab=realtime',
|
|
lambda data: [
|
|
{'title': item.get('word', ''), 'url': item.get('url', ''),
|
|
'heat_score': int(item.get('hotScore', 0)),
|
|
'summary': item.get('desc', '')}
|
|
for item in data.get('data', {}).get('cards', [{}])[0].get('content', []) if isinstance(item, dict)
|
|
],
|
|
{'User-Agent': 'Mozilla/5.0'})
|
|
|
|
# -- 头条热榜 --
|
|
_reg('toutiao_hot',
|
|
'https://www.toutiao.com/hot-event/hot-board/?origin=toutiao_pc',
|
|
lambda data: [
|
|
{'title': item.get('Title', ''), 'url': item.get('Url', ''),
|
|
'heat_score': int(float(item.get('HotValue', 0))),
|
|
'summary': item.get('Desc', '')}
|
|
for item in (data.get('data', []) if isinstance(data.get('data'), list) else [])[:50]
|
|
],
|
|
{'User-Agent': 'Mozilla/5.0', 'Referer': 'https://www.toutiao.com/'})
|
|
|
|
# -- B站热门 --
|
|
_reg('bilibili_hot',
|
|
'https://api.bilibili.com/x/web-interface/popular?ps=50',
|
|
lambda data: [
|
|
{'title': item.get('title', ''), 'url': f'https://www.bilibili.com/video/{item.get("bvid","")}',
|
|
'heat_score': item.get('stat', {}).get('view', 0),
|
|
'summary': item.get('desc', '')[:500],
|
|
'engagement_count': item.get('stat', {}).get('like', 0),
|
|
'comment_count': item.get('stat', {}).get('reply', 0),
|
|
'share_count': item.get('stat', {}).get('share', 0),
|
|
'category': item.get('tname', '')}
|
|
for item in data.get('data', {}).get('list', []) if isinstance(item, dict)
|
|
],
|
|
{'User-Agent': 'Mozilla/5.0', 'Referer': 'https://www.bilibili.com/'})
|
|
|
|
# -- 抖音热点 (通过第三方聚合接口) --
|
|
_reg('douyin_hot',
|
|
'https://dy.feigua.cn/api/hot/board',
|
|
lambda data: [
|
|
{'title': item.get('title', item.get('word', '')),
|
|
'url': item.get('url', ''),
|
|
'heat_score': int(item.get('hot_value', item.get('heat', 0))),
|
|
'summary': item.get('desc', '')[:500]}
|
|
for item in (data.get('data', []) if isinstance(data.get('data'), list) else [])[:50]
|
|
],
|
|
{'User-Agent': 'Mozilla/5.0'})
|
|
|
|
# -- 36氪热榜 --
|
|
_reg('36kr_hot',
|
|
'https://www.36kr.com/api/search/list/info-flow/hot?pageSize=30',
|
|
lambda data: [
|
|
{'title': item.get('title', ''), 'url': f'https://www.36kr.com/p/{item.get("id","")}',
|
|
'heat_score': item.get('total_count', 0),
|
|
'summary': item.get('summary', '')[:500],
|
|
'engagement_count': item.get('total_count', 0)}
|
|
for item in data.get('data', {}).get('items', []) if isinstance(item, dict)
|
|
],
|
|
{'User-Agent': 'Mozilla/5.0'})
|
|
|
|
# -- GitHub Trending (HTML parsing via API proxy) --
|
|
_reg('github_trending',
|
|
'https://api.github.com/search/repositories?q=stars:>1&sort=stars&order=desc&per_page=20',
|
|
lambda data: [
|
|
{'title': item.get('full_name', ''), 'url': item.get('html_url', ''),
|
|
'heat_score': item.get('stargazers_count', 0),
|
|
'summary': item.get('description', '')[:500],
|
|
'category': item.get('language', ''),
|
|
'engagement_count': item.get('forks_count', 0)}
|
|
for item in data.get('items', []) if isinstance(item, dict)
|
|
],
|
|
{'Accept': 'application/vnd.github.v3+json', 'User-Agent': 'HotspotRadar/1.0'})
|
|
|
|
# -- HackerNews --
|
|
_reg('hackernews',
|
|
'https://hacker-news.firebaseio.com/v0/topstories.json',
|
|
None, # special: needs 2-step fetch
|
|
{'User-Agent': 'Mozilla/5.0'})
|
|
|
|
# -- V2EX 热帖 --
|
|
_reg('v2ex_hot',
|
|
'https://www.v2ex.com/api/topics/hot.json',
|
|
lambda data: [
|
|
{'title': item.get('title', ''), 'url': item.get('url', ''),
|
|
'heat_score': item.get('replies', 0) * 10,
|
|
'summary': (item.get('content', '') or '')[:500],
|
|
'comment_count': item.get('replies', 0),
|
|
'category': item.get('node', {}).get('title', '') if isinstance(item.get('node'), dict) else ''}
|
|
for item in data if isinstance(item, dict)
|
|
],
|
|
{'User-Agent': 'Mozilla/5.0'})
|
|
|
|
# -- RSSHub (标准格式) --
|
|
_reg('rsshub',
|
|
None, # url comes from source config
|
|
None, # parsed as RSS/Atom feed
|
|
{})
|
|
|
|
|
|
# ============================================================
|
|
# 可配置提取器
|
|
# ============================================================
|
|
|
|
def extract_json_path(data, config):
|
|
"""JSONPath 提取 (简易实现: $.a.b[*] 格式)"""
|
|
item_path = config.get('item_path', '$[*]')
|
|
field_map = config.get('field_map', {})
|
|
|
|
# 解析路径: $.data.list[*] → ['data', 'list', '*']
|
|
path_parts = []
|
|
if item_path.startswith('$.'):
|
|
item_path = item_path[2:]
|
|
elif item_path.startswith('$'):
|
|
item_path = item_path[1:]
|
|
|
|
# 按 . 分割，处理 [*] 通配
|
|
for part in item_path.split('.'):
|
|
part = part.strip()
|
|
if not part:
|
|
continue
|
|
if part.endswith('[*]'):
|
|
path_parts.append(part[:-3])
|
|
path_parts.append('*')
|
|
else:
|
|
path_parts.append(part)
|
|
|
|
# 导航到 items 数组
|
|
items = data
|
|
for part in path_parts:
|
|
if part == '*':
|
|
if isinstance(items, list):
|
|
pass # wildcard means iterate
|
|
continue
|
|
if isinstance(items, dict):
|
|
items = items.get(part)
|
|
elif isinstance(items, list):
|
|
# 对每个元素取 part
|
|
items = [i.get(part) for i in items if isinstance(i, dict)]
|
|
items = [x for sub in items for x in (sub if isinstance(sub, list) else [sub])]
|
|
else:
|
|
return []
|
|
|
|
if not isinstance(items, list):
|
|
items = [items] if items else []
|
|
|
|
result = []
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
entry = {
|
|
'title': _extract_field(item, field_map.get('title', 'title')),
|
|
'url': _extract_field(item, field_map.get('url', 'url')),
|
|
'heat_score': float(_extract_field(item, field_map.get('heat_score', '0')) or 0),
|
|
'summary': str(_extract_field(item, field_map.get('summary', '')) or '')[:500],
|
|
'category': str(_extract_field(item, field_map.get('category', '')) or ''),
|
|
'tags': str(_extract_field(item, field_map.get('tags', '')) or ''),
|
|
'engagement_count': int(_extract_field(item, field_map.get('engagement_count', '0')) or 0),
|
|
'comment_count': int(_extract_field(item, field_map.get('comment_count', '0')) or 0),
|
|
'share_count': int(_extract_field(item, field_map.get('share_count', '0')) or 0),
|
|
}
|
|
if entry['title']:
|
|
result.append(entry)
|
|
return result
|
|
|
|
|
|
def _extract_field(obj, path):
|
|
"""从嵌套字典中按路径取值: 'target.title' 或 'stat.view'"""
|
|
if not path or not isinstance(obj, dict):
|
|
return obj if not isinstance(obj, dict) else ''
|
|
parts = path.split('.')
|
|
val = obj
|
|
for p in parts:
|
|
if isinstance(val, dict):
|
|
val = val.get(p, '')
|
|
else:
|
|
return ''
|
|
return val
|
|
|
|
|
|
def extract_css(html, config):
|
|
"""
|
|
CSS 选择器提取 (简易实现: 正则 + 文本匹配)
|
|
完整版需要 BeautifulSoup/pyquery — 这里做轻量正则提取
|
|
"""
|
|
item_sel = config.get('item_selector', '')
|
|
field_map = config.get('field_map', {})
|
|
results = []
|
|
|
|
# 尝试按常见的 HTML 模式分割
|
|
# 匹配 <a> 标签中的 title 和 href
|
|
link_pattern = re.compile(
|
|
r'<a[^>]+href=["\']([^"\']+)["\'][^>]*>([^<]*)</a>',
|
|
re.IGNORECASE | re.DOTALL
|
|
)
|
|
for m in link_pattern.finditer(html):
|
|
url, title = m.group(1), m.group(2).strip()
|
|
if len(title) < 3:
|
|
continue
|
|
results.append({
|
|
'title': title[:200],
|
|
'url': url[:2000],
|
|
'heat_score': 60,
|
|
'summary': '',
|
|
'category': '',
|
|
'tags': '',
|
|
'engagement_count': 0,
|
|
'comment_count': 0,
|
|
'share_count': 0,
|
|
})
|
|
return results[:100]
|
|
|
|
|
|
def extract_regex(text, config):
|
|
"""正则提取"""
|
|
pattern = config.get('item_regex', '')
|
|
field_map = config.get('field_map', {})
|
|
if not pattern:
|
|
return []
|
|
results = []
|
|
for m in re.finditer(pattern, text, re.DOTALL):
|
|
entry = {
|
|
'title': m.groupdict().get(field_map.get('title', 'title'), '')[:200],
|
|
'url': m.groupdict().get(field_map.get('url', 'url'), ''),
|
|
'heat_score': float(m.groupdict().get(field_map.get('heat_score', '0'), 0) or 0),
|
|
'summary': str(m.groupdict().get(field_map.get('summary', ''), ''))[:500],
|
|
'category': '',
|
|
'tags': '',
|
|
'engagement_count': 0,
|
|
'comment_count': 0,
|
|
'share_count': 0,
|
|
}
|
|
if entry['title']:
|
|
results.append(entry)
|
|
return results
|
|
|
|
|
|
def extract_rss(xml_text):
|
|
"""RSS/Atom 提取"""
|
|
items = []
|
|
try:
|
|
root = ET.fromstring(xml_text)
|
|
ns_atom = 'http://www.w3.org/2005/Atom'
|
|
|
|
for item in root.findall('.//item'):
|
|
items.append({
|
|
'title': item.findtext('title', '')[:200],
|
|
'url': item.findtext('link', ''),
|
|
'summary': (item.findtext('description', '') or '')[:500],
|
|
'heat_score': 80,
|
|
'category': item.findtext('category', ''),
|
|
'tags': '',
|
|
'engagement_count': 0, 'comment_count': 0, 'share_count': 0,
|
|
})
|
|
|
|
if not items:
|
|
for entry in root.findall(f'.//{{{ns_atom}}}entry'):
|
|
link_el = entry.find(f'{{{ns_atom}}}link')
|
|
items.append({
|
|
'title': entry.findtext(f'{{{ns_atom}}}title', '')[:200],
|
|
'url': link_el.get('href', '') if link_el is not None else '',
|
|
'summary': (entry.findtext(f'{{{ns_atom}}}summary', '') or '')[:500],
|
|
'heat_score': 70,
|
|
'category': '',
|
|
'tags': '',
|
|
'engagement_count': 0, 'comment_count': 0, 'share_count': 0,
|
|
})
|
|
except ET.ParseError:
|
|
pass
|
|
return items
|
|
|
|
|
|
# ============================================================
|
|
# 主逻辑: fetch + extract
|
|
# ============================================================
|
|
|
|
async def fetch_and_extract(session, src):
|
|
"""根据 source 配置，抓取并提取 items"""
|
|
url = src.get('url', '')
|
|
headers = {}
|
|
if src.get('headers'):
|
|
try:
|
|
headers = json.loads(src.get('headers', '{}'))
|
|
except:
|
|
pass
|
|
|
|
# 解析 parser_config
|
|
parser_config = {}
|
|
if src.get('parser_config'):
|
|
try:
|
|
parser_config = json.loads(src.get('parser_config', '{}'))
|
|
except:
|
|
pass
|
|
|
|
parser_type = parser_config.get('parser', 'json_path')
|
|
builtin_name = parser_config.get('builtin_name', '')
|
|
|
|
import aiohttp
|
|
|
|
# === 内置解析器路径 ===
|
|
if parser_type == 'builtin' and builtin_name in BUILTIN_PARSERS:
|
|
bp = BUILTIN_PARSERS[builtin_name]
|
|
fetch_url = bp['url'] or url
|
|
fetch_headers = {**headers, **bp.get('headers', {})}
|
|
|
|
if builtin_name == 'hackernews':
|
|
# 两步: 先拿 top IDs, 再批量拿详情
|
|
try:
|
|
async with aiohttp.ClientSession() as hn_session:
|
|
async with hn_session.get(
|
|
'https://hacker-news.firebaseio.com/v0/topstories.json',
|
|
timeout=aiohttp.ClientTimeout(total=15)
|
|
) as resp:
|
|
ids = await resp.json()
|
|
items = []
|
|
for story_id in ids[:30]:
|
|
try:
|
|
async with hn_session.get(
|
|
f'https://hacker-news.firebaseio.com/v0/item/{story_id}.json',
|
|
timeout=aiohttp.ClientTimeout(total=10)
|
|
) as r2:
|
|
story = await r2.json()
|
|
if story and story.get('title'):
|
|
items.append({
|
|
'title': story.get('title', '')[:200],
|
|
'url': story.get('url', f'https://news.ycombinator.com/item?id={story_id}'),
|
|
'heat_score': story.get('score', 0),
|
|
'engagement_count': story.get('score', 0),
|
|
'comment_count': story.get('descendants', 0),
|
|
'summary': '',
|
|
'category': story.get('type', ''),
|
|
'tags': '',
|
|
'share_count': 0,
|
|
})
|
|
except:
|
|
continue
|
|
return items, 200, 0
|
|
except Exception as e:
|
|
return [], 0, 0
|
|
|
|
elif builtin_name == 'rsshub':
|
|
# 来源的 url 就是 RSSHub 地址
|
|
try:
|
|
async with session.get(
|
|
url, headers=fetch_headers,
|
|
timeout=aiohttp.ClientTimeout(total=30)
|
|
) as resp:
|
|
text = await resp.text()
|
|
size = len(text.encode())
|
|
items = extract_rss(text)
|
|
return items, resp.status, size
|
|
except Exception as e:
|
|
return [], 0, 0
|
|
|
|
else:
|
|
try:
|
|
async with session.get(
|
|
fetch_url, headers=fetch_headers,
|
|
timeout=aiohttp.ClientTimeout(total=30)
|
|
) as resp:
|
|
if resp.status != 200:
|
|
return [], resp.status, 0
|
|
text = await resp.text()
|
|
size = len(text.encode())
|
|
try:
|
|
data = json.loads(text)
|
|
except json.JSONDecodeError:
|
|
return [], resp.status, size
|
|
|
|
items = bp['parser'](data)
|
|
return items, resp.status, size
|
|
except Exception as e:
|
|
return [], 0, 0
|
|
|
|
# === 通用提取器路径 ===
|
|
try:
|
|
async with session.get(
|
|
url, headers=headers,
|
|
timeout=aiohttp.ClientTimeout(total=30)
|
|
) as resp:
|
|
if resp.status != 200:
|
|
return [], resp.status, 0
|
|
text = await resp.text()
|
|
size = len(text.encode())
|
|
|
|
if parser_type == 'json_path':
|
|
try:
|
|
data = json.loads(text)
|
|
except json.JSONDecodeError:
|
|
return [], 0, size
|
|
items = extract_json_path(data, parser_config)
|
|
return items, resp.status, size
|
|
|
|
elif parser_type == 'css':
|
|
items = extract_css(text, parser_config)
|
|
return items, resp.status, size
|
|
|
|
elif parser_type == 'regex':
|
|
items = extract_regex(text, parser_config)
|
|
return items, resp.status, size
|
|
|
|
elif parser_type == 'rss':
|
|
items = extract_rss(text)
|
|
return items, resp.status, size
|
|
|
|
else:
|
|
return [], resp.status, size
|
|
|
|
except Exception as e:
|
|
return [], 0, 0
|
|
|
|
|
|
# ============================================================
|
|
# 入库
|
|
# ============================================================
|
|
|
|
async def save_items(db, src_id, items, fetch_time):
|
|
new_count = 0
|
|
dup_count = 0
|
|
|
|
async with db.sqlorContext('sage') as sor:
|
|
for item in items:
|
|
url = item.get('url', '')
|
|
if url:
|
|
existing = await sor.sqlExe(
|
|
"select id from hotspot_item where url=${u}$ limit 1", {'u': url})
|
|
if existing:
|
|
dup_count += 1
|
|
continue
|
|
|
|
await sor.C('hotspot_item', {
|
|
'id': getID(),
|
|
'source_id': src_id,
|
|
'title': item.get('title', '')[:500],
|
|
'url': url[:2000] if url else '',
|
|
'summary': item.get('summary', '')[:1000],
|
|
'heat_score': item.get('heat_score', 100),
|
|
'heat_velocity': 0,
|
|
'publish_time': fetch_time,
|
|
'first_seen': fetch_time,
|
|
'last_updated': fetch_time,
|
|
'status': 'emerging',
|
|
'category': item.get('category', ''),
|
|
'tags': item.get('tags', ''),
|
|
'engagement_count': item.get('engagement_count', 0),
|
|
'comment_count': item.get('comment_count', 0),
|
|
'share_count': item.get('share_count', 0),
|
|
'sentiment': 'neutral',
|
|
})
|
|
new_count += 1
|
|
|
|
return new_count, dup_count
|
|
|
|
|
|
# ============================================================
|
|
# 入口
|
|
# ============================================================
|
|
|
|
async def main(request):
|
|
db = DBPools()
|
|
now = now_str()
|
|
results = {'total': 0, 'new_items': 0, 'errors': 0, 'details': []}
|
|
|
|
async with db.sqlorContext('sage') as sor:
|
|
sources = await sor.R('hotspot_source', {})
|
|
|
|
import aiohttp
|
|
async with aiohttp.ClientSession() as session:
|
|
for src in sources:
|
|
if src.get('enabled') != '1':
|
|
continue
|
|
|
|
results['total'] += 1
|
|
src_id = src['id']
|
|
t0 = time.time()
|
|
|
|
try:
|
|
items, code, size = await fetch_and_extract(session, src)
|
|
except Exception as e:
|
|
async with db.sqlorContext('sage') as sor:
|
|
await sor.C('hotspot_fetch_log', {
|
|
'id': getID(), 'source_id': src_id,
|
|
'start_time': now, 'end_time': now_str(),
|
|
'duration_ms': int((time.time() - t0) * 1000),
|
|
'status': 'failed', 'items_total': 0, 'items_new': 0,
|
|
'items_duplicate': 0,
|
|
'error_msg': str(e)[:1000],
|
|
'response_code': 0, 'response_size_bytes': 0,
|
|
})
|
|
results['errors'] += 1
|
|
continue
|
|
|
|
elapsed = int((time.time() - t0) * 1000)
|
|
status = 'success' if code in (0, 200) else ('partial' if items else 'failed')
|
|
|
|
new_count, dup_count = await save_items(db, src_id, items, now)
|
|
|
|
async with db.sqlorContext('sage') as sor:
|
|
await sor.U('hotspot_source', {'id': src_id, 'last_fetch': now})
|
|
await sor.C('hotspot_fetch_log', {
|
|
'id': getID(), 'source_id': src_id,
|
|
'start_time': now, 'end_time': now_str(),
|
|
'duration_ms': elapsed, 'status': status,
|
|
'items_total': len(items), 'items_new': new_count,
|
|
'items_duplicate': dup_count, 'error_msg': '',
|
|
'response_code': code, 'response_size_bytes': size,
|
|
})
|
|
|
|
results['new_items'] += new_count
|
|
results['details'].append({
|
|
'source': src.get('name', src_id),
|
|
'items': len(items), 'new': new_count, 'dup': dup_count,
|
|
'elapsed_ms': elapsed, 'code': code,
|
|
})
|
|
|
|
return results
|