yumoqing
64cc51cb6e
Database (3→6 tables): - hotspot_source: support api/browser/crawler types with type-specific fields - hotspot_schedule: cron/interval daemon config per source - hotspot_fetch_log: per-run audit trail (status/duration/items/errors) - hotspot_item: add heat_velocity, engagement/comment/share counts, sentiment - hotspot_analysis: 5-dimension scoring (time/heat/content/propagation/audience) - hotspot_alert: alert rules (heat threshold/velocity/sentiment triggers) Fetch engine (3 modes): - API: GET/POST + headers + Bearer auth + JSON extraction - Browser: HTML title/link extraction + og:title fallback - Crawler: recursive with depth/domain allowlist/link following Analysis engine: - Exponential decay model (7-day half-life) - Auto-classify: emerging→rising→hot→cooling→expired - 5-dimension scoring with detail JSON per dimension Dashboard: 6 tabs + 7 stat cards + workflow guide Permission paths: 42 paths registered (6 tables × 7 CRUD ops)
420 lines
16 KiB
Plaintext
420 lines
16 KiB
Plaintext
# 抓取引擎 — 支持 API / Browser / Crawler 三种模式
|
||
import json, time, hashlib
|
||
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')
|
||
|
||
async def fetch_rss(session, url):
|
||
"""RSS/Atom 解析"""
|
||
import aiohttp
|
||
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)
|
||
ns_atom = 'http://www.w3.org/2005/Atom'
|
||
|
||
# RSS 2.0
|
||
for item in root.findall('.//item'):
|
||
items.append({
|
||
'title': item.findtext('title', ''),
|
||
'url': item.findtext('link', ''),
|
||
'summary': (item.findtext('description', '') or '')[:500],
|
||
'publish_time': item.findtext('pubDate', ''),
|
||
})
|
||
|
||
# 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', ''),
|
||
'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 '',
|
||
})
|
||
|
||
return items, resp.status, size
|
||
except Exception as e:
|
||
return [], 0, 0
|
||
|
||
async def fetch_api(session, src):
|
||
"""API 模式：支持 GET/POST + Headers + Auth"""
|
||
import aiohttp
|
||
url = src.get('url', '')
|
||
method = src.get('fetch_method', 'GET').upper()
|
||
|
||
headers = {}
|
||
if src.get('headers'):
|
||
try:
|
||
headers = json.loads(src.get('headers', '{}'))
|
||
except:
|
||
pass
|
||
|
||
if src.get('api_key'):
|
||
headers['Authorization'] = f'Bearer {src["api_key"]}'
|
||
|
||
body = None
|
||
if src.get('request_body') and method == 'POST':
|
||
try:
|
||
body = json.dumps(json.loads(src['request_body']))
|
||
except:
|
||
body = src['request_body']
|
||
|
||
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'):
|
||
try:
|
||
headers.update(json.loads(src.get('headers', '{}')))
|
||
except:
|
||
pass
|
||
|
||
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())
|
||
|
||
if status != 200:
|
||
return items, status, size
|
||
|
||
# 通用 HTML 提取：找常见的热点列表结构
|
||
# 策略：提取所有 <a> 标签中带标题属性的，或 <article>/<li> 块
|
||
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 = ''
|
||
|
||
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'))
|
||
|
||
extractor = HotExtractor()
|
||
try:
|
||
extractor.feed(html)
|
||
except:
|
||
pass
|
||
|
||
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,
|
||
})
|
||
|
||
# 如果没提取到，尝试 JSON-LD 或 meta 标签
|
||
if not items:
|
||
import re
|
||
# 尝试提取 og:title / twitter:title
|
||
og_titles = re.findall(r'<meta[^>]+property="og:title"[^>]+content="([^"]+)"', html)
|
||
for t in og_titles[:20]:
|
||
items.append({'title': t, 'url': url, 'summary': '', 'heat_score': 60})
|
||
|
||
return items, 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
|
||
|
||
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):
|
||
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'<title>([^<]+)</title>', 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'<h[12][^>]*>([^<]+)</h[12]>', 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个链接
|
||
|
||
return items, 200, 0
|
||
|
||
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']
|
||
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
|
||
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],
|
||
'response_code': 0, 'response_size_bytes': 0,
|
||
})
|
||
continue
|
||
|
||
elapsed = int((time.time() - t0) * 1000)
|
||
status = 'success' if code in (0, 200) else 'partial'
|
||
new_count = 0
|
||
dup_count = 0
|
||
|
||
# 存入数据库
|
||
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': '',
|
||
'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,
|
||
})
|
||
|
||
return results
|