#!/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 split_sentences(text): text = re.sub(r'\s+', ' ', text).strip() if not text: return [] parts = re.split(r'(?<=[\.!?])\s+', text) return [p.strip() for p in parts if p.strip()] def pick_sentence(sentences, cues): for s in sentences: low = s.lower() if any(c in low for c in cues): return s return sentences[0] if sentences else '' def zh_simplify(text): if not text: return '（摘要未提供）' m = { 'this paper': '本文', 'we propose': '提出了', 'we present': '提出了', 'we introduce': '引入了', 'our method': '该方法', 'results show': '结果显示', 'experiments show': '实验显示', 'state-of-the-art': 'SOTA', 'benchmark': '基准测试', 'model': '模型', 'models': '模型', 'dataset': '数据集', 'datasets': '数据集', 'training': '训练', 'inference': '推理', 'diffusion': '扩散', 'transformer': 'Transformer', 'vision-language-action': '视觉-语言-动作', 'large language model': '大语言模型', 'llm': 'LLM', 'video generation': '视频生成' } out = text for k, v in m.items(): out = re.sub(k, v, out, flags=re.IGNORECASE) if len(out) > 140: out = out[:137] + '...' return out def title_zh(title): t = title repl = { 'Benchmark': '基准', 'Accelerating': '加速', 'Generation': '生成', 'Controlling': '控制', 'Features': '特征', 'Vision-Language-Action': '视觉-语言-动作', 'Models': '模型', 'Unbiased': '无偏', 'Evaluation': '评估', 'medical': '医疗', 'neural network': '神经网络' } for k, v in repl.items(): t = re.sub(k, v, t, flags=re.IGNORECASE) return t def paper_brief(p): sents = split_sentences(p.get('summary', '')) problem = pick_sentence(sents, ['challenge', 'problem', 'critical', 'limited', 'suffer']) method = pick_sentence(sents, ['we propose', 'we present', 'we introduce', 'our method']) result = pick_sentence(sents, ['results show', 'experiments show', 'outperform', 'improve', 'achieve']) impact = pick_sentence(sents, ['enables', 'useful', 'applications', 'real-world', 'towards']) if not method: method = sents[0] if sents else '' if not result: result = '文中给出了实验结果来验证方法有效性。' if not impact: impact = '对相关方向的研究和落地应用有参考价值。' return { 'title_zh': title_zh(p['title']), 'problem': zh_simplify(problem), 'method': zh_simplify(method), 'result': zh_simplify(result), 'impact': zh_simplify(impact), } 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 3（中文可读版）') for i, p in enumerate(hot[:3], 1): b = paper_brief(p) lines.append(f"{i}. **{p['title']}**") lines.append(f" - 中文题目（意译）: {b['title_zh']}") lines.append(f" - 这篇在讲什么: {b['problem']}") lines.append(f" - 它怎么做: {b['method']}") lines.append(f" - 得出了什么结果: {b['result']}") lines.append(f" - 可能的影响: {b['impact']}") lines.append(f" - arXiv: {p['id']}") 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()