First Commit
This commit is contained in:
@@ -0,0 +1,579 @@
|
||||
import argparse
|
||||
import requests
|
||||
import json
|
||||
import logging
|
||||
|
||||
from paper_monitor.circuit_breaker import protected_search
|
||||
from collections import OrderedDict
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
from .config import load_config, env_value, resolve_path
|
||||
from .db import PaperDB
|
||||
from .util import paper_key, clean_text
|
||||
from .scoring import classify_and_score
|
||||
from .render import render_markdown, write_markdown
|
||||
from .joplin import JoplinClient
|
||||
from .ai_enrich import (
|
||||
enrich_with_gemini,
|
||||
enrich_with_gemini_full_text,
|
||||
)
|
||||
from .collectors.ieee import (
|
||||
IEEECollector,
|
||||
IEEEAPIError,
|
||||
)
|
||||
from .collectors.semantic_scholar import SemanticScholarCollector
|
||||
from .collectors.crossref import CrossrefCollector
|
||||
from .collectors.scholar_gmail import ScholarGmailCollector
|
||||
from .oa_resolver import UnpaywallResolver
|
||||
from .pdf_utils import download_and_extract_pdf
|
||||
|
||||
log = logging.getLogger('paper_monitor')
|
||||
|
||||
|
||||
def merge_paper(dst, src):
|
||||
def longer(a, b):
|
||||
return b if len(b or '') > len(a or '') else a
|
||||
if not dst.doi and src.doi:
|
||||
dst.doi = src.doi
|
||||
dst.title = longer(dst.title, src.title)
|
||||
if len(src.authors) > len(dst.authors):
|
||||
dst.authors = src.authors
|
||||
dst.abstract = longer(dst.abstract, src.abstract)
|
||||
if not dst.url and src.url:
|
||||
dst.url = src.url
|
||||
dst.venue = longer(dst.venue, src.venue)
|
||||
if not dst.publication_date and src.publication_date:
|
||||
dst.publication_date = src.publication_date
|
||||
if not dst.year and src.year:
|
||||
dst.year = src.year
|
||||
if src.citation_count is not None:
|
||||
dst.citation_count = max(dst.citation_count or 0, src.citation_count)
|
||||
if src.source and src.source not in dst.source:
|
||||
dst.source = f'{dst.source} + {src.source}' if dst.source else src.source
|
||||
return dst
|
||||
|
||||
|
||||
def period(now, mode):
|
||||
if mode == 'weekly':
|
||||
iso = now.isocalendar()
|
||||
return f'{iso.year}-W{iso.week:02d}'
|
||||
return now.date().isoformat()
|
||||
|
||||
|
||||
def _status_code(exc):
|
||||
status = getattr(exc, 'status_code', None)
|
||||
|
||||
if status is not None:
|
||||
return status
|
||||
|
||||
response = getattr(exc, 'response', None)
|
||||
|
||||
if response is not None:
|
||||
return response.status_code
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def run(config_path, dry_run=False):
|
||||
cfg = load_config(config_path)
|
||||
app = cfg['app']
|
||||
search_cfg = cfg['search']
|
||||
sources = cfg.get('sources', {})
|
||||
logging.basicConfig(
|
||||
level=getattr(logging, app.get('log_level', 'INFO').upper(), logging.INFO),
|
||||
format='%(asctime)s %(levelname)s %(message)s',
|
||||
)
|
||||
|
||||
now = datetime.now(ZoneInfo(app.get('timezone', 'Asia/Seoul')))
|
||||
local_date = now.date().isoformat()
|
||||
lookback = int(app.get('lookback_days', 14))
|
||||
limit = int(app.get('max_papers_per_source_per_query', 50))
|
||||
queries = search_cfg.get('queries', [])
|
||||
collected = []
|
||||
|
||||
if sources.get('ieee', {}).get('enabled'):
|
||||
s = sources['ieee']
|
||||
c = IEEECollector(s['base_url'], env_value(s.get('api_key_env')), lookback, limit)
|
||||
for q in queries:
|
||||
try:
|
||||
x = protected_search(c, q)
|
||||
log.info('IEEE %r -> %d', q, len(x))
|
||||
collected += x
|
||||
|
||||
except Exception as exc:
|
||||
log.exception(
|
||||
'IEEE failed: %r',
|
||||
q,
|
||||
)
|
||||
|
||||
if isinstance(
|
||||
exc,
|
||||
IEEEAPIError,
|
||||
):
|
||||
if exc.status_code is None:
|
||||
log.error(
|
||||
'IEEE network/request error. '
|
||||
'Stopping IEEE queries for this run.'
|
||||
)
|
||||
else:
|
||||
log.error(
|
||||
'IEEE HTTP %s. '
|
||||
'Stopping IEEE queries for this run.',
|
||||
exc.status_code,
|
||||
)
|
||||
|
||||
break
|
||||
|
||||
status = _status_code(exc)
|
||||
|
||||
if status in (
|
||||
401,
|
||||
403,
|
||||
429,
|
||||
):
|
||||
log.error(
|
||||
'IEEE HTTP %s. '
|
||||
'Stopping IEEE queries for this run.',
|
||||
status,
|
||||
)
|
||||
break
|
||||
|
||||
log.error(
|
||||
'Unexpected IEEE error. '
|
||||
'Stopping IEEE queries for this run.'
|
||||
)
|
||||
break
|
||||
|
||||
# Network problems
|
||||
if isinstance(
|
||||
exc,
|
||||
(
|
||||
requests.exceptions.Timeout,
|
||||
requests.exceptions.ConnectionError,
|
||||
),
|
||||
):
|
||||
log.error(
|
||||
'IEEE network error. '
|
||||
'Stopping IEEE queries for this run.'
|
||||
)
|
||||
break
|
||||
|
||||
if sources.get('semantic_scholar', {}).get('enabled'):
|
||||
s = sources['semantic_scholar']
|
||||
c = SemanticScholarCollector(s['base_url'], env_value(s.get('api_key_env')), lookback, limit)
|
||||
for q in queries:
|
||||
try:
|
||||
x = protected_search(c, q)
|
||||
log.info('Semantic Scholar %r -> %d', q, len(x))
|
||||
collected += x
|
||||
except Exception as exc:
|
||||
log.exception('Semantic Scholar failed: %r', q)
|
||||
status = _status_code(exc)
|
||||
# Authentication / authorization / rate-limit errors:
|
||||
# Remaining queries are unlikely to succeed in this run.
|
||||
if status in (401, 403, 429):
|
||||
log.error(
|
||||
'Semantic Scholar HTTP %s. '
|
||||
'Stopping Semantic Scholar queries for this run.',
|
||||
status,
|
||||
)
|
||||
break
|
||||
# Network problem:
|
||||
# Do not repeat the same timeout for every query.
|
||||
if isinstance(
|
||||
exc,
|
||||
(
|
||||
requests.exceptions.Timeout,
|
||||
requests.exceptions.ConnectionError,
|
||||
),
|
||||
):
|
||||
log.error(
|
||||
'Semantic Scholar network error. '
|
||||
'Stopping Semantic Scholar queries for this run.'
|
||||
)
|
||||
break
|
||||
|
||||
if sources.get('crossref', {}).get('enabled'):
|
||||
s = sources['crossref']
|
||||
c = CrossrefCollector(s['base_url'], env_value(s.get('mailto_env')), lookback, limit)
|
||||
for q in queries:
|
||||
try:
|
||||
x = protected_search(c, q)
|
||||
log.info('Crossref %r -> %d', q, len(x))
|
||||
collected += x
|
||||
|
||||
except Exception as exc:
|
||||
log.exception('Crossref failed: %r', q)
|
||||
|
||||
status = _status_code(exc)
|
||||
|
||||
# HTTP errors that are unlikely to recover during this run
|
||||
if status in (401, 403, 429):
|
||||
log.error(
|
||||
'Crossref HTTP %s. '
|
||||
'Stopping Crossref queries for this run.',
|
||||
status,
|
||||
)
|
||||
break
|
||||
|
||||
# Network problems
|
||||
if isinstance(
|
||||
exc,
|
||||
(
|
||||
requests.exceptions.Timeout,
|
||||
requests.exceptions.ConnectionError,
|
||||
),
|
||||
):
|
||||
log.error(
|
||||
'Crossref network error. '
|
||||
'Stopping Crossref queries for this run.'
|
||||
)
|
||||
break
|
||||
|
||||
if sources.get('google_scholar_alert', {}).get('enabled'):
|
||||
s = sources['google_scholar_alert']
|
||||
c = ScholarGmailCollector(
|
||||
env_value(s.get('gmail_address_env')),
|
||||
env_value(s.get('gmail_app_password_env')),
|
||||
s.get('imap_host', 'imap.gmail.com'),
|
||||
s.get('mailbox', 'INBOX'),
|
||||
s.get('sender_contains', ''),
|
||||
s.get('subject_contains', ''),
|
||||
lookback,
|
||||
)
|
||||
try:
|
||||
x = c.collect()
|
||||
log.info('Google Scholar Alert -> %d', len(x))
|
||||
collected += x
|
||||
except Exception:
|
||||
log.exception('Google Scholar Alert failed')
|
||||
|
||||
merged = OrderedDict()
|
||||
for p in collected:
|
||||
p.title = clean_text(p.title)
|
||||
if not p.title:
|
||||
continue
|
||||
k = paper_key(p.doi, p.title)
|
||||
merged[k] = merge_paper(merged[k], p) if k in merged else p
|
||||
|
||||
papers = [classify_and_score(p, search_cfg) for p in merged.values()]
|
||||
papers = [p for p in papers if p.relevance >= int(app.get('min_relevance', 1))]
|
||||
|
||||
db_path = resolve_path(cfg, app.get('database_path', './data/papers.db'))
|
||||
db = PaperDB(db_path)
|
||||
new = []
|
||||
|
||||
try:
|
||||
# Determine which accepted papers are genuinely new before AI processing.
|
||||
new_candidates = [
|
||||
p for p in papers
|
||||
if not db.exists(p)
|
||||
]
|
||||
|
||||
new_keys = {
|
||||
paper_key(p.doi, p.title)
|
||||
for p in new_candidates
|
||||
}
|
||||
|
||||
# Resolve Open Access information for new papers only.
|
||||
oa = cfg.get('oa', {})
|
||||
|
||||
if oa.get('enabled'):
|
||||
email = env_value(
|
||||
oa.get('email_env', 'UNPAYWALL_EMAIL')
|
||||
)
|
||||
|
||||
if not email:
|
||||
log.warning(
|
||||
'Unpaywall email missing; '
|
||||
'OA lookup skipped'
|
||||
)
|
||||
|
||||
else:
|
||||
resolver = UnpaywallResolver(email)
|
||||
|
||||
oa_candidates = [
|
||||
p for p in new_candidates
|
||||
if (
|
||||
p.doi
|
||||
and p.doi.strip()
|
||||
and not (
|
||||
p.pdf_url
|
||||
and p.pdf_url.strip()
|
||||
)
|
||||
)
|
||||
]
|
||||
|
||||
log.info(
|
||||
'Unpaywall candidates: %d / %d new papers',
|
||||
len(oa_candidates),
|
||||
len(new_candidates),
|
||||
)
|
||||
|
||||
for p in oa_candidates:
|
||||
try:
|
||||
resolver.resolve(p)
|
||||
|
||||
log.info(
|
||||
'Unpaywall resolved: status=%s pdf=%s title=%s',
|
||||
p.oa_status or '-',
|
||||
'YES' if p.pdf_url else 'NO',
|
||||
p.title,
|
||||
)
|
||||
|
||||
except Exception:
|
||||
# OA lookup failure must never stop paper collection.
|
||||
log.exception(
|
||||
'Unpaywall lookup failed; continuing without OA information: %s',
|
||||
p.title,
|
||||
)
|
||||
|
||||
# ============================================================
|
||||
# 3. AI enrichment
|
||||
#
|
||||
# Candidates:
|
||||
#
|
||||
# A. Papers discovered for the first time in this run
|
||||
# B. Papers from earlier runs whose ai_status == "failed"
|
||||
#
|
||||
# Historical papers with ai_status == "" are NOT automatically
|
||||
# processed. This prevents unexpectedly processing the entire DB.
|
||||
# ============================================================
|
||||
|
||||
|
||||
ai = cfg.get('ai', {})
|
||||
|
||||
if ai.get('enabled'):
|
||||
api_key = env_value(
|
||||
ai.get('api_key_env', 'GEMINI_API_KEY')
|
||||
)
|
||||
|
||||
model = ai.get(
|
||||
'model',
|
||||
'gemini-3.6-flash',
|
||||
)
|
||||
|
||||
max_papers = int(
|
||||
ai.get('max_papers_per_run', 20)
|
||||
)
|
||||
|
||||
# Retry only papers explicitly marked as failed.
|
||||
# Existing historical papers with ai_status="" are NOT retried.
|
||||
retry_candidates = db.list_ai_failed(
|
||||
limit=max_papers
|
||||
)
|
||||
|
||||
# Avoid processing the same paper twice if it is somehow
|
||||
# present in both new_candidates and retry_candidates.
|
||||
candidate_map = {}
|
||||
|
||||
for p in new_candidates:
|
||||
candidate_map[paper_key(p.doi, p.title)] = p
|
||||
|
||||
for p in retry_candidates:
|
||||
k = paper_key(p.doi, p.title)
|
||||
|
||||
if k not in candidate_map:
|
||||
candidate_map[k] = p
|
||||
|
||||
ai_candidates = list(
|
||||
candidate_map.values()
|
||||
)
|
||||
|
||||
ai_candidates = sorted(
|
||||
ai_candidates,
|
||||
key=lambda x: x.relevance,
|
||||
reverse=True,
|
||||
)[:max_papers]
|
||||
|
||||
if not api_key:
|
||||
log.warning(
|
||||
'Gemini API key missing; AI enrichment skipped'
|
||||
)
|
||||
|
||||
else:
|
||||
log.info(
|
||||
'Gemini enrichment candidates: %d '
|
||||
'(new=%d retry=%d)',
|
||||
len(ai_candidates),
|
||||
len(new_candidates),
|
||||
len(retry_candidates),
|
||||
)
|
||||
|
||||
for p in ai_candidates:
|
||||
full_text_done = False
|
||||
abstract_done = False
|
||||
had_failure = False
|
||||
|
||||
# -----------------------------------------
|
||||
# 1. Prefer OA PDF full-text analysis
|
||||
# -----------------------------------------
|
||||
if p.pdf_url and p.pdf_url.strip():
|
||||
try:
|
||||
log.info(
|
||||
'Downloading OA PDF: %s',
|
||||
p.title,
|
||||
)
|
||||
|
||||
full_text = download_and_extract_pdf(
|
||||
p.pdf_url
|
||||
)
|
||||
|
||||
log.info(
|
||||
'PDF extracted: chars=%d title=%s',
|
||||
len(full_text),
|
||||
p.title,
|
||||
)
|
||||
|
||||
enrich_with_gemini_full_text(
|
||||
p,
|
||||
full_text,
|
||||
api_key,
|
||||
model,
|
||||
)
|
||||
|
||||
full_text_done = True
|
||||
p.ai_status = 'done'
|
||||
|
||||
log.info(
|
||||
'Gemini full-text enriched: '
|
||||
'relevance=%d categories=%s title=%s',
|
||||
p.relevance,
|
||||
p.categories,
|
||||
p.title,
|
||||
)
|
||||
|
||||
except Exception:
|
||||
had_failure = True
|
||||
|
||||
log.exception(
|
||||
'Full-text AI failed; '
|
||||
'trying abstract fallback: %s',
|
||||
p.title,
|
||||
)
|
||||
|
||||
# -----------------------------------------
|
||||
# 2. Abstract fallback
|
||||
# -----------------------------------------
|
||||
if (
|
||||
not full_text_done
|
||||
and p.abstract
|
||||
and p.abstract.strip()
|
||||
):
|
||||
try:
|
||||
enrich_with_gemini(
|
||||
p,
|
||||
api_key,
|
||||
model,
|
||||
)
|
||||
|
||||
abstract_done = True
|
||||
p.ai_status = 'done'
|
||||
|
||||
log.info(
|
||||
'Gemini abstract enriched: '
|
||||
'relevance=%d categories=%s title=%s',
|
||||
p.relevance,
|
||||
p.categories,
|
||||
p.title,
|
||||
)
|
||||
|
||||
except Exception:
|
||||
had_failure = True
|
||||
|
||||
log.exception(
|
||||
'Gemini abstract enrichment failed; '
|
||||
'keeping original metadata: %s',
|
||||
p.title,
|
||||
)
|
||||
|
||||
# -----------------------------------------
|
||||
# 3. Final AI status
|
||||
# -----------------------------------------
|
||||
if not full_text_done and not abstract_done:
|
||||
if had_failure:
|
||||
p.ai_status = 'failed'
|
||||
|
||||
log.warning(
|
||||
'AI status=failed: %s',
|
||||
p.title,
|
||||
)
|
||||
|
||||
else:
|
||||
p.ai_status = 'skipped'
|
||||
|
||||
log.info(
|
||||
'AI status=skipped; '
|
||||
'no usable PDF or abstract: %s',
|
||||
p.title,
|
||||
)
|
||||
|
||||
k = paper_key(p.doi, p.title)
|
||||
|
||||
if k not in new_keys:
|
||||
db.upsert(
|
||||
p,
|
||||
local_date=local_date,
|
||||
)
|
||||
|
||||
# Store every accepted paper, regardless of AI success/failure.
|
||||
for p in papers:
|
||||
if db.upsert(p, local_date=local_date):
|
||||
new.append(p)
|
||||
|
||||
# Daily report is cumulative for the current local date, so reruns safely replace
|
||||
# the same Joplin note without losing papers discovered earlier that day.
|
||||
report_papers = db.list_first_seen_on(local_date)
|
||||
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
jcfg = cfg.get('joplin', {})
|
||||
suffix = period(now, jcfg.get('note_mode', 'daily'))
|
||||
title = f'{jcfg.get("note_title_prefix", "Radar Literature")} - {suffix}'
|
||||
body = render_markdown(report_papers, title, now)
|
||||
out_dir = resolve_path(cfg, app.get('output_dir', './data/outbox'))
|
||||
filename = title.replace('/', '-') + '.md'
|
||||
out = write_markdown(out_dir, filename, body)
|
||||
|
||||
note_id = ''
|
||||
if jcfg.get('enabled') and not dry_run:
|
||||
token = env_value(jcfg.get('token_env'))
|
||||
if not token:
|
||||
log.warning('Joplin token missing; Markdown only')
|
||||
else:
|
||||
try:
|
||||
jc = JoplinClient(jcfg.get('base_url', 'http://127.0.0.1:41184'), token)
|
||||
jc.ping()
|
||||
folder = jc.ensure_folder_path(jcfg.get('notebook_path', ['Research', 'Radar Papers']))
|
||||
note_id = jc.create_or_update_note(title, body, folder, bool(jcfg.get('update_existing_note', True)))
|
||||
for tag in jcfg.get('tags', []):
|
||||
jc.add_tag_to_note(jc.ensure_tag(tag), note_id)
|
||||
except Exception:
|
||||
log.exception('Joplin failed; Markdown kept at %s', out)
|
||||
|
||||
result = {
|
||||
'collected': len(collected),
|
||||
'merged': len(merged),
|
||||
'accepted': len(papers),
|
||||
'new': len(new),
|
||||
'report_count': len(report_papers),
|
||||
'report_title': title,
|
||||
'markdown': out,
|
||||
'joplin_note_id': note_id,
|
||||
}
|
||||
result_path = Path(resolve_path(cfg, './data/last_result.json'))
|
||||
result_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
result_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
|
||||
log.info('Result: %s', result)
|
||||
return result
|
||||
|
||||
|
||||
def cli():
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument('--config', default='config.yaml')
|
||||
p.add_argument('--dry-run', action='store_true')
|
||||
a = p.parse_args()
|
||||
run(a.config, a.dry_run)
|
||||
Reference in New Issue
Block a user