250 lines
9.3 KiB
Python
Executable File
250 lines
9.3 KiB
Python
Executable File
#!/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()
|