hello-agents/Co-creation-projects/DeLunnLi-PaperGraph/backend/app/services/papers/papers_library_service.py

"""文献库管理服务 —— 论文存储、分类、标签、阅读记录与 PDF 下载管理."""

import logging
import os
import time

from fastapi import BackgroundTasks, HTTPException, Request

from ...settings import get_settings
from ...models.schemas import (
    DeletePaperResponse,
    LibraryCategoriesResponse,
    LibraryCategoryFolder,
    Paper,
    PapersResponse,
    SavePapersRequest,
    SavePapersResponse,
    UpdatePaperRequest,
    UpdatePaperResponse,
)

logger = logging.getLogger(__name__)

def _merge_tag_lists(base: list[str], extra: list[str], max_n: int = 24) -> list[str]:
    out: list[str] = []
    seen: set[str] = set()
    for t in (base or []) + (extra or []):
        k = (t or "").strip()
        if not k:
            continue
        low = k.lower()
        if low in seen:
            continue
        seen.add(low)
        out.append(k)
        if len(out) >= max_n:
            break
    return out

def list_library_categories(*, db) -> LibraryCategoriesResponse:
    try:
        from app.core.paper_paths import LIBRARY_PDF_ROOT_DIR

        items = db.list_library_category_folders()
        return LibraryCategoriesResponse(
            success=True,
            store_root=LIBRARY_PDF_ROOT_DIR,
            folders=[LibraryCategoryFolder(**x) for x in items],
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

def get_library(
    *,
    db,
    litpaper_to_api_paper_fn,
    limit: int = 50,
    offset: int = 0,
    q: str | None = None,
    year_from: int | None = None,
    year_to: int | None = None,
    read_status=None,
    tags: str | None = None,
    category: str | None = None,
) -> PapersResponse:
    try:

        tag_list = [t.strip() for t in tags.split(",")] if tags else None
        cat = (category or "").strip() or None
        if q or year_from or year_to or read_status or tag_list or cat:
            papers_data = db.search_library(
                query=q,
                tags=tag_list,
                year_from=year_from,
                year_to=year_to,
                read_status=read_status.value if read_status else None,
                category=cat,
                limit=limit,
                offset=offset,
            )
        else:
            papers_data = db.get_all_papers(limit=limit, offset=offset, order_by="created_at DESC")

        ids_missing_pdf = [
            int(p.id)
            for p in papers_data
            if p.id is not None and not (getattr(p, "local_pdf_path", None) or "").strip()
        ]
        if ids_missing_pdf:
            repaired = db.repair_library_local_pdf_paths_batch(ids_missing_pdf)
            for p in papers_data:
                if p.id is not None and int(p.id) in repaired:
                    p.local_pdf_path = repaired[int(p.id)]

        total = db.count_papers() if not (q or year_from or year_to or read_status or tag_list or cat) else len(papers_data) + (1 if len(papers_data) >= limit else 0)
        papers = [litpaper_to_api_paper_fn(p) for p in papers_data]
        return PapersResponse(success=True, total=total or len(papers), papers=papers)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

async def save_papers(
    *,
    db,
    request: SavePapersRequest,
    background_tasks: BackgroundTasks,
    api_to_lit_fn,
    litpaper_to_api_paper_fn,
) -> SavePapersResponse:
    try:
        t0 = time.perf_counter()
        from ..graph.kg_relations import build_relations_for_new_paper
        from app.core.paper_paths import (
            LIBRARY_PDF_ROOT_DIR,
            library_pdf_relative_path,
            normalize_library_category_display,
        )
        from app.core.pdf_download import download_paper_pdf_to_path, resolve_paper_pdf_url

        lit_list = [api_to_lit_fn(p) for p in request.papers]
        for api_p, lit_p in zip(request.papers, lit_list):
            if (api_p.source_url or "").strip():
                lit_p.source_url = (api_p.source_url or "").strip()
            if (api_p.pdf_url or "").strip():
                lit_p.pdf_url = (api_p.pdf_url or "").strip()
            if (api_p.doi or "").strip():
                lit_p.doi = (api_p.doi or "").strip()
            if (api_p.arxiv_id or "").strip():
                lit_p.arxiv_id = (api_p.arxiv_id or "").strip()
            # Fill missing arXiv ID from DOI/source URL.
            if not (getattr(lit_p, "arxiv_id", None) or "").strip():
                import re
                for field_val in ((api_p.doi or "").strip(), (api_p.source_url or "").strip()):
                    m = re.search(r"arxiv/([\d.]+)", field_val, re.I)
                    if m:
                        lit_p.arxiv_id = m.group(1)
                        break
            # Infer venue type from journal name when absent.
            if not (getattr(lit_p, "venue_type", None) or "").strip():
                j = (getattr(lit_p, "journal", None) or "").strip()
                if j:
                    import re
                    if re.search(r"(?i)\b(journal|transactions|letters|magazine|review|annals|acta|bulletin)\b", j):
                        lit_p.venue_type = "journal"
                    else:
                        lit_p.venue_type = "conference"

            # Fill missing DOI from source URL.
            if not (getattr(lit_p, "doi", None) or "").strip():
                su = (api_p.source_url or "").strip()
                import re
                m = re.search(r"doi\.org/(10\.\S+)", su, re.I)
                if m:
                    lit_p.doi = m.group(1)

            if (api_p.category or "").strip():
                lit_p.category = normalize_library_category_display(api_p.category)

        llm_classified = 0

        if lit_list and getattr(request, "llm_classify", True):
            try:
                existing_categories = []
                try:
                    existing_categories = db.list_library_categories_by_count(limit=80)
                except Exception:
                    existing_categories = []

                from ...agents import get_paper_analysis_agent

                agent = get_paper_analysis_agent()
                for lit_p in lit_list:
                    cat, extra = agent.classify_for_library(
                        lit_p.title,
                        lit_p.abstract,
                        lit_p.journal,
                        getattr(lit_p, "keywords", None) or [],
                        existing_categories=existing_categories,
                    )
                    lit_p.category = cat
                    lit_p.tags = _merge_tag_lists(lit_p.tags or [], extra)
                    if not getattr(lit_p, "venue_type", None):
                        lit_p.venue_type = agent.classify_venue_type(lit_p.journal)
                    llm_classified += 1
            except Exception as e:
                logger.warning("大模型归类未执行：%s", e)
                for lit_p in lit_list:
                    lit_p.category = normalize_library_category_display(getattr(lit_p, "category", None))
        elif lit_list:

            for lit_p in lit_list:
                lit_p.category = normalize_library_category_display(getattr(lit_p, "category", None))

        for lit_p in lit_list:
            lit_p.category = normalize_library_category_display(getattr(lit_p, "category", None))

        t_after_classify = time.perf_counter()

        # Backfill missing abstracts via Tavily.
        for lit_p in lit_list:
            if not (lit_p.abstract or "").strip() and lit_p.title:
                try:
                    from ...settings import get_settings as _gs
                    _ak = getattr(_gs(), "tavily_api_key", "").strip()
                    if _ak:
                        import httpx
                        _resp = httpx.post("https://api.tavily.com/search", json={
                            "api_key": _ak, "query": f"{lit_p.title} paper abstract", "max_results": 3}, timeout=15.0)
                        for _it in (_resp.json().get("results") or []):
                            if len(_it.get("content","")) > 100:
                                lit_p.abstract = _it["content"][:2000]
                                break
                except Exception: pass

        ids, added_new, updated_existing = db.add_papers(lit_list)
        t_after_db = time.perf_counter()

        t_after_memory = time.perf_counter()

        for pid in ids or []:
            try:
                if pid is None or int(pid) <= 0:
                    continue
                build_relations_for_new_paper(db.db_path, int(pid))
            except Exception:
                continue

        pdf_downloaded = 0
        if request.download_pdfs and lit_list and ids:
            s = get_settings()
            mail = (s.ncbi_email or "").strip()
            data_root = os.path.dirname(os.path.abspath(db.db_path))
            os.makedirs(os.path.join(data_root, LIBRARY_PDF_ROOT_DIR), exist_ok=True)
            for lit_p, pid in zip(lit_list, ids):
                if pid is None or pid < 0:
                    continue
                relpath = library_pdf_relative_path(
                    getattr(lit_p, "category", None), int(pid), getattr(lit_p, "title", None)
                )
                dest = os.path.join(data_root, relpath)
                os.makedirs(os.path.dirname(dest), exist_ok=True)
                try:
                    resolved = resolve_paper_pdf_url(lit_p, email=mail)
                except Exception as ex:
                    logger.warning("解析 PDF 链接异常（已跳过该条 PDF）: %s", ex, exc_info=True)
                    resolved = None
                if not resolved:
                    logger.warning(
                        "保存跳过 PDF：无可用链接 title=%r doi=%r",
                        lit_p.title,
                        lit_p.doi,
                    )
                if os.path.isfile(dest) and os.path.getsize(dest) >= 256:
                    db.set_local_pdf_path(int(pid), relpath)
                    pdf_downloaded += 1
                    continue
                if resolved and download_paper_pdf_to_path(lit_p, dest, email=mail):
                    db.set_local_pdf_path(int(pid), relpath)
                    pdf_downloaded += 1
                elif resolved:
                    logger.warning("保存 PDF 下载失败 title=%r", lit_p.title)
        t_after_pdf = time.perf_counter()

        ids_ok = [int(x) for x in ids if x is not None and int(x) >= 0]
        if ids_ok:
            need_repair = []
            for pid in ids_ok:
                row = db.get_paper_by_id(pid)
                if row and not (getattr(row, "local_pdf_path", None) or "").strip():
                    need_repair.append(pid)
            if need_repair:
                db.repair_library_local_pdf_paths_batch(need_repair)

        msg = None
        if (
            request.download_pdfs
            and lit_list
            and pdf_downloaded == 0
            and any(pid is not None and pid >= 0 for pid in ids)
        ):
            msg = "未能写入本地 PDF：请确认含 arXiv / pdf_url 等可下载链接。"

        logger.info(
            "POST /api/papers/save timing total=%.3fs classify=%.3fs db=%.3fs memory=%.3fs pdf=%.3fs"
            " papers=%d llm_classify=%s download_pdfs=%s",
            (t_after_pdf - t0),
            (t_after_classify - t0),
            (t_after_db - t_after_classify),
            (t_after_memory - t_after_db),
            (t_after_pdf - t_after_memory),
            len(lit_list),
            getattr(request, "llm_classify", True),
            getattr(request, "download_pdfs", False),
        )

        return SavePapersResponse(
            success=True,
            added=int(added_new),
            updated=int(updated_existing),
            ids=ids,
            pdf_downloaded=pdf_downloaded,
            llm_classified=llm_classified,
            message=msg,
        )
    except Exception as e:
        logger.exception("POST /api/papers/save 失败")
        raise HTTPException(status_code=500, detail=str(e))

def get_paper_by_id(*, db, paper_id: int, litpaper_to_api_paper_fn) -> Paper:
    p = db.get_paper_by_id(paper_id)
    if not p:
        raise HTTPException(status_code=404, detail="文献不存在")

    if p.id is not None and not (getattr(p, "local_pdf_path", None) or "").strip():
        repaired = db.repair_library_local_pdf_paths_batch([int(p.id)])
        if repaired:
            p2 = db.get_paper_by_id(paper_id)
            if p2:
                p = p2
    return litpaper_to_api_paper_fn(p)

def update_paper_by_id(*, db, paper_id: int, body: UpdatePaperRequest) -> UpdatePaperResponse:
    try:
        from app.core.paper_paths import normalize_library_category_display

        fields = {}
        if body.notes is not None:
            fields["notes"] = body.notes
        if body.tags is not None:
            fields["tags"] = body.tags
        if body.category is not None:
            fields["category"] = normalize_library_category_display(body.category)
        if body.rating is not None:
            fields["rating"] = body.rating
        if body.read_status is not None:
            fields["read_status"] = body.read_status.value
        if body.importance is not None:
            fields["importance"] = body.importance
        ok = db.update_paper(paper_id, **fields)
        if not ok:
            raise HTTPException(status_code=404, detail="未更新或文献不存在")
        return UpdatePaperResponse(success=True, updated_fields=list(fields.keys()))
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

def delete_paper_by_id(*, db, paper_id: int) -> DeletePaperResponse:
    ok = db.delete_paper(paper_id)
    if not ok:
        raise HTTPException(status_code=404, detail="文献不存在")
    return DeletePaperResponse(success=True, message="已删除")

def build_library_pdf_response_service(*, paper_id: int, request: Request, db_path: str, logger_obj):
    from ..pdf.pdf_service import build_library_pdf_response

    return build_library_pdf_response(
        paper_id=int(paper_id),
        request=request,
        db_path=db_path,
        logger=logger_obj,
    )