Paper-Monitoring-SW/src/paper_monitor/runner.py

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()]

    min_rel = int(app.get("min_relevance", 1))
    scholar_min_rel = int(
        app.get("google_scholar_min_relevance", min_rel)
    )

    papers = [
        p for p in papers
        if p.relevance >= (
            scholar_min_rel
            if p.source == "Google Scholar Alert"
            else min_rel
        )
    ]

    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)