gr/kefu
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
ai-x2-8-21
kefu/wechat_rpa/archive_store.py
T
Your Name 334890171e gengx
2026-08-27 14:04:28 +08:00

2836 lines
126 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""企业微信聊天归档的数据层、COS 对象层与流式导出。
这个模块刻意独立于原来的模型配置后台：它只复用 ``admin_backend.Database``
提供的连接、审计和主密钥，不修改桌面端同步、模型配置或用户管理的数据结构。
当前部署仍然使用项目已有的 SQLite；表结构和所有查询都带租户键、业务唯一键与
游标索引，后续切 MySQL 时 API 和前端无需重做。大文件本体永远不进数据库，只保存
COS 对象定位、哈希与校验状态。
"""
from __future__ import annotations
import csv
import base64
import hashlib
import hmac
import json
import os
import re
import shutil
import threading
import uuid
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable, Iterator
from urllib.parse import urlparse
from archive_content_parser import (
decode_hex_protobuf_text,
file_message_content,
mini_program_content,
parse_file_message_metadata,
parse_mini_program_metadata,
)
DEFAULT_TENANT = "default"
MAX_DIRECT_UPLOAD_BYTES = 5 * 1024 * 1024 * 1024
MULTIPART_THRESHOLD_BYTES = 8 * 1024 * 1024
MULTIPART_PART_BYTES = 2 * 1024 * 1024
EXCEL_SHEET_DATA_ROWS = 900_000
SAFE_SCOPE_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
SCHEMA = """
CREATE TABLE IF NOT EXISTS archive_storage_config (
id INTEGER PRIMARY KEY CHECK(id = 1),
provider TEXT NOT NULL DEFAULT 'cos',
bucket TEXT NOT NULL DEFAULT '',
region TEXT NOT NULL DEFAULT '',
custom_domain TEXT NOT NULL DEFAULT '',
media_prefix TEXT NOT NULL DEFAULT 'archive/media',
export_prefix TEXT NOT NULL DEFAULT 'archive/exports',
encryption_mode TEXT NOT NULL DEFAULT 'AES256',
secret_id_enc TEXT NOT NULL DEFAULT '',
secret_key_enc TEXT NOT NULL DEFAULT '',
enabled INTEGER NOT NULL DEFAULT 0,
updated_at TEXT NOT NULL,
updated_by INTEGER REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS archive_source_account (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
external_account_id TEXT NOT NULL,
display_name TEXT NOT NULL DEFAULT '',
corp_scope_id TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
UNIQUE(tenant_id, external_account_id)
);
CREATE TABLE IF NOT EXISTS archive_import_batch (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
source_account_id TEXT NOT NULL REFERENCES archive_source_account(id),
status TEXT NOT NULL,
received_rows INTEGER NOT NULL DEFAULT 0,
inserted_rows INTEGER NOT NULL DEFAULT 0,
duplicate_rows INTEGER NOT NULL DEFAULT 0,
error_rows INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL,
completed_at TEXT NOT NULL DEFAULT ''
);
CREATE TABLE IF NOT EXISTS archive_checkpoint (
source_account_id TEXT NOT NULL REFERENCES archive_source_account(id),
source_table TEXT NOT NULL,
cursor_json TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY(source_account_id, source_table)
);
CREATE TABLE IF NOT EXISTS archive_raw_event (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
source_account_id TEXT NOT NULL REFERENCES archive_source_account(id),
batch_id TEXT NOT NULL REFERENCES archive_import_batch(id),
source_table TEXT NOT NULL DEFAULT 'message_table',
source_message_key TEXT NOT NULL,
payload_json TEXT NOT NULL,
payload_hash TEXT NOT NULL,
parser_version TEXT NOT NULL DEFAULT 'archive-v1',
parse_status TEXT NOT NULL DEFAULT 'parsed',
ingested_at TEXT NOT NULL,
UNIQUE(tenant_id, source_account_id, source_message_key)
);
CREATE TABLE IF NOT EXISTS archive_person (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
display_name TEXT NOT NULL DEFAULT '',
real_name TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS archive_person_identity (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
person_id TEXT NOT NULL REFERENCES archive_person(id),
identity_type TEXT NOT NULL,
scope_id TEXT NOT NULL DEFAULT '',
external_id TEXT NOT NULL,
external_id_hash TEXT NOT NULL,
verified INTEGER NOT NULL DEFAULT 0,
source TEXT NOT NULL DEFAULT 'local_db',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
UNIQUE(tenant_id, identity_type, scope_id, external_id_hash)
);
CREATE TABLE IF NOT EXISTS archive_conversation (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
source_account_id TEXT NOT NULL REFERENCES archive_source_account(id),
external_id TEXT NOT NULL,
conversation_type TEXT NOT NULL DEFAULT 'unknown',
name TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'active',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
last_message_at TEXT NOT NULL DEFAULT '',
raw_json TEXT NOT NULL DEFAULT '{}',
UNIQUE(tenant_id, source_account_id, external_id)
);
CREATE TABLE IF NOT EXISTS archive_conversation_member (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
conversation_id TEXT NOT NULL REFERENCES archive_conversation(id),
person_id TEXT NOT NULL REFERENCES archive_person(id),
member_role TEXT NOT NULL DEFAULT 'member',
nickname TEXT NOT NULL DEFAULT '',
valid_from TEXT NOT NULL DEFAULT '',
valid_to TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
UNIQUE(conversation_id, person_id, valid_from)
);
CREATE TABLE IF NOT EXISTS archive_message (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
conversation_id TEXT NOT NULL REFERENCES archive_conversation(id),
source_account_id TEXT NOT NULL REFERENCES archive_source_account(id),
sender_person_id TEXT REFERENCES archive_person(id),
raw_event_id TEXT REFERENCES archive_raw_event(id),
source_message_id TEXT NOT NULL DEFAULT '',
server_id TEXT NOT NULL DEFAULT '',
client_id TEXT NOT NULL DEFAULT '',
sequence_no INTEGER,
message_type TEXT NOT NULL DEFAULT 'unknown',
content TEXT NOT NULL DEFAULT '',
direction TEXT NOT NULL DEFAULT 'unknown',
status TEXT NOT NULL DEFAULT 'normal',
sent_at TEXT NOT NULL,
sent_at_epoch INTEGER,
dedup_key TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
UNIQUE(tenant_id, dedup_key)
);
CREATE TABLE IF NOT EXISTS archive_message_version (
id TEXT PRIMARY KEY,
message_id TEXT NOT NULL REFERENCES archive_message(id),
version_no INTEGER NOT NULL,
status TEXT NOT NULL,
content TEXT NOT NULL DEFAULT '',
raw_event_id TEXT REFERENCES archive_raw_event(id),
created_at TEXT NOT NULL,
UNIQUE(message_id, version_no)
);
CREATE TABLE IF NOT EXISTS archive_message_relation (
id TEXT PRIMARY KEY,
message_id TEXT NOT NULL REFERENCES archive_message(id),
relation_type TEXT NOT NULL,
target_source_id TEXT NOT NULL DEFAULT '',
target_message_id TEXT REFERENCES archive_message(id),
created_at TEXT NOT NULL,
UNIQUE(message_id, relation_type, target_source_id)
);
CREATE TABLE IF NOT EXISTS archive_media_object (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
provider TEXT NOT NULL DEFAULT 'cos',
bucket TEXT NOT NULL DEFAULT '',
region TEXT NOT NULL DEFAULT '',
object_key TEXT NOT NULL DEFAULT '',
version_id TEXT NOT NULL DEFAULT '',
sha256 TEXT NOT NULL,
crc64 TEXT NOT NULL DEFAULT '',
etag TEXT NOT NULL DEFAULT '',
size_bytes INTEGER NOT NULL,
mime_type TEXT NOT NULL DEFAULT 'application/octet-stream',
original_filename TEXT NOT NULL DEFAULT '',
media_type TEXT NOT NULL DEFAULT 'file',
storage_class TEXT NOT NULL DEFAULT 'STANDARD',
encryption_mode TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'reserved',
last_error TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
verified_at TEXT NOT NULL DEFAULT '',
UNIQUE(tenant_id, sha256)
);
CREATE TABLE IF NOT EXISTS archive_media_upload_session (
media_id TEXT PRIMARY KEY REFERENCES archive_media_object(id) ON DELETE CASCADE,
upload_id TEXT NOT NULL,
part_size INTEGER NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS archive_message_attachment (
message_id TEXT NOT NULL REFERENCES archive_message(id),
media_id TEXT NOT NULL REFERENCES archive_media_object(id),
attachment_index INTEGER NOT NULL DEFAULT 0,
attachment_role TEXT NOT NULL DEFAULT 'attachment',
match_method TEXT NOT NULL DEFAULT 'source',
match_confidence REAL NOT NULL DEFAULT 1.0,
created_at TEXT NOT NULL,
PRIMARY KEY(message_id, media_id, attachment_index)
);
CREATE TABLE IF NOT EXISTS archive_pending_attachment (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
message_id TEXT NOT NULL REFERENCES archive_message(id) ON DELETE CASCADE,
source_account_id TEXT NOT NULL REFERENCES archive_source_account(id),
source_message_id TEXT NOT NULL DEFAULT '',
original_filename TEXT NOT NULL,
size_bytes INTEGER NOT NULL DEFAULT 0,
checksum TEXT NOT NULL DEFAULT '',
media_type TEXT NOT NULL DEFAULT 'file',
source_reference_sha256 TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'source_not_cached',
media_id TEXT REFERENCES archive_media_object(id),
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
UNIQUE(message_id, original_filename, checksum)
);
CREATE TABLE IF NOT EXISTS archive_export_job (
id TEXT PRIMARY KEY,
tenant_id TEXT NOT NULL,
status TEXT NOT NULL,
formats_json TEXT NOT NULL,
filters_json TEXT NOT NULL,
cutoff_at TEXT NOT NULL,
progress INTEGER NOT NULL DEFAULT 0,
total_rows INTEGER NOT NULL DEFAULT 0,
error_message TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
started_at TEXT NOT NULL DEFAULT '',
completed_at TEXT NOT NULL DEFAULT '',
created_by INTEGER REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS archive_export_file (
id TEXT PRIMARY KEY,
job_id TEXT NOT NULL REFERENCES archive_export_job(id) ON DELETE CASCADE,
file_format TEXT NOT NULL,
file_name TEXT NOT NULL,
local_path TEXT NOT NULL DEFAULT '',
object_key TEXT NOT NULL DEFAULT '',
size_bytes INTEGER NOT NULL DEFAULT 0,
sha256 TEXT NOT NULL DEFAULT '',
storage_status TEXT NOT NULL DEFAULT 'local',
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS archive_maintenance_state (
maintenance_key TEXT PRIMARY KEY,
maintenance_value TEXT NOT NULL DEFAULT '',
updated_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_archive_raw_batch
ON archive_raw_event(batch_id, ingested_at);
CREATE INDEX IF NOT EXISTS idx_archive_raw_source
ON archive_raw_event(tenant_id, source_account_id, source_message_key, ingested_at);
CREATE INDEX IF NOT EXISTS idx_archive_identity_person
ON archive_person_identity(person_id, identity_type);
CREATE INDEX IF NOT EXISTS idx_archive_member_person
ON archive_conversation_member(person_id, conversation_id);
CREATE INDEX IF NOT EXISTS idx_archive_conv_recent
ON archive_conversation(tenant_id, last_message_at DESC, id);
CREATE INDEX IF NOT EXISTS idx_archive_msg_timeline
ON archive_message(tenant_id, conversation_id, sent_at DESC, id DESC);
CREATE INDEX IF NOT EXISTS idx_archive_msg_sender
ON archive_message(tenant_id, sender_person_id, sent_at DESC, id DESC);
CREATE INDEX IF NOT EXISTS idx_archive_msg_source
ON archive_message(source_account_id, source_message_id);
CREATE INDEX IF NOT EXISTS idx_archive_media_status
ON archive_media_object(tenant_id, status, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_archive_pending_source
ON archive_pending_attachment(source_account_id, status, source_message_id);
CREATE INDEX IF NOT EXISTS idx_archive_pending_retry
ON archive_pending_attachment(source_account_id, status, updated_at, source_message_id);
CREATE INDEX IF NOT EXISTS idx_archive_export_created
ON archive_export_job(tenant_id, created_at DESC);
"""
MYSQL_EXPORT_SCHEMA = """SET NAMES utf8mb4;
CREATE TABLE IF NOT EXISTS archive_people (
person_id VARCHAR(40) NOT NULL,
display_name VARCHAR(255) NOT NULL DEFAULT '',
real_name VARCHAR(255) NOT NULL DEFAULT '',
PRIMARY KEY (person_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE TABLE IF NOT EXISTS archive_conversations (
conversation_id VARCHAR(40) NOT NULL,
source_account VARCHAR(191) NOT NULL,
external_id VARCHAR(255) NOT NULL,
conversation_type VARCHAR(32) NOT NULL,
name VARCHAR(255) NOT NULL DEFAULT '',
last_message_at DATETIME(3) NULL,
PRIMARY KEY (conversation_id),
KEY idx_archive_conversation_source (source_account, external_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE TABLE IF NOT EXISTS archive_messages (
message_id VARCHAR(40) NOT NULL,
dedup_key CHAR(64) NOT NULL,
conversation_id VARCHAR(40) NOT NULL,
sender_person_id VARCHAR(40) NULL,
source_message_id VARCHAR(255) NOT NULL DEFAULT '',
server_id VARCHAR(255) NOT NULL DEFAULT '',
client_id VARCHAR(255) NOT NULL DEFAULT '',
sequence_no BIGINT NULL,
message_type VARCHAR(64) NOT NULL,
direction VARCHAR(16) NOT NULL,
status VARCHAR(32) NOT NULL,
sent_at DATETIME(3) NOT NULL,
content LONGTEXT NOT NULL,
PRIMARY KEY (message_id),
UNIQUE KEY uk_archive_message_dedup (dedup_key),
KEY idx_archive_message_timeline (conversation_id, sent_at, message_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE TABLE IF NOT EXISTS archive_media (
media_id VARCHAR(40) NOT NULL,
bucket VARCHAR(191) NOT NULL DEFAULT '',
region VARCHAR(64) NOT NULL DEFAULT '',
object_key VARCHAR(1024) NOT NULL DEFAULT '',
version_id VARCHAR(255) NOT NULL DEFAULT '',
sha256 CHAR(64) NOT NULL,
size_bytes BIGINT UNSIGNED NOT NULL,
mime_type VARCHAR(191) NOT NULL,
original_filename VARCHAR(512) NOT NULL DEFAULT '',
status VARCHAR(32) NOT NULL,
PRIMARY KEY (media_id),
KEY idx_archive_media_sha256 (sha256)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE TABLE IF NOT EXISTS archive_message_media (
message_id VARCHAR(40) NOT NULL,
media_id VARCHAR(40) NOT NULL,
attachment_index INT NOT NULL DEFAULT 0,
attachment_role VARCHAR(32) NOT NULL DEFAULT 'attachment',
match_method VARCHAR(32) NOT NULL DEFAULT 'source',
match_confidence DECIMAL(6,5) NOT NULL DEFAULT 1,
PRIMARY KEY (message_id, media_id, attachment_index),
KEY idx_archive_message_media_media (media_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE TABLE IF NOT EXISTS archive_pending_attachments (
pending_id VARCHAR(40) NOT NULL,
message_id VARCHAR(40) NOT NULL,
source_message_id VARCHAR(255) NOT NULL DEFAULT '',
original_filename VARCHAR(512) NOT NULL,
size_bytes BIGINT UNSIGNED NOT NULL DEFAULT 0,
checksum VARCHAR(128) NOT NULL DEFAULT '',
media_type VARCHAR(32) NOT NULL DEFAULT 'file',
status VARCHAR(32) NOT NULL,
media_id VARCHAR(40) NULL,
PRIMARY KEY (pending_id),
KEY idx_archive_pending_message (message_id),
KEY idx_archive_pending_status (status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
"""
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat(timespec="milliseconds")
def new_id() -> str:
return uuid.uuid4().hex
def json_text(value: Any) -> str:
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
def sha256_text(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def safe_scope(value: str, *, label: str = "标识") -> str:
text = str(value or "").strip()
if not SAFE_SCOPE_RE.fullmatch(text):
raise ValueError(f"{label}只能包含字母、数字、下划线和连字符，长度 1–64")
return text
def safe_prefix(value: str, default: str) -> str:
text = str(value or default).strip().strip("/")
if not text or ".." in text or "\\" in text:
raise ValueError("COS 路径前缀不合法")
if any(part in {"", ".", ".."} for part in text.split("/")):
raise ValueError("COS 路径前缀不合法")
return text
def normalize_sent_at(value: Any, epoch: Any = None) -> tuple[str, int | None]:
"""把秒/毫秒时间戳或 ISO 文本统一成 UTC ISO；同时保留原始秒级 epoch。"""
raw = epoch if epoch not in (None, "") else value
if isinstance(raw, (int, float)) or str(raw or "").strip().isdigit():
number = int(float(raw))
if number > 100_000_000_000: # 毫秒
number //= 1000
if number > 0:
return datetime.fromtimestamp(number, tz=timezone.utc).isoformat(
timespec="milliseconds"
), number
text = str(value or "").strip()
if not text:
raise ValueError("消息时间不能为空")
try:
parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
except ValueError as exc:
raise ValueError(f"无法识别消息时间：{text}") from exc
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
parsed = parsed.astimezone(timezone.utc)
return parsed.isoformat(timespec="milliseconds"), int(parsed.timestamp())
def conversation_type(external_id: str, explicit: str = "") -> str:
if explicit:
return explicit
prefix = str(external_id or "").split(":", 1)[0].upper()
return {
"M": "direct_wechat",
"S": "direct_wecom",
"R": "group",
"Y": "application",
"O": "service",
}.get(prefix, "unknown")
def media_type(mime_type: str) -> str:
value = str(mime_type or "").lower()
if value.startswith("image/"):
return "image"
if value.startswith("audio/"):
return "audio"
if value.startswith("video/"):
return "video"
return "file"
def _is_placeholder_name(value: Any, external_id: Any) -> bool:
"""判断名称是否只是源库里的技术 ID，占位值不能覆盖真实昵称。"""
text = str(value or "").strip()
external = str(external_id or "").strip()
if not text or text == external:
return True
if text.startswith(("S:", "M:", "R:", "Y:", "O:")):
return True
return text.isdigit()
def _display_message_content(
content: Any, message_type_name: Any, *, has_attachment: bool = False
) -> str:
"""隐藏媒体 protobuf 的十六进制残留，保留真正可读的文本。"""
text = str(content or "").strip()
type_name = str(message_type_name or "未知消息").strip() or "未知消息"
decoded = decode_hex_protobuf_text(text, type_name)
if decoded:
return decoded
compact = re.sub(r"[\s|]+", "", text)
looks_binary = bool(
len(compact) >= 80
and len(compact) % 2 == 0
and re.fullmatch(r"[0-9a-fA-F]+", compact)
)
if text and not looks_binary:
return text
if has_attachment or type_name in {
"图片", "截图", "语音", "视频", "文件", "文件回复", "文件预览",
"文件分享", "群文件", "表情",
}:
return f"[{type_name}]"
return text
def excel_safe(value: Any) -> Any:
if not isinstance(value, str):
return value
cleaned = "".join(ch for ch in value if ch in "\t\n\r" or ord(ch) >= 32)
if cleaned.startswith(("=", "+", "-", "@")):
cleaned = "'" + cleaned
if len(cleaned) > 32_767:
cleaned = cleaned[:32_730] + "…[完整内容见 SQL/CSV]"
return cleaned
def mysql_literal(value: Any) -> str:
if value is None:
return "NULL"
if isinstance(value, bool):
return "1" if value else "0"
if isinstance(value, (int, float)):
return str(value)
text = str(value).replace("\\", "\\\\").replace("'", "''")
text = text.replace("\x00", "").replace("\r", "\\r").replace("\n", "\\n")
return f"'{text}'"
def mysql_datetime(value: Any) -> str | None:
"""将库内 UTC ISO 时间转成 MySQL DATETIME(3) 可直接导入的文本。"""
text = str(value or "").strip()
if not text:
return None
parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
parsed = parsed.astimezone(timezone.utc)
return parsed.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
class ArchiveStore:
def __init__(self, database: Any):
self.database = database
self.export_root = Path(database.path).parent / "archive_exports"
self._export_lock = threading.Lock()
self._media_upload_lock = threading.Lock()
def initialize(self) -> None:
now = utc_now()
with self.database.connect() as db:
db.executescript(SCHEMA)
# 已存在的 admin 角色不会被 rbac.seed 重新灌新权限，这里补齐它的
# 全管理能力。operator/viewer 不自动获得敏感聊天权限，由角色页显式授予。
grants = {
"admin": (
"im:read", "im:content:read", "im:import", "im:export",
"im:storage:write", "im:identity:write",
),
}
for role, codes in grants.items():
if db.execute("SELECT 1 FROM roles WHERE code=?", (role,)).fetchone():
db.executemany(
"INSERT OR IGNORE INTO role_permissions(role_code,permission_code) "
"VALUES (?,?)",
[(role, code) for code in codes],
)
db.execute(
"""INSERT OR IGNORE INTO archive_storage_config
(id,updated_at) VALUES (1,?)""",
(now,),
)
self._repair_hex_protobuf_text(db, now)
self._repair_file_message_metadata(db, now)
self._repair_mini_program_metadata(db, now)
db.commit()
@staticmethod
def _repair_hex_protobuf_text(db: Any, now: str) -> int:
"""一次性回填旧版漏掉的单字符/短文本，原始事件仍完整保留。"""
repair_key = "hex-protobuf-text-v1"
if db.execute(
"SELECT 1 FROM archive_maintenance_state WHERE maintenance_key=?",
(repair_key,),
).fetchone():
return 0
rows = db.execute(
"""SELECT id,message_type,content FROM archive_message
WHERE message_type IN ('文本','text','Text')
AND length(content) BETWEEN 2 AND 131072"""
).fetchall()
repaired = 0
for row in rows:
original = str(row["content"] or "")
decoded = decode_hex_protobuf_text(original, row["message_type"])
if not decoded or decoded == original:
continue
db.execute(
"UPDATE archive_message SET content=?,updated_at=? WHERE id=?",
(decoded, now, row["id"]),
)
db.execute(
"""UPDATE archive_message_version SET content=?
WHERE message_id=? AND content=?""",
(decoded, row["id"], original),
)
repaired += 1
db.execute(
"""INSERT INTO archive_maintenance_state
(maintenance_key,maintenance_value,updated_at) VALUES (?,?,?)""",
(repair_key, str(repaired), now),
)
return repaired
@staticmethod
def _repair_file_message_metadata(db: Any, now: str) -> int:
"""从原始事件回填文件名/大小；未下载到本机的文件进入待补传表。"""
repair_key = "file-message-metadata-v1"
if db.execute(
"SELECT 1 FROM archive_maintenance_state WHERE maintenance_key=?",
(repair_key,),
).fetchone():
return 0
rows = db.execute(
"""SELECT m.id,m.source_account_id,m.source_message_id,m.content,
m.message_type,r.payload_json
FROM archive_message m
JOIN archive_raw_event r ON r.id=m.raw_event_id
WHERE m.message_type IN ('合并转发','文件')"""
).fetchall()
repaired = 0
for row in rows:
try:
payload = json.loads(row["payload_json"])
raw_fields = payload.get("raw_fields") or {}
encoded = raw_fields.get("content") or {}
if encoded.get("encoding") != "base64":
continue
raw_content = base64.b64decode(encoded.get("data") or "", validate=True)
metadata = parse_file_message_metadata(
raw_content, raw_fields.get("content_type")
)
except (TypeError, ValueError, json.JSONDecodeError):
continue
if not metadata:
continue
linked = db.execute(
"""SELECT mo.id FROM archive_message_attachment ma
JOIN archive_media_object mo ON mo.id=ma.media_id
WHERE ma.message_id=? AND mo.original_filename=?
AND mo.size_bytes=? AND mo.status='ready' LIMIT 1""",
(
row["id"],
metadata["original_filename"],
metadata["size_bytes"],
),
).fetchone()
cached = linked is not None
normalized = file_message_content(metadata, cached=cached)
db.execute(
"""UPDATE archive_message SET message_type='文件',content=?,updated_at=?
WHERE id=?""",
(normalized, now, row["id"]),
)
db.execute(
"""UPDATE archive_message_version SET content=?
WHERE message_id=? AND content=?""",
(normalized, row["id"], row["content"]),
)
if not cached:
db.execute(
"""INSERT INTO archive_pending_attachment
(id,tenant_id,message_id,source_account_id,source_message_id,
original_filename,size_bytes,checksum,media_type,
source_reference_sha256,status,created_at,updated_at)
VALUES (?,?,?,?,?,?,?,?,?,?,'source_not_cached',?,?)
ON CONFLICT(message_id,original_filename,checksum) DO UPDATE SET
size_bytes=excluded.size_bytes,
source_reference_sha256=excluded.source_reference_sha256,
updated_at=excluded.updated_at""",
(
new_id(), DEFAULT_TENANT, row["id"], row["source_account_id"],
row["source_message_id"], metadata["original_filename"],
metadata["size_bytes"], metadata["checksum"], "file",
metadata["source_reference_sha256"], now, now,
),
)
repaired += 1
db.execute(
"""INSERT INTO archive_maintenance_state
(maintenance_key,maintenance_value,updated_at) VALUES (?,?,?)""",
(repair_key, str(repaired), now),
)
return repaired
@staticmethod
def _repair_mini_program_metadata(db: Any, now: str) -> int:
"""回填旧版误标为“修改群名”的 content_type=78 小程序卡片。"""
repair_key = "mini-program-metadata-v1"
if db.execute(
"SELECT 1 FROM archive_maintenance_state WHERE maintenance_key=?",
(repair_key,),
).fetchone():
return 0
rows = db.execute(
"""SELECT m.id,m.content,r.payload_json
FROM archive_message m
JOIN archive_raw_event r ON r.id=m.raw_event_id
WHERE m.message_type IN ('修改群名','小程序')"""
).fetchall()
repaired = 0
for row in rows:
try:
payload = json.loads(row["payload_json"])
raw_fields = payload.get("raw_fields") or {}
encoded = raw_fields.get("content") or {}
if encoded.get("encoding") != "base64":
continue
raw_content = base64.b64decode(encoded.get("data") or "", validate=True)
metadata = parse_mini_program_metadata(
raw_content, raw_fields.get("content_type")
)
except (TypeError, ValueError, json.JSONDecodeError):
continue
if not metadata:
continue
normalized = mini_program_content(metadata)
db.execute(
"""UPDATE archive_message SET message_type='小程序',content=?,updated_at=?
WHERE id=?""",
(normalized, now, row["id"]),
)
db.execute(
"""UPDATE archive_message_version SET content=?
WHERE message_id=? AND content=?""",
(normalized, row["id"], row["content"]),
)
repaired += 1
db.execute(
"""INSERT INTO archive_maintenance_state
(maintenance_key,maintenance_value,updated_at) VALUES (?,?,?)""",
(repair_key, str(repaired), now),
)
return repaired
# ── COS 配置 ─────────────────────────────────────────────────────────
def _decrypt_storage_row(self, row: Any) -> dict[str, Any]:
import secret_box
item = {key: row[key] for key in row.keys()}
secret_id = ""
secret_key = ""
try:
if item.get("secret_id_enc"):
secret_id = secret_box.decrypt(
item["secret_id_enc"], self.database._secret_key()
)
if item.get("secret_key_enc"):
secret_key = secret_box.decrypt(
item["secret_key_enc"], self.database._secret_key()
)
except Exception as exc:
raise RuntimeError("COS 凭证无法解密，请重新填写") from exc
item["secret_id"] = secret_id
item["secret_key"] = secret_key
return item
def storage_config(self, *, include_secrets: bool = False) -> dict[str, Any]:
import secret_box
with self.database.connect() as db:
row = db.execute("SELECT * FROM archive_storage_config WHERE id=1").fetchone()
if row is None:
self.initialize()
return self.storage_config(include_secrets=include_secrets)
item = self._decrypt_storage_row(row)
item.pop("secret_id_enc", None)
item.pop("secret_key_enc", None)
item["enabled"] = bool(item.get("enabled"))
item["secret_id_present"] = bool(item["secret_id"])
item["secret_key_present"] = bool(item["secret_key"])
if not include_secrets:
item["secret_id_masked"] = secret_box.masked(item.pop("secret_id", ""))
item["secret_key_masked"] = secret_box.masked(item.pop("secret_key", ""))
return item
def save_storage_config(self, values: dict[str, Any], user_id: int, ip: str) -> dict:
import secret_box
bucket = str(values.get("bucket") or "").strip()
region = str(values.get("region") or "").strip()
if bucket and not re.fullmatch(r"[a-z0-9][a-z0-9.-]{0,62}-\d+", bucket):
raise ValueError("Bucket 格式不正确，应包含 APPID 后缀")
if region and not re.fullmatch(r"[a-z0-9-]{2,64}", region):
raise ValueError("Region 格式不正确")
custom_domain = str(values.get("custom_domain") or "").strip().rstrip("/")
if custom_domain:
parsed = urlparse(custom_domain)
if parsed.scheme != "https" or not parsed.netloc or parsed.username:
raise ValueError("自定义域名必须是无账号信息的完整 HTTPS 地址")
if parsed.path not in ("", "/") or parsed.query or parsed.fragment:
raise ValueError("自定义域名不能包含路径、查询参数或片段")
encryption = str(values.get("encryption_mode") or "AES256").strip()
if encryption not in {"", "AES256", "cos/kms"}:
raise ValueError("加密模式只支持 AES256、cos/kms 或留空")
media_prefix = safe_prefix(values.get("media_prefix", ""), "archive/media")
export_prefix = safe_prefix(values.get("export_prefix", ""), "archive/exports")
now = utc_now()
with self.database.connect() as db:
db.execute("BEGIN IMMEDIATE")
current = db.execute(
"SELECT secret_id_enc,secret_key_enc FROM archive_storage_config WHERE id=1"
).fetchone()
raw_id = str(values.get("secret_id") or "").strip()
raw_key = str(values.get("secret_key") or "").strip()
secret_id_enc = (
secret_box.encrypt(raw_id, self.database._secret_key())
if raw_id
else (current["secret_id_enc"] if current else "")
)
secret_key_enc = (
secret_box.encrypt(raw_key, self.database._secret_key())
if raw_key
else (current["secret_key_enc"] if current else "")
)
enabled = bool(values.get("enabled"))
if enabled and not (bucket and region and secret_id_enc and secret_key_enc):
raise ValueError("开启 COS 前必须填写 Bucket、Region、SecretId 和 SecretKey")
db.execute(
"""INSERT INTO archive_storage_config
(id,provider,bucket,region,custom_domain,media_prefix,
export_prefix,encryption_mode,secret_id_enc,secret_key_enc,
enabled,updated_at,updated_by)
VALUES (1,'cos',?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(id) DO UPDATE SET
bucket=excluded.bucket,region=excluded.region,
custom_domain=excluded.custom_domain,
media_prefix=excluded.media_prefix,
export_prefix=excluded.export_prefix,
encryption_mode=excluded.encryption_mode,
secret_id_enc=excluded.secret_id_enc,
secret_key_enc=excluded.secret_key_enc,
enabled=excluded.enabled,updated_at=excluded.updated_at,
updated_by=excluded.updated_by""",
(
bucket, region, custom_domain, media_prefix, export_prefix,
encryption, secret_id_enc, secret_key_enc, 1 if enabled else 0,
now, user_id,
),
)
self.database._audit(
db, user_id, "archive.storage.save",
f"bucket={bucket} region={region} enabled={enabled}", ip,
)
db.commit()
return self.storage_config()
def _cos_client(self):
config = self.storage_config(include_secrets=True)
if not config.get("enabled"):
raise RuntimeError("COS 尚未启用")
if not all(config.get(key) for key in ("bucket", "region", "secret_id", "secret_key")):
raise RuntimeError("COS 配置不完整")
from qcloud_cos import CosConfig, CosS3Client
sdk_config = CosConfig(
Region=config["region"],
SecretId=config["secret_id"],
SecretKey=config["secret_key"],
Timeout=60,
)
return config, CosS3Client(sdk_config)
def test_storage(self) -> dict[str, Any]:
config, client = self._cos_client()
response = client.head_bucket(Bucket=config["bucket"])
return {
"ok": True,
"bucket": config["bucket"],
"region": config["region"],
"request_id": str(response.get("x-cos-request-id") or ""),
}
# ── COS 素材 ─────────────────────────────────────────────────────────
def prepare_media(self, values: dict[str, Any]) -> dict[str, Any]:
tenant = safe_scope(values.get("tenant_id") or DEFAULT_TENANT, label="租户")
digest = str(values.get("sha256") or "").strip().lower()
if not SHA256_RE.fullmatch(digest):
raise ValueError("sha256 必须是 64 位小写十六进制")
size = int(values.get("size_bytes") or 0)
if size <= 0 or size > MAX_DIRECT_UPLOAD_BYTES:
raise ValueError("文件大小必须大于 0 且不超过 5GB")
mime = str(values.get("mime_type") or "application/octet-stream").strip()[:191]
filename = Path(str(values.get("original_filename") or "file")).name[:512]
config, client = self._cos_client()
now = utc_now()
with self.database.connect() as db:
row = db.execute(
"SELECT * FROM archive_media_object WHERE tenant_id=? AND sha256=?",
(tenant, digest),
).fetchone()
if row is not None and row["status"] == "ready":
return {
"media": self._public_media(row),
"reused": True,
"upload_mode": "reused",
"upload_url": "",
"required_headers": {},
}
# 上次可能已经把完整对象传到 COS，只在最终校验/回写时中断。先尝试
# 修复并复用，避免大视频因一次接口中断从头上传。
if row is not None:
try:
repaired = self._verify_media(row["id"], config, client)
except Exception:
pass
else:
return {
"media": repaired,
"reused": True,
"upload_mode": "reused",
"upload_url": "",
"required_headers": {},
}
with self.database.connect() as db:
row = db.execute(
"SELECT * FROM archive_media_object WHERE tenant_id=? AND sha256=?",
(tenant, digest),
).fetchone()
media_id = row["id"] if row is not None else new_id()
object_key = (
f"{safe_prefix(config['media_prefix'], 'archive/media')}/"
f"{tenant}/original/{digest[:2]}/{digest}"
)
if row is None:
db.execute(
"""INSERT INTO archive_media_object
(id,tenant_id,bucket,region,object_key,sha256,size_bytes,
mime_type,original_filename,media_type,encryption_mode,
status,created_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?,'reserved',?)""",
(
media_id, tenant, config["bucket"], config["region"], object_key,
digest, size, mime, filename, media_type(mime),
config.get("encryption_mode") or "", now,
),
)
else:
db.execute(
"""UPDATE archive_media_object SET bucket=?,region=?,object_key=?,
size_bytes=?,mime_type=?,original_filename=?,media_type=?,
encryption_mode=?,status='reserved',last_error='' WHERE id=?""",
(
config["bucket"], config["region"], object_key, size, mime,
filename, media_type(mime), config.get("encryption_mode") or "",
media_id,
),
)
db.commit()
row = db.execute(
"SELECT * FROM archive_media_object WHERE id=?", (media_id,)
).fetchone()
encryption = config.get("encryption_mode") or ""
if size >= MULTIPART_THRESHOLD_BYTES:
return self._prepare_multipart_media(
config, client, row, mime=mime, encryption=encryption
)
headers = {"Content-Type": mime, "x-cos-meta-sha256": digest}
if encryption:
headers["x-cos-server-side-encryption"] = encryption
upload_url = client.get_presigned_url(
Bucket=config["bucket"], Key=object_key, Method="PUT", Expired=900,
Headers=headers,
)
return {
"media": self._public_media(row),
"reused": False,
"upload_mode": "single",
"upload_url": upload_url,
"expires_in": 900,
"required_headers": headers,
}
def _prepare_multipart_media(
self,
config: dict[str, Any],
client: Any,
row: Any,
*,
mime: str,
encryption: str,
) -> dict[str, Any]:
size = int(row["size_bytes"])
part_size = max(
MULTIPART_PART_BYTES,
((size + 9999) // 10000 + 1024 * 1024 - 1) // (1024 * 1024)
* (1024 * 1024),
)
with self._media_upload_lock:
with self.database.connect() as db:
session = db.execute(
"SELECT * FROM archive_media_upload_session WHERE media_id=?",
(row["id"],),
).fetchone()
if session is not None and int(session["part_size"]) != part_size:
try:
client.abort_multipart_upload(
Bucket=row["bucket"],
Key=row["object_key"],
UploadId=str(session["upload_id"]),
)
finally:
with self.database.connect() as db:
db.execute(
"DELETE FROM archive_media_upload_session WHERE media_id=?",
(row["id"],),
)
db.commit()
session = None
if session is None:
options: dict[str, Any] = {
"ContentType": mime,
"Metadata": {"x-cos-meta-sha256": row["sha256"]},
}
if encryption:
options["ServerSideEncryption"] = encryption
created = client.create_multipart_upload(
Bucket=row["bucket"], Key=row["object_key"], **options
)
upload_id = str(created.get("UploadId") or "")
if not upload_id:
raise RuntimeError("COS 没有返回分块上传 ID")
with self.database.connect() as db:
db.execute(
"""INSERT OR REPLACE INTO archive_media_upload_session
(media_id,upload_id,part_size,created_at) VALUES (?,?,?,?)""",
(row["id"], upload_id, part_size, utc_now()),
)
db.commit()
else:
upload_id = str(session["upload_id"])
part_size = int(session["part_size"])
part_count = (size + part_size - 1) // part_size
completed_parts: list[dict[str, Any]] = []
marker = 0
while True:
listed = client.list_parts(
Bucket=row["bucket"],
Key=row["object_key"],
UploadId=upload_id,
MaxParts=1000,
PartNumberMarker=marker,
)
listed_parts = listed.get("Part") or []
if isinstance(listed_parts, dict):
listed_parts = [listed_parts]
for part in listed_parts:
number = int(part.get("PartNumber") or 0)
etag = str(part.get("ETag") or "").strip().strip('"')
part_bytes = int(part.get("Size") or 0)
if number > 0 and etag and part_bytes > 0:
completed_parts.append(
{
"part_number": number,
"etag": etag,
"size_bytes": part_bytes,
}
)
truncated = str(listed.get("IsTruncated") or "").lower() == "true"
next_marker = int(listed.get("NextPartNumberMarker") or 0)
if not truncated or next_marker <= marker:
break
marker = next_marker
parts = []
for number in range(1, part_count + 1):
parts.append(
{
"part_number": number,
"size_bytes": min(part_size, size - (number - 1) * part_size),
"upload_url": client.get_presigned_url(
Bucket=row["bucket"],
Key=row["object_key"],
Method="PUT",
Expired=1800,
Params={"partNumber": number, "uploadId": upload_id},
),
}
)
return {
"media": self._public_media(row),
"reused": False,
"upload_mode": "multipart",
"upload_url": "",
"expires_in": 1800,
"required_headers": {},
"multipart": {
"upload_id": upload_id,
"part_size": part_size,
"parts": parts,
"completed_parts": completed_parts,
},
}
def complete_media(self, media_id: str) -> dict[str, Any]:
config, client = self._cos_client()
return self._verify_media(media_id, config, client)
def complete_multipart_media(
self, media_id: str, upload_id: str, parts: Iterable[dict[str, Any]]
) -> dict[str, Any]:
config, client = self._cos_client()
with self.database.connect() as db:
row = db.execute(
"SELECT * FROM archive_media_object WHERE id=?", (media_id,)
).fetchone()
session = db.execute(
"SELECT * FROM archive_media_upload_session WHERE media_id=?",
(media_id,),
).fetchone()
if row is None or session is None:
raise KeyError("分块上传任务不存在")
if not upload_id or not hmac.compare_digest(str(session["upload_id"]), upload_id):
raise ValueError("分块上传 ID 不匹配")
expected = (int(row["size_bytes"]) + int(session["part_size"]) - 1) // int(
session["part_size"]
)
normalized: list[dict[str, Any]] = []
seen: set[int] = set()
for item in parts:
number = int(item.get("part_number") or 0)
etag = str(item.get("etag") or "").strip().strip('"')
if not 1 <= number <= expected or number in seen or not etag:
raise ValueError("分块编号或 ETag 不正确")
seen.add(number)
normalized.append({"PartNumber": number, "ETag": etag})
if seen != set(range(1, expected + 1)):
raise ValueError("分块数量不完整")
normalized.sort(key=lambda item: item["PartNumber"])
client.complete_multipart_upload(
Bucket=row["bucket"],
Key=row["object_key"],
UploadId=upload_id,
MultipartUpload={"Part": normalized},
)
with self.database.connect() as db:
db.execute(
"DELETE FROM archive_media_upload_session WHERE media_id=?", (media_id,)
)
db.commit()
return self._verify_media(media_id, config, client)
def _verify_media(
self, media_id: str, config: dict[str, Any], client: Any
) -> dict[str, Any]:
with self.database.connect() as db:
row = db.execute(
"SELECT * FROM archive_media_object WHERE id=?", (media_id,)
).fetchone()
if row is None:
raise KeyError("素材不存在")
if row["bucket"] != config["bucket"] or row["region"] != config["region"]:
raise RuntimeError("素材所属 COS 配置已变化，请重新准备上传")
try:
head = client.head_object(Bucket=row["bucket"], Key=row["object_key"])
remote_size = int(head.get("Content-Length") or head.get("content-length") or 0)
remote_hash = str(
head.get("x-cos-meta-sha256")
or head.get("X-Cos-Meta-Sha256")
or ""
).lower()
if remote_size != int(row["size_bytes"]):
raise RuntimeError(
f"COS 文件大小校验失败：本地 {row['size_bytes']}，远端 {remote_size}"
)
if not remote_hash:
options: dict[str, Any] = {
"ContentType": row["mime_type"],
"Metadata": {"x-cos-meta-sha256": row["sha256"]},
}
if row["encryption_mode"]:
options["ServerSideEncryption"] = row["encryption_mode"]
client.copy_object(
Bucket=row["bucket"],
Key=row["object_key"],
CopySource={
"Bucket": row["bucket"],
"Region": row["region"],
"Key": row["object_key"],
},
CopyStatus="Replaced",
**options,
)
head = client.head_object(Bucket=row["bucket"], Key=row["object_key"])
remote_hash = str(
head.get("x-cos-meta-sha256")
or head.get("X-Cos-Meta-Sha256")
or ""
).lower()
if not remote_hash:
raise RuntimeError("COS 对象缺少 SHA-256 元数据，服务端修复失败")
if remote_hash != row["sha256"]:
raise RuntimeError("COS SHA-256 元数据校验失败")
except Exception as exc:
with self.database.connect() as db:
db.execute(
"UPDATE archive_media_object SET status='failed',last_error=? WHERE id=?",
(str(exc)[:1000], media_id),
)
db.commit()
raise
now = utc_now()
with self.database.connect() as db:
db.execute(
"""UPDATE archive_media_object SET status='ready',last_error='',
crc64=?,etag=?,version_id=?,verified_at=? WHERE id=?""",
(
str(head.get("x-cos-hash-crc64ecma") or ""),
str(head.get("ETag") or head.get("etag") or "").strip('"'),
str(head.get("x-cos-version-id") or ""), now, media_id,
),
)
db.commit()
ready = db.execute(
"SELECT * FROM archive_media_object WHERE id=?", (media_id,)
).fetchone()
return self._public_media(ready)
@staticmethod
def _public_media(row: Any) -> dict[str, Any]:
keys = (
"id", "tenant_id", "provider", "bucket", "region", "object_key",
"version_id", "sha256", "crc64", "etag", "size_bytes", "mime_type",
"original_filename", "media_type", "storage_class", "encryption_mode",
"status", "last_error", "created_at", "verified_at",
)
return {key: row[key] for key in keys}
def media_items(self, limit: int = 100) -> list[dict[str, Any]]:
limit = max(1, min(int(limit), 500))
with self.database.connect() as db:
rows = db.execute(
"""SELECT * FROM archive_media_object WHERE tenant_id=?
ORDER BY created_at DESC LIMIT ?""",
(DEFAULT_TENANT, limit),
).fetchall()
return [self._public_media(row) for row in rows]
def media_download_url(self, media_id: str, expires: int = 300) -> str:
config, client = self._cos_client()
with self.database.connect() as db:
row = db.execute(
"SELECT * FROM archive_media_object WHERE id=? AND status='ready'",
(media_id,),
).fetchone()
if row is None:
raise KeyError("素材不存在或尚未上传完成")
return client.get_presigned_download_url(
Bucket=row["bucket"], Key=row["object_key"], Expired=max(60, min(expires, 900))
)
def media_download_urls(
self, media_ids: Iterable[str], expires: int = 300
) -> list[dict[str, Any]]:
"""批量签发素材访问地址，避免消息页为每个附件重复初始化 COS。"""
unique_ids = list(dict.fromkeys(str(item).strip() for item in media_ids if item))
if not unique_ids:
return []
if len(unique_ids) > 200:
raise ValueError("单次最多获取 200 个素材访问地址")
valid_expires = max(60, min(int(expires), 900))
placeholders = ",".join("?" for _ in unique_ids)
with self.database.connect() as db:
rows = db.execute(
f"""SELECT id,bucket,object_key FROM archive_media_object
WHERE tenant_id=? AND status='ready' AND id IN ({placeholders})""",
(DEFAULT_TENANT, *unique_ids),
).fetchall()
if not rows:
return []
_, client = self._cos_client()
by_id = {str(row["id"]): row for row in rows}
result: list[dict[str, Any]] = []
for media_id in unique_ids:
row = by_id.get(media_id)
if row is None:
continue
result.append(
{
"id": media_id,
"url": client.get_presigned_download_url(
Bucket=row["bucket"],
Key=row["object_key"],
Expired=valid_expires,
),
"expires_in": valid_expires,
}
)
return result
# ── 导入与 IM 结构 ───────────────────────────────────────────────────
def source_checkpoint(
self, external_account_id: str, source_table: str = "message_table"
) -> dict[str, Any]:
with self.database.connect() as db:
row = db.execute(
"""SELECT c.cursor_json FROM archive_checkpoint c
JOIN archive_source_account a ON a.id=c.source_account_id
WHERE a.tenant_id=? AND a.external_account_id=?
AND c.source_table=?""",
(DEFAULT_TENANT, str(external_account_id), str(source_table)),
).fetchone()
if row is None:
return {}
try:
value = json.loads(row["cursor_json"])
return value if isinstance(value, dict) else {}
except (TypeError, ValueError):
return {}
def advance_source_checkpoint(
self,
source_account: dict[str, Any],
source_table: str,
checkpoint: dict[str, Any],
) -> dict[str, Any]:
"""只推进桌面导入游标，用于整批消息均被规则过滤的情况。"""
if not isinstance(checkpoint, dict) or not checkpoint:
raise ValueError("checkpoint 不能为空")
table = str(source_table or "message_table")[:128]
now = utc_now()
with self.database.connect() as db:
db.execute("BEGIN IMMEDIATE")
account = self._source_account(db, DEFAULT_TENANT, source_account)
db.execute(
"""INSERT INTO archive_checkpoint
(source_account_id,source_table,cursor_json,updated_at)
VALUES (?,?,?,?) ON CONFLICT(source_account_id,source_table)
DO UPDATE SET cursor_json=excluded.cursor_json,
updated_at=excluded.updated_at""",
(account["id"], table, json_text(checkpoint), now),
)
db.commit()
return checkpoint
def pending_attachment_source_ids(
self, external_account_id: str, limit: int = 500
) -> list[str]:
limit = max(1, min(int(limit), 1000))
with self.database.connect() as db:
rows = db.execute(
"""SELECT DISTINCT pa.source_message_id
FROM archive_pending_attachment pa
JOIN archive_source_account sa ON sa.id=pa.source_account_id
JOIN archive_message m ON m.id=pa.message_id
JOIN archive_conversation c ON c.id=m.conversation_id
WHERE sa.tenant_id=? AND sa.external_account_id=?
AND c.conversation_type<>'application'
AND pa.status='source_not_cached' AND pa.source_message_id<>''
ORDER BY pa.updated_at,pa.source_message_id LIMIT ?""",
(DEFAULT_TENANT, str(external_account_id), limit),
).fetchall()
return [str(row["source_message_id"]) for row in rows]
def claim_pending_attachment_source_ids(
self, external_account_id: str, limit: int = 500
) -> list[str]:
"""轮转领取待补传附件，避免海量队列中靠后的记录长期得不到检查。"""
limit = max(1, min(int(limit), 1000))
now = utc_now()
with self.database.connect() as db:
db.execute("BEGIN IMMEDIATE")
rows = db.execute(
"""SELECT pa.id,pa.source_message_id
FROM archive_pending_attachment pa
JOIN archive_source_account sa ON sa.id=pa.source_account_id
JOIN archive_message m ON m.id=pa.message_id
JOIN archive_conversation c ON c.id=m.conversation_id
WHERE sa.tenant_id=? AND sa.external_account_id=?
AND c.conversation_type<>'application'
AND pa.status='source_not_cached' AND pa.source_message_id<>''
ORDER BY pa.updated_at,pa.source_message_id LIMIT ?""",
(DEFAULT_TENANT, str(external_account_id), limit),
).fetchall()
if rows:
db.executemany(
"UPDATE archive_pending_attachment SET updated_at=? WHERE id=?",
[(now, row["id"]) for row in rows],
)
db.commit()
return [str(row["source_message_id"]) for row in rows]
def _source_account(self, db: Any, tenant: str, values: dict[str, Any]) -> Any:
external = str(values.get("external_account_id") or "").strip()
if not external:
raise ValueError("source_account.external_account_id 不能为空")
now = utc_now()
row = db.execute(
"""SELECT * FROM archive_source_account
WHERE tenant_id=? AND external_account_id=?""",
(tenant, external),
).fetchone()
if row is None:
account_id = new_id()
db.execute(
"""INSERT INTO archive_source_account
(id,tenant_id,external_account_id,display_name,corp_scope_id,
created_at,updated_at) VALUES (?,?,?,?,?,?,?)""",
(
account_id, tenant, external,
str(values.get("display_name") or ""),
str(values.get("corp_scope_id") or ""), now, now,
),
)
row = db.execute(
"SELECT * FROM archive_source_account WHERE id=?", (account_id,)
).fetchone()
else:
db.execute(
"""UPDATE archive_source_account SET display_name=?,corp_scope_id=?,
updated_at=? WHERE id=?""",
(
str(values.get("display_name") or row["display_name"]),
str(values.get("corp_scope_id") or row["corp_scope_id"]),
now, row["id"],
),
)
return row
def _person_for_sender(
self, db: Any, tenant: str, account: Any, sender: dict[str, Any]
) -> str | None:
external_id = str(sender.get("external_id") or sender.get("id") or "").strip()
if not external_id:
return None
identity_type = str(sender.get("identity_type") or "wecom_local_uid").strip()
scope_id = str(
sender.get("scope_id") or account["corp_scope_id"] or account["id"]
)
hashed = sha256_text(f"{identity_type}\x1f{scope_id}\x1f{external_id}")
row = db.execute(
"""SELECT i.person_id FROM archive_person_identity i
WHERE i.tenant_id=? AND i.identity_type=? AND i.scope_id=?
AND i.external_id_hash=?""",
(tenant, identity_type, scope_id, hashed),
).fetchone()
now = utc_now()
display = str(sender.get("display_name") or sender.get("name") or external_id)
real_name = str(sender.get("real_name") or "")
if row is not None:
current = db.execute(
"SELECT display_name,real_name FROM archive_person WHERE id=?",
(row["person_id"],),
).fetchone()
if (
current is not None
and _is_placeholder_name(display, external_id)
and not _is_placeholder_name(current["display_name"], external_id)
):
display = str(current["display_name"])
if current is not None and not real_name:
real_name = str(current["real_name"] or "")
db.execute(
"""UPDATE archive_person SET display_name=?,real_name=?,updated_at=?
WHERE id=?""",
(display, real_name, now, row["person_id"]),
)
return str(row["person_id"])
person_id = new_id()
db.execute(
"""INSERT INTO archive_person
(id,tenant_id,display_name,real_name,created_at,updated_at)
VALUES (?,?,?,?,?,?)""",
(person_id, tenant, display, real_name, now, now),
)
db.execute(
"""INSERT INTO archive_person_identity
(id,tenant_id,person_id,identity_type,scope_id,external_id,
external_id_hash,verified,source,created_at,updated_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
(
new_id(), tenant, person_id, identity_type, scope_id, external_id,
hashed, 1 if sender.get("verified") else 0,
str(sender.get("source") or "local_db"), now, now,
),
)
return person_id
def _conversation(
self, db: Any, tenant: str, account: Any, values: dict[str, Any]
) -> Any:
external = str(values.get("external_id") or values.get("conversation_id") or "").strip()
if not external:
raise ValueError("conversation.external_id 不能为空")
now = utc_now()
row = db.execute(
"""SELECT * FROM archive_conversation
WHERE tenant_id=? AND source_account_id=? AND external_id=?""",
(tenant, account["id"], external),
).fetchone()
name = str(values.get("name") or external)
kind = conversation_type(external, str(values.get("conversation_type") or ""))
if row is None:
conv_id = new_id()
db.execute(
"""INSERT INTO archive_conversation
(id,tenant_id,source_account_id,external_id,conversation_type,
name,created_at,updated_at,raw_json)
VALUES (?,?,?,?,?,?,?,?,?)""",
(
conv_id, tenant, account["id"], external, kind, name, now, now,
json_text(values),
),
)
row = db.execute(
"SELECT * FROM archive_conversation WHERE id=?", (conv_id,)
).fetchone()
else:
if (
_is_placeholder_name(name, external)
and not _is_placeholder_name(row["name"], external)
):
name = str(row["name"])
db.execute(
"""UPDATE archive_conversation SET name=?,conversation_type=?,
updated_at=?,raw_json=? WHERE id=?""",
(name, kind, now, json_text(values), row["id"]),
)
return row
def sync_metadata(
self, values: dict[str, Any], user_id: int | None, ip: str
) -> dict[str, int]:
"""独立同步通讯录昵称和会话名；没有新消息时也会执行。"""
tenant = safe_scope(values.get("tenant_id") or DEFAULT_TENANT, label="租户")
people = list(values.get("people") or [])
conversations = list(values.get("conversations") or [])
if len(people) > 5000 or len(conversations) > 5000:
raise ValueError("单次元数据同步每类最多 5000 条")
now = utc_now()
people_synced = conversations_updated = 0
with self.database.connect() as db:
db.execute("BEGIN IMMEDIATE")
account = self._source_account(
db, tenant, dict(values.get("source_account") or {})
)
for raw_person in people:
person = dict(raw_person or {})
person.setdefault("identity_type", "wecom_userid")
person.setdefault("scope_id", account["corp_scope_id"] or account["id"])
person.setdefault("source", "wxwork_metadata")
if self._person_for_sender(db, tenant, account, person):
people_synced += 1
for raw_conversation in conversations:
conversation = dict(raw_conversation or {})
external = str(
conversation.get("external_id")
or conversation.get("conversation_id")
or ""
).strip()
if not external:
continue
row = db.execute(
"""SELECT id,name FROM archive_conversation
WHERE tenant_id=? AND source_account_id=? AND external_id=?""",
(tenant, account["id"], external),
).fetchone()
if row is None:
continue
name = str(conversation.get("name") or external)
if (
_is_placeholder_name(name, external)
and not _is_placeholder_name(row["name"], external)
):
name = str(row["name"])
db.execute(
"""UPDATE archive_conversation SET name=?,conversation_type=?,
raw_json=?,updated_at=? WHERE id=?""",
(
name,
conversation_type(
external, str(conversation.get("conversation_type") or "")
),
json_text(conversation),
now,
row["id"],
),
)
conversations_updated += 1
self.database._audit(
db,
user_id,
"archive.metadata.sync",
f"people={people_synced} conversations={conversations_updated}",
ip,
)
db.commit()
return {
"people_synced": people_synced,
"conversations_updated": conversations_updated,
}
def _sync_conversation_members(
self,
db: Any,
tenant: str,
account: Any,
conversation: Any,
members: Iterable[dict[str, Any]],
) -> None:
now = utc_now()
for raw_member in members:
member = dict(raw_member or {})
person_id = self._person_for_sender(db, tenant, account, member)
if not person_id:
continue
valid_from = str(member.get("valid_from") or "")
db.execute(
"""INSERT INTO archive_conversation_member
(id,tenant_id,conversation_id,person_id,member_role,nickname,
valid_from,valid_to,created_at)
VALUES (?,?,?,?,?,?,?,?,?)
ON CONFLICT(conversation_id,person_id,valid_from) DO UPDATE SET
member_role=excluded.member_role,nickname=excluded.nickname,
valid_to=excluded.valid_to""",
(
new_id(), tenant, conversation["id"], person_id,
str(member.get("member_role") or "member"),
str(member.get("nickname") or member.get("display_name") or ""),
valid_from, str(member.get("valid_to") or ""), now,
),
)
@staticmethod
def _source_message_key(message: dict[str, Any]) -> str:
for key in ("source_message_id", "server_id", "client_id"):
value = str(message.get(key) or "").strip()
if value:
return f"{key}:{value}"
payload = {
"conversation": message.get("conversation") or message.get("conversation_id"),
"sender": message.get("sender"),
"sent_at": message.get("sent_at") or message.get("send_time"),
"type": message.get("message_type") or message.get("msg_type"),
"content": message.get("content"),
}
return "fingerprint:" + sha256_text(json_text(payload))
def import_messages(
self, values: dict[str, Any], user_id: int | None, ip: str
) -> dict[str, Any]:
tenant = safe_scope(values.get("tenant_id") or DEFAULT_TENANT, label="租户")
rows = list(values.get("messages") or [])
if not rows:
raise ValueError("消息批次不能为空")
if len(rows) > 5000:
raise ValueError("单批最多 5000 条消息")
batch_id = str(values.get("batch_id") or new_id()).strip()
inserted = duplicates = errors = 0
now = utc_now()
with self.database.connect() as db:
db.execute("BEGIN IMMEDIATE")
account = self._source_account(
db, tenant, dict(values.get("source_account") or {})
)
existing_batch = db.execute(
"SELECT id FROM archive_import_batch WHERE id=?", (batch_id,)
).fetchone()
if existing_batch is None:
db.execute(
"""INSERT INTO archive_import_batch
(id,tenant_id,source_account_id,status,received_rows,created_at)
VALUES (?,?,?,?,?,?)""",
(batch_id, tenant, account["id"], "processing", len(rows), now),
)
for message in rows:
try:
raw = dict(message)
source_key = self._source_message_key(raw)
raw_json = json_text(raw)
raw_hash = sha256_text(raw_json)
# 同一源消息撤回、编辑后 payload 会变；原始事件必须保留每个不同版本。
# 归一消息的去重仍只用 source_key，不会因版本变化复制出新消息。
raw_event_key = f"{source_key}:payload:{raw_hash}"
raw_event = db.execute(
"""SELECT id FROM archive_raw_event WHERE tenant_id=?
AND source_account_id=? AND source_message_key=?""",
(tenant, account["id"], raw_event_key),
).fetchone()
if raw_event is None:
raw_id = new_id()
db.execute(
"""INSERT INTO archive_raw_event
(id,tenant_id,source_account_id,batch_id,source_table,
source_message_key,payload_json,payload_hash,ingested_at)
VALUES (?,?,?,?,?,?,?,?,?)""",
(
raw_id, tenant, account["id"], batch_id,
str(raw.get("source_table") or "message_table"),
raw_event_key, raw_json, raw_hash, now,
),
)
else:
raw_id = raw_event["id"]
conv_values = dict(raw.get("conversation") or {})
if not conv_values:
conv_values = {
"external_id": raw.get("conversation_id"),
"name": raw.get("conversation_name"),
}
conv = self._conversation(db, tenant, account, conv_values)
sender = dict(raw.get("sender") or {})
if not sender and raw.get("sender_id") not in (None, ""):
sender = {
"external_id": raw.get("sender_id"),
"display_name": raw.get("sender_name"),
"identity_type": raw.get("sender_identity_type")
or "wecom_local_uid",
"scope_id": raw.get("sender_scope_id") or "",
}
sender_id = self._person_for_sender(db, tenant, account, sender)
members = list(
conv_values.get("members")
or raw.get("conversation_members")
or []
)
if sender:
members.append(sender)
self._sync_conversation_members(
db, tenant, account, conv, members
)
sent_at, sent_epoch = normalize_sent_at(
raw.get("sent_at") or raw.get("send_time"),
raw.get("sent_at_epoch"),
)
dedup = sha256_text(
f"{tenant}\x1f{account['id']}\x1f{source_key}"
)
existing = db.execute(
"SELECT id,status,content FROM archive_message WHERE tenant_id=? "
"AND dedup_key=?",
(tenant, dedup),
).fetchone()
content = str(raw.get("content") or "")
content = (
decode_hex_protobuf_text(
content,
raw.get("message_type")
or raw.get("msg_type_name")
or raw.get("msg_type"),
)
or content
)
status = str(raw.get("status") or "normal")
if existing is not None:
duplicates += 1
if existing["status"] != status or existing["content"] != content:
version = db.execute(
"""SELECT COALESCE(MAX(version_no),0)+1 AS n
FROM archive_message_version WHERE message_id=?""",
(existing["id"],),
).fetchone()["n"]
db.execute(
"""INSERT INTO archive_message_version
(id,message_id,version_no,status,content,raw_event_id,created_at)
VALUES (?,?,?,?,?,?,?)""",
(
new_id(), existing["id"], version, status, content,
raw_id, now,
),
)
db.execute(
"""UPDATE archive_message SET status=?,content=?,raw_event_id=?,
updated_at=? WHERE id=?""",
(status, content, raw_id, now, existing["id"]),
)
message_id = existing["id"]
else:
message_id = new_id()
db.execute(
"""INSERT INTO archive_message
(id,tenant_id,conversation_id,source_account_id,
sender_person_id,raw_event_id,source_message_id,server_id,
client_id,sequence_no,message_type,content,direction,status,
sent_at,sent_at_epoch,dedup_key,created_at,updated_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(
message_id, tenant, conv["id"], account["id"], sender_id,
raw_id, str(raw.get("source_message_id") or ""),
str(raw.get("server_id") or ""),
str(raw.get("client_id") or ""), raw.get("sequence_no")
if raw.get("sequence_no") is not None
else raw.get("message_seq"),
str(raw.get("message_type") or raw.get("msg_type_name")
or raw.get("msg_type") or "unknown"),
content, str(raw.get("direction") or "unknown"), status,
sent_at, sent_epoch, dedup, now, now,
),
)
db.execute(
"""INSERT INTO archive_message_version
(id,message_id,version_no,status,content,raw_event_id,created_at)
VALUES (?,?,1,?,?,?,?)""",
(new_id(), message_id, status, content, raw_id, now),
)
inserted += 1
parent = str(
raw.get("parent_source_message_id")
or raw.get("parent_message_id") or ""
).strip()
if parent:
db.execute(
"""INSERT OR IGNORE INTO archive_message_relation
(id,message_id,relation_type,target_source_id,created_at)
VALUES (?,?,?,?,?)""",
(new_id(), message_id, "reply", parent, now),
)
linked_media: list[Any] = []
for index, media_id in enumerate(raw.get("media_ids") or []):
media = db.execute(
"""SELECT id,original_filename,size_bytes,status
FROM archive_media_object WHERE id=? AND tenant_id=?""",
(str(media_id), tenant),
).fetchone()
if media is not None:
linked_media.append(media)
db.execute(
"""INSERT OR IGNORE INTO archive_message_attachment
(message_id,media_id,attachment_index,created_at)
VALUES (?,?,?,?)""",
(message_id, media["id"], index, now),
)
for raw_attachment in raw.get("attachment_metadata") or []:
attachment = dict(raw_attachment or {})
filename = Path(
str(attachment.get("original_filename") or "")
).name[:512]
if not filename:
continue
size_bytes = max(0, int(attachment.get("size_bytes") or 0))
checksum = str(attachment.get("checksum") or "")[:128]
reference_hash = str(
attachment.get("source_reference_sha256") or ""
)[:64]
matched = next(
(
item for item in linked_media
if str(item["original_filename"]) == filename
and int(item["size_bytes"]) == size_bytes
and str(item["status"]) == "ready"
),
None,
)
pending_status = "ready" if matched is not None else str(
attachment.get("status") or "source_not_cached"
)
db.execute(
"""INSERT INTO archive_pending_attachment
(id,tenant_id,message_id,source_account_id,
source_message_id,original_filename,size_bytes,checksum,
media_type,source_reference_sha256,status,media_id,
created_at,updated_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(message_id,original_filename,checksum) DO UPDATE SET
size_bytes=excluded.size_bytes,
source_reference_sha256=excluded.source_reference_sha256,
status=excluded.status,media_id=excluded.media_id,
updated_at=excluded.updated_at""",
(
new_id(), tenant, message_id, account["id"],
str(raw.get("source_message_id") or ""), filename,
size_bytes, checksum,
str(attachment.get("media_type") or "file"),
reference_hash, pending_status,
matched["id"] if matched is not None else None,
now, now,
),
)
db.execute(
"""UPDATE archive_conversation SET
last_message_at=CASE WHEN last_message_at='' OR last_message_at<?
THEN ? ELSE last_message_at END,
updated_at=? WHERE id=?""",
(sent_at, sent_at, now, conv["id"]),
)
except Exception:
errors += 1
raise
checkpoint = values.get("checkpoint")
if checkpoint is not None:
db.execute(
"""INSERT INTO archive_checkpoint
(source_account_id,source_table,cursor_json,updated_at)
VALUES (?,?,?,?) ON CONFLICT(source_account_id,source_table)
DO UPDATE SET cursor_json=excluded.cursor_json,
updated_at=excluded.updated_at""",
(
account["id"], str(values.get("source_table") or "message_table"),
json_text(checkpoint), now,
),
)
db.execute(
"""UPDATE archive_import_batch SET status='completed',inserted_rows=?,
duplicate_rows=?,error_rows=?,completed_at=? WHERE id=?""",
(inserted, duplicates, errors, utc_now(), batch_id),
)
self.database._audit(
db, user_id, "archive.import",
f"batch={batch_id} inserted={inserted} duplicate={duplicates}", ip,
)
db.commit()
return {
"batch_id": batch_id,
"received": len(rows),
"inserted": inserted,
"duplicates": duplicates,
"errors": errors,
}
# ── 人员唯一标识 ──────────────────────────────────────────────────────────────────────
def people(self, limit: int = 100, keyword: str = "") -> list[dict[str, Any]]:
limit = max(1, min(int(limit), 500))
search = str(keyword or "").strip()
params: list[Any] = [DEFAULT_TENANT]
where = "p.tenant_id=?"
if search:
where += " AND (p.display_name LIKE ? OR p.real_name LIKE ?)"
params.extend([f"%{search}%", f"%{search}%"])
params.append(limit)
with self.database.connect() as db:
rows = db.execute(
f"""SELECT p.id,p.display_name,p.real_name,p.created_at,p.updated_at,
(SELECT COUNT(*) FROM archive_person_identity i
WHERE i.person_id=p.id) AS identity_count,
(SELECT COUNT(DISTINCT cm.conversation_id)
FROM archive_conversation_member cm
WHERE cm.person_id=p.id) AS conversation_count,
(SELECT COUNT(*) FROM archive_message m
WHERE m.sender_person_id=p.id) AS message_count
FROM archive_person p WHERE {where}
ORDER BY p.updated_at DESC,p.id DESC LIMIT ?""",
tuple(params),
).fetchall()
return [
{
"id": row["id"],
"display_name": row["display_name"],
"real_name": row["real_name"],
"identity_count": int(row["identity_count"]),
"conversation_count": int(row["conversation_count"]),
"message_count": int(row["message_count"]),
"created_at": row["created_at"],
"updated_at": row["updated_at"],
}
for row in rows
]
def person_detail(self, person_id: str) -> dict[str, Any]:
with self.database.connect() as db:
person = db.execute(
"SELECT * FROM archive_person WHERE id=? AND tenant_id=?",
(person_id, DEFAULT_TENANT),
).fetchone()
if person is None:
raise KeyError("人员不存在")
identities = db.execute(
"""SELECT id,identity_type,scope_id,external_id,verified,source,
created_at,updated_at
FROM archive_person_identity WHERE person_id=?
ORDER BY verified DESC,identity_type,created_at""",
(person_id,),
).fetchall()
return {
"id": person["id"],
"display_name": person["display_name"],
"real_name": person["real_name"],
"identities": [
{
"id": row["id"],
"identity_type": row["identity_type"],
"scope_id": row["scope_id"],
"external_id": row["external_id"],
"verified": bool(row["verified"]),
"source": row["source"],
"created_at": row["created_at"],
"updated_at": row["updated_at"],
}
for row in identities
],
}
def bind_identity(
self, person_id: str, values: dict[str, Any], user_id: int, ip: str
) -> dict[str, Any]:
identity_type = str(values.get("identity_type") or "").strip()
scope_id = str(values.get("scope_id") or "").strip()
external_id = str(values.get("external_id") or "").strip()
if not re.fullmatch(r"[A-Za-z0-9_.:-]{2,64}", identity_type):
raise ValueError("标识类型格式不正确")
if not external_id or len(external_id) > 512:
raise ValueError("外部人员 ID 长度必须在 1–512 之间")
hashed = sha256_text(f"{identity_type}\x1f{scope_id}\x1f{external_id}")
now = utc_now()
with self.database.connect() as db:
db.execute("BEGIN IMMEDIATE")
if db.execute(
"SELECT 1 FROM archive_person WHERE id=? AND tenant_id=?",
(person_id, DEFAULT_TENANT),
).fetchone() is None:
raise KeyError("人员不存在")
existing = db.execute(
"""SELECT person_id FROM archive_person_identity
WHERE tenant_id=? AND identity_type=? AND scope_id=?
AND external_id_hash=?""",
(DEFAULT_TENANT, identity_type, scope_id, hashed),
).fetchone()
if existing is not None and existing["person_id"] != person_id:
raise ValueError("该企微人员 ID 已绑定到另一个人员")
db.execute(
"""INSERT INTO archive_person_identity
(id,tenant_id,person_id,identity_type,scope_id,external_id,
external_id_hash,verified,source,created_at,updated_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(tenant_id,identity_type,scope_id,external_id_hash)
DO UPDATE SET verified=excluded.verified,source=excluded.source,
updated_at=excluded.updated_at""",
(
new_id(), DEFAULT_TENANT, person_id, identity_type, scope_id,
external_id, hashed, 1 if values.get("verified", True) else 0,
"manual", now, now,
),
)
self.database._audit(
db, user_id, "archive.identity.bind",
f"person={person_id} type={identity_type} scope={scope_id}", ip,
)
db.commit()
return self.person_detail(person_id)
# ── 查询 ─────────────────────────────────────────────────────────────
def stats(self) -> dict[str, Any]:
with self.database.connect() as db:
result = {}
for key, table in (
("messages", "archive_message"),
("conversations", "archive_conversation"),
("people", "archive_person"),
("media", "archive_media_object"),
("imports", "archive_import_batch"),
("exports", "archive_export_job"),
):
result[key] = int(
db.execute(
f"SELECT COUNT(*) AS n FROM {table} WHERE tenant_id=?",
(DEFAULT_TENANT,),
).fetchone()["n"]
)
result["media_ready"] = int(
db.execute(
"""SELECT COUNT(*) AS n FROM archive_media_object
WHERE tenant_id=? AND status='ready'""",
(DEFAULT_TENANT,),
).fetchone()["n"]
)
result["media_failed"] = int(
db.execute(
"""SELECT COUNT(*) AS n FROM archive_media_object
WHERE tenant_id=? AND status='failed'""",
(DEFAULT_TENANT,),
).fetchone()["n"]
)
result["last_message_at"] = str(
db.execute(
"SELECT COALESCE(MAX(sent_at),'') AS v FROM archive_message WHERE tenant_id=?",
(DEFAULT_TENANT,),
).fetchone()["v"]
)
return result
def conversations(self, limit: int = 50, cursor: str = "") -> dict[str, Any]:
limit = max(1, min(int(limit), 200))
params: list[Any] = [DEFAULT_TENANT]
where = "c.tenant_id=?"
if cursor:
where += " AND (c.last_message_at < ? OR (c.last_message_at=? AND c.id<?))"
try:
cursor_time, cursor_id = cursor.split("|", 1)
except ValueError as exc:
raise ValueError("会话游标格式不正确") from exc
params.extend([cursor_time, cursor_time, cursor_id])
params.append(limit + 1)
with self.database.connect() as db:
rows = db.execute(
f"""SELECT c.*,a.external_account_id,
(SELECT COUNT(*) FROM archive_message m
WHERE m.conversation_id=c.id) AS message_count,
(SELECT m.content FROM archive_message m
WHERE m.conversation_id=c.id
ORDER BY m.sent_at DESC,m.id DESC LIMIT 1) AS last_content,
(SELECT m.message_type FROM archive_message m
WHERE m.conversation_id=c.id
ORDER BY m.sent_at DESC,m.id DESC LIMIT 1) AS last_message_type,
(SELECT COUNT(*) FROM archive_message_attachment ma
JOIN archive_message m ON m.id=ma.message_id
WHERE m.conversation_id=c.id
AND m.id=(SELECT lm.id FROM archive_message lm
WHERE lm.conversation_id=c.id
ORDER BY lm.sent_at DESC,lm.id DESC LIMIT 1))
AS last_attachment_count
FROM archive_conversation c
JOIN archive_source_account a ON a.id=c.source_account_id
WHERE {where}
ORDER BY c.last_message_at DESC,c.id DESC LIMIT ?""",
tuple(params),
).fetchall()
has_more = len(rows) > limit
rows = rows[:limit]
items = [
{
"id": row["id"],
"external_id": row["external_id"],
"name": row["name"],
"conversation_type": row["conversation_type"],
"source_account": row["external_account_id"],
"last_message_at": row["last_message_at"],
"last_content": _display_message_content(
row["last_content"],
row["last_message_type"],
has_attachment=bool(row["last_attachment_count"]),
),
"message_count": int(row["message_count"]),
"status": row["status"],
}
for row in rows
]
next_cursor = ""
if has_more and rows:
next_cursor = f"{rows[-1]['last_message_at']}|{rows[-1]['id']}"
return {"items": items, "next_cursor": next_cursor, "has_more": has_more}
def messages(
self, conversation_id: str, limit: int = 100, cursor: str = ""
) -> dict[str, Any]:
limit = max(1, min(int(limit), 500))
params: list[Any] = [DEFAULT_TENANT, conversation_id]
where = "m.tenant_id=? AND m.conversation_id=?"
if cursor:
try:
cursor_time, cursor_id = cursor.split("|", 1)
except ValueError as exc:
raise ValueError("消息游标格式不正确") from exc
where += " AND (m.sent_at > ? OR (m.sent_at=? AND m.id>?))"
params.extend([cursor_time, cursor_time, cursor_id])
params.append(limit + 1)
with self.database.connect() as db:
rows = db.execute(
f"""SELECT m.*,p.display_name AS sender_name,
(SELECT COUNT(*) FROM archive_message_attachment ma
WHERE ma.message_id=m.id) AS attachment_count
FROM archive_message m
LEFT JOIN archive_person p ON p.id=m.sender_person_id
WHERE {where}
ORDER BY m.sent_at ASC,m.id ASC LIMIT ?""",
tuple(params),
).fetchall()
has_more = len(rows) > limit
rows = rows[:limit]
attachments_by_message: dict[str, list[dict[str, Any]]] = {
str(row["id"]): [] for row in rows
}
if rows:
message_ids = [str(row["id"]) for row in rows]
placeholders = ",".join("?" for _ in message_ids)
attachment_rows = db.execute(
f"""SELECT ma.message_id,ma.attachment_index,ma.attachment_role,
ma.match_method,ma.match_confidence,
mo.id,mo.media_type,mo.mime_type,mo.original_filename,
mo.size_bytes,mo.status,mo.sha256
FROM archive_message_attachment ma
JOIN archive_media_object mo ON mo.id=ma.media_id
WHERE ma.message_id IN ({placeholders})
ORDER BY ma.message_id,ma.attachment_index,mo.id""",
tuple(message_ids),
).fetchall()
for attachment in attachment_rows:
attachments_by_message[str(attachment["message_id"])].append(
{
"id": attachment["id"],
"media_type": attachment["media_type"],
"mime_type": attachment["mime_type"],
"original_filename": attachment["original_filename"],
"size_bytes": int(attachment["size_bytes"]),
"status": attachment["status"],
"sha256": attachment["sha256"],
"attachment_index": int(attachment["attachment_index"]),
"attachment_role": attachment["attachment_role"],
"match_method": attachment["match_method"],
"match_confidence": float(attachment["match_confidence"]),
}
)
pending_rows = db.execute(
f"""SELECT id,message_id,original_filename,size_bytes,checksum,
media_type,status
FROM archive_pending_attachment
WHERE status<>'ready' AND message_id IN ({placeholders})
ORDER BY message_id,created_at,id""",
tuple(message_ids),
).fetchall()
for pending in pending_rows:
items_for_message = attachments_by_message[str(pending["message_id"])]
items_for_message.append(
{
"id": f"pending:{pending['id']}",
"media_type": pending["media_type"],
"mime_type": "application/octet-stream",
"original_filename": pending["original_filename"],
"size_bytes": int(pending["size_bytes"]),
"status": pending["status"],
"sha256": "",
"checksum": pending["checksum"],
"attachment_index": len(items_for_message),
"attachment_role": "attachment",
"match_method": "source_metadata",
"match_confidence": 1.0,
}
)
items = [
{
"id": row["id"],
"source_message_id": row["source_message_id"],
"sender_person_id": row["sender_person_id"],
"sender_name": row["sender_name"] or "",
"message_type": row["message_type"],
"content": _display_message_content(
row["content"],
row["message_type"],
has_attachment=bool(attachments_by_message[str(row["id"])]),
),
"direction": row["direction"],
"status": row["status"],
"sent_at": row["sent_at"],
"sequence_no": row["sequence_no"],
"attachment_count": len(attachments_by_message[str(row["id"])]),
"attachments": attachments_by_message[str(row["id"])],
}
for row in rows
]
next_cursor = ""
if has_more and rows:
next_cursor = f"{rows[-1]['sent_at']}|{rows[-1]['id']}"
return {"items": items, "next_cursor": next_cursor, "has_more": has_more}
# ── 导出 ─────────────────────────────────────────────────────────────
def create_export_job(
self, formats: Iterable[str], filters: dict[str, Any], user_id: int, ip: str
) -> dict[str, Any]:
normalized = sorted({str(item).lower() for item in formats})
unsupported = set(normalized) - {"sql", "csv", "xlsx"}
if not normalized or unsupported:
raise ValueError("导出格式只能选择 sql、csv、xlsx")
job_id = new_id()
now = utc_now()
with self.database.connect() as db:
db.execute(
"""INSERT INTO archive_export_job
(id,tenant_id,status,formats_json,filters_json,cutoff_at,
created_at,created_by)
VALUES (?,?,'queued',?,?,?,?,?)""",
(
job_id, DEFAULT_TENANT, json_text(normalized), json_text(filters or {}),
now, now, user_id,
),
)
self.database._audit(
db, user_id, "archive.export.create",
f"job={job_id} formats={','.join(normalized)}", ip,
)
db.commit()
return self.export_job(job_id)
def export_job(self, job_id: str) -> dict[str, Any]:
with self.database.connect() as db:
row = db.execute(
"SELECT * FROM archive_export_job WHERE id=?", (job_id,)
).fetchone()
if row is None:
raise KeyError("导出任务不存在")
files = db.execute(
"SELECT * FROM archive_export_file WHERE job_id=? ORDER BY file_name",
(job_id,),
).fetchall()
item = {key: row[key] for key in row.keys()}
item["formats"] = json.loads(item.pop("formats_json"))
item["filters"] = json.loads(item.pop("filters_json"))
item["files"] = [
{
"id": file["id"],
"format": file["file_format"],
"file_name": file["file_name"],
"size_bytes": file["size_bytes"],
"sha256": file["sha256"],
"storage_status": file["storage_status"],
}
for file in files
]
return item
def export_jobs(self, limit: int = 100) -> list[dict[str, Any]]:
with self.database.connect() as db:
ids = [
row["id"]
for row in db.execute(
"""SELECT id FROM archive_export_job WHERE tenant_id=?
ORDER BY created_at DESC LIMIT ?""",
(DEFAULT_TENANT, max(1, min(int(limit), 500))),
).fetchall()
]
return [self.export_job(job_id) for job_id in ids]
def _export_where(self, filters: dict[str, Any], cutoff: str) -> tuple[str, list[Any]]:
where = ["m.tenant_id=?", "m.created_at<=?"]
params: list[Any] = [DEFAULT_TENANT, cutoff]
if filters.get("conversation_id"):
where.append("m.conversation_id=?")
params.append(str(filters["conversation_id"]))
if filters.get("date_from"):
where.append("m.sent_at>=?")
params.append(str(filters["date_from"]) + "T00:00:00.000+00:00")
if filters.get("date_to"):
where.append("m.sent_at<=?")
params.append(str(filters["date_to"]) + "T23:59:59.999+00:00")
return " AND ".join(where), params
def _message_rows(
self, filters: dict[str, Any], cutoff: str, *, batch_size: int = 2000
) -> Iterator[dict[str, Any]]:
where, params = self._export_where(filters, cutoff)
conn = self.database.connect()
try:
cursor = conn.execute(
f"""SELECT m.id,m.dedup_key,m.conversation_id,m.sender_person_id,
m.source_message_id,m.server_id,m.client_id,m.sequence_no,
m.message_type,m.direction,m.status,m.sent_at,m.content,
p.display_name AS sender_name,c.name AS conversation_name,
a.external_account_id AS source_account
FROM archive_message m
LEFT JOIN archive_person p ON p.id=m.sender_person_id
JOIN archive_conversation c ON c.id=m.conversation_id
JOIN archive_source_account a ON a.id=m.source_account_id
WHERE {where} ORDER BY m.sent_at,m.id""",
tuple(params),
)
while True:
rows = cursor.fetchmany(batch_size)
if not rows:
break
for row in rows:
yield {key: row[key] for key in row.keys()}
finally:
conn.close()
def _count_export_rows(self, filters: dict[str, Any], cutoff: str) -> int:
where, params = self._export_where(filters, cutoff)
with self.database.connect() as db:
return int(
db.execute(
f"SELECT COUNT(*) AS n FROM archive_message m WHERE {where}",
tuple(params),
).fetchone()["n"]
)
def _write_csv(self, path: Path, filters: dict[str, Any], cutoff: str) -> None:
columns = [
"message_id", "dedup_key", "source_account", "conversation_id",
"conversation_name", "sender_person_id", "sender_name", "source_message_id",
"server_id", "client_id", "sequence_no", "message_type", "direction",
"status", "sent_at", "content",
]
mapping = {"message_id": "id"}
with path.open("w", encoding="utf-8-sig", newline="") as handle:
writer = csv.DictWriter(handle, fieldnames=columns)
writer.writeheader()
for row in self._message_rows(filters, cutoff):
writer.writerow({key: row.get(mapping.get(key, key), "") for key in columns})
def _write_sql(self, path: Path, filters: dict[str, Any], cutoff: str) -> None:
where, params = self._export_where(filters, cutoff)
def query_rows(sql: str, values: Iterable[Any]) -> Iterator[dict[str, Any]]:
connection = self.database.connect()
try:
cursor = connection.execute(sql, tuple(values))
while True:
rows = cursor.fetchmany(2000)
if not rows:
break
for row in rows:
yield {key: row[key] for key in row.keys()}
finally:
connection.close()
with path.open("w", encoding="utf-8", newline="\n") as handle:
handle.write("-- 企业微信聊天归档 MySQL 8 导出\n")
handle.write(f"-- 截止水位: {cutoff}\n\n")
handle.write(MYSQL_EXPORT_SCHEMA)
handle.write("\nSTART TRANSACTION;\n")
def write_insert(
table: str,
columns: list[str],
rows: Iterable[dict[str, Any]],
updates: list[str],
) -> None:
batch: list[str] = []
batch_bytes = 0
def flush() -> None:
nonlocal batch, batch_bytes
if not batch:
return
suffix = ""
if updates:
suffix = "\nON DUPLICATE KEY UPDATE " + ",".join(
f"`{column}`=VALUES(`{column}`)" for column in updates
)
handle.write(
f"INSERT INTO `{table}` (`"
+ "`,`".join(columns)
+ "`) VALUES\n"
+ ",\n".join(batch)
+ suffix
+ ";\n"
)
batch = []
batch_bytes = 0
for row in rows:
literal = "(" + ",".join(
mysql_literal(row.get(column)) for column in columns
) + ")"
size = len(literal.encode("utf-8"))
if batch and (len(batch) >= 500 or batch_bytes + size > 4_000_000):
flush()
batch.append(literal)
batch_bytes += size
flush()
people = query_rows(
f"""SELECT DISTINCT p.id AS person_id,p.display_name,p.real_name
FROM archive_person p JOIN archive_message m
ON m.sender_person_id=p.id WHERE {where}
ORDER BY p.id""",
params,
)
write_insert(
"archive_people",
["person_id", "display_name", "real_name"],
people,
["display_name", "real_name"],
)
conversations = query_rows(
f"""SELECT DISTINCT c.id AS conversation_id,
a.external_account_id AS source_account,c.external_id,
c.conversation_type,c.name,c.last_message_at
FROM archive_conversation c
JOIN archive_source_account a ON a.id=c.source_account_id
JOIN archive_message m ON m.conversation_id=c.id
WHERE {where} ORDER BY c.id""",
params,
)
write_insert(
"archive_conversations",
[
"conversation_id", "source_account", "external_id",
"conversation_type", "name", "last_message_at",
],
(
{**row, "last_message_at": mysql_datetime(row["last_message_at"])}
for row in conversations
),
["conversation_type", "name", "last_message_at"],
)
media = query_rows(
f"""SELECT DISTINCT mo.id AS media_id,mo.bucket,mo.region,
mo.object_key,mo.version_id,mo.sha256,mo.size_bytes,
mo.mime_type,mo.original_filename,mo.status
FROM archive_media_object mo
JOIN archive_message_attachment ma ON ma.media_id=mo.id
JOIN archive_message m ON m.id=ma.message_id
WHERE {where} ORDER BY mo.id""",
params,
)
write_insert(
"archive_media",
[
"media_id", "bucket", "region", "object_key", "version_id",
"sha256", "size_bytes", "mime_type", "original_filename", "status",
],
media,
["version_id", "status"],
)
message_columns = [
"message_id", "dedup_key", "conversation_id", "sender_person_id",
"source_message_id", "server_id", "client_id", "sequence_no",
"message_type", "direction", "status", "sent_at", "content",
]
message_rows = (
{
**row,
"message_id": row["id"],
"sent_at": mysql_datetime(row["sent_at"]),
}
for row in self._message_rows(filters, cutoff)
)
write_insert(
"archive_messages", message_columns, message_rows, ["status", "content"]
)
links = query_rows(
f"""SELECT ma.message_id,ma.media_id,ma.attachment_index,
ma.attachment_role,ma.match_method,ma.match_confidence
FROM archive_message_attachment ma
JOIN archive_message m ON m.id=ma.message_id
WHERE {where} ORDER BY ma.message_id,ma.attachment_index""",
params,
)
write_insert(
"archive_message_media",
[
"message_id", "media_id", "attachment_index", "attachment_role",
"match_method", "match_confidence",
],
links,
["attachment_role", "match_method", "match_confidence"],
)
pending_attachments = query_rows(
f"""SELECT pa.id AS pending_id,pa.message_id,
pa.source_message_id,pa.original_filename,pa.size_bytes,
pa.checksum,pa.media_type,pa.status,pa.media_id
FROM archive_pending_attachment pa
JOIN archive_message m ON m.id=pa.message_id
WHERE {where} ORDER BY pa.message_id,pa.created_at,pa.id""",
params,
)
write_insert(
"archive_pending_attachments",
[
"pending_id", "message_id", "source_message_id",
"original_filename", "size_bytes", "checksum", "media_type",
"status", "media_id",
],
pending_attachments,
["status", "media_id"],
)
handle.write("COMMIT;\n")
def _write_xlsx(self, path: Path, filters: dict[str, Any], cutoff: str) -> None:
from openpyxl import Workbook
workbook = Workbook(write_only=True)
def append_sheets(
base_name: str, headers: list[str], rows: Iterable[Iterable[Any]]
) -> None:
index = 1
sheet = workbook.create_sheet(f"{base_name}_{index:03d}")
sheet.append(headers)
count = 0
for row in rows:
if count and count % EXCEL_SHEET_DATA_ROWS == 0:
index += 1
sheet = workbook.create_sheet(f"{base_name}_{index:03d}")
sheet.append(headers)
sheet.append([excel_safe(value) for value in row])
count += 1
append_sheets(
"消息",
[
"消息ID", "去重键", "账号", "会话ID", "会话名称", "发送者ID",
"发送者", "源消息ID", "服务器ID", "客户端ID", "序号", "类型",
"方向", "状态", "发送时间(UTC)", "内容",
],
(
(
row["id"], row["dedup_key"], row["source_account"],
row["conversation_id"], row["conversation_name"],
row["sender_person_id"] or "", row["sender_name"] or "",
row["source_message_id"], row["server_id"], row["client_id"],
row["sequence_no"], row["message_type"], row["direction"],
row["status"], row["sent_at"], row["content"],
)
for row in self._message_rows(filters, cutoff)
),
)
where, params = self._export_where(filters, cutoff)
def query_rows(sql: str) -> Iterator[tuple[Any, ...]]:
connection = self.database.connect()
try:
cursor = connection.execute(sql, tuple(params))
while True:
batch = cursor.fetchmany(2000)
if not batch:
break
for row in batch:
yield tuple(row)
finally:
connection.close()
append_sheets(
"会话",
["会话ID", "账号", "源会话ID", "类型", "名称", "最后消息时间"],
query_rows(
f"""SELECT DISTINCT c.id,a.external_account_id,c.external_id,
c.conversation_type,c.name,c.last_message_at
FROM archive_conversation c
JOIN archive_source_account a ON a.id=c.source_account_id
JOIN archive_message m ON m.conversation_id=c.id
WHERE {where} ORDER BY c.id"""
),
)
append_sheets(
"联系人",
["联系人ID", "显示名称", "真实姓名", "创建时间"],
query_rows(
f"""SELECT DISTINCT p.id,p.display_name,p.real_name,p.created_at
FROM archive_person p JOIN archive_message m
ON m.sender_person_id=p.id
WHERE {where} ORDER BY p.id"""
),
)
append_sheets(
"素材",
[
"素材ID", "消息ID", "Bucket", "Region", "ObjectKey", "VersionId",
"SHA256", "大小", "MIME", "原文件名", "状态",
],
query_rows(
f"""SELECT mo.id,ma.message_id,mo.bucket,mo.region,mo.object_key,
mo.version_id,mo.sha256,mo.size_bytes,mo.mime_type,
mo.original_filename,mo.status
FROM archive_media_object mo
JOIN archive_message_attachment ma ON ma.media_id=mo.id
JOIN archive_message m ON m.id=ma.message_id
WHERE {where} ORDER BY ma.message_id,ma.attachment_index"""
),
)
append_sheets(
"待补传附件",
[
"待补传ID", "消息ID", "源消息ID", "原文件名", "大小",
"校验值", "素材类型", "状态", "已关联素材ID",
],
query_rows(
f"""SELECT pa.id,pa.message_id,pa.source_message_id,
pa.original_filename,pa.size_bytes,pa.checksum,
pa.media_type,pa.status,COALESCE(pa.media_id,'')
FROM archive_pending_attachment pa
JOIN archive_message m ON m.id=pa.message_id
WHERE {where} ORDER BY pa.message_id,pa.created_at,pa.id"""
),
)
workbook.save(path)
def _upload_export_file(self, path: Path, job: Any) -> tuple[str, str]:
if not self.storage_config(include_secrets=True).get("enabled"):
return "", "local"
config, client = self._cos_client()
created = datetime.fromisoformat(job["created_at"])
object_key = (
f"{safe_prefix(config['export_prefix'], 'archive/exports')}/"
f"{job['tenant_id']}/{created:%Y/%m}/{job['id']}/{path.name}"
)
kwargs: dict[str, Any] = {}
if config.get("encryption_mode"):
kwargs["ServerSideEncryption"] = config["encryption_mode"]
client.upload_file(
Bucket=config["bucket"], Key=object_key, LocalFilePath=str(path),
PartSize=16, MAXThread=3, EnableMD5=True, **kwargs,
)
head = client.head_object(Bucket=config["bucket"], Key=object_key)
remote_size = int(head.get("Content-Length") or head.get("content-length") or 0)
if remote_size != path.stat().st_size:
raise RuntimeError(f"导出文件上传后大小不一致：{path.name}")
return object_key, "cos"
def run_export_job(self, job_id: str) -> None:
# 单进程开发模式串行导出，避免两个大 XLSX 同时抢满内存和磁盘。生产环境可把
# 这个同名任务交给独立 Worker，数据库状态机保持不变。
with self._export_lock:
with self.database.connect() as db:
db.execute("BEGIN IMMEDIATE")
claimed = db.execute(
"""UPDATE archive_export_job SET status='running',started_at=?,
progress=1,error_message='' WHERE id=?
AND status IN ('queued','failed')""",
(utc_now(), job_id),
)
if claimed.rowcount != 1:
db.rollback()
return
db.commit()
job = db.execute(
"SELECT * FROM archive_export_job WHERE id=?", (job_id,)
).fetchone()
try:
formats = json.loads(job["formats_json"])
filters = json.loads(job["filters_json"])
total = self._count_export_rows(filters, job["cutoff_at"])
directory = self.export_root / job_id
if directory.exists():
shutil.rmtree(directory)
directory.mkdir(parents=True, exist_ok=True)
created_files: list[tuple[str, Path]] = []
if "sql" in formats:
path = directory / "archive.sql"
self._write_sql(path, filters, job["cutoff_at"])
created_files.append(("sql", path))
if "csv" in formats:
path = directory / "messages.csv"
self._write_csv(path, filters, job["cutoff_at"])
created_files.append(("csv", path))
if "xlsx" in formats:
path = directory / "archive.xlsx"
self._write_xlsx(path, filters, job["cutoff_at"])
created_files.append(("xlsx", path))
manifest = {
"job_id": job_id,
"cutoff_at": job["cutoff_at"],
"filters": filters,
"message_count": total,
"files": [
{"name": path.name, "sha256": sha256_file(path), "size": path.stat().st_size}
for _fmt, path in created_files
],
}
manifest_path = directory / "manifest.json"
manifest_path.write_text(
json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8"
)
created_files.append(("manifest", manifest_path))
zip_path = directory / "archive_bundle.zip"
with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as bundle:
for _fmt, path in created_files:
bundle.write(path, path.name)
created_files.append(("zip", zip_path))
with self.database.connect() as db:
db.execute("DELETE FROM archive_export_file WHERE job_id=?", (job_id,))
for file_format, path in created_files:
object_key, storage_status = self._upload_export_file(path, job)
db.execute(
"""INSERT INTO archive_export_file
(id,job_id,file_format,file_name,local_path,object_key,
size_bytes,sha256,storage_status,created_at)
VALUES (?,?,?,?,?,?,?,?,?,?)""",
(
new_id(), job_id, file_format, path.name, str(path),
object_key, path.stat().st_size, sha256_file(path),
storage_status, utc_now(),
),
)
db.execute(
"""UPDATE archive_export_job SET status='completed',progress=100,
total_rows=?,completed_at=? WHERE id=?""",
(total, utc_now(), job_id),
)
db.commit()
except Exception as exc:
with self.database.connect() as db:
db.execute(
"""UPDATE archive_export_job SET status='failed',
error_message=?,completed_at=? WHERE id=?""",
(str(exc)[:2000], utc_now(), job_id),
)
db.commit()
def export_file(self, file_id: str) -> dict[str, Any]:
with self.database.connect() as db:
row = db.execute(
"""SELECT f.*,j.tenant_id FROM archive_export_file f
JOIN archive_export_job j ON j.id=f.job_id WHERE f.id=?""",
(file_id,),
).fetchone()
if row is None:
raise KeyError("导出文件不存在")
return {key: row[key] for key in row.keys()}
def export_download_url(self, file_id: str) -> str:
row = self.export_file(file_id)
if row["storage_status"] != "cos" or not row["object_key"]:
return ""
config, client = self._cos_client()
return client.get_presigned_download_url(
Bucket=config["bucket"], Key=row["object_key"], Expired=300
)
Reference in New Issue View Git Blame Copy Permalink