From 297a97974dbcdf220a82d9176843e5b81e0dfa8a Mon Sep 17 00:00:00 2001 From: yumoqing Date: Sat, 1 Aug 2026 12:09:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(hotspot):=20real=20extraction=20engine=20?= =?UTF-8?q?=E2=80=94=20pluggable=20parsers=20+=2011=20builtins?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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) --- json/hotspot_source.json | 2 +- models/mysql.ddl.sql | 1 + wwwroot/hotspot/fetch_now.dspy | 787 ++++++++++++++++++++------------- wwwroot/hotspot/index.ui | 2 +- 4 files changed, 472 insertions(+), 320 deletions(-) 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;" } }