前言：冒險再起，電子郵件的魔法門

今天的挑戰，是我作為工程師兼探險者的另一段旅程——打造一條可靠的 Email Pipeline。就像在荒野中探險，我手上握著一張充滿未知的地圖：訂閱者名單、最新論文、摘要生成、以及最終的郵件傳送。每一個節點都是一道小小的魔法門，必須正確解鎖，才能將知識寶石順利送達使用者手中 📬。

坦白說，我今天一度想按下暫停鍵，好好休息…但是，發現如果不拆解這套 Email Pipeline，我的訂閱系統就像缺了魔法鑰匙的寶箱，永遠打不開。主動的人先享受世界嘛。於是乎，身上背著行囊、召喚 ChatGPT 陪我加班（我這慣老闆，完全不給薪水還強迫加班 🤣），再次踏上冒險。

技術解說：每日摘要的魔法流程

先看流程

# Fetch all subscribed user emails, user_id, translate, and user_language
users = get_users_task()
papers = fetch_papers_task()

if not papers:
    return

content_map = fetch_paper_content_task(papers)

for user in users:
    user_unsent_papers = filter_already_sent_papers(
        user["user_id"], papers
    )
    if not user_unsent_papers:
        continue

    # 取 top_k
    assigned_papers = user_unsent_papers[:top_k]
    # 發送給使用者
    result = process_user_task(user, assigned_papers, content_map)

這是一條每天自動執行的知識傳送路徑：抓論文、聚合內容、生成摘要、分發給訂閱者。每個節點都是一個關卡，必須通過才能順利前進。

1. 使用者抓取（Get Subscribed Users）

def get_users_task():
    with db_session() as db:
        users = get_subscribed_users(db)
    return users

取得訂閱者名單，並帶入語言、翻譯偏好等設定，確保每個使用者能收到個人化的摘要。

2. 論文抓取（Fetch Papers）

這段程式碼的核心是 fetch_papers_task：

def fetch_papers_task(days: int = 1) -> list[dict]:
    with db_session() as db:
        papers = fetch_new_papers(
            db, since_date=datetime.utcnow() - timedelta(days=days)
        )
        return [
            {
                "title": p.title or "No Title",
                "authors": p.authors or [],
                "abstract": p.abstract or "",
                "pdf_url": getattr(p, "pdf_url", None),
                "arxiv_id": getattr(p, "arxiv_id", None),
            }
            for p in papers
        ]

只抓取 since_date 之後的新論文。

3. 內容聚合（Fetch Paper Content）

def fetch_paper_content_task(papers: list[dict]) -> dict[str, str]:
    content_map = {}
    for p in papers:
        arxiv_id = p.get("arxiv_id")
        if not arxiv_id:
            continue
        raw_content = fetch_paper_content_from_qdrant(
            arxiv_id=arxiv_id,
            title=p.get("title"),
        )
        if raw_content:
            content_map[arxiv_id] = raw_content
    return content_map

這步解決了摘要生成所需原始內容的問題。沒有全文內容，摘要就只是空洞的描述。

4. 使用者處理與發送（Process User Task）

def process_user_task(user: dict, papers: list[dict], content_map: dict):
    if not user.get("email") or not papers:
        return {"user_id": user.get("user_id"), "status": "skipped"}
    summary = generate_summary((papers, content_map), user)
    send_email(subject="Daily Paper Summary", recipients=[user["email"]], body=summary)
    record_sent_papers(user["user_id"], [p["arxiv_id"] for p in papers if p.get("arxiv_id")])
    return {"user_id": user["user_id"], "status": "success", "sent_count": len(papers)}

生成 summary --> 發送 email
每位使用者的摘要生成與郵件發送都封裝成單獨 Task。

小結

這樣的流程完整串起每日 Email Pipeline 的核心四部曲：

抓論文 → 聚合內容 → 生成摘要 → 發送

每個步驟都像探索荒野的小冒險，必須精準、可靠，才能把寶石（知識）安全送到目的地。