379 lines
16 KiB
Python
379 lines
16 KiB
Python
"""The Radar of a project (docs/projects.md): once a week the new papers (projects/papers.py)
|
|
go to Claude in a batch (half price, done within the hour as a rule): one request sorts every
|
|
paper into the config's groups and writes about the worth-reading ones (JSON schema), one looks
|
|
for the week's news (web search). The answer becomes a RadarDigest — the bundle format of
|
|
F:\\neuropolitik\\tools\\digest_batch.py (version 1), so a digest made elsewhere can come in too.
|
|
|
|
Runs from the host's crontab: `manage.py radar` every half hour submits what is due and
|
|
collects what is done. Needs ANTHROPIC_API_KEY in the container's environment.
|
|
"""
|
|
import json
|
|
import os
|
|
from datetime import date, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from django.db import transaction
|
|
from django.http import JsonResponse
|
|
from django.utils import timezone
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
from django.views.decorators.http import require_http_methods
|
|
|
|
from . import papers
|
|
from .models import Project, Radar, RadarDigest
|
|
|
|
DEFAULT_MODEL = 'claude-opus-5-5'
|
|
# Batch prices: half of Opus 5.5's list prices ($4 / $20 per million tokens); web search $10 / 1000.
|
|
PRICE_IN, PRICE_OUT, PRICE_SEARCH = 2.00 / 1e6, 10.00 / 1e6, 10.00 / 1000
|
|
SEEN_MAX = 20000
|
|
# The weekday and hour of a config are the user's (the server runs on UTC).
|
|
ZONE = ZoneInfo('Europe/Berlin')
|
|
DIGESTS_SHOWN = 12
|
|
|
|
|
|
def papers_schema(groups):
|
|
return {
|
|
'type': 'object',
|
|
'properties': {
|
|
'auf_einen_blick': {'type': 'string'},
|
|
'items': {'type': 'array', 'items': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'id': {'type': 'string'},
|
|
'gruppe': {'type': 'string', 'enum': list(groups)},
|
|
'worum_es_geht': {'type': 'string'},
|
|
'einordnung': {'type': 'string'},
|
|
'politischer_haken': {'type': 'string'},
|
|
'kurz': {'type': 'string'},
|
|
},
|
|
'required': ['id', 'gruppe', 'worum_es_geht', 'einordnung', 'politischer_haken', 'kurz'],
|
|
'additionalProperties': False,
|
|
}},
|
|
'grundlagenbegriff': {
|
|
'type': 'object',
|
|
'properties': {'begriff': {'type': 'string'}, 'erklaerung': {'type': 'string'}},
|
|
'required': ['begriff', 'erklaerung'],
|
|
'additionalProperties': False,
|
|
},
|
|
},
|
|
'required': ['auf_einen_blick', 'items', 'grundlagenbegriff'],
|
|
'additionalProperties': False,
|
|
}
|
|
|
|
|
|
def _model_record(r):
|
|
"""What the model needs of a paper to sort it."""
|
|
return {k: r.get(k) for k in ('id', 'source', 'topic', 'journal', 'first_author', 'n_authors', 'pub_types',
|
|
'date', 'title', 'abstract', 'institution') if r.get(k) not in (None, '', [])}
|
|
|
|
|
|
def _terms(radar):
|
|
"""The basic terms explained before: not again."""
|
|
out = []
|
|
for b in radar.digests.values_list('bundle', flat=True)[:52]:
|
|
t = ((b or {}).get('grundlagenbegriff') or {}).get('begriff')
|
|
if t:
|
|
out.append(t)
|
|
return out
|
|
|
|
|
|
def build_requests(config, records, terms, today):
|
|
model = config.get('model') or DEFAULT_MODEL
|
|
groups = config.get('groups') or ['lesenswert', 'rauschen']
|
|
prompt = config['papers_task'].format(
|
|
n=len(records), glossar=', '.join(terms) if terms else '(noch keine)',
|
|
records=json.dumps([_model_record(r) for r in records], ensure_ascii=False))
|
|
reqs = [{'custom_id': 'papers', 'params': {
|
|
'model': model, 'max_tokens': 48000, 'thinking': {'type': 'adaptive'},
|
|
'output_config': {'effort': 'high', 'format': {'type': 'json_schema', 'schema': papers_schema(groups)}},
|
|
'system': config['system'], 'messages': [{'role': 'user', 'content': prompt}],
|
|
}}]
|
|
if config.get('news_task'):
|
|
reqs.append({'custom_id': 'news', 'params': {
|
|
'model': model, 'max_tokens': 16000, 'thinking': {'type': 'adaptive'},
|
|
'output_config': {'effort': 'medium'}, 'system': config['system'],
|
|
'tools': [{'type': 'web_search_20260209', 'name': 'web_search', 'max_uses': 4}],
|
|
'messages': [{'role': 'user', 'content': config['news_task'].format(today=today)}],
|
|
}})
|
|
return reqs
|
|
|
|
|
|
def build_bundle(config, pending, papers_json, news_md, usage, batch_id, project_name=''):
|
|
groups = config.get('groups') or ['lesenswert', 'rauschen']
|
|
records = {r['id']: r for r in pending['records']}
|
|
items, seen = [], set()
|
|
for it in papers_json.get('items') or []:
|
|
r = records.get(it.get('id'))
|
|
if not r or it['id'] in seen or it.get('gruppe') not in groups:
|
|
continue # an id or group the model made up, or twice: left out
|
|
seen.add(it['id'])
|
|
items.append({**r, **it})
|
|
# What the model forgot shows, unsorted, among the rest.
|
|
fallback = 'technik' if 'technik' in groups else groups[-1]
|
|
for rid, r in records.items():
|
|
if rid not in seen:
|
|
items.append({**r, 'gruppe': fallback, 'worum_es_geht': '', 'einordnung': '', 'politischer_haken': '',
|
|
'kurz': '(nicht eingeordnet)'})
|
|
per_source = {}
|
|
for r in pending['records']:
|
|
per_source[r['source']] = per_source.get(r['source'], 0) + 1
|
|
cost = (usage['input_tokens'] * PRICE_IN + usage['output_tokens'] * PRICE_OUT
|
|
+ usage['web_search_requests'] * PRICE_SEARCH)
|
|
return {
|
|
'version': 1, 'project': project_name, 'generated': pending['generated'], 'days': pending['days'],
|
|
'model': config.get('model') or DEFAULT_MODEL, 'batch_id': batch_id,
|
|
'auf_einen_blick': papers_json.get('auf_einen_blick', ''),
|
|
'grundlagenbegriff': papers_json.get('grundlagenbegriff') or {'begriff': '', 'erklaerung': ''},
|
|
'items': items, 'news_markdown': news_md,
|
|
'stats': {'records': len(records), 'per_source': per_source, 'errors': pending.get('errors', []),
|
|
'usage': usage, 'cost_usd': round(cost, 4)},
|
|
}
|
|
|
|
|
|
def _client():
|
|
import anthropic # only where the Radar runs
|
|
return anthropic.Anthropic()
|
|
|
|
|
|
# ---- running ---------------------------------------------------------------------------------
|
|
|
|
def due(radar, now=None):
|
|
"""Its weekday and hour have come (local time), and this week has no digest yet."""
|
|
if not radar.enabled or radar.pending or not radar.config.get('papers_task'):
|
|
return False
|
|
now = (now or timezone.now()).astimezone(ZONE)
|
|
c = radar.config
|
|
weekday, hour = int(c.get('weekday', 6)), int(c.get('hour', 18))
|
|
start = (now - timedelta(days=(now.weekday() - weekday) % 7)).replace(hour=hour, minute=0, second=0,
|
|
microsecond=0)
|
|
if start > now:
|
|
start -= timedelta(days=7)
|
|
last = radar.last_run_at.astimezone(ZONE) if radar.last_run_at else None
|
|
return last is None or last < start
|
|
|
|
|
|
def submit(radar, force=False):
|
|
"""Fetch the new papers and send the batch. Returns what happened, in words."""
|
|
if radar.pending and not force:
|
|
return 'Es läuft schon ein Abruf.'
|
|
c = radar.config
|
|
days = int(c.get('days', 7))
|
|
records, errors = papers.collect(c, days, set(radar.seen))
|
|
radar.last_run_at = timezone.now()
|
|
today = date.today().isoformat()
|
|
if not records:
|
|
radar.last_error = '; '.join(errors)
|
|
radar.pending = None
|
|
radar.save(update_fields=['last_run_at', 'last_error', 'pending'])
|
|
return 'Keine neuen Papers.' + (' Fehler: ' + radar.last_error if errors else '')
|
|
batch = _client().messages.batches.create(requests=build_requests(c, records, _terms(radar), today))
|
|
radar.pending = {'batch_id': batch.id, 'generated': today, 'days': days, 'records': records,
|
|
'errors': errors, 'submitted': timezone.now().isoformat(timespec='seconds')}
|
|
radar.last_error = ''
|
|
radar.save(update_fields=['last_run_at', 'last_error', 'pending'])
|
|
return f'{len(records)} Papers an Claude geschickt ({batch.id}).'
|
|
|
|
|
|
def _text(message):
|
|
return ''.join(b.text for b in message.content if b.type == 'text')
|
|
|
|
|
|
def _add_usage(total, message):
|
|
u = message.usage
|
|
total['input_tokens'] += (u.input_tokens or 0) + (getattr(u, 'cache_read_input_tokens', 0) or 0) \
|
|
+ (getattr(u, 'cache_creation_input_tokens', 0) or 0)
|
|
total['output_tokens'] += u.output_tokens or 0
|
|
stu = getattr(u, 'server_tool_use', None)
|
|
if stu:
|
|
total['web_search_requests'] += getattr(stu, 'web_search_requests', 0) or 0
|
|
|
|
|
|
def collect(radar):
|
|
"""The batch's answer, if it's there: a digest. Returns 'done', 'waiting' or an error text."""
|
|
p = radar.pending
|
|
if not p:
|
|
return 'done'
|
|
client = _client()
|
|
batch = client.messages.batches.retrieve(p['batch_id'])
|
|
if batch.processing_status != 'ended':
|
|
return 'waiting'
|
|
usage = {'input_tokens': 0, 'output_tokens': 0, 'web_search_requests': 0}
|
|
papers_json, news_md, problems = None, '', []
|
|
for res in client.messages.batches.results(batch.id):
|
|
if res.result.type != 'succeeded':
|
|
detail = getattr(getattr(res.result, 'error', None), 'type', '')
|
|
problems.append(f'{res.custom_id}: {res.result.type} {detail}'.strip())
|
|
continue
|
|
msg = res.result.message
|
|
_add_usage(usage, msg)
|
|
if msg.stop_reason == 'refusal':
|
|
problems.append(f'{res.custom_id}: abgelehnt')
|
|
elif res.custom_id == 'papers':
|
|
if msg.stop_reason == 'max_tokens':
|
|
problems.append('papers: max_tokens erreicht, Ausgabe unvollständig')
|
|
else:
|
|
try:
|
|
papers_json = json.loads(_text(msg))
|
|
except ValueError:
|
|
problems.append('papers: keine gültige JSON-Antwort')
|
|
elif res.custom_id == 'news':
|
|
news_md = _text(msg)
|
|
if papers_json is None:
|
|
# Nothing remembered as seen: the next run gets the papers again.
|
|
radar.pending = None
|
|
radar.last_error = 'Einordnung fehlgeschlagen: ' + '; '.join(problems)
|
|
radar.save(update_fields=['pending', 'last_error'])
|
|
return radar.last_error
|
|
p['errors'] = list(p.get('errors') or []) + problems
|
|
bundle = build_bundle(radar.config, p, papers_json, news_md, usage, batch.id)
|
|
store(radar, bundle, mark_seen=[r['id'] for r in p['records']])
|
|
return 'done'
|
|
|
|
|
|
def store(radar, bundle, mark_seen=None):
|
|
"""A digest in; what it reports counts as seen. Replaces one with the same batch."""
|
|
with transaction.atomic():
|
|
if bundle.get('batch_id'):
|
|
radar.digests.filter(batch_id=bundle['batch_id']).delete()
|
|
d = RadarDigest.objects.create(radar=radar, generated=date.fromisoformat(str(bundle['generated'])[:10]),
|
|
batch_id=bundle.get('batch_id') or '', bundle=bundle)
|
|
seen = list(dict.fromkeys(list(radar.seen) + list(mark_seen or [i.get('id') for i in bundle.get('items', [])])))
|
|
radar.seen = [s for s in seen if s][-SEEN_MAX:]
|
|
radar.pending = None
|
|
radar.last_error = ''
|
|
radar.save(update_fields=['seen', 'pending', 'last_error'])
|
|
return d
|
|
|
|
|
|
def tick(now=None):
|
|
"""What the cron job does: collect what's done, submit what's due. Returns lines to log."""
|
|
out = []
|
|
for radar in Radar.objects.filter(enabled=True).select_related('project'):
|
|
try:
|
|
if radar.pending:
|
|
r = collect(radar)
|
|
if r != 'waiting':
|
|
out.append(f'radar {radar.project_id}: {r}')
|
|
elif due(radar, now):
|
|
out.append(f'radar {radar.project_id}: {submit(radar)}')
|
|
except Exception as e: # noqa: BLE001 - one radar's trouble mustn't stop the others
|
|
radar.last_error = str(e)[:2000]
|
|
radar.save(update_fields=['last_error'])
|
|
out.append(f'radar {radar.project_id}: Fehler {e}')
|
|
return out
|
|
|
|
|
|
# ---- API ---------------------------------------------------------------------------------------
|
|
|
|
def _auth(request):
|
|
if not request.user.is_authenticated:
|
|
return JsonResponse({'error': 'authentication required'}, status=401)
|
|
return None
|
|
|
|
|
|
def next_run(radar, now=None):
|
|
"""When the cron job will fetch next: now (it's due), else the coming weekday and hour."""
|
|
if not radar.enabled:
|
|
return None
|
|
now = (now or timezone.now()).astimezone(ZONE)
|
|
if radar.pending:
|
|
return None
|
|
if due(radar, now):
|
|
return now.isoformat()
|
|
c = radar.config
|
|
weekday, hour = int(c.get('weekday', 6)), int(c.get('hour', 18))
|
|
t = (now + timedelta(days=(weekday - now.weekday()) % 7)).replace(hour=hour, minute=0, second=0, microsecond=0)
|
|
if t <= now:
|
|
t += timedelta(days=7)
|
|
return t.isoformat()
|
|
|
|
|
|
def _radar_json(radar):
|
|
c = radar.config
|
|
return {'enabled': radar.enabled, 'title': c.get('title', 'Radar'), 'groups': c.get('groups', []),
|
|
'group_labels': c.get('group_labels', {}), 'weekday': c.get('weekday', 6), 'hour': c.get('hour', 18),
|
|
'days': c.get('days', 7), 'sources': [s for s, on in (('PubMed', c.get('pubmed')), ('arXiv', c.get('arxiv')),
|
|
('bioRxiv/medRxiv', c.get('keywords'))) if on],
|
|
'running': bool(radar.pending), 'submitted': (radar.pending or {}).get('submitted'),
|
|
'last_run_at': radar.last_run_at.isoformat() if radar.last_run_at else None,
|
|
'last_error': radar.last_error, 'next_run': next_run(radar),
|
|
'key': bool(os.environ.get('ANTHROPIC_API_KEY'))}
|
|
|
|
|
|
def _digest_json(d):
|
|
return {'id': d.id, 'generated': d.generated.isoformat(), 'bundle': d.bundle}
|
|
|
|
|
|
@require_http_methods(['GET'])
|
|
def radar_view(request, pk):
|
|
"""The project's Radar and its latest digests (or radar: null)."""
|
|
err = _auth(request)
|
|
if err:
|
|
return err
|
|
p = Project.objects.filter(pk=pk, user=request.user).first()
|
|
if not p:
|
|
return JsonResponse({'error': 'not found'}, status=404)
|
|
radar = Radar.objects.filter(project=p).first()
|
|
if not radar:
|
|
return JsonResponse({'radar': None, 'digests': []})
|
|
return JsonResponse({'radar': _radar_json(radar),
|
|
'digests': [_digest_json(d) for d in radar.digests.all()[:DIGESTS_SHOWN]]})
|
|
|
|
|
|
@csrf_exempt
|
|
@require_http_methods(['POST'])
|
|
def radar_settings(request, pk):
|
|
"""{enabled}."""
|
|
err = _auth(request)
|
|
if err:
|
|
return err
|
|
radar = Radar.objects.filter(project__pk=pk, project__user=request.user).first()
|
|
if not radar:
|
|
return JsonResponse({'error': 'not found'}, status=404)
|
|
try:
|
|
body = json.loads(request.body or b'{}')
|
|
except ValueError:
|
|
return JsonResponse({'error': 'invalid JSON'}, status=400)
|
|
if 'enabled' in body:
|
|
radar.enabled = bool(body['enabled'])
|
|
radar.save(update_fields=['enabled'])
|
|
return JsonResponse({'radar': _radar_json(radar)})
|
|
|
|
|
|
@csrf_exempt
|
|
@require_http_methods(['POST'])
|
|
def radar_run(request, pk):
|
|
"""Fetch now rather than on the weekday: the cron job takes it within half an hour."""
|
|
err = _auth(request)
|
|
if err:
|
|
return err
|
|
radar = Radar.objects.filter(project__pk=pk, project__user=request.user).first()
|
|
if not radar:
|
|
return JsonResponse({'error': 'not found'}, status=404)
|
|
if not radar.pending:
|
|
radar.last_run_at = None
|
|
radar.enabled = True
|
|
radar.save(update_fields=['last_run_at', 'enabled'])
|
|
return JsonResponse({'radar': _radar_json(radar)})
|
|
|
|
|
|
@csrf_exempt
|
|
@require_http_methods(['POST'])
|
|
def radar_import(request, pk):
|
|
"""A digest made elsewhere (digest_batch.py --diora-url, Bearer token): the bundle."""
|
|
err = _auth(request)
|
|
if err:
|
|
return err
|
|
radar = Radar.objects.filter(project__pk=pk, project__user=request.user).first()
|
|
if not radar:
|
|
return JsonResponse({'error': 'not found'}, status=404)
|
|
try:
|
|
bundle = json.loads(request.body or b'{}')
|
|
date.fromisoformat(str(bundle['generated'])[:10])
|
|
if bundle.get('version') != 1 or not isinstance(bundle.get('items'), list):
|
|
raise ValueError
|
|
except (ValueError, KeyError, TypeError):
|
|
return JsonResponse({'error': 'kein Digest (version 1)'}, status=400)
|
|
if len(request.body) > 5_000_000:
|
|
return JsonResponse({'error': 'zu groß'}, status=400)
|
|
d = store(radar, bundle)
|
|
return JsonResponse({'digest': d.id})
|