187 lines
6.7 KiB
Python
Executable File
187 lines
6.7 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 plain_summary(p):
|
||
s = p.get('summary', '')
|
||
s = re.sub(r'\s+', ' ', s).strip()
|
||
if not s:
|
||
return '这篇主要在提出一个新方法，并用实验验证它是否更好。'
|
||
# take first sentence-like chunk
|
||
chunk = re.split(r'(?<=[\.!?])\s+', s)[0]
|
||
if len(chunk) > 120:
|
||
chunk = chunk[:117] + '...'
|
||
return f"它在做的事：{chunk}"
|
||
|
||
|
||
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):
|
||
lines.append(f"{i}. **{p['title']}**")
|
||
lines.append(f" - 一句话看懂: {plain_summary(p)}")
|
||
lines.append(f" - 你可能会关心: {one_liner(p)}")
|
||
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()
|