diff --git a/yunke-server/app/api/main.py b/yunke-server/app/api/main.py index 9dd6d4d..01ac3e5 100644 --- a/yunke-server/app/api/main.py +++ b/yunke-server/app/api/main.py @@ -17,6 +17,7 @@ Includes: * ``GET /api/sms`` (see :mod:`app.api.routers.sms`). * ``GET /api/sync/status`` (see :mod:`app.api.routers.sync`). * ``GET /api/search`` (see :mod:`app.api.routers.search`). +* ``GET /api/export/chat-html`` (see :mod:`app.api.routers.export`). CORS is configured permissively (wildcard origin, no credentials) so the local frontend can call the API during development. @@ -38,6 +39,7 @@ from app.api.auth import require_auth from app.api.routers import ( calls, customer, + export, groups, media, messages, @@ -65,6 +67,7 @@ app.include_router(calls.router) app.include_router(sms.router) app.include_router(sync.router) app.include_router(search.router) +app.include_router(export.router) @app.get("/api/health") diff --git a/yunke-server/app/api/routers/export.py b/yunke-server/app/api/routers/export.py new file mode 100644 index 0000000..a4660ac --- /dev/null +++ b/yunke-server/app/api/routers/export.py @@ -0,0 +1,96 @@ +"""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_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 = escape(getattr(m, "text", "") or "") + bubbles.append( + f'
{ts}
{text}
' + ) + title = f"聊天记录 · {escape(channel)} · {escape(owner)} / {escape(peer)}" + return f""" +{title}

{title}

+{''.join(bubbles) or '

暂无消息

'} +""" + + +@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.replace("@", "_").replace("/", "_") 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}"'}, + ) diff --git a/yunke-server/tests/test_api_export.py b/yunke-server/tests/test_api_export.py new file mode 100644 index 0000000..37b56f9 --- /dev/null +++ b/yunke-server/tests/test_api_export.py @@ -0,0 +1,59 @@ +"""Tests for GET /api/export/chat-html — 导出 1:1/群聊天为自包含 HTML。 + +对齐云客微信客户/群列表「导出HTML聊天记录」：按 (channel, owner, talker_id|room_id) +拉消息（可选 start/end 时间范围），渲染为左右气泡 HTML 作为 attachment 返回下载。 +""" + +from __future__ import annotations + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import Message + + +async def _seed(session: AsyncSession) -> None: + session.add_all([ + Message(msg_id="dm1", channel="wechat", owner_account_id="wx_owner", + talker_id="wx_friend", room_id=None, text="你好", + timestamp=1000, is_send=False, type="text", type_raw=1), + Message(msg_id="dm2", channel="wechat", owner_account_id="wx_owner", + talker_id="wx_friend", room_id=None, text="在的", + timestamp=2000, is_send=True, type="text", type_raw=1), + Message(msg_id="g1", channel="wechat", owner_account_id="wx_owner", + talker_id="room_topic", room_id="g1@chatroom", text="群消息", + timestamp=3000, is_send=True, type="text", type_raw=1), + Message(msg_id="g2", channel="wechat", owner_account_id="wx_owner", + talker_id="room_topic", room_id="g1@chatroom", text="范围外", + timestamp=9999999, is_send=True, type="text", type_raw=1), + ]) + await session.commit() + + +async def test_export_dm_returns_html(make_client): + client = await make_client(_seed) + r = await client.get("/api/export/chat-html", params={ + "channel": "wechat", "owner_account_id": "wx_owner", "talker_id": "wx_friend"}) + assert r.status_code == 200 + assert r.headers["content-type"].startswith("text/html") + assert "attachment" in r.headers["content-disposition"] + body = r.text + assert "你好" in body and "在的" in body + assert "wx_owner" in body + + +async def test_export_group_with_time_range(make_client): + client = await make_client(_seed) + r = await client.get("/api/export/chat-html", params={ + "channel": "wechat", "owner_account_id": "wx_owner", "room_id": "g1@chatroom", + "start": 0, "end": 5000}) + assert r.status_code == 200 + assert "群消息" in r.text + assert "范围外" not in r.text # end=5000 过滤掉 timestamp=9999999 + + +async def test_export_empty(make_client): + client = await make_client(_seed) + r = await client.get("/api/export/chat-html", params={ + "channel": "wechat", "owner_account_id": "wx_owner", "talker_id": "none"}) + assert r.status_code == 200 + assert r.headers["content-type"].startswith("text/html")