callback_server.py

"""Card callback handler: listen for button clicks via WebSocket long-connection.

Uses Feishu's long-connection mode — no public URL or port needed.
The SDK maintains a WebSocket to Feishu's server and receives events in real-time.

Can be used standalone or started as a background thread from main.py.
"""

import json
import logging
import re
import threading

import lark_oapi as lark
from lark_oapi.api.im.v1 import (
    CreateMessageRequest,
    CreateMessageRequestBody,
    ReplyMessageRequest,
    ReplyMessageRequestBody,
)
from lark_oapi.event.callback.model.p2_card_action_trigger import (
    CallBackToast,
    P2CardActionTrigger,
    P2CardActionTriggerResponse,
)

from config import FEISHU_APP_ID, FEISHU_APP_SECRET, FEISHU_BOT_CHAT_ID, FEISHU_BOT_OPEN_ID, FEISHU_BOT_RECEIVE_ID_TYPE, FEISHU_REPORT_FOLDER_TOKEN
from push.feishu_bitable import update_record_bookmark, update_record_status
from push.feishu_doc import create_report_doc, delete_doc
from push.feishu_msg import send_report_notification
from reports.deep_report import generate_report
from storage.db import get_bitable_record_id, get_paper, get_mark_status, init_db, update_mark_status, upsert_user_mark

logger = logging.getLogger(__name__)

# Bot's own open_id — used to detect @mentions of the bot in group chats and
# to ignore self-sent messages. Set FEISHU_BOT_OPEN_ID in .env (see .env.example).
BOT_OPEN_ID = FEISHU_BOT_OPEN_ID

# arXiv id forms we accept: bare "2510.01174", URLs containing it, optional v1/v2 suffix.
_ARXIV_ID_RE = re.compile(r"\b(\d{4}\.\d{4,5})(?:v\d+)?\b")

# 用户在 @bot 消息里要求重新生成报告的关键词。
_REGEN_RE = re.compile(r"重新|重做|重生成|regen(?:erate)?", re.IGNORECASE)


def handle_card_action(data: P2CardActionTrigger) -> P2CardActionTriggerResponse | None:
    """Handle a card button click event."""
    event = data.event
    action = event.action
    value = action.value  # already a dict, not a string

    if not isinstance(value, dict):
        return None

    action_type = value.get("action", "")
    arxiv_id = value.get("arxiv_id", "")
    if not arxiv_id:
        return None

    if action_type == "bookmark":
        return _handle_bookmark(arxiv_id)
    elif action_type == "interested":
        return _handle_interested(event, arxiv_id)
    else:
        return None


def _handle_bookmark(arxiv_id: str) -> P2CardActionTriggerResponse:
    """Handle bookmark button click."""
    toast = CallBackToast()
    resp = P2CardActionTriggerResponse()

    record_id = get_bitable_record_id(arxiv_id)
    if record_id:
        update_record_bookmark(record_id, True)
        toast.type = "success"
        toast.content = "已收藏！可在多维表格「收藏」视图中查看"
    else:
        toast.type = "error"
        toast.content = "收藏失败：未找到对应记录"
        logger.error("Bookmark failed: no bitable record for %s", arxiv_id)

    resp.toast = toast
    return resp


def _handle_interested(event, arxiv_id: str) -> P2CardActionTriggerResponse:
    """Handle interested button click — trigger report generation."""
    user_id = event.operator.open_id if event.operator else "unknown"
    message_id = event.context.open_message_id if event.context else ""
    chat_id = event.context.open_chat_id if event.context else FEISHU_BOT_CHAT_ID

    # Check if already generated or in progress
    status, doc_url = get_mark_status(arxiv_id)
    toast = CallBackToast()
    resp = P2CardActionTriggerResponse()

    if status in ("done", "done_no_doc"):
        logger.info("Paper %s already has report (status=%s)", arxiv_id, status)
        toast.type = "info"
        toast.content = "报告已生成，请查看之前的通知消息"
        resp.toast = toast
        return resp

    if status == "generating":
        logger.info("Paper %s report is already being generated", arxiv_id)
        toast.type = "info"
        toast.content = "报告正在生成中，请耐心等待..."
        resp.toast = toast
        return resp

    logger.info("User %s marked paper %s as interested", user_id, arxiv_id)

    threading.Thread(
        target=_process_interested,
        args=(arxiv_id, chat_id, message_id, user_id),
        daemon=True,
    ).start()

    toast.type = "info"
    toast.content = "正在生成阅读报告，请稍候..."
    resp.toast = toast
    return resp


def _process_interested(arxiv_id: str, chat_id: str, message_id: str, user_open_id: str = "") -> None:
    """Generate report for interested paper and send back."""
    from datetime import UTC, datetime

    upsert_user_mark(arxiv_id, datetime.now(UTC).isoformat(), "generating")

    paper = get_paper(arxiv_id)
    if not paper:
        logger.error("Paper %s not found in DB", arxiv_id)
        _send_error(chat_id, f"论文 {arxiv_id} 未找到")
        update_mark_status(arxiv_id, "error")
        return

    logger.info("Generating report for: %s", paper["title"])

    try:
        report_md = generate_report(paper)
    except Exception:
        logger.exception("Report generation failed for %s", arxiv_id)
        _send_error(chat_id, f"报告生成失败: {paper['title'][:50]}")
        update_mark_status(arxiv_id, "error")
        return

    doc_title = f"📄 {paper['title'][:80]}"
    editor_open_ids = [user_open_id] if user_open_id and user_open_id != "unknown" else None
    doc_url = create_report_doc(
        doc_title,
        report_md,
        folder_token=FEISHU_REPORT_FOLDER_TOKEN,
        editor_open_ids=editor_open_ids,
    )

    # Update both SQLite and Bitable
    record_id = get_bitable_record_id(arxiv_id)

    if doc_url:
        update_mark_status(arxiv_id, "done", doc_url)
        if record_id:
            update_record_status(record_id, "已完成", doc_url)
        send_report_notification(paper["title"], doc_url, message_id, arxiv_id=arxiv_id)
        logger.info("Report done for %s: %s", arxiv_id, doc_url)
    else:
        update_mark_status(arxiv_id, "done_no_doc")
        if record_id:
            update_record_status(record_id, "报告已生成（文档创建失败）")
        _send_error(chat_id, f"报告已生成但文档创建失败: {paper['title'][:50]}")


def _extract_arxiv_id(text: str) -> str | None:
    m = _ARXIV_ID_RE.search(text)
    return m.group(1) if m else None


def _reply_text(message_id: str, text: str) -> None:
    """Reply to a specific message with plain text. Best-effort, errors logged."""
    client = lark.Client.builder().app_id(FEISHU_APP_ID).app_secret(FEISHU_APP_SECRET).build()
    request = (
        ReplyMessageRequest.builder()
        .message_id(message_id)
        .request_body(
            ReplyMessageRequestBody.builder()
            .content(json.dumps({"text": text}))
            .msg_type("text")
            .reply_in_thread(False)
            .build()
        )
        .build()
    )
    try:
        resp = client.im.v1.message.reply(request)
        if not resp.success():
            logger.error("Reply failed: code=%s msg=%s", resp.code, resp.msg)
    except Exception:
        logger.exception("Reply exception")


def handle_im_message(data) -> None:
    """Handle an incoming chat message: parse arXiv id from @bot or DM, queue analysis."""
    msg = data.event.message
    sender = data.event.sender
    sender_open_id = getattr(getattr(sender, "sender_id", None), "open_id", None) if sender else None

    if sender_open_id == BOT_OPEN_ID:
        logger.debug("Ignoring bot's own message")
        return

    if msg.message_type != "text":
        logger.info("Ignoring non-text message (type=%s) from %s", msg.message_type, sender_open_id)
        return

    try:
        content = json.loads(msg.content or "{}")
        text = content.get("text", "")
    except json.JSONDecodeError:
        logger.warning("Could not parse message content: %r", msg.content)
        return

    chat_type = getattr(msg, "chat_type", "")
    logger.info("Got message from %s (chat_type=%s): %r", sender_open_id, chat_type, text[:100])

    if chat_type == "group":
        mentions = msg.mentions or []
        bot_mentioned = any(
            getattr(getattr(m, "id", None), "open_id", None) == BOT_OPEN_ID for m in mentions
        )
        if not bot_mentioned:
            logger.info("Group message did not @bot, ignoring")
            return
        for m in mentions:
            key = getattr(m, "key", None)
            if key:
                text = text.replace(key, " ")

    arxiv_id = _extract_arxiv_id(text)
    chat_id = msg.chat_id
    message_id = msg.message_id
    force = bool(_REGEN_RE.search(text))

    if not arxiv_id:
        _reply_text(message_id, "🤔 没识别到 arXiv ID 或链接。试试发：https://arxiv.org/abs/2510.01174")
        return

    # Dedup against already-done / in-progress reports（force 重新生成时跳过）
    if not force:
        status, doc_url = get_mark_status(arxiv_id)
        if status == "done" and doc_url:
            _reply_text(
                message_id,
                f"📄 这篇之前已经读过了：{doc_url}\n如需重做，回复"
                f"「重新生成 {arxiv_id}」",
            )
            return
        if status == "generating":
            _reply_text(message_id, f"⏳ {arxiv_id} 正在分析中，稍等几分钟。")
            return

    prefix = "🔁 重新生成" if force else "📖 收到"
    _reply_text(message_id, f"{prefix} {arxiv_id}，正在分析（约 2 分钟）...")
    threading.Thread(
        target=_process_arxiv_request,
        args=(arxiv_id, chat_id, message_id, force, sender_open_id or ""),
        daemon=True,
    ).start()


def _process_arxiv_request(
    arxiv_id: str,
    chat_id: str,
    message_id: str,
    force: bool = False,
    sender_open_id: str = "",
) -> None:
    """Run the full single-paper pipeline triggered by an @bot message."""
    try:
        # Lazy import to avoid pulling argparse + heavy modules at startup
        from run_one_paper import run as run_one_paper

        editor_ids = [sender_open_id] if sender_open_id else None
        rc = run_one_paper(arxiv_id, force=force, editor_open_ids=editor_ids)
        if rc != 0:
            _reply_text(message_id, f"❌ {arxiv_id} 处理失败（rc={rc}），看 logs/daily.log 排查")
    except Exception as e:
        logger.exception("@bot arxiv request failed for %s", arxiv_id)
        _reply_text(message_id, f"❌ {arxiv_id} 处理异常: {type(e).__name__}: {str(e)[:100]}")


def _send_error(chat_id: str, msg: str) -> None:
    """Send an error message."""
    client = lark.Client.builder().app_id(FEISHU_APP_ID).app_secret(FEISHU_APP_SECRET).build()
    card = {
        "header": {"template": "red", "title": {"tag": "plain_text", "content": "❌ 处理失败"}},
        "elements": [{"tag": "markdown", "content": msg}],
    }
    request = (
        CreateMessageRequest.builder()
        .receive_id_type(FEISHU_BOT_RECEIVE_ID_TYPE)
        .request_body(
            CreateMessageRequestBody.builder()
            .receive_id(chat_id or FEISHU_BOT_CHAT_ID)
            .msg_type("interactive")
            .content(json.dumps(card))
            .build()
        )
        .build()
    )
    client.im.v1.message.create(request)


def start_listener() -> None:
    """Start the WebSocket callback listener in a background thread."""
    event_handler = (
        lark.EventDispatcherHandler.builder("", "")
        .register_p2_card_action_trigger(handle_card_action)
        .register_p2_im_message_receive_v1(handle_im_message)
        .build()
    )

    ws_client = lark.ws.Client(
        FEISHU_APP_ID,
        FEISHU_APP_SECRET,
        event_handler=event_handler,
        log_level=lark.LogLevel.INFO,
    )

    thread = threading.Thread(target=ws_client.start, daemon=True)
    thread.start()
    logger.info("Card callback listener started (WebSocket, background thread)")
    return thread