hello-agents/Co-creation-projects/DeLunnLi-PaperGraph/backend/app/services/daily/daily_auto_refresh.py

"""每日自动刷新 —— 后台定时拉取 arXiv 新论文、智能缓存与用户行为触发."""

from __future__ import annotations

import asyncio
import logging
import time
from typing import TYPE_CHECKING

from ...settings import get_settings
import contextlib

if TYPE_CHECKING:
    from fastapi import FastAPI

logger = logging.getLogger(__name__)

_daily_compute_lock: asyncio.Lock | None = None

def get_daily_compute_lock() -> asyncio.Lock:
    global _daily_compute_lock
    if _daily_compute_lock is None:
        _daily_compute_lock = asyncio.Lock()
    return _daily_compute_lock

_EXCLUDE_MEANINGFUL_PREFIXES: tuple[str, ...] = (
    "/health",
    "/api/papers/meta/summary",
    "/api/papers/reading/calendar",
)

def request_updates_meaningful_activity(method: str, path: str) -> bool:
    p = path or ""
    if p in ("/", "/health"):
        return False
    for pref in _EXCLUDE_MEANINGFUL_PREFIXES:
        if p.startswith(pref):
            return False
    return not (method.upper() == "GET" and p.startswith("/api/papers/daily"))

def touch_meaningful_activity_if_needed(app: FastAPI, method: str, path: str) -> None:
    if not request_updates_meaningful_activity(method, path):
        return
    with contextlib.suppress(Exception):
        app.state.last_meaningful_activity_monotonic = time.monotonic()

async def daily_auto_refresh_loop(app: FastAPI) -> None:
    s = get_settings()
    if not s.papergraph_daily_auto_refresh:
        logger.info("每日论文后台自动刷新已关闭（PAPERGRAPH_DAILY_AUTO_REFRESH=0）")
        return

    idle = max(15, s.papergraph_daily_auto_refresh_idle_sec)
    poll = max(30, s.papergraph_daily_auto_refresh_poll_sec)
    grace = max(10, s.papergraph_daily_auto_refresh_startup_grace_sec)
    logger.info("每日论文后台自动刷新已启用：idle=%ss poll=%ss startup_grace=%ss", idle, poll, grace)
    await asyncio.sleep(grace)

    from starlette.concurrency import run_in_threadpool
    from ...api.dependencies import get_db_path, get_searcher
    from ...models.schemas import DailyPapersRequest
    from ...services.daily.daily_cache_store import get_cache
    from ...services.daily.daily_service import compute_daily_papers as compute_daily
    import datetime as _dt
    from ...services.papers.papers_helpers import daily_paper_identity_sig

    def _cache_nonempty(cached) -> bool:
        if not cached:
            return False
        try:
            return bool(cached.get("arxiv_selected") or []) or bool(cached.get("personalized") or [])
        except Exception:
            return False

    lock = get_daily_compute_lock()
    date_key = _dt.datetime.now().strftime("%Y-%m-%d")

    while True:
        try:
            await asyncio.sleep(poll)
            ts = getattr(app.state, "last_meaningful_activity_monotonic", None)
            if ts is not None and time.monotonic() - ts < idle:
                continue
            db_path = get_db_path()
            if _cache_nonempty(await run_in_threadpool(get_cache, db_path, date_key=date_key, cache_key='default')):
                continue
            if lock.locked():
                continue

            async with lock:
                if _cache_nonempty(await run_in_threadpool(get_cache, db_path, date_key=date_key, cache_key='default')):
                    continue
                ts2 = getattr(app.state, "last_meaningful_activity_monotonic", None)
                if ts2 is not None and time.monotonic() - ts2 < idle:
                    continue
                settings = get_settings()
                from ...services.papers import papers_converters
                body = DailyPapersRequest(force_refresh=False)
                logger.info("每日论文：后台自动拉取开始（当日无有效缓存且系统空闲）")
                await compute_daily(
                    body=body, db_path=db_path, searcher=get_searcher(),
                    daily_paper_identity_sig_fn=daily_paper_identity_sig,
                    daily_arxiv_cs_categories=settings.get_daily_arxiv_cs_categories(),
                    papergraph_to_api_fn=papers_converters.litpaper_to_api_paper,
                    logger=logger,
                )
                logger.info("每日论文：后台自动拉取完成")
        except asyncio.CancelledError:
            logger.info("每日论文后台自动刷新任务已取消")
            raise
        except Exception:
            logger.exception("每日论文后台自动拉取失败（将按 poll 间隔重试）")

def spawn_daily_auto_refresh(app: FastAPI) -> asyncio.Task:
    return asyncio.create_task(daily_auto_refresh_loop(app), name="papergraph_daily_auto_refresh")