94 lines
3.9 KiB
Python
94 lines
3.9 KiB
Python
import json
|
|
import os
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
DEFAULT_DATA = Path(__file__).resolve().parent.parent / 'data'
|
|
|
|
|
|
def connect(root=None):
|
|
root = Path(root or os.environ.get('WECHAT_ANALYSIS_DATA', DEFAULT_DATA))
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
root.chmod(0o700)
|
|
except OSError:
|
|
pass
|
|
db = sqlite3.connect(root / 'workbench.sqlite3', timeout=30)
|
|
db.row_factory = sqlite3.Row
|
|
db.execute('PRAGMA foreign_keys=ON')
|
|
db.execute('PRAGMA journal_mode=WAL')
|
|
db.executescript('''
|
|
CREATE TABLE IF NOT EXISTS messages (
|
|
id INTEGER PRIMARY KEY, fingerprint TEXT UNIQUE NOT NULL,
|
|
source_id TEXT NOT NULL, conversation TEXT NOT NULL,
|
|
conversation_id TEXT NOT NULL, kind TEXT NOT NULL,
|
|
sender TEXT NOT NULL, sender_id TEXT NOT NULL,
|
|
timestamp TEXT NOT NULL, content TEXT NOT NULL,
|
|
message_type TEXT NOT NULL, quote TEXT NOT NULL, source TEXT NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS msg_time ON messages(timestamp);
|
|
CREATE INDEX IF NOT EXISTS msg_conv ON messages(conversation_id, timestamp);
|
|
CREATE TABLE IF NOT EXISTS runs (
|
|
id TEXT PRIMARY KEY, created TEXT NOT NULL, mode TEXT NOT NULL,
|
|
filters TEXT NOT NULL, cards TEXT NOT NULL, coverage TEXT NOT NULL,
|
|
status TEXT NOT NULL, error TEXT NOT NULL DEFAULT ''
|
|
);
|
|
CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
|
''')
|
|
if 'events' not in {row[1] for row in db.execute('PRAGMA table_info(runs)')}:
|
|
db.execute("ALTER TABLE runs ADD COLUMN events TEXT NOT NULL DEFAULT '[]'")
|
|
db.commit()
|
|
return db
|
|
|
|
|
|
def setting(db, key, default=None):
|
|
row = db.execute('SELECT value FROM settings WHERE key=?', (key,)).fetchone()
|
|
return json.loads(row[0]) if row else default
|
|
|
|
|
|
def set_setting(db, key, value):
|
|
db.execute('INSERT OR REPLACE INTO settings VALUES (?,?)', (key, json.dumps(value, ensure_ascii=False)))
|
|
db.commit()
|
|
|
|
|
|
def where(filters):
|
|
clauses, args = ['1=1'], []
|
|
# Stored wall-clock times are normalized to the configured source UTC offset.
|
|
from .importer import parse_time
|
|
for key, op in [('start', '>='), ('end', '<=')]:
|
|
val = filters.get(key)
|
|
if val:
|
|
if len(val) == 10:
|
|
val += ' 00:00:00' if key == 'start' else ' 23:59:59'
|
|
clauses.append(f'timestamp {op} ?')
|
|
args.append(parse_time(val, filters.get('utc_offset', '+08:00')))
|
|
if filters.get('kind') in ('group', 'private'):
|
|
clauses.append('kind=?')
|
|
args.append(filters['kind'])
|
|
for key, negate in [('conversations', False), ('exclude', True)]:
|
|
values = filters.get(key) or []
|
|
if not isinstance(values, list) or len(values) > 500:
|
|
raise ValueError('会话筛选必须是数组，最多 500 项')
|
|
if values:
|
|
clauses.append('conversation_id ' + ('NOT IN' if negate else 'IN') + '(' + ','.join('?' for _ in values) + ')')
|
|
args.extend(values)
|
|
if filters.get('search'):
|
|
clauses.append('content LIKE ? ESCAPE \'\\\'')
|
|
term = filters['search'].replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_')
|
|
args.append('%' + term + '%')
|
|
return ' AND '.join(clauses), args
|
|
|
|
|
|
def messages(db, filters):
|
|
clause, args = where(filters)
|
|
return [dict(row) for row in db.execute('SELECT * FROM messages WHERE ' + clause + ' ORDER BY timestamp,id', args)]
|
|
|
|
|
|
def overview(db):
|
|
rows = db.execute('''SELECT conversation_id, MAX(conversation) AS conversation, kind,
|
|
COUNT(*) AS count, MIN(timestamp) AS first, MAX(timestamp) AS last,
|
|
SUM(CASE WHEN message_type!='text' THEN 1 ELSE 0 END) AS media
|
|
FROM messages GROUP BY conversation_id,kind ORDER BY count DESC''').fetchall()
|
|
total = db.execute('SELECT COUNT(*),MIN(timestamp),MAX(timestamp) FROM messages').fetchone()
|
|
return {'count': total[0], 'first': total[1], 'last': total[2], 'conversations': [dict(r) for r in rows]}
|