298 lines
11 KiB
Python
298 lines
11 KiB
Python
"""
|
|
内置平台解析器 — 每个返回 (fetch_url, headers_override, extract_fn)
|
|
|
|
添加新平台: 在模块底部调用 _reg() 注册即可。
|
|
"""
|
|
import json
|
|
import re
|
|
import xml.etree.ElementTree as ET
|
|
import aiohttp
|
|
|
|
BUILTIN = {}
|
|
|
|
def _reg(name, url, parser_fn, headers=None):
|
|
BUILTIN[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))}
|
|
for item in data.get('data', {}).get('realtime', [])[:50]
|
|
if isinstance(item, dict) and item.get('word')
|
|
],
|
|
{'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(str(item.get('detail_text', '0'))
|
|
.replace('万','0000').replace('亿','00000000')
|
|
.replace('热度','').strip() or 0),
|
|
'summary': (item.get('target', {}).get('excerpt', '') or '')[: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)))}
|
|
for item in (data.get('data', []) if isinstance(data.get('data'), list) else [])[:50]
|
|
if isinstance(item, dict) and item.get('Title')
|
|
],
|
|
{'User-Agent': 'Mozilla/5.0', 'Referer': 'https://www.toutiao.com/'})
|
|
|
|
_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) and item.get('title')
|
|
],
|
|
{'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)))}
|
|
for item in (data.get('data', []) if isinstance(data.get('data'), list) else [])[:50]
|
|
if isinstance(item, dict)
|
|
],
|
|
{'User-Agent': 'Mozilla/5.0'})
|
|
|
|
_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', '') or '')[:500]}
|
|
for item in data.get('data', {}).get('items', [])
|
|
if isinstance(item, dict) and item.get('title')
|
|
],
|
|
{'User-Agent': 'Mozilla/5.0'})
|
|
|
|
# ---- 海外平台 ----
|
|
|
|
_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', '') or '')[:500],
|
|
'category': item.get('language', '')}
|
|
for item in data.get('items', []) if isinstance(item, dict)
|
|
],
|
|
{'Accept': 'application/vnd.github.v3+json', 'User-Agent': 'HotspotRadar/1.0'})
|
|
|
|
_reg('hackernews',
|
|
'https://hacker-news.firebaseio.com/v0/topstories.json',
|
|
None, # 特殊: 两步抓取, 见 engine.py
|
|
{'User-Agent': 'Mozilla/5.0'})
|
|
|
|
_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) and item.get('title')
|
|
],
|
|
{'User-Agent': 'Mozilla/5.0'})
|
|
|
|
_reg('rsshub', None, None, {}) # url 来自 source 配置
|
|
|
|
# ---- 通用提取器 ----
|
|
|
|
def extract_json_path(data, config):
|
|
"""简易 JSONPath: $.data.list[*] → {title, url, heat_score, ...}"""
|
|
path = config.get('item_path', '$[*]')
|
|
fm = config.get('field_map', {})
|
|
|
|
if path.startswith('$.'):
|
|
path = path[2:]
|
|
elif path.startswith('$'):
|
|
path = path[1:]
|
|
|
|
parts = []
|
|
for p in path.split('.'):
|
|
p = p.strip()
|
|
if not p: continue
|
|
if p.endswith('[*]'):
|
|
parts.append(p[:-3]); parts.append('*')
|
|
else:
|
|
parts.append(p)
|
|
|
|
items = data
|
|
for part in parts:
|
|
if part == '*':
|
|
continue
|
|
if isinstance(items, dict):
|
|
items = items.get(part)
|
|
elif isinstance(items, list):
|
|
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': _field(item, fm.get('title', 'title')),
|
|
'url': _field(item, fm.get('url', 'url')),
|
|
'heat_score': float(_field(item, fm.get('heat_score', '0')) or 0),
|
|
'summary': str(_field(item, fm.get('summary', '')) or '')[:500],
|
|
'category': str(_field(item, fm.get('category', '')) or ''),
|
|
'tags': str(_field(item, fm.get('tags', '')) or ''),
|
|
'engagement_count': int(_field(item, fm.get('engagement_count', '0')) or 0),
|
|
'comment_count': int(_field(item, fm.get('comment_count', '0')) or 0),
|
|
'share_count': int(_field(item, fm.get('share_count', '0')) or 0),
|
|
}
|
|
if entry['title']:
|
|
result.append(entry)
|
|
return result
|
|
|
|
def _field(obj, path):
|
|
if not path or not isinstance(obj, dict):
|
|
return obj if not isinstance(obj, dict) else ''
|
|
val = obj
|
|
for p in path.split('.'):
|
|
if isinstance(val, dict):
|
|
val = val.get(p, '')
|
|
else:
|
|
return ''
|
|
return val
|
|
|
|
def extract_css(html, config):
|
|
"""CSS 选择器提取 (轻量正则版)"""
|
|
results = []
|
|
for m in re.finditer(
|
|
r'<a[^>]+href=["\']([^"\']+)["\'][^>]*>([^<]*)</a>',
|
|
html, re.IGNORECASE | re.DOTALL
|
|
):
|
|
url, title = m.group(1), m.group(2).strip()
|
|
if len(title) >= 3:
|
|
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', '')
|
|
fm = config.get('field_map', {})
|
|
if not pattern: return []
|
|
results = []
|
|
for m in re.finditer(pattern, text, re.DOTALL):
|
|
gd = m.groupdict()
|
|
entry = {
|
|
'title': gd.get(fm.get('title', 'title'), '')[:200],
|
|
'url': gd.get(fm.get('url', 'url'), ''),
|
|
'heat_score': float(gd.get(fm.get('heat_score', '0'), 0) or 0),
|
|
'summary': str(gd.get(fm.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):
|
|
items = []
|
|
try:
|
|
root = ET.fromstring(xml_text)
|
|
ns_a = '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 e in root.findall(f'.//{{{ns_a}}}entry'):
|
|
link = e.find(f'{{{ns_a}}}link')
|
|
items.append({
|
|
'title': e.findtext(f'{{{ns_a}}}title', '')[:200],
|
|
'url': link.get('href', '') if link is not None else '',
|
|
'summary': (e.findtext(f'{{{ns_a}}}summary', '') or '')[:500],
|
|
'heat_score': 70, 'category': '',
|
|
'tags': '', 'engagement_count': 0, 'comment_count': 0, 'share_count': 0,
|
|
})
|
|
except ET.ParseError:
|
|
pass
|
|
return items
|
|
|
|
async def fetch_hn(session):
|
|
"""HackerNews 两步抓取"""
|
|
items = []
|
|
try:
|
|
async with session.get(
|
|
'https://hacker-news.firebaseio.com/v0/topstories.json',
|
|
timeout=aiohttp.ClientTimeout(total=15)
|
|
) as resp:
|
|
ids = await resp.json()
|
|
for sid in ids[:30]:
|
|
try:
|
|
async with session.get(
|
|
f'https://hacker-news.firebaseio.com/v0/item/{sid}.json',
|
|
timeout=aiohttp.ClientTimeout(total=10)
|
|
) as r2:
|
|
s = await r2.json()
|
|
if s and s.get('title'):
|
|
items.append({
|
|
'title': s.get('title', '')[:200],
|
|
'url': s.get('url', f'https://news.ycombinator.com/item?id={sid}'),
|
|
'heat_score': s.get('score', 0),
|
|
'engagement_count': s.get('score', 0),
|
|
'comment_count': s.get('descendants', 0),
|
|
'summary': '', 'category': s.get('type', ''),
|
|
'tags': '', 'share_count': 0,
|
|
})
|
|
except Exception:
|
|
continue
|
|
except Exception:
|
|
pass
|
|
return items
|