162 lines
7.3 KiB
Python
162 lines
7.3 KiB
Python
"""Explicit adapters for readable exports. Never modifies a WeChat database."""
|
||
import csv
|
||
import hashlib
|
||
import io
|
||
import json
|
||
import re
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
ALIASES = {
|
||
'source_id': ['id', 'msg_id', 'msgId', 'MsgSvrID', 'msgSvrId', '消息ID'],
|
||
'conversation': ['conversation', 'chat_name', 'talkerName', 'room_name', '群名', '会话'],
|
||
'conversation_id': ['conversation_id', 'talker', 'StrTalker', 'chat_id', 'room_id', '会话ID'],
|
||
'kind': ['kind', 'chat_type', '会话类型'],
|
||
'sender': ['sender', 'sender_name', 'senderName', 'NickName', '发送人'],
|
||
'sender_id': ['sender_id', 'senderUsername', 'from_user', '发送人ID'],
|
||
'timestamp': ['timestamp', 'time', 'createTime', 'CreateTime', 'datetime', '时间'],
|
||
'content': ['content', 'text', 'StrContent', 'Content', '内容'],
|
||
'message_type': ['message_type', 'type', 'Type', '消息类型'],
|
||
'quote': ['quote', '引用'],
|
||
}
|
||
|
||
|
||
def offset_zone(offset):
|
||
if not re.fullmatch(r'[+-](?:0\d|1[0-4]):[0-5]\d', str(offset)):
|
||
raise ValueError('时区偏移格式应为 +08:00 或 -06:00')
|
||
hours, minutes = map(int, offset[1:].split(':'))
|
||
if hours == 14 and minutes:
|
||
raise ValueError('时区偏移不能超过 14 小时')
|
||
return timezone((1 if offset[0] == '+' else -1) * timedelta(hours=hours, minutes=minutes))
|
||
|
||
|
||
def parse_time(value, offset='+08:00'):
|
||
zone = offset_zone(offset)
|
||
value = str(value).strip()
|
||
if re.fullmatch(r'\d{10}(?:\.\d+)?|\d{13}', value):
|
||
number = float(value)
|
||
if number > 100_000_000_000:
|
||
number /= 1000
|
||
dt = datetime.fromtimestamp(number, timezone.utc)
|
||
else:
|
||
value = value.replace('/', '-').replace('年', '-').replace('月', '-').replace('日', '')
|
||
dt = datetime.fromisoformat(value.replace('Z', '+00:00'))
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=zone)
|
||
# One canonical UTC representation, independent of the importing computer.
|
||
return dt.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
|
||
|
||
|
||
def decode(data):
|
||
encodings = ('utf-16',) if data.startswith((b'\xff\xfe', b'\xfe\xff')) else ('utf-8-sig', 'gb18030')
|
||
for encoding in encodings:
|
||
try:
|
||
return data.decode(encoding)
|
||
except UnicodeError:
|
||
continue
|
||
raise ValueError('无法识别文件编码，请另存为 UTF-8')
|
||
|
||
|
||
def read_rows(data, filename):
|
||
text = decode(data)
|
||
ext = filename.lower().rsplit('.', 1)[-1]
|
||
if ext == 'json':
|
||
rows = json.loads(text)
|
||
if isinstance(rows, dict):
|
||
rows = rows.get('messages', rows.get('items', rows.get('data')))
|
||
if isinstance(rows, dict):
|
||
rows = rows.get('messages', rows.get('items'))
|
||
if not isinstance(rows, list):
|
||
raise ValueError('JSON 应为消息数组或包含 messages 数组的对象')
|
||
elif ext in ('jsonl', 'ndjson'):
|
||
rows = [json.loads(line) for line in text.splitlines() if line.strip()]
|
||
elif ext in ('csv', 'tsv'):
|
||
rows = list(csv.DictReader(io.StringIO(text), delimiter='\t' if ext == 'tsv' else ','))
|
||
elif ext == 'txt':
|
||
# Deliberately strict: no guessing ambiguous sender/time boundaries.
|
||
rows = []
|
||
current = None
|
||
for line in text.splitlines():
|
||
match = re.match(r'^\[(\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2})\]\s+([^:：]+)[:：]\s?(.*)$', line)
|
||
if match:
|
||
current = dict(timestamp=match[1], sender=match[2], content=match[3])
|
||
rows.append(current)
|
||
elif current:
|
||
current['content'] += '\n' + line
|
||
elif line.strip():
|
||
raise ValueError('TXT 格式应为：[2026-10-01 09:00:00] 张三: 消息内容。其他格式请先转换为 CSV/JSON。')
|
||
else:
|
||
raise ValueError('支持 JSON / JSONL / CSV / TSV / TXT；SQLite 请使用数据库导入页')
|
||
if any(not isinstance(row, dict) for row in rows):
|
||
raise ValueError('每条消息必须是对象')
|
||
return rows
|
||
|
||
|
||
def normalize(row, options):
|
||
mapping = options.get('mapping') or {}
|
||
|
||
def get(field, default=''):
|
||
keys = [mapping[field]] if mapping.get(field) else ALIASES[field]
|
||
for key in keys:
|
||
if row.get(key) is not None and str(row[key]).strip():
|
||
val = row[key]
|
||
if isinstance(val, bytes):
|
||
raise ValueError('字段 ' + key + ' 含二进制内容，需先解码为可读文本')
|
||
return json.dumps(val, ensure_ascii=False) if isinstance(val, (dict, list)) else str(val).strip()
|
||
return default
|
||
|
||
cid = get('conversation_id', options.get('conversation_id', options.get('conversation', '')))
|
||
name = get('conversation', options.get('conversation', cid))
|
||
if not cid:
|
||
cid = name
|
||
if not cid:
|
||
raise ValueError('缺少会话名称/ID，请填写默认会话或字段映射')
|
||
kind = get('kind', options.get('kind', ''))
|
||
if kind in ('群', '群聊', 'chatroom'):
|
||
kind = 'group'
|
||
if kind in ('私聊', '个人', 'direct'):
|
||
kind = 'private'
|
||
if not kind:
|
||
kind = 'group' if cid.endswith('@chatroom') else 'private'
|
||
if kind not in ('group', 'private'):
|
||
raise ValueError('会话类型应为 group 或 private')
|
||
sender = get('sender', get('sender_id'))
|
||
if not sender:
|
||
raise ValueError('缺少发送人，请配置 sender 字段映射')
|
||
content = get('content')
|
||
typ = get('message_type', 'text')
|
||
typ = 'text' if typ.lower() in ('1', 'text', '文本', '文字') else typ
|
||
if not content and typ == 'text':
|
||
raise ValueError('文字消息内容为空')
|
||
msg = dict(source_id=get('source_id'), conversation=name or cid, conversation_id=cid,
|
||
kind=kind, sender=sender, sender_id=get('sender_id', sender),
|
||
timestamp=parse_time(get('timestamp'), options.get('utc_offset', '+08:00')),
|
||
content=content, message_type=typ, quote=get('quote'))
|
||
# Authoritative IDs are scoped by conversation; no-ID files use content identity.
|
||
identity = [cid, msg['source_id']] if msg['source_id'] else [cid, msg['sender_id'], msg['timestamp'], content, typ, msg['quote']]
|
||
msg['fingerprint'] = hashlib.sha256(json.dumps(identity, ensure_ascii=False).encode()).hexdigest()
|
||
return msg
|
||
|
||
|
||
def import_rows(db, rows, source, options):
|
||
added = duplicates = rejected = without_ids = 0
|
||
errors = []
|
||
for index, row in enumerate(rows, 1):
|
||
try:
|
||
msg = normalize(row, options)
|
||
except (ValueError, TypeError, OverflowError, OSError) as exc:
|
||
rejected += 1
|
||
if len(errors) < 30:
|
||
errors.append({'row': index, 'reason': str(exc)})
|
||
continue
|
||
msg['source'] = source
|
||
without_ids += not bool(msg['source_id'])
|
||
fields = list(msg)
|
||
result = db.execute('INSERT OR IGNORE INTO messages (' + ','.join(fields) + ') VALUES (' + ','.join('?' for _ in fields) + ')', [msg[f] for f in fields])
|
||
if result.rowcount:
|
||
added += 1
|
||
else:
|
||
duplicates += 1
|
||
db.commit()
|
||
return dict(added=added, duplicates=duplicates, rejected=rejected, errors=errors, total=len(rows),
|
||
without_ids=without_ids, note='无消息 ID 的记录按会话、发送人、时间、内容、类型与引用去重；同秒完全相同的真实消息可能被合并。' if without_ids else '')
|