houhou/wechat-robot-skills
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 导出聊天记录技能

Browse Source
This commit is contained in:
hp0912 2026-07-29 22:29:30 +08:00
parent 48d446c27e
commit 4a046d9e74
6 changed files with 1208 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
skills/export-chat-history/SKILL.md Normal file
View File

@ -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` 时，告知用户导出成功；脚本已经发送文件，不要再次调用发送文件技能。
- 没有符合条件的记录时，说明该时间范围没有可导出的聊天记录，并建议用户调整日期或范围。
- 查询、生成或发送失败时，按脚本错误说明原因，不要声称文件已发送。

4
skills/export-chat-history/agents/openai.yaml Normal file
View File

@ -0,0 +1,4 @@
interface:
display_name: "导出聊天记录"
short_description: "将当前群聊指定时间范围的聊天记录整理为 Excel 并发送"
default_prompt: "使用 $export-chat-history 将当前群聊今天的聊天记录导出为 Excel 并发送给我。"

131
skills/export-chat-history/scripts/bootstrap.py Normal file
View File

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

637
skills/export-chat-history/scripts/export_chat_history.py Normal file
View File

@ -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())

3
skills/export-chat-history/scripts/requirements.txt Normal file
View File

@ -0,0 +1,3 @@
cryptography
openpyxl>=3.1,<4
pymysql>=1.1,<2

351
tests/test_export_chat_history.py Normal file
View File

@ -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()