diff --git a/skills/export-chat-history/SKILL.md b/skills/export-chat-history/SKILL.md new file mode 100644 index 0000000..d5452f1 --- /dev/null +++ b/skills/export-chat-history/SKILL.md @@ -0,0 +1,82 @@ +--- +name: export-chat-history +description: "将当前微信群聊指定日期或时间范围内的聊天记录导出为 Excel 并发送到当前群聊。当用户要求导出、下载、整理或备份今天、昨天或某段时间的群聊记录、聊天记录、群消息时使用。" +--- + +# 导出群聊记录 + +运行 `scripts/export_chat_history.py`，查询当前微信群聊的聊天记录，生成 Excel 工作簿并直接发送到当前群聊。 + +## 入参 + +```json +{ + "type": "object", + "properties": { + "date": { + "type": "string", + "description": "按自然日导出，格式为 YYYY-MM-DD。" + }, + "start_time": { + "type": "string", + "description": "自定义范围的开始时间（包含），格式为 YYYY-MM-DD HH:mm 或 YYYY-MM-DD HH:mm:ss。" + }, + "end_time": { + "type": "string", + "description": "自定义范围的结束时间（不包含），格式为 YYYY-MM-DD HH:mm 或 YYYY-MM-DD HH:mm:ss。" + } + }, + "additionalProperties": false +} +``` + +遵守以下时间规则： + +- 使用 Asia/Shanghai 时区。 +- 用户未指定时间时，不传时间参数，默认导出今天 00:00 至脚本执行时的记录。 +- 用户说“今天”时传当天 `--date`；说“昨天”时传昨天的 `--date`。 +- 自定义范围必须同时传 `--start_time` 和 `--end_time`，结束时间必须晚于开始时间。 +- `--date` 不能与自定义范围同时使用。 + +## 执行 + +导出今天： + +```bash +python3 scripts/export_chat_history.py --date 2026-07-29 +``` + +导出自定义范围： + +```bash +python3 scripts/export_chat_history.py --start_time '2026-07-28 09:00' --end_time '2026-07-28 18:00' +``` + +未指定时间时： + +```bash +python3 scripts/export_chat_history.py +``` + +脚本由 `execute_skill_script` 在技能根目录运行。运行时自动读取客户端注入的以下环境变量： + +- `ROBOT_FROM_WX_ID`：当前群聊 ID。 +- `ROBOT_WX_ID`：机器人自身微信 ID，用于排除机器人消息。 +- `ROBOT_WECHAT_CLIENT_PORT`：机器人客户端端口，用于发送 Excel。 +- `ROBOT_CODE`、`MYSQL_HOST`、`MYSQL_PORT`、`MYSQL_USER`、`MYSQL_PASSWORD`：机器人实例数据库连接。 + +## 查询与导出规则 + +- 仅允许导出当前群聊 `ROBOT_FROM_WX_ID`，不接受外部传入群 ID。 +- 复用客户端群聊总结的查询语义：查询 `messages`，联表 `chat_room_members` 取得群成员备注或昵称，排除机器人自身发送的消息。 +- 导出普通文本消息，以及引用消息、网页分享和文件消息中可读的标题或描述。 +- 按消息时间、消息 ID 升序排列，结束时间不包含在结果中。 +- Excel 包含“导出说明”和“聊天记录”两个工作表；聊天记录列为序号、发送时间、发送人、发送人微信 ID、消息内容。 +- 单次最多导出 50,000 条记录；超过上限时停止并提示缩短时间范围。 +- 生成后直接调用客户端本地文件发送接口；发送成功后删除临时文件。 + +## 结果处理 + +- 仅当脚本返回 JSON 且 `ok` 为 `true` 时，告知用户导出成功；脚本已经发送文件，不要再次调用发送文件技能。 +- 没有符合条件的记录时，说明该时间范围没有可导出的聊天记录，并建议用户调整日期或范围。 +- 查询、生成或发送失败时，按脚本错误说明原因，不要声称文件已发送。 diff --git a/skills/export-chat-history/agents/openai.yaml b/skills/export-chat-history/agents/openai.yaml new file mode 100644 index 0000000..1058b86 --- /dev/null +++ b/skills/export-chat-history/agents/openai.yaml @@ -0,0 +1,4 @@ +interface: + display_name: "导出聊天记录" + short_description: "将当前群聊指定时间范围的聊天记录整理为 Excel 并发送" + default_prompt: "使用 $export-chat-history 将当前群聊今天的聊天记录导出为 Excel 并发送给我。" diff --git a/skills/export-chat-history/scripts/bootstrap.py b/skills/export-chat-history/scripts/bootstrap.py new file mode 100644 index 0000000..697f1f7 --- /dev/null +++ b/skills/export-chat-history/scripts/bootstrap.py @@ -0,0 +1,131 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import hashlib +import shutil +import subprocess +import sys +import traceback +from pathlib import Path + +sys.stderr = sys.stdout + + +def _skill_root(script_dir: Path) -> Path: + return script_dir.parent + + +def _venv_dir(script_dir: Path) -> Path: + return _skill_root(script_dir) / ".venv" + + +def _venv_python(venv_dir: Path) -> Path: + if sys.platform == "win32": + return venv_dir / "Scripts" / "python.exe" + return venv_dir / "bin" / "python" + + +def _stamp_file(venv_dir: Path) -> Path: + return venv_dir / ".req_hash" + + +def _file_hash(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _deps_up_to_date(requirements_file: Path, venv_dir: Path) -> bool: + stamp = _stamp_file(venv_dir) + return stamp.is_file() and stamp.read_text().strip() == _file_hash( + requirements_file + ) + + +def _write_stamp(requirements_file: Path, venv_dir: Path) -> None: + _stamp_file(venv_dir).write_text(_file_hash(requirements_file)) + + +def _ensure_venv(venv_dir: Path, venv_python: Path) -> int: + if venv_python.is_file(): + return 0 + + sys.stdout.write(f"未检测到技能虚拟环境，正在创建: {venv_dir}\n") + python_executable = sys.executable or next( + ( + shutil.which(candidate) + for candidate in ("python3", "python") + if shutil.which(candidate) + ), + None, + ) + if not python_executable: + raise RuntimeError("无法找到 Python 解释器路径") + + try: + subprocess.run( + [python_executable, "-m", "venv", str(venv_dir)], + check=True, + stdout=sys.stdout, + stderr=sys.stdout, + ) + except subprocess.CalledProcessError as exc: + sys.stdout.write(f"创建虚拟环境失败，退出码: {exc.returncode}\n") + return exc.returncode or 1 + return 0 + + +def main() -> int: + script_dir = Path(__file__).resolve().parent + requirements_file = script_dir / "requirements.txt" + venv_dir = _venv_dir(script_dir) + venv_python = _venv_python(venv_dir) + + if not requirements_file.is_file(): + sys.stdout.write(f"未找到依赖文件: {requirements_file}\n") + return 1 + + ensure_result = _ensure_venv(venv_dir, venv_python) + if ensure_result != 0: + return ensure_result + + if _deps_up_to_date(requirements_file, venv_dir): + sys.stdout.write("依赖已是最新，跳过安装\n") + return 0 + + try: + subprocess.run( + [str(venv_python), "-m", "pip", "install", "--upgrade", "pip"], + check=True, + stdout=sys.stdout, + stderr=sys.stdout, + ) + subprocess.run( + [ + str(venv_python), + "-m", + "pip", + "install", + "-r", + str(requirements_file), + ], + check=True, + stdout=sys.stdout, + stderr=sys.stdout, + ) + except subprocess.CalledProcessError as exc: + sys.stdout.write(f"安装依赖失败，退出码: {exc.returncode}\n") + return exc.returncode or 1 + + _write_stamp(requirements_file, venv_dir) + sys.stdout.write(f"依赖安装完成，当前技能虚拟环境: {venv_dir}\n") + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except SystemExit: + raise + except Exception: + traceback.print_exc(file=sys.stdout) + raise SystemExit(1) diff --git a/skills/export-chat-history/scripts/export_chat_history.py b/skills/export-chat-history/scripts/export_chat_history.py new file mode 100644 index 0000000..4383bb5 --- /dev/null +++ b/skills/export-chat-history/scripts/export_chat_history.py @@ -0,0 +1,637 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import argparse +import json +import os +import re +import shutil +import subprocess +import sys +import tempfile +import urllib.error +import urllib.request +import uuid +from datetime import date, datetime, time, timedelta, timezone +from pathlib import Path +from typing import Any, NoReturn + +try: + from zoneinfo import ZoneInfo +except ImportError: # pragma: no cover - Python 3.8 fallback + ZoneInfo = None # type: ignore[assignment,misc] + +sys.stderr = sys.stdout + +MAX_EXPORT_ROWS = 50_000 +MAX_EXCEL_FILE_SIZE = 25 * 1024 * 1024 +EXCEL_CELL_MAX_LENGTH = 32_767 +ILLEGAL_XML_CHARACTERS = re.compile( + r"[\x00-\x08\x0B-\x0C\x0E-\x1F\x7F-\x84\x86-\x9F]" +) +DATETIME_FORMATS = ( + "%Y-%m-%d %H:%M", + "%Y-%m-%d %H:%M:%S", + "%Y-%m-%dT%H:%M", + "%Y-%m-%dT%H:%M:%S", +) + + +class SkillArgumentParser(argparse.ArgumentParser): + def error(self, message: str) -> NoReturn: + raise ValueError(f"参数错误：{message}") + + +def _shanghai_timezone(): + if ZoneInfo is not None: + try: + return ZoneInfo("Asia/Shanghai") + except Exception: + pass + return timezone(timedelta(hours=8), name="Asia/Shanghai") + + +SHANGHAI_TZ = _shanghai_timezone() + + +def _skill_root() -> Path: + return Path(__file__).resolve().parent.parent + + +def _skill_venv_python() -> Path: + venv_dir = _skill_root() / ".venv" + if sys.platform == "win32": + return venv_dir / "Scripts" / "python.exe" + return venv_dir / "bin" / "python" + + +def _python_executable() -> str: + if sys.executable: + return sys.executable + for candidate in ("python3", "python"): + executable = shutil.which(candidate) + if executable: + return executable + raise RuntimeError("无法找到 Python 解释器路径") + + +def _run_bootstrap() -> None: + bootstrap = Path(__file__).resolve().parent / "bootstrap.py" + result = subprocess.run( + [_python_executable(), str(bootstrap)], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + ) + if result.returncode != 0: + output = result.stdout.strip() + detail = f"：{output}" if output else "" + raise RuntimeError( + f"安装技能依赖失败，退出码: {result.returncode}{detail}" + ) + + +def _ensure_runtime_dependencies() -> None: + try: + import openpyxl # noqa: F401 + import pymysql # noqa: F401 + + return + except ModuleNotFoundError: + pass + + _run_bootstrap() + venv_python = _skill_venv_python() + if not venv_python.is_file(): + raise RuntimeError("安装依赖后仍未找到技能虚拟环境") + + venv_dir = (_skill_root() / ".venv").resolve() + if Path(sys.prefix).resolve() == venv_dir: + try: + import openpyxl # noqa: F401 + import pymysql # noqa: F401 + + return + except ModuleNotFoundError as exc: + raise RuntimeError( + "技能虚拟环境缺少 pymysql 或 openpyxl" + ) from exc + + os.execv( + str(venv_python), + [str(venv_python), str(Path(__file__).resolve()), *sys.argv[1:]], + ) + + +def _parse_args(argv: list[str]) -> argparse.Namespace: + parser = SkillArgumentParser(description="导出当前群聊的聊天记录") + parser.add_argument("--date", dest="export_date", default="") + parser.add_argument( + "--start-time", + "--start_time", + dest="start_time", + default="", + ) + parser.add_argument( + "--end-time", + "--end_time", + dest="end_time", + default="", + ) + args = parser.parse_args(argv) + + if args.export_date and (args.start_time or args.end_time): + raise ValueError("date 不能与 start_time 或 end_time 同时使用") + if bool(args.start_time) != bool(args.end_time): + raise ValueError("自定义时间范围必须同时提供 start_time 和 end_time") + return args + + +def _parse_datetime(value: str, field_name: str) -> datetime: + text = value.strip() + for pattern in DATETIME_FORMATS: + try: + return datetime.strptime(text, pattern).replace(tzinfo=SHANGHAI_TZ) + except ValueError: + continue + raise ValueError( + f"{field_name} 必须使用 YYYY-MM-DD HH:mm 或 " + "YYYY-MM-DD HH:mm:ss 格式" + ) + + +def _resolve_time_range( + args: argparse.Namespace, + now: datetime | None = None, +) -> tuple[datetime, datetime]: + current = now or datetime.now(SHANGHAI_TZ) + if current.tzinfo is None: + current = current.replace(tzinfo=SHANGHAI_TZ) + else: + current = current.astimezone(SHANGHAI_TZ) + + if args.export_date: + try: + selected_date = date.fromisoformat(args.export_date.strip()) + except ValueError as exc: + raise ValueError("date 必须使用 YYYY-MM-DD 格式") from exc + start = datetime.combine( + selected_date, + time.min, + tzinfo=SHANGHAI_TZ, + ) + end = start + timedelta(days=1) + elif args.start_time and args.end_time: + start = _parse_datetime(args.start_time, "start_time") + end = _parse_datetime(args.end_time, "end_time") + else: + start = current.replace(hour=0, minute=0, second=0, microsecond=0) + end = current + + if end <= start: + raise ValueError("结束时间必须晚于开始时间") + return start, end + + +def _require_env(name: str) -> str: + value = os.environ.get(name, "").strip() + if not value: + raise ValueError(f"环境变量 {name} 未配置") + return value + + +def _conversation_context() -> tuple[str, str, str]: + chat_room_id = _require_env("ROBOT_FROM_WX_ID") + if not chat_room_id.endswith("@chatroom"): + raise ValueError("聊天记录导出仅支持在微信群聊中使用") + self_wx_id = _require_env("ROBOT_WX_ID") + client_port = _require_env("ROBOT_WECHAT_CLIENT_PORT") + if not client_port.isdigit(): + raise ValueError("环境变量 ROBOT_WECHAT_CLIENT_PORT 格式不正确") + return chat_room_id, self_wx_id, client_port + + +def _mysql_connect() -> Any: + try: + import pymysql + except ModuleNotFoundError as exc: # pragma: no cover - runtime guard + raise RuntimeError("pymysql 依赖未安装") from exc + + try: + port = int(os.environ.get("MYSQL_PORT", "3306")) + except ValueError as exc: + raise ValueError("环境变量 MYSQL_PORT 格式不正确") from exc + + database = _require_env("ROBOT_CODE") + return pymysql.connect( + host=os.environ.get("MYSQL_HOST", "127.0.0.1"), + port=port, + user=os.environ.get("MYSQL_USER", "root"), + password=os.environ.get("MYSQL_PASSWORD", ""), + database=database, + charset="utf8mb4", + connect_timeout=10, + read_timeout=120, + cursorclass=pymysql.cursors.DictCursor, + ) + + +def _fetch_chat_room_name(connection: Any, chat_room_id: str) -> str: + sql = """ + SELECT nickname + FROM contacts + WHERE wechat_id = %s + AND deleted_at IS NULL + LIMIT 1 + """ + with connection.cursor() as cursor: + cursor.execute(sql, (chat_room_id,)) + row = cursor.fetchone() + if isinstance(row, dict): + nickname = str(row.get("nickname") or "").strip() + if nickname: + return nickname + return chat_room_id + + +def _fetch_messages( + connection: Any, + chat_room_id: str, + self_wx_id: str, + start_timestamp: int, + end_timestamp: int, +) -> list[dict[str, Any]]: + sql = """ + SELECT + messages.id, + messages.sender_wxid, + COALESCE( + NULLIF(chat_room_members.remark, ''), + NULLIF(chat_room_members.nickname, ''), + messages.sender_wxid + ) AS nickname, + CASE + WHEN messages.type = 49 THEN + CASE + WHEN EXTRACTVALUE( + messages.content, + '/msg/appmsg/type' + ) = '57' THEN EXTRACTVALUE( + messages.content, + '/msg/appmsg/title' + ) + WHEN EXTRACTVALUE( + messages.content, + '/msg/appmsg/type' + ) IN ('4', '5') THEN CONCAT( + '网页分享消息，标题: ', + EXTRACTVALUE( + messages.content, + '/msg/appmsg/title' + ), + '，描述：', + EXTRACTVALUE( + messages.content, + '/msg/appmsg/des' + ) + ) + WHEN EXTRACTVALUE( + messages.content, + '/msg/appmsg/type' + ) = '6' THEN CONCAT( + '文件消息，文件名: ', + EXTRACTVALUE( + messages.content, + '/msg/appmsg/title' + ) + ) + ELSE EXTRACTVALUE( + messages.content, + '/msg/appmsg/des' + ) + END + ELSE messages.content + END AS message, + messages.created_at + FROM messages + LEFT JOIN chat_room_members + ON chat_room_members.wechat_id = messages.sender_wxid + AND chat_room_members.chat_room_id = messages.from_wxid + WHERE messages.from_wxid = %s + AND ( + messages.type = 1 + OR ( + messages.type = 49 + AND EXTRACTVALUE( + messages.content, + '/msg/appmsg/type' + ) IN ('57', '4', '5', '6') + ) + ) + AND messages.sender_wxid != %s + AND messages.created_at >= %s + AND messages.created_at < %s + ORDER BY messages.created_at ASC, messages.id ASC + LIMIT %s + """ + with connection.cursor() as cursor: + cursor.execute( + sql, + ( + chat_room_id, + self_wx_id, + start_timestamp, + end_timestamp, + MAX_EXPORT_ROWS + 1, + ), + ) + rows = cursor.fetchall() + + if len(rows) > MAX_EXPORT_ROWS: + raise ValueError( + f"聊天记录超过 {MAX_EXPORT_ROWS:,} 条，请缩短导出时间范围" + ) + return [row for row in rows if isinstance(row, dict)] + + +def _safe_excel_text(value: Any) -> str: + text_value = "" if value is None else str(value) + text_value = ILLEGAL_XML_CHARACTERS.sub("", text_value) + prefix = "'" if text_value.startswith(("=", "+", "-", "@")) else "" + maximum_text_length = EXCEL_CELL_MAX_LENGTH - len(prefix) + if len(text_value) > maximum_text_length: + suffix = "\n……（内容过长，已截断）" + text_value = text_value[ + : maximum_text_length - len(suffix) + ] + suffix + return prefix + text_value + + +def _safe_filename_part(value: str) -> str: + sanitized = re.sub(r'[\\/:*?"<>|\x00-\x1F]', "_", value) + sanitized = re.sub(r"\s+", "_", sanitized).strip("._") + return sanitized[:40] or "群聊" + + +def _create_output_path( + chat_room_name: str, + start: datetime, + end: datetime, +) -> Path: + output_dir = Path(tempfile.gettempdir()) / "wechat-robot-chat-history" + output_dir.mkdir(mode=0o700, parents=True, exist_ok=True) + try: + output_dir.chmod(0o700) + except OSError: + pass + + filename = ( + f"聊天记录_{_safe_filename_part(chat_room_name)}_" + f"{start:%Y%m%d_%H%M}-{end:%Y%m%d_%H%M}_" + f"{uuid.uuid4().hex[:8]}.xlsx" + ) + return output_dir / filename + + +def _build_workbook( + output_path: Path, + chat_room_id: str, + chat_room_name: str, + start: datetime, + end: datetime, + messages: list[dict[str, Any]], + exported_at: datetime | None = None, +) -> None: + try: + from openpyxl import Workbook + from openpyxl.cell import WriteOnlyCell + from openpyxl.styles import Alignment, Font, PatternFill + except ModuleNotFoundError as exc: # pragma: no cover - runtime guard + raise RuntimeError("openpyxl 依赖未安装") from exc + + workbook = Workbook(write_only=True) + workbook.properties.creator = "wechat-robot" + workbook.properties.title = f"{chat_room_name}聊天记录" + workbook.properties.subject = "微信群聊记录导出" + + title_fill = PatternFill("solid", fgColor="1F4E78") + header_fill = PatternFill("solid", fgColor="5B9BD5") + title_font = Font(color="FFFFFF", bold=True, size=14) + header_font = Font(color="FFFFFF", bold=True) + label_font = Font(bold=True, color="1F4E78") + wrapped_alignment = Alignment( + vertical="top", + wrap_text=True, + ) + centered_alignment = Alignment( + horizontal="center", + vertical="center", + ) + + info_sheet = workbook.create_sheet("导出说明") + info_sheet.column_dimensions["A"].width = 18 + info_sheet.column_dimensions["B"].width = 72 + + info_title = WriteOnlyCell(info_sheet, value="群聊记录导出说明") + info_title.fill = title_fill + info_title.font = title_font + info_title.alignment = centered_alignment + info_sheet.append([info_title, ""]) + + current = exported_at or datetime.now(SHANGHAI_TZ) + info_rows = ( + ("群聊名称", chat_room_name), + ("群聊 ID", chat_room_id), + ("开始时间（包含）", start.strftime("%Y-%m-%d %H:%M:%S")), + ("结束时间（不包含）", end.strftime("%Y-%m-%d %H:%M:%S")), + ("消息数量", len(messages)), + ("导出时间", current.strftime("%Y-%m-%d %H:%M:%S")), + ) + for label, value in info_rows: + label_cell = WriteOnlyCell(info_sheet, value=label) + label_cell.font = label_font + value_cell = WriteOnlyCell( + info_sheet, + value=_safe_excel_text(value), + ) + value_cell.alignment = wrapped_alignment + info_sheet.append([label_cell, value_cell]) + + history_sheet = workbook.create_sheet("聊天记录") + history_sheet.freeze_panes = "A2" + widths = {"A": 9, "B": 21, "C": 24, "D": 30, "E": 88} + for column, width in widths.items(): + history_sheet.column_dimensions[column].width = width + + headers = ( + "序号", + "发送时间", + "发送人", + "发送人微信 ID", + "消息内容", + ) + header_cells = [] + for header in headers: + cell = WriteOnlyCell(history_sheet, value=header) + cell.fill = header_fill + cell.font = header_font + cell.alignment = centered_alignment + header_cells.append(cell) + history_sheet.append(header_cells) + + for index, message in enumerate(messages, start=1): + created_at = int(message.get("created_at") or 0) + sent_at = datetime.fromtimestamp( + created_at, + tz=SHANGHAI_TZ, + ).strftime("%Y-%m-%d %H:%M:%S") + row_values = ( + index, + sent_at, + _safe_excel_text(message.get("nickname")), + _safe_excel_text(message.get("sender_wxid")), + _safe_excel_text(message.get("message")), + ) + row_cells = [] + for column_index, value in enumerate(row_values): + cell = WriteOnlyCell(history_sheet, value=value) + cell.alignment = ( + centered_alignment + if column_index in (0, 1) + else wrapped_alignment + ) + row_cells.append(cell) + history_sheet.append(row_cells) + + history_sheet.auto_filter.ref = f"A1:E{len(messages) + 1}" + + temporary_path = output_path.with_name( + f".{output_path.stem}.{uuid.uuid4().hex}.tmp.xlsx" + ) + try: + workbook.save(temporary_path) + if temporary_path.stat().st_size > MAX_EXCEL_FILE_SIZE: + raise ValueError("生成的 Excel 文件超过 25MB，请缩短导出时间范围") + temporary_path.replace(output_path) + finally: + temporary_path.unlink(missing_ok=True) + + +def _raise_for_client_error(payload: dict[str, Any]) -> None: + code = payload.get("code") + if code is not None and str(code) != "200": + message = str(payload.get("message") or f"客户端返回错误码 {code}") + raise RuntimeError(message) + + +def _send_file(client_port: str, chat_room_id: str, path: Path) -> None: + url = ( + f"http://127.0.0.1:{client_port}" + "/api/v1/robot/message/send/file/local" + ) + body = json.dumps( + {"to_wxid": chat_room_id, "file_path": str(path)} + ).encode("utf-8") + request = urllib.request.Request( + url, + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=300) as response: + response_text = response.read().decode("utf-8") + except urllib.error.HTTPError as exc: + error_body = exc.read().decode("utf-8", errors="replace") + raise RuntimeError(f"发送 Excel 失败，HTTP {exc.code}: {error_body}") from exc + except urllib.error.URLError as exc: + raise RuntimeError(f"发送 Excel 失败: {exc.reason}") from exc + + if not response_text.strip(): + return + try: + payload = json.loads(response_text) + except json.JSONDecodeError as exc: + raise RuntimeError("客户端文件发送接口返回了无效 JSON") from exc + if not isinstance(payload, dict): + raise RuntimeError("客户端文件发送接口返回格式不正确") + _raise_for_client_error(payload) + + +def _emit_success( + chat_room_name: str, + start: datetime, + end: datetime, + message_count: int, +) -> None: + result = { + "ok": True, + "message": "聊天记录已导出为 Excel 并发送到当前群聊", + "chat_room_name": chat_room_name, + "start_time": start.strftime("%Y-%m-%d %H:%M:%S"), + "end_time": end.strftime("%Y-%m-%d %H:%M:%S"), + "message_count": message_count, + } + sys.stdout.write(json.dumps(result, ensure_ascii=False) + "\n") + + +def main() -> int: + output_path: Path | None = None + file_sent = False + try: + args = _parse_args(sys.argv[1:]) + start, end = _resolve_time_range(args) + chat_room_id, self_wx_id, client_port = _conversation_context() + _ensure_runtime_dependencies() + + connection = _mysql_connect() + try: + chat_room_name = _fetch_chat_room_name( + connection, + chat_room_id, + ) + messages = _fetch_messages( + connection, + chat_room_id, + self_wx_id, + int(start.timestamp()), + int(end.timestamp()), + ) + finally: + connection.close() + + if not messages: + raise ValueError("该时间范围没有可导出的聊天记录") + + output_path = _create_output_path(chat_room_name, start, end) + _build_workbook( + output_path, + chat_room_id, + chat_room_name, + start, + end, + messages, + ) + _send_file(client_port, chat_room_id, output_path) + file_sent = True + _emit_success( + chat_room_name, + start, + end, + len(messages), + ) + return 0 + except (ValueError, RuntimeError) as exc: + sys.stdout.write(f"导出聊天记录失败: {exc}\n") + return 1 + except Exception as exc: + sys.stdout.write(f"导出聊天记录失败: {exc}\n") + return 1 + finally: + if file_sent and output_path is not None: + output_path.unlink(missing_ok=True) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/export-chat-history/scripts/requirements.txt b/skills/export-chat-history/scripts/requirements.txt new file mode 100644 index 0000000..21f2710 --- /dev/null +++ b/skills/export-chat-history/scripts/requirements.txt @@ -0,0 +1,3 @@ +cryptography +openpyxl>=3.1,<4 +pymysql>=1.1,<2 diff --git a/tests/test_export_chat_history.py b/tests/test_export_chat_history.py new file mode 100644 index 0000000..3d6fa57 --- /dev/null +++ b/tests/test_export_chat_history.py @@ -0,0 +1,351 @@ +from __future__ import annotations + +import argparse +import contextlib +import importlib.util +import io +import json +import os +import sys +import tempfile +import unittest +from datetime import datetime +from pathlib import Path +from unittest import mock + + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +SCRIPT_PATH = ( + REPOSITORY_ROOT + / "skills" + / "export-chat-history" + / "scripts" + / "export_chat_history.py" +) + + +def load_script(): + module_name = "_test_export_chat_history" + spec = importlib.util.spec_from_file_location(module_name, SCRIPT_PATH) + if spec is None or spec.loader is None: + raise RuntimeError(f"无法加载测试脚本：{SCRIPT_PATH}") + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + spec.loader.exec_module(module) + return module + + +class FakeCursor: + def __init__( + self, + *, + fetchone_result=None, + fetchall_result=None, + ) -> None: + self.fetchone_result = fetchone_result + self.fetchall_result = fetchall_result or [] + self.sql = "" + self.params = () + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, traceback): + return False + + def execute(self, sql, params) -> None: + self.sql = sql + self.params = params + + def fetchone(self): + return self.fetchone_result + + def fetchall(self): + return self.fetchall_result + + +class FakeConnection: + def __init__(self, cursors: list[FakeCursor]) -> None: + self.cursors = list(cursors) + self.closed = False + + def cursor(self): + return self.cursors.pop(0) + + def close(self) -> None: + self.closed = True + + +class ExportChatHistoryTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.module = load_script() + + def test_resolve_time_range_supports_date_and_today_default(self) -> None: + now = datetime( + 2026, + 7, + 29, + 15, + 30, + tzinfo=self.module.SHANGHAI_TZ, + ) + date_args = argparse.Namespace( + export_date="2026-07-28", + start_time="", + end_time="", + ) + start, end = self.module._resolve_time_range(date_args, now) + self.assertEqual(start.isoformat(), "2026-07-28T00:00:00+08:00") + self.assertEqual(end.isoformat(), "2026-07-29T00:00:00+08:00") + + default_args = argparse.Namespace( + export_date="", + start_time="", + end_time="", + ) + start, end = self.module._resolve_time_range(default_args, now) + self.assertEqual(start.isoformat(), "2026-07-29T00:00:00+08:00") + self.assertEqual(end, now) + + def test_parse_args_rejects_conflicting_or_partial_ranges(self) -> None: + with self.assertRaisesRegex(ValueError, "不能与"): + self.module._parse_args( + [ + "--date", + "2026-07-29", + "--start_time", + "2026-07-29 09:00", + "--end_time", + "2026-07-29 10:00", + ] + ) + with self.assertRaisesRegex(ValueError, "必须同时提供"): + self.module._parse_args( + ["--start_time", "2026-07-29 09:00"] + ) + + def test_query_matches_chat_room_summary_scope(self) -> None: + rows = [ + { + "id": 1, + "sender_wxid": "member", + "nickname": "成员", + "message": "你好", + "created_at": 1_753_776_000, + } + ] + cursor = FakeCursor(fetchall_result=rows) + connection = FakeConnection([cursor]) + + result = self.module._fetch_messages( + connection, + "group@chatroom", + "robot", + 1_753_776_000, + 1_753_862_400, + ) + + self.assertEqual(result, rows) + self.assertIn("LEFT JOIN chat_room_members", cursor.sql) + self.assertIn("messages.sender_wxid != %s", cursor.sql) + self.assertIn("EXTRACTVALUE", cursor.sql) + self.assertIn("messages.created_at >= %s", cursor.sql) + self.assertIn("messages.created_at < %s", cursor.sql) + self.assertEqual( + cursor.params, + ( + "group@chatroom", + "robot", + 1_753_776_000, + 1_753_862_400, + self.module.MAX_EXPORT_ROWS + 1, + ), + ) + + def test_excel_text_limit_includes_formula_protection_prefix(self) -> None: + raw = "=" + ("x" * self.module.EXCEL_CELL_MAX_LENGTH) + result = self.module._safe_excel_text(raw) + + self.assertEqual(len(result), self.module.EXCEL_CELL_MAX_LENGTH) + self.assertTrue(result.startswith("'=")) + self.assertTrue(result.endswith("……（内容过长，已截断）")) + + @unittest.skipUnless( + importlib.util.find_spec("openpyxl"), + "openpyxl is not installed", + ) + def test_workbook_contains_metadata_and_escapes_formula_text(self) -> None: + from openpyxl import load_workbook + + start = datetime( + 2026, + 7, + 29, + 9, + 0, + tzinfo=self.module.SHANGHAI_TZ, + ) + end = datetime( + 2026, + 7, + 29, + 10, + 0, + tzinfo=self.module.SHANGHAI_TZ, + ) + messages = [ + { + "id": 1, + "sender_wxid": "wxid_1", + "nickname": "@成员", + "message": "=HYPERLINK(\"https://example.test\")\x00", + "created_at": int(start.timestamp()), + } + ] + + with tempfile.TemporaryDirectory() as tmp: + output = Path(tmp) / "history.xlsx" + self.module._build_workbook( + output, + "group@chatroom", + "测试群", + start, + end, + messages, + exported_at=end, + ) + + workbook = load_workbook(output, read_only=True, data_only=False) + self.assertEqual(workbook.sheetnames, ["导出说明", "聊天记录"]) + info_rows = list(workbook["导出说明"].iter_rows(values_only=True)) + self.assertIn(("群聊名称", "测试群"), info_rows) + history_rows = list( + workbook["聊天记录"].iter_rows(values_only=True) + ) + self.assertEqual( + history_rows[0], + ( + "序号", + "发送时间", + "发送人", + "发送人微信 ID", + "消息内容", + ), + ) + self.assertEqual(history_rows[1][2], "'@成员") + self.assertEqual( + history_rows[1][4], + "'=HYPERLINK(\"https://example.test\")", + ) + workbook.close() + + def test_main_sends_generated_file_and_removes_it_after_success(self) -> None: + start = datetime( + 2026, + 7, + 29, + 9, + 0, + tzinfo=self.module.SHANGHAI_TZ, + ) + end = datetime( + 2026, + 7, + 29, + 10, + 0, + tzinfo=self.module.SHANGHAI_TZ, + ) + messages = [ + { + "id": 1, + "sender_wxid": "wxid_1", + "nickname": "成员", + "message": "你好", + "created_at": int(start.timestamp()), + } + ] + connection = FakeConnection([]) + stdout = io.StringIO() + + with tempfile.TemporaryDirectory() as tmp: + output = Path(tmp) / "history.xlsx" + + def build_workbook(path, *args, **kwargs): + del args, kwargs + path.write_bytes(b"xlsx") + + def send_file(client_port, chat_room_id, path): + self.assertEqual(client_port, "9100") + self.assertEqual(chat_room_id, "group@chatroom") + self.assertTrue(path.is_file()) + + patches = ( + mock.patch.object( + self.module, + "_resolve_time_range", + return_value=(start, end), + ), + mock.patch.object( + self.module, + "_ensure_runtime_dependencies", + ), + mock.patch.object( + self.module, + "_mysql_connect", + return_value=connection, + ), + mock.patch.object( + self.module, + "_fetch_chat_room_name", + return_value="测试群", + ), + mock.patch.object( + self.module, + "_fetch_messages", + return_value=messages, + ), + mock.patch.object( + self.module, + "_create_output_path", + return_value=output, + ), + mock.patch.object( + self.module, + "_build_workbook", + side_effect=build_workbook, + ), + mock.patch.object( + self.module, + "_send_file", + side_effect=send_file, + ), + ) + environment = { + "ROBOT_FROM_WX_ID": "group@chatroom", + "ROBOT_WX_ID": "robot", + "ROBOT_WECHAT_CLIENT_PORT": "9100", + } + with contextlib.ExitStack() as stack: + for patcher in patches: + stack.enter_context(patcher) + stack.enter_context( + mock.patch.dict(os.environ, environment, clear=False) + ) + stack.enter_context( + mock.patch.object(sys, "argv", ["export_chat_history.py"]) + ) + with contextlib.redirect_stdout(stdout): + return_code = self.module.main() + + self.assertEqual(return_code, 0) + self.assertTrue(connection.closed) + self.assertFalse(output.exists()) + result = json.loads(stdout.getvalue()) + self.assertTrue(result["ok"]) + self.assertEqual(result["message_count"], 1) + + +if __name__ == "__main__": + unittest.main()