val-blog/org/cases/arxiv_digest/arxiv_daily_digest.py

#!/usr/bin/env python3
import datetime as dt
import json
import re
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from pathlib import Path

BASE = Path('/Users/guchen/.openclaw/workspace/org/cases/arxiv_digest')
OUT_DIR = BASE / 'output'
STATE_DIR = BASE / 'state'
OUT_DIR.mkdir(parents=True, exist_ok=True)
STATE_DIR.mkdir(parents=True, exist_ok=True)

CATS = ['cs.AI', 'cs.LG', 'cs.CL', 'cs.CV', 'cs.RO', 'stat.ML']
MAX_RESULTS = 120
HOT_KEYWORDS = ['agent', 'reasoning', 'multimodal', 'alignment', 'rl', 'diffusion', 'transformer', 'benchmark']


def fetch(url, timeout=20):
    req = urllib.request.Request(url, headers={'User-Agent': 'Val-ArXiv-Digest/1.0'})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return r.read().decode('utf-8', errors='ignore')


def query_arxiv():
    q = ' OR '.join([f'cat:{c}' for c in CATS])
    params = {
        'search_query': q,
        'start': 0,
        'max_results': MAX_RESULTS,
        'sortBy': 'submittedDate',
        'sortOrder': 'descending',
    }
    url = 'http://export.arxiv.org/api/query?' + urllib.parse.urlencode(params)
    xml = fetch(url)
    ns = {'a': 'http://www.w3.org/2005/Atom', 'arxiv': 'http://arxiv.org/schemas/atom'}
    root = ET.fromstring(xml)
    papers = []
    for e in root.findall('a:entry', ns):
        pid = (e.findtext('a:id', default='', namespaces=ns) or '').strip()
        title = re.sub(r'\s+', ' ', (e.findtext('a:title', default='', namespaces=ns) or '').strip())
        summary = re.sub(r'\s+', ' ', (e.findtext('a:summary', default='', namespaces=ns) or '').strip())
        published = e.findtext('a:published', default='', namespaces=ns)
        updated = e.findtext('a:updated', default='', namespaces=ns)
        authors = [re.sub(r'\s+', ' ', (a.findtext('a:name', default='', namespaces=ns) or '').strip())
                   for a in e.findall('a:author', ns)]
        primary = ''
        pcat = e.find('arxiv:primary_category', ns)
        if pcat is not None:
            primary = pcat.attrib.get('term', '')
        all_cats = [c.attrib.get('term', '') for c in e.findall('a:category', ns)]
        papers.append({
            'id': pid,
            'arxiv_id': pid.split('/abs/')[-1] if '/abs/' in pid else pid,
            'title': title,
            'summary': summary,
            'published': published,
            'updated': updated,
            'authors': authors,
            'primary': primary,
            'categories': all_cats,
        })
    return papers


def hours_since(iso):
    try:
        t = dt.datetime.fromisoformat(iso.replace('Z', '+00:00'))
        now = dt.datetime.now(dt.timezone.utc)
        return max(0.0, (now - t).total_seconds() / 3600)
    except Exception:
        return 999.0


def hn_hits_for(paper):
    # lightweight buzz proxy
    q = paper['arxiv_id']
    url = 'https://hn.algolia.com/api/v1/search?' + urllib.parse.urlencode({'query': q, 'tags': 'story'})
    try:
        data = json.loads(fetch(url, timeout=3))
        return int(data.get('nbHits', 0))
    except Exception:
        return 0


def score(p):
    h = hours_since(p['published'])
    recency = max(0, 72 - h) / 72 * 60
    kw = sum(1 for k in HOT_KEYWORDS if k in (p['title'] + ' ' + p['summary']).lower())
    keyword_score = min(20, kw * 4)
    github_bonus = 8 if ('github.com' in p['summary'].lower() or 'code:' in p['summary'].lower()) else 0
    hn = hn_hits_for(p)
    hn_score = min(12, hn * 2)
    cat_bonus = 5 if p['primary'] in ['cs.AI', 'cs.LG', 'cs.CL'] else 2
    total = recency + keyword_score + github_bonus + hn_score + cat_bonus
    return round(total, 2), {'recency': round(recency,2), 'keywords': keyword_score, 'github': github_bonus, 'hn': hn_score, 'cat': cat_bonus}


def one_liner(p):
    t = p['title']
    if len(t) > 90:
        t = t[:87] + '...'
    return f"{t}（{p['primary']}）"


def build_digest(papers):
    # limit scoring set for speed/stability
    recent_pool = sorted(papers, key=lambda x: x['published'], reverse=True)[:40]
    scored = []
    for p in recent_pool:
        s, detail = score(p)
        p2 = dict(p)
        p2['hot_score'] = s
        p2['score_detail'] = detail
        scored.append(p2)
    hot = sorted(scored, key=lambda x: x['hot_score'], reverse=True)[:5]
    latest = sorted(scored, key=lambda x: x['published'], reverse=True)[:10]
    return hot, latest, scored


def to_md(hot, latest):
    today = dt.datetime.now().strftime('%Y-%m-%d')
    lines = []
    lines.append(f"# ArXiv Daily Brief - {today}")
    lines.append('')
    lines.append('## 🔥 今日热度 Top 5（新鲜度+关键词+HN提及+代码线索）')
    for i, p in enumerate(hot, 1):
        lines.append(f"{i}. **{p['title']}**")
        lines.append(f"   - arXiv: {p['id']}")
        lines.append(f"   - 类别: {p['primary']} | HotScore: {p['hot_score']} | 作者: {', '.join(p['authors'][:3])}")
        lines.append(f"   - 速读: {one_liner(p)}")
    lines.append('')
    lines.append('## 🆕 最新上新 Top 10')
    for i, p in enumerate(latest, 1):
        lines.append(f"{i}. {p['title']} ({p['primary']}) - {p['id']}")
    lines.append('')
    lines.append('## Val 今日建议')
    lines.append('- 先读 Top 5 里的 1-2 篇，优先看是否有可直接复用的方法/代码。')
    lines.append('- 若你愿意，我下一步可对 Top 3 产出“中文三段式精读卡”（问题-方法-可落地点）。')
    return '\n'.join(lines) + '\n'


def main():
    papers = query_arxiv()
    hot, latest, scored = build_digest(papers)
    md = to_md(hot, latest)

    day = dt.datetime.now().strftime('%Y-%m-%d')
    out = OUT_DIR / f'{day}.md'
    out.write_text(md, encoding='utf-8')
    (OUT_DIR / 'latest.md').write_text(md, encoding='utf-8')

    state = {
        'updatedAt': dt.datetime.now().isoformat(),
        'date': day,
        'papersFetched': len(papers),
        'topHot': [
            {'title': p['title'], 'arxiv_id': p['arxiv_id'], 'score': p['hot_score']} for p in hot
        ]
    }
    (STATE_DIR / 'latest_state.json').write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding='utf-8')
    print(str(out))


if __name__ == '__main__':
    main()