sage/wwwroot/hotspot/fetch_now.dspy

# 从所有启用的来源抓取热点数据
import json
import xml.etree.ElementTree as ET
from datetime import datetime
from appPublic.uniqueID import getID
from sqlor.dbpools import DBPools

async def fetch_rss(url):
    """解析 RSS/Atom feed"""
    import aiohttp
    items = []
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
                if resp.status != 200:
                    return items
                text = await resp.text()
        
        root = ET.fromstring(text)
        ns = {'atom': 'http://www.w3.org/2005/Atom'}
        
        # RSS 2.0
        for item in root.findall('.//item'):
            title = item.findtext('title', '')
            link = item.findtext('link', '')
            desc = item.findtext('description', '')[:500]
            pub_date = item.findtext('pubDate', '')
            items.append({
                'title': title,
                'url': link,
                'summary': desc,
                'publish_time': pub_date,
            })
        
        # Atom
        if not items:
            for entry in root.findall('.//atom:entry', ns):
                title = entry.findtext('atom:title', '', ns)
                link_el = entry.find('atom:link', ns)
                link = link_el.get('href', '') if link_el is not None else ''
                summary = entry.findtext('atom:summary', '', ns) or ''
                updated = entry.findtext('atom:updated', '', ns) or ''
                items.append({
                    'title': title,
                    'url': link,
                    'summary': summary[:500],
                    'publish_time': updated,
                })
    except Exception as e:
        print(f'RSS fetch error for {url}: {e}')
    return items

async def fetch_api(url, method='GET', headers=None, json_path=None):
    """调用 API 端点获取热点数据"""
    import aiohttp
    items = []
    try:
        hdrs = {}
        if headers:
            try:
                hdrs = json.loads(headers)
            except:
                pass
        
        async with aiohttp.ClientSession() as session:
            if method.upper() == 'GET':
                async with session.get(url, headers=hdrs, timeout=aiohttp.ClientTimeout(total=30)) as resp:
                    if resp.status == 200:
                        data = await resp.json()
            else:
                async with session.post(url, headers=hdrs, timeout=aiohttp.ClientTimeout(total=30)) as resp:
                    if resp.status == 200:
                        data = await resp.json()
        
        # 尝试从 JSON 中提取 items
        if isinstance(data, list):
            entries = data
        elif isinstance(data, dict):
            # 常见热点 API 格式
            entries = data.get('data', data.get('items', data.get('list', [])))
        else:
            entries = []
        
        for entry in entries[:50]:  # limit
            if isinstance(entry, dict):
                title = entry.get('title', entry.get('name', ''))
                url_val = entry.get('url', entry.get('link', ''))
                summary = entry.get('summary', entry.get('description', entry.get('desc', '')))
                heat = entry.get('heat', entry.get('hot', entry.get('score', 0)))
                items.append({
                    'title': str(title),
                    'url': str(url_val),
                    'summary': str(summary)[:500],
                    'heat_score': int(heat) if heat else 0,
                })
    except Exception as e:
        print(f'API fetch error for {url}: {e}')
    return items

async def main(request):
    db = DBPools()
    now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    new_items = 0
    total_sources = 0
    
    async with db.sqlorContext('sage') as sor:
        sources = await sor.R('hotspot_source', {})
    
    for src in sources:
        if src.get('enabled') != '1':
            continue
        
        total_sources += 1
        src_type = src.get('type', 'rss')
        src_url = src.get('url', '')
        
        if not src_url:
            continue
        
        if src_type == 'rss':
            items = await fetch_rss(src_url)
        elif src_type == 'api':
            fetch_method = src.get('fetch_method', 'GET')
            headers = src.get('headers', '')
            items = await fetch_api(src_url, fetch_method, headers)
        else:
            continue
        
        # 存入数据库
        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 是否已存在
                existing = await sor.sqlExe(
                    "select id from hotspot_item where url=${url}$",
                    {'url': item.get('url', '')}
                )
                if existing:
                    continue
                
                new_id = getID()
                status = 'rising'  # 新抓取的默认 rising
                await sor.C('hotspot_item', {
                    'id': new_id,
                    'source_id': src['id'],
                    'title': item.get('title', ''),
                    'url': item.get('url', ''),
                    'summary': item.get('summary', '')[:500],
                    'heat_score': item.get('heat_score', 100),
                    'publish_time': item.get('publish_time', now),
                    'fetch_time': now,
                    'status': status,
                })
                new_items += 1
    
    return {
        'total': total_sources,
        'new_items': new_items,
        'time': now,
    }