diff --git a/json/hotspot_source.json b/json/hotspot_source.json
index 2e43c6e1..ab994b36 100644
--- a/json/hotspot_source.json
+++ b/json/hotspot_source.json
@@ -7,7 +7,7 @@
"params": {
"sortby": "priority asc",
"browserfields": {
- "exclouded": ["id", "api_secret", "headers", "request_body", "crawler_rules"],
+ "exclouded": ["id", "api_secret", "headers", "request_body", "crawler_rules", "parser_config"],
"cwidth": {"name": 150, "url": 300}
},
"editexclouded": ["id", "created_at", "last_fetch"],
diff --git a/models/mysql.ddl.sql b/models/mysql.ddl.sql
index f1eed647..ce8f5111 100644
--- a/models/mysql.ddl.sql
+++ b/models/mysql.ddl.sql
@@ -6607,6 +6607,7 @@ CREATE TABLE hotspot_source
-- 通用
`proxy_enabled` VARCHAR(1) DEFAULT '0' comment '启用代理',
`proxy_url` VARCHAR(500) comment '代理地址',
+ `parser_config` TEXT comment '提取规则 JSON: parser/item_path/field_map/builtin_name',
`enabled` VARCHAR(1) DEFAULT '1' comment '是否启用',
`priority` INT DEFAULT 0 comment '优先级(越大越高)',
`last_fetch` VARCHAR(50) comment '上次抓取时间',
diff --git a/wwwroot/hotspot/fetch_now.dspy b/wwwroot/hotspot/fetch_now.dspy
index 4f1ea8e2..c6885bfb 100644
--- a/wwwroot/hotspot/fetch_now.dspy
+++ b/wwwroot/hotspot/fetch_now.dspy
@@ -1,5 +1,17 @@
-# 抓取引擎 — 支持 API / Browser / Crawler 三种模式
-import json, time, hashlib
+"""
+抓取引擎 — 可配置提取规则 + 内置平台解析器
+
+每条 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
@@ -8,50 +20,317 @@ from sqlor.dbpools import DBPools
def now_str():
return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
-async def fetch_rss(session, url):
- """RSS/Atom 解析"""
- import aiohttp
+# ============================================================
+# 内置平台解析器 — 每个返回 (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 模式分割
+ # 匹配 标签中的 title 和 href
+ link_pattern = re.compile(
+ r']+href=["\']([^"\']+)["\'][^>]*>([^<]*)',
+ 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:
- async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
- if resp.status != 200:
- return items, resp.status, 0
- text = await resp.text()
- size = len(text.encode())
-
- root = ET.fromstring(text)
+ root = ET.fromstring(xml_text)
ns_atom = 'http://www.w3.org/2005/Atom'
- # RSS 2.0
for item in root.findall('.//item'):
items.append({
- 'title': item.findtext('title', ''),
+ 'title': item.findtext('title', '')[:200],
'url': item.findtext('link', ''),
'summary': (item.findtext('description', '') or '')[:500],
- 'publish_time': item.findtext('pubDate', ''),
+ 'heat_score': 80,
+ 'category': item.findtext('category', ''),
+ 'tags': '',
+ 'engagement_count': 0, 'comment_count': 0, 'share_count': 0,
})
- # Atom fallback
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', ''),
+ '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],
- 'publish_time': entry.findtext(f'{{{ns_atom}}}updated', '') or '',
+ 'heat_score': 70,
+ 'category': '',
+ 'tags': '',
+ 'engagement_count': 0, 'comment_count': 0, 'share_count': 0,
})
-
- return items, resp.status, size
- except Exception as e:
- return [], 0, 0
+ except ET.ParseError:
+ pass
+ return items
-async def fetch_api(session, src):
- """API 模式：支持 GET/POST + Headers + Auth"""
- import aiohttp
+
+# ============================================================
+# 主逻辑: fetch + extract
+# ============================================================
+
+async def fetch_and_extract(session, src):
+ """根据 source 配置，抓取并提取 items"""
url = src.get('url', '')
- method = src.get('fetch_method', 'GET').upper()
-
headers = {}
if src.get('headers'):
try:
@@ -59,256 +338,177 @@ async def fetch_api(session, src):
except:
pass
- if src.get('api_key'):
- headers['Authorization'] = f'Bearer {src["api_key"]}'
-
- body = None
- if src.get('request_body') and method == 'POST':
+ # 解析 parser_config
+ parser_config = {}
+ if src.get('parser_config'):
try:
- body = json.dumps(json.loads(src['request_body']))
+ parser_config = json.loads(src.get('parser_config', '{}'))
except:
- body = src['request_body']
+ pass
+
+ parser_type = parser_config.get('parser', 'json_path')
+ builtin_name = parser_config.get('builtin_name', '')
- items = []
- try:
- if method == 'POST':
- async with session.post(url, headers=headers, data=body or '', timeout=aiohttp.ClientTimeout(total=30)) as resp:
- status = resp.status
- text = await resp.text()
- size = len(text.encode())
- if status == 200:
- try:
- data = json.loads(text)
- except:
- data = {}
- else:
- async with session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as resp:
- status = resp.status
- text = await resp.text()
- size = len(text.encode())
- if status == 200:
- try:
- data = json.loads(text)
- except:
- # Maybe it's XML/RSS — try RSS parser
- data = {}
-
- # Extract items from JSON response
- if isinstance(data, list):
- entries = data
- elif isinstance(data, dict):
- entries = data.get('data', data.get('items', data.get('list', data.get('result', []))))
- else:
- entries = []
-
- for entry in entries[:100]:
- if not isinstance(entry, dict):
- continue
- items.append({
- 'title': str(entry.get('title', entry.get('name', ''))),
- 'url': str(entry.get('url', entry.get('link', ''))),
- 'summary': str(entry.get('summary', entry.get('description', entry.get('desc', ''))))[:500],
- 'heat_score': float(entry.get('heat', entry.get('hot', entry.get('score', 0))) or 0),
- 'category': str(entry.get('category', entry.get('type', ''))),
- 'tags': str(entry.get('tags', '')),
- 'engagement_count': int(entry.get('engagement', entry.get('interactions', 0)) or 0),
- 'comment_count': int(entry.get('comments', entry.get('replies', 0)) or 0),
- 'share_count': int(entry.get('shares', entry.get('forwards', 0)) or 0),
- })
-
- return items, status, size
- except Exception as e:
- return [], 0, 0
-
-async def fetch_browser(session, src):
- """
- Browser 模式：使用 HTTP 请求 + 从 HTML 提取结构化数据
- (完整 Headless Browser 需要 playwright/selenium — 这里用 HTTP+解析做轻量版)
- """
import aiohttp
- from html.parser import HTMLParser
- url = src.get('url', '')
- items = []
-
- try:
- headers = {'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'}
- if src.get('headers'):
+ # === 内置解析器路径 ===
+ 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:
- headers.update(json.loads(src.get('headers', '{}')))
- except:
- pass
+ 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
- async with session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as resp:
- status = resp.status
- html = await resp.text()
- size = len(html.encode())
+ 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
- if status != 200:
- return items, status, size
-
- # 通用 HTML 提取：找常见的热点列表结构
- # 策略：提取所有 标签中带标题属性的，或 / 块
- from html.parser import HTMLParser as HP
-
- class HotExtractor(HP):
- def __init__(self):
- super().__init__()
- self.items = []
- self.current = {}
- self.in_item = False
- self.tag_stack = []
- self.text_buf = ''
+ 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
- def handle_starttag(self, tag, attrs):
- attrs_d = dict(attrs)
- self.tag_stack.append(tag)
- # 检测列表项容器
- if tag in ('article', 'li') and 'class' in attrs_d:
- cls = attrs_d['class'].lower()
- if any(k in cls for k in ('post', 'item', 'entry', 'card', 'topic', 'hot', 'trend')):
- self.in_item = True
- self.current = {'url': attrs_d.get('href', '')}
-
- if tag == 'a' and ('title' in attrs_d or self._is_hot_class(attrs_d.get('class', ''))):
- href = attrs_d.get('href', '')
- title = attrs_d.get('title', '')
- if href and title:
- self.items.append({'title': title, 'url': href, 'summary': ''})
-
- def handle_endtag(self, tag):
- if self.tag_stack:
- self.tag_stack.pop()
- if self.in_item and tag in ('article', 'li', 'div'):
- if self.current.get('title'):
- self.items.append(self.current)
- self.in_item = False
- self.current = {}
-
- def handle_data(self, data):
- if self.in_item and not self.current.get('title'):
- d = data.strip()
- if len(d) > 3:
- self.current['title'] = d[:200]
-
- def _is_hot_class(self, cls):
- if not cls:
- return False
- cl = cls.lower()
- return any(k in cl for k in ('title', 'hot', 'trending', 'headline'))
+ 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())
- extractor = HotExtractor()
- try:
- extractor.feed(html)
- except:
- pass
+ 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
- for it in extractor.items:
- if it.get('title') and len(it['title']) > 2:
- items.append({
- 'title': it['title'][:200],
- 'url': it.get('url', ''),
- 'summary': it.get('summary', '')[:500],
- 'heat_score': 80,
- })
+ elif parser_type == 'css':
+ items = extract_css(text, parser_config)
+ return items, resp.status, size
- # 如果没提取到，尝试 JSON-LD 或 meta 标签
- if not items:
- import re
- # 尝试提取 og:title / twitter:title
- og_titles = re.findall(r']+property="og:title"[^>]+content="([^"]+)"', html)
- for t in og_titles[:20]:
- items.append({'title': t, 'url': url, 'summary': '', 'heat_score': 60})
+ elif parser_type == 'regex':
+ items = extract_regex(text, parser_config)
+ return items, resp.status, size
- return items, 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 fetch_crawler(session, src):
- """
- Crawler 模式：从起始URL递归爬取，遵守深度和规则
- """
- import aiohttp
- from urllib.parse import urljoin, urlparse
+
+# ============================================================
+# 入库
+# ============================================================
+
+async def save_items(db, src_id, items, fetch_time):
+ new_count = 0
+ dup_count = 0
- start_urls = [src.get('url', '')]
- if src.get('crawler_start_urls'):
- try:
- start_urls = json.loads(src.get('crawler_start_urls', '[]'))
- except:
- pass
-
- depth = int(src.get('crawler_depth', 1))
- rules = {}
- if src.get('crawler_rules'):
- try:
- rules = json.loads(src.get('crawler_rules', '{}'))
- except:
- pass
-
- # 规则：allowed_domains, link_selector, item_selector, title_selector
- allowed_domains = rules.get('allowed_domains', [])
-
- visited = set()
- to_visit = list(start_urls)
- items = []
- headers = {'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'}
-
- for d in range(depth):
- next_urls = []
- for url in to_visit:
- if url in visited or len(items) >= 200:
- continue
- visited.add(url)
-
- # 域名白名单检查
- if allowed_domains:
- domain = urlparse(url).netloc
- if not any(ad in domain for ad in allowed_domains):
+ 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
- try:
- async with session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=20)) as resp:
- if resp.status != 200:
- continue
- html = await resp.text()
-
- import re
- # 提取页面内链接
- links = re.findall(r'href=["\']([^"\']+)["\']', html)
- for link in links:
- full = urljoin(url, link)
- if full not in visited and full.startswith(('http://', 'https://')):
- next_urls.append(full)
-
- # 提取标题
- titles = re.findall(r'([^<]+)', html)
- if titles:
- items.append({
- 'title': titles[0][:200],
- 'url': url,
- 'summary': '',
- 'heat_score': 50 + (3 - d) * 15,
- })
-
- # 尝试提取 h1/h2 标题
- h_tags = re.findall(r']*>([^<]+)', html)
- for h in h_tags[:10]:
- h = h.strip()
- if len(h) > 5:
- items.append({
- 'title': h[:200],
- 'url': url,
- 'summary': '',
- 'heat_score': 40 + (3 - d) * 10,
- })
-
- except:
- continue
-
- to_visit = next_urls[:50] # 每层限制50个链接
+ 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 items, 200, 0
+ return new_count, dup_count
+
+
+# ============================================================
+# 入口
+# ============================================================
async def main(request):
db = DBPools()
@@ -326,94 +526,45 @@ async def main(request):
results['total'] += 1
src_id = src['id']
- src_type = src.get('type', 'rss')
t0 = time.time()
- # 执行抓取
try:
- if src_type == 'rss':
- items, code, size = await fetch_rss(session, src.get('url', ''))
- elif src_type == 'api':
- items, code, size = await fetch_api(session, src)
- elif src_type == 'browser':
- items, code, size = await fetch_browser(session, src)
- elif src_type == 'crawler':
- items, code, size = await fetch_crawler(session, src)
- else:
- items, code, size = [], 0, 0
+ items, code, size = await fetch_and_extract(session, src)
except Exception as e:
- items, code, size = [], 0, 0
- results['errors'] += 1
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],
+ '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'
- new_count = 0
- dup_count = 0
+ 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:
- # 更新来源 last_fetch
await sor.U('hotspot_source', {'id': src_id, 'last_fetch': now})
-
- 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
-
- new_id = getID()
- await sor.C('hotspot_item', {
- 'id': new_id, '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': item.get('publish_time', now),
- 'first_seen': now, 'last_updated': now,
- '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
-
- # 写抓取日志
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': '',
+ '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),
- 'type': src_type,
- 'total': len(items),
- 'new': new_count,
- 'dup': dup_count,
- 'elapsed_ms': elapsed,
- 'code': code,
+ 'items': len(items), 'new': new_count, 'dup': dup_count,
+ 'elapsed_ms': elapsed, 'code': code,
})
return results
diff --git a/wwwroot/hotspot/index.ui b/wwwroot/hotspot/index.ui
index c69350f3..d4d0c83f 100644
--- a/wwwroot/hotspot/index.ui
+++ b/wwwroot/hotspot/index.ui
@@ -173,7 +173,7 @@
{
"widgettype": "Text",
"options": {
- "text": "工作流: ① 添加来源(type=api/browser/crawler) → ② 配置调度(cron/interval) → ③ 查看抓取日志(成功率/错误) → ④ 浏览热点(emerging/rising/hot/cooling/expired) → ⑤ 五维分析 → ⑥ 设置预警",
+ "text": "工作流: ① 添加来源 → parser_config选parser=builtin,builtin_name=weibo/zhihu/baidu/bilibili/... 或自定义json_path/css/regex → ② 调度配置 → ③ 抓取 → ④ 分析 → ⑤ 浏览 | 内置解析器: weibo_hot zhihu_hot baidu_hot toutiao_hot bilibili_hot douyin_hot 36kr_hot github_trending hackernews v2ex_hot rsshub",
"style": "font-size:12px;color:#bbb;padding:16px;border-top:1px solid #eee;margin-top:8px;"
}
}