yunke/yunke-server/app/api/routers/export.py

"""GET /api/export/chat-html — 导出一段 1:1 / 群聊天为独立 HTML 文件。

对齐云客微信客户列表「导出HTML聊天记录」：按 (channel, owner, talker_id|room_id)
拉消息（可选 start/end 时间范围），渲染自包含 HTML 聊天页（is_send 决定左右气泡），
作为 attachment 返回下载。
"""
from __future__ import annotations

from datetime import datetime, timezone
from html import escape
from typing import Annotated

import sqlalchemy as sa
from fastapi import APIRouter, Depends, Query, Response
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.deps import get_session
from app.db.models import Message

router = APIRouter(prefix="/api/export", tags=["export"])

SessionDep = Annotated[AsyncSession, Depends(get_session)]

# Hard cap on rows an HTML export will load + render. An account-wide export
# with no date range can match millions of messages; rendering all of them in
# memory risks OOM. The cap is a safety valve, not a tight budget — at 5k rows
# the HTML is already unwieldy in any browser. When hit, the response carries
# a visible warning banner so the operator knows to narrow the date range.
EXPORT_MAX_ROWS = 5000


def _ts_text(ts: int | None) -> str:
    if not ts:
        return ""
    try:
        return datetime.fromtimestamp(ts / 1000, tz=timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S")
    except (ValueError, OSError, OverflowError):
        return str(ts)


def _render_media(m) -> str:
    """Render media (image/voice/video/file) HTML for a message, or empty string.

    OSS URLs (http/https) embed directly. Local /api/ paths can't resolve in a
    standalone HTML file (opened from disk, not served by backend) → show a note.
    """
    msg_type = getattr(m, "type", None) or ""
    file_url = getattr(m, "file_url", None) or ""
    if not file_url:
        return ""

    # Local backend path → not accessible from a downloaded HTML file
    if file_url.startswith("/api/"):
        return '<div class="media">📎 [本地媒体]</div>'

    # Only embed full URLs; anything else (bare path, data URI, etc.) → note
    if not (file_url.startswith("http://") or file_url.startswith("https://")):
        return '<div class="media">📎 [本地媒体]</div>'

    url = escape(file_url, quote=True)
    if msg_type == "image":
        return (
            f'<div class="media"><img src="{url}" alt="图片" loading="lazy"'
            f' style="max-width:100%;border-radius:4px;margin:4px 0"/></div>'
        )
    if msg_type in ("voice", "audio"):
        duration = getattr(m, "duration", None)
        dur_label = f" ({duration}秒)" if duration else ""
        return (
            f'<div class="media"><audio controls preload="none" src="{url}"'
            f' style="max-width:100%"></audio><span class="dur">{dur_label}</span></div>'
        )
    if msg_type == "video":
        return (
            f'<div class="media"><video controls preload="none" src="{url}"'
            f' style="max-width:100%;border-radius:4px"></video></div>'
        )
    if msg_type == "file":
        return f'<div class="media"><a href="{url}" target="_blank">📎 附件</a></div>'
    return ""


def _render_chat_html(
    rows: list[Message],
    channel: str,
    owner: str,
    peer: str,
    *,
    warning: str = "",
) -> str:
    bubbles = []
    for m in rows:
        side = "right" if m.is_send else "left"
        ts = escape(_ts_text(getattr(m, "timestamp", None)))
        text_html = escape(getattr(m, "text", "") or "")
        media_html = _render_media(m)
        if text_html or media_html:
            body = f'<div class="b">{text_html}</div>{media_html}'
        else:
            body = '<div class="b">(空消息)</div>'
        bubbles.append(f'<div class="bubble {side}"><div class="t">{ts}</div>{body}</div>')
    title = f"聊天记录 · {escape(channel)} · {escape(owner)} / {escape(peer)}"
    return f"""<!doctype html><html lang="zh"><head><meta charset="utf-8">
<title>{title}</title><style>
body{{font-family:-apple-system,Segoe UI,sans-serif;background:#f5f5f5;margin:0;padding:16px}}
h1{{font-size:16px;margin:0 0 12px}}
.export-warning{{background:#fff3cd;border:1px solid #ffeeba;color:#856404;padding:8px 12px;border-radius:4px;margin:0 0 12px;font-size:13px}}
.bubble{{max-width:70%;margin:6px 0;padding:8px 12px;border-radius:8px;clear:both}}
.bubble.right{{float:right;background:#95ec69;text-align:right}}
.bubble.left{{float:left;background:#fff}}
.bubble .t{{font-size:11px;color:#999;margin-bottom:2px}}
.bubble .b{{font-size:14px;white-space:pre-wrap;word-break:break-word}}
.bubble .media{{margin-top:4px;font-size:13px}}
.bubble .media audio,.bubble .media video,.bubble .media img{{display:block;max-width:100%}}
.bubble .media .dur{{font-size:12px;color:#666;margin-left:4px}}
</style></head><body>{warning}<h1>{title}</h1>
{''.join(bubbles) or '<p>暂无消息</p>'}
</body></html>"""


@router.get("/chat-html")
async def export_chat_html(
    session: SessionDep,
    channel: Annotated[str, Query()],
    owner_account_id: Annotated[str, Query()],
    talker_id: Annotated[str | None, Query()] = None,
    room_id: Annotated[str | None, Query()] = None,
    start: Annotated[int | None, Query(ge=0)] = None,
    end: Annotated[int | None, Query(ge=0)] = None,
) -> Response:
    q = (
        sa.select(Message)
        .where(Message.channel == channel)
        .where(Message.owner_account_id == owner_account_id)
    )
    peer = ""
    if room_id:
        q = q.where(Message.room_id == room_id)
        peer = room_id
    elif talker_id:
        # DM：限定 room_id 为空，避免误捞群消息
        q = q.where(Message.talker_id == talker_id).where(
            (Message.room_id == None) | (Message.room_id == "")  # noqa: E711
        )
        peer = talker_id
    if start is not None:
        q = q.where(Message.timestamp >= start)
    if end is not None:
        q = q.where(Message.timestamp <= end)
    q = q.order_by(sa.nullslast(Message.timestamp.asc()), Message.msg_id.asc()).limit(
        EXPORT_MAX_ROWS
    )
    rows = (await session.execute(q)).scalars().all()
    # If we filled the cap there may be more rows beyond it — surface a visible
    # warning so the operator knows the export was truncated and can re-run
    # with a tighter date range.
    capped = len(rows) >= EXPORT_MAX_ROWS
    warning = (
        f'<div class="export-warning">⚠️ 结果超过 {EXPORT_MAX_ROWS} 条，'
        f'仅导出前 {EXPORT_MAX_ROWS} 条。请缩小日期范围。</div>'
        if capped
        else ""
    )
    html = _render_chat_html(list(rows), channel, owner_account_id, peer, warning=warning)
    safe = peer or "all"
    filename = f"chat-{channel}-{owner_account_id}-{safe}.html".replace("@", "_").replace("/", "_")
    return Response(
        content=html,
        media_type="text/html; charset=utf-8",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )