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)]


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) -> 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}}
.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><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())
    rows = (await session.execute(q)).scalars().all()
    html = _render_chat_html(list(rows), channel, owner_account_id, peer)
    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}"'},
    )