diff --git a/mcp_server.py b/mcp_server.py index 8da488b..8211f5f 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -16,8 +16,8 @@ import xml.etree.ElementTree as ET from Crypto.Cipher import AES from mcp.server.fastmcp import FastMCP import zstandard as zstd -from config import _config_file_path, _DEFAULT -from decode_image import ImageResolver +from config import _config_file_path, _DEFAULT +from decode_image import ImageResolver from key_utils import get_key_info, key_path_variants, strip_key_metadata # ============ 加密常量 ============ @@ -30,17 +30,17 @@ WAL_HEADER_SZ = 32 WAL_FRAME_HEADER_SZ = 24 # ============ 配置加载 ============ -SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) -CONFIG_FILE = _config_file_path() - -try: - with open(CONFIG_FILE, encoding="utf-8") as f: - _cfg = json.load(f) -except FileNotFoundError: - _cfg = dict(_DEFAULT) -for _key in ("keys_file", "decrypted_dir"): - if _key in _cfg and not os.path.isabs(_cfg[_key]): - _cfg[_key] = os.path.join(os.path.dirname(CONFIG_FILE), _cfg[_key]) +SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) +CONFIG_FILE = _config_file_path() + +try: + with open(CONFIG_FILE, encoding="utf-8") as f: + _cfg = json.load(f) +except FileNotFoundError: + _cfg = dict(_DEFAULT) +for _key in ("keys_file", "decrypted_dir"): + if _key in _cfg and not os.path.isabs(_cfg[_key]): + _cfg[_key] = os.path.join(os.path.dirname(CONFIG_FILE), _cfg[_key]) DB_DIR = _cfg["db_dir"] KEYS_FILE = _cfg["keys_file"] @@ -59,11 +59,11 @@ if not DECODED_IMAGE_DIR: elif not os.path.isabs(DECODED_IMAGE_DIR): DECODED_IMAGE_DIR = os.path.join(SCRIPT_DIR, DECODED_IMAGE_DIR) -try: - with open(KEYS_FILE, encoding="utf-8") as f: - ALL_KEYS = strip_key_metadata(json.load(f)) -except FileNotFoundError: - ALL_KEYS = {} +try: + with open(KEYS_FILE, encoding="utf-8") as f: + ALL_KEYS = strip_key_metadata(json.load(f)) +except FileNotFoundError: + ALL_KEYS = {} # ============ 解密函数 ============ @@ -229,7 +229,7 @@ atexit.register(_cache.cleanup) # ============ 联系人缓存 ============ _contact_names = None # {username: display_name} -_contact_full = None # [{username, nick_name, remark, alias, description, phone}] +_contact_full = None # [{username, nick_name, remark, alias, description, phone}] _contact_tags = None # {label_id: {name, sort_order, members: [{username, display_name}]}} _self_username = None _contact_db_mtime = 0 # mtime of the decrypted contact.db when caches were last populated @@ -247,55 +247,60 @@ _QUERY_LIMIT_MAX = 500 _HISTORY_QUERY_BATCH_SIZE = 500 -def _load_contacts_from(db_path): - names = {} - full = [] - conn = sqlite3.connect(db_path) - try: - columns = { - row[1] for row in conn.execute("PRAGMA table_info(contact)").fetchall() - } - optional_columns = { - "alias": "", - "description": "", - "phone": "", - "phone_number": "", - "mobile": "", - "mobile_phone": "", - "telephone": "", - } - select_columns = ["username", "nick_name", "remark"] - select_columns.extend( - col for col in optional_columns - if col in columns and col not in select_columns - ) - rows = conn.execute( - "SELECT " + ", ".join(f"[{col}]" for col in select_columns) - + " FROM contact" - ).fetchall() - for r in rows: - data = dict(zip(select_columns, r)) - uname = data.get("username") - nick = data.get("nick_name") - remark = data.get("remark") - display = remark if remark else nick if nick else uname - names[uname] = display - phone = "" - for col in ("phone", "phone_number", "mobile", "mobile_phone", "telephone"): - if data.get(col): - phone = data.get(col) or "" - break - full.append({ - 'username': uname, - 'nick_name': nick or '', - 'remark': remark or '', - 'alias': data.get("alias") or '', - 'description': data.get("description") or '', - 'phone': phone, - }) - finally: - conn.close() - return names, full +def _load_contacts_from(db_path): + names = {} + full = [] + conn = sqlite3.connect(db_path) + try: + columns = { + row[1] for row in conn.execute("PRAGMA table_info(contact)").fetchall() + } + optional_columns = { + "alias": "", + "description": "", + "phone": "", + "phone_number": "", + "mobile": "", + "mobile_phone": "", + "telephone": "", + } + select_columns = ["username", "nick_name", "remark"] + select_columns.extend( + col for col in optional_columns + if col in columns and col not in select_columns + ) + # 过滤群成员 (local_type=3, 每个群每个成员一条 → 数量爆炸 → 共 9416 假联系人) + # 仅在 local_type 列存在时加 WHERE, 兼容老版本 schema (#117 fix) + sql = ( + "SELECT " + ", ".join(f"[{col}]" for col in select_columns) + + " FROM contact" + ) + if "local_type" in columns: + sql += " WHERE local_type != 3" + rows = conn.execute(sql).fetchall() + for r in rows: + data = dict(zip(select_columns, r)) + uname = data.get("username") + nick = data.get("nick_name") + remark = data.get("remark") + display = remark if remark else nick if nick else uname + names[uname] = display + phone = "" + for col in ("phone", "phone_number", "mobile", "mobile_phone", "telephone"): + if data.get(col): + phone = data.get(col) or "" + break + full.append({ + 'username': uname, + 'nick_name': nick or '', + 'remark': remark or '', + 'alias': data.get("alias") or '', + 'description': data.get("description") or '', + 'phone': phone, + }) + finally: + conn.close() + return names, full def _get_contact_db_path(): @@ -344,23 +349,23 @@ def get_contact_names(): return {} -def get_contact_full(): - get_contact_names() - return _contact_full or [] - - -def get_contact_tag_names_by_username(): - tags = _load_contact_tags() - by_username = {} - for tag in tags.values(): - name = tag.get('name') or '' - if not name: - continue - for member in tag.get('members', []): - username = member.get('username') - if username: - by_username.setdefault(username, []).append(name) - return by_username +def get_contact_full(): + get_contact_names() + return _contact_full or [] + + +def get_contact_tag_names_by_username(): + tags = _load_contact_tags() + by_username = {} + for tag in tags.values(): + name = tag.get('name') or '' + if not name: + continue + for member in tag.get('members', []): + username = member.get('username') + if username: + by_username.setdefault(username, []).append(name) + return by_username def _extract_pb_field_30(data):