# -*- coding: utf-8 -*- """ 🔐 混沌加密聊天 v6 - 单文件版 (PyQt5 客户端 + 全部后端) ============================================================ · 后端: 混沌加密/每用户密钥认证/账户SQLite/历史SQLite/媒体持久存储 · 界面: PyQt5 现代圆角UI, 自适应消息气泡 · 两种群: 🔗端到端(P2P) + ☁️中央服务器(解决NAT) · 中央服务器单独运行 server.py 运行: python dvs_enc.py """ import os os.environ.setdefault('PYGAME_HIDE_SUPPORT_PROMPT', '1') import io import sys import time import queue import wave import json import base64 import struct import socket import hashlib import hmac import threading import tempfile import sqlite3 import shutil from datetime import datetime # ---------------- 路径 ---------------- APP_DIR = os.path.dirname(os.path.abspath(__file__)) DATA_DIR = os.path.join(APP_DIR, 'data') try: os.makedirs(DATA_DIR, exist_ok=True) except Exception: DATA_DIR = APP_DIR USERS_DB = os.path.join(DATA_DIR, 'users.db') HISTORY_DB = os.path.join(DATA_DIR, 'chat_history.db') SESSION_FILE = os.path.join(DATA_DIR, 'session.dat') SERVER_DB = os.path.join(DATA_DIR, 'server_state.db') # 媒体持久目录: 文档/ChaosChat/media (用户可在 GUI 清理, 清理后数据消失) MEDIA_DIR = os.path.join(os.path.expanduser('~'), 'Documents', 'ChaosChat', 'media') try: os.makedirs(MEDIA_DIR, exist_ok=True) except Exception: MEDIA_DIR = os.path.join(DATA_DIR, 'media') try: os.makedirs(MEDIA_DIR, exist_ok=True) except Exception: pass # ---------------- 可选依赖 ---------------- try: import tkinter as tk from tkinter import scrolledtext, messagebox, simpledialog, filedialog GUI_AVAILABLE = True except Exception: GUI_AVAILABLE = False try: import pygame PYGAME_AVAILABLE = True except Exception: PYGAME_AVAILABLE = False try: from PIL import Image, ImageTk PIL_AVAILABLE = True except Exception: PIL_AVAILABLE = False # 录音后端: 优先 sounddevice, 其次 pyaudio RECORDER = None try: import sounddevice as sd import numpy as np RECORDER = 'sounddevice' except Exception: RECORDER = None if RECORDER is None: try: import pyaudio RECORDER = 'pyaudio' except Exception: RECORDER = None try: import qrcode QR_AVAILABLE = True except Exception: QR_AVAILABLE = False # ==================== 混沌加密核心 ==================== class ChaosCryptCore: def __init__(self, debug=False): self.debug = debug self.special_encrypt = {'+': '.', '/': "'", '=': ''} self.special_decrypt = {'.': '+', "'": '/'} self.flip_pattern = [1, 1, 0, 0, 1, 0, 0, 0, 1, 0] def _generate_random_alphabet(self, lowercase=False): chars = list("abcdefghijklmnopqrstuvwxyz" if lowercase else "ABCDEFGHIJKLMNOPQRSTUVWXYZ") n = len(chars) for i in range(n-1, 0, -1): j = int.from_bytes(os.urandom(1), 'big') % (i+1) chars[i], chars[j] = chars[j], chars[i] return ''.join(chars) def _generate_random_digit_mapping(self): digits = list("0123456789") pool = list("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz") n = len(pool) for i in range(n-1, 0, -1): j = int.from_bytes(os.urandom(1), 'big') % (i+1) pool[i], pool[j] = pool[j], pool[i] return {d: pool[i] for i, d in enumerate(digits)} def _generate_random_equal_mapping(self): chars = list("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz") n = len(chars) for i in range(n-1, 0, -1): j = int.from_bytes(os.urandom(1), 'big') % (i+1) chars[i], chars[j] = chars[j], chars[i] return {'0': chars[0], '1': chars[1], '2': chars[2], '3': chars[3]} def _generate_random_flip_pattern(self): return [int.from_bytes(os.urandom(1), 'big') % 2 for _ in range(10)] def _generate_short_key(self, length=512): chars = "0123456789abcdef" return ''.join(chars[int.from_bytes(os.urandom(1), 'big') % 16] for _ in range(length)) def _build_maps(self, upper_mapping, lower_mapping, digit_mapping, equal_mapping): upper_orig = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" lower_orig = "abcdefghijklmnopqrstuvwxyz" enc = {} dec = {} for i in range(26): enc[upper_orig[i]] = upper_mapping[i] enc[lower_orig[i]] = lower_mapping[i] dec[upper_mapping[i]] = upper_orig[i] dec[lower_mapping[i]] = lower_orig[i] digit_rev = {v: k for k, v in digit_mapping.items()} equal_rev = {v: k for k, v in equal_mapping.items()} return enc, dec, digit_rev, equal_rev def _ensure_printable(self, text): res = [] for c in text: v = ord(c) if v == 124: res.append('|') elif v < 32 or v > 126: v = v % 95 + 32 res.append(chr(v)) else: res.append(c) return ''.join(res) def _mix_keys(self, user_password, short_key, target_length): mixed = [] max_len = max(len(user_password), len(short_key)) for i in range(max_len): if i < len(user_password): mixed.append(ord(user_password[i])) if i < len(short_key): mixed.append(ord(short_key[i])) for i in range(len(mixed)): if i % 3 == 0: mixed[i] = (mixed[i] << 1) & 0xFF elif i % 3 == 1: mixed[i] = (mixed[i] >> 1) & 0xFF else: mixed[i] = mixed[i] ^ 0x5A for i in range(0, len(mixed)-3, 4): mixed[i], mixed[i+3] = mixed[i+3], mixed[i] mixed[i+1], mixed[i+2] = mixed[i+2], mixed[i+1] mixed.reverse() res = [] for x in mixed: if x < 32 or x > 126: x = x % 95 + 32 res.append(chr(x)) res_str = ''.join(res) if len(res_str) < target_length: base = res_str final = base it = 0 while len(final) < target_length: if it % 3 == 0: chunk = base[::-1] elif it % 3 == 1: chunk = base[::-1] chunk = ''.join([chr((ord(c) + it) % 95 + 32) for c in chunk]) else: chunk = base[::-1] chunk = ''.join([chr((ord(c) ^ it) % 95 + 32) for c in chunk]) final += chunk it += 1 res_str = final[:target_length] return self._ensure_printable(res_str) def _derive_final_key(self, user_password, short_key, text_length): mixed = self._mix_keys(user_password, short_key, text_length * 2) hex_key = ''.join([f"{ord(c):02x}" for c in mixed]) rev_hex = hex_key[::-1] final = [] for i in range(0, len(rev_hex), 2): if i+1 < len(rev_hex): byte = rev_hex[i:i+2] try: val = int(byte, 16) if val < 32 or val > 126: val = val % 95 + 32 final.append(chr(val)) except Exception: final.append('x') final_str = ''.join(final) if len(final_str) < text_length: base = final_str it = 0 while len(final_str) < text_length: if it % 3 == 0: chunk = base[::-1] elif it % 3 == 1: chunk = base[::-1] chunk = ''.join([chr((ord(c) + it) % 95 + 32) for c in chunk]) else: chunk = base[::-1] chunk = ''.join([chr((ord(c) ^ it) % 95 + 32) for c in chunk]) final_str += chunk it += 1 final_str = final_str[:text_length] return self._ensure_printable(final_str) def _apply_flip(self, text, flip_pattern): res = [] for i, ch in enumerate(text): idx = i % len(flip_pattern) if ch.isalpha() and flip_pattern[idx] == 1: res.append(ch.swapcase()) else: res.append(ch) return ''.join(res) def _xor_encrypt_to_hex(self, text, key): key_len = len(key) res = [] for i, ch in enumerate(text): res.append(f"{ord(ch) ^ ord(key[i % key_len]):02x}") return ''.join(res) def _xor_decrypt_to_string(self, hex_text, key): try: key_len = len(key) res = [] for i in range(0, len(hex_text), 2): if i+1 < len(hex_text): byte = hex_text[i:i+2] val = int(byte, 16) ^ ord(key[(i//2) % key_len]) if val < 32 or val > 126: val = val % 95 + 32 res.append(chr(val)) return ''.join(res) except Exception: return None def _substitute_letters(self, text, enc_map): return ''.join([enc_map.get(c, c) for c in text]) def _inverse_substitute(self, text, dec_map): return ''.join([dec_map.get(c, c) for c in text]) def encrypt(self, plaintext, session_key, params): b64 = base64.b64encode(plaintext.encode('utf-8')).decode('ascii') processed = b64 for old, new in self.special_encrypt.items(): processed = processed.replace(old, new) processed = processed.rstrip('=') enc_map, _, _, _ = self._build_maps( params['upper_mapping'], params['lower_mapping'], params['digit_mapping'], params['equal_mapping'] ) sub = self._substitute_letters(processed, enc_map) flipped = self._apply_flip(sub, params['flip_pattern']) rev = flipped[::-1] eq_count = b64.count('=') eq_char = params['equal_mapping'][str(eq_count)] with_equal = f"{rev}|{eq_char}" final_key = self._derive_final_key( session_key[:16], params['short_key_fragment'], len(with_equal) ) return self._xor_encrypt_to_hex(with_equal, final_key) def decrypt(self, ciphertext, session_key, params): try: final_key = self._derive_final_key( session_key[:16], params['short_key_fragment'], len(ciphertext) // 2 ) decrypted = self._xor_decrypt_to_string(ciphertext, final_key) if decrypted is None: return None if '|' in decrypted: main, eq_char = decrypted.split('|') eq_rev = {v: k for k, v in params['equal_mapping'].items()} eq_count = int(eq_rev.get(eq_char, '0')) else: main = decrypted eq_count = 0 rev = main[::-1] flipped = self._apply_flip(rev, params['flip_pattern']) _, dec_map, _, _ = self._build_maps( params['upper_mapping'], params['lower_mapping'], params['digit_mapping'], params['equal_mapping'] ) sub = self._inverse_substitute(flipped, dec_map) for old, new in self.special_decrypt.items(): sub = sub.replace(old, new) b64_str = sub + '=' * eq_count return base64.b64decode(b64_str.encode('ascii')).decode('utf-8') except Exception: return None # ==================== 会话密钥 (共享加密) ==================== class SecureSession: """基于 ChaosCryptCore 的共享会话加密 (群聊同密钥)""" def __init__(self): self.core = ChaosCryptCore() self.session_key = None self.params = None def generate(self): self.session_key = os.urandom(32).hex() self.params = { 'upper_mapping': self.core._generate_random_alphabet(False), 'lower_mapping': self.core._generate_random_alphabet(True), 'digit_mapping': self.core._generate_random_digit_mapping(), 'equal_mapping': self.core._generate_random_equal_mapping(), 'flip_pattern': self.core._generate_random_flip_pattern(), 'short_key_fragment': self.core._generate_short_key(32), 'generated_at': datetime.now().isoformat() } return self.session_key def export_key(self): data = json.dumps({'session_key': self.session_key, 'params': self.params}, ensure_ascii=False) return base64.urlsafe_b64encode(data.encode('utf-8')).decode('ascii') def import_key(self, key_str): try: raw = base64.urlsafe_b64decode(key_str.encode('ascii')) data = json.loads(raw.decode('utf-8')) self.session_key = data['session_key'] self.params = data['params'] return True except Exception: return False def has_key(self): return self.session_key is not None and self.params is not None def encrypt(self, plaintext): if not self.has_key(): raise ValueError("会话密钥未建立") nonce = os.urandom(16).hex() msg_key = hashlib.sha256((self.session_key + nonce).encode('utf-8')).hexdigest() cipher = self.core.encrypt(plaintext, msg_key, self.params) hmac = hashlib.sha256((cipher + self.session_key).encode('utf-8')).hexdigest()[:16] return json.dumps({ 'nonce': nonce, 'ciphertext': cipher, 'hmac': hmac, 'ts': int(time.time()) }, ensure_ascii=False) def decrypt(self, packet_json): try: pkt = json.loads(packet_json) hmac_calc = hashlib.sha256((pkt['ciphertext'] + self.session_key).encode('utf-8')).hexdigest()[:16] if hmac_calc != pkt['hmac']: return None msg_key = hashlib.sha256((self.session_key + pkt['nonce']).encode('utf-8')).hexdigest() return self.core.decrypt(pkt['ciphertext'], msg_key, self.params) except Exception: return None # ---- 二进制数据加密 (文件/图片/语音) ---- def _keystream(self, nonce, length): return hashlib.shake_256((self.session_key + nonce).encode('utf-8')).digest(length) @staticmethod def _fast_xor(a, b, chunk=1 << 20): out = bytearray() n = len(a) for i in range(0, n, chunk): size = min(chunk, n - i) x = int.from_bytes(a[i:i+size], 'big') ^ int.from_bytes(b[i:i+size], 'big') out += x.to_bytes(size, 'big') return bytes(out) def encrypt_bytes(self, data): if not self.has_key(): raise ValueError("会话密钥未建立") nonce = os.urandom(16).hex() ks = self._keystream(nonce, len(data)) cipher = self._fast_xor(data, ks) hmac = hashlib.sha256( b'BIN|' + self.session_key.encode('utf-8') + b'|' + nonce.encode('utf-8') + b'|' + cipher).hexdigest() return json.dumps({'alg': 'xor', 'nonce': nonce, 'data': base64.b64encode(cipher).decode('ascii'), 'hmac': hmac}, ensure_ascii=False) def decrypt_bytes(self, packet): try: pkt = json.loads(packet) nonce = pkt['nonce'] cipher = base64.b64decode(pkt['data']) hmac_calc = hashlib.sha256( b'BIN|' + self.session_key.encode('utf-8') + b'|' + nonce.encode('utf-8') + b'|' + cipher).hexdigest() if hmac_calc != pkt['hmac']: return None ks = self._keystream(nonce, len(cipher)) return self._fast_xor(cipher, ks) except Exception: return None def auth_secret(self): """从会话密钥派生的群认证密钥 (证明是群成员)""" return hashlib.sha256(b'CHAOS_AUTH|' + self.session_key.encode('utf-8')).hexdigest() def user_key_enc(self): """用于加密下发每用户密钥的密钥 (客户端/服务器各自计算一致)""" return hashlib.sha256(b'CHAOS_USERKEY_ENC|' + self.session_key.encode('utf-8')).hexdigest() # ==================== 传输协议 (帧封装) ==================== MAX_FRAME = 256 * 1024 * 1024 def send_msg(sock, obj): payload = json.dumps(obj, ensure_ascii=False).encode('utf-8') sock.sendall(struct.pack('>I', len(payload)) + payload) def recv_exact(sock, n): data = b'' while len(data) < n: try: chunk = sock.recv(n - len(data)) except socket.timeout: continue except Exception: return None if not chunk: return None data += chunk return data def recv_msg(sock): header = recv_exact(sock, 4) if header is None: return None length = struct.unpack('>I', header)[0] if length <= 0 or length > MAX_FRAME: return None payload = recv_exact(sock, length) if payload is None: return None try: return json.loads(payload.decode('utf-8')) except Exception: return None # ==================== 通用工具 ==================== def stream_xor(data, key_bytes): """shake256 密钥流异或 (用于加密下发每用户密钥)""" ks = hashlib.shake_256(key_bytes).digest(len(data)) out = bytearray() n = len(data) chunk = 1 << 20 for i in range(0, n, chunk): size = min(chunk, n - i) x = int.from_bytes(data[i:i+size], 'big') ^ int.from_bytes(ks[i:i+size], 'big') out += x.to_bytes(size, 'big') return bytes(out) def gen_mid(): return f"{int(time.time()*1000)}-{os.urandom(4).hex()}" def fmt_time(ts): return datetime.fromtimestamp(ts).strftime('%H:%M:%S') # ==================== 文件类型识别 ==================== FILE_ICONS = { 'image': '🖼️', 'audio': '🎵', 'video': '🎬', 'archive': '📦', 'exe': '🛠️', 'lnk': '🔗', 'link': '🔗', 'doc': '📝', 'sheet': '📊', 'slide': '📽️', 'pdf': '📃', 'code': '💻', 'text': '📄', 'default': '📄', } def file_icon(ext): """按扩展名返回可识别图标""" e = (ext or '').lower().lstrip('.') if e in ('png', 'jpg', 'jpeg', 'gif', 'bmp', 'webp', 'ico'): return FILE_ICONS['image'] if e in ('wav', 'mp3', 'ogg', 'flac', 'aac', 'm4a'): return FILE_ICONS['audio'] if e in ('mp4', 'avi', 'mkv', 'mov', 'wmv', 'flv'): return FILE_ICONS['video'] if e in ('zip', 'rar', '7z', 'tar', 'gz', 'bz2'): return FILE_ICONS['archive'] if e in ('exe', 'msi', 'bat', 'cmd'): return FILE_ICONS['exe'] if e == 'lnk': return FILE_ICONS['lnk'] if e in ('doc', 'docx', 'rtf', 'odt'): return FILE_ICONS['doc'] if e in ('xls', 'xlsx', 'csv', 'ods'): return FILE_ICONS['sheet'] if e in ('ppt', 'pptx', 'odp'): return FILE_ICONS['slide'] if e == 'pdf': return FILE_ICONS['pdf'] if e in ('py', 'js', 'html', 'css', 'c', 'cpp', 'java', 'go', 'rs', 'json', 'xml'): return FILE_ICONS['code'] if e in ('txt', 'md', 'log'): return FILE_ICONS['text'] return FILE_ICONS['default'] # ==================== 媒体持久存储 ==================== def save_media(data, ext='.bin'): """保存媒体文件到 文档/ChaosChat/media, 返回绝对路径""" fname = f"{int(time.time()*1000)}-{os.urandom(4).hex()}{ext}" path = os.path.join(MEDIA_DIR, fname) try: with open(path, 'wb') as f: f.write(data) return path except Exception: return '' def delete_media(path): try: if path and os.path.exists(path): os.remove(path) except Exception: pass def clear_media_cache(): """清空媒体缓存目录 (数据不可恢复)""" if not os.path.isdir(MEDIA_DIR): return 0 count = 0 for name in os.listdir(MEDIA_DIR): p = os.path.join(MEDIA_DIR, name) try: if os.path.isfile(p): os.remove(p) count += 1 except Exception: pass return count # ==================== 账户数据库 (SQLite) ==================== class AccountDB: """账户 + 会话 + 已保存群""" def __init__(self, db_path=USERS_DB): self.path = db_path self.conn = sqlite3.connect(db_path, check_same_thread=False) self.conn.row_factory = sqlite3.Row self.lock = threading.Lock() self._init() def _init(self): with self.lock: cur = self.conn.cursor() cur.execute('''CREATE TABLE IF NOT EXISTS accounts( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, nickname TEXT, salt TEXT NOT NULL, pass_hash TEXT NOT NULL, created_at REAL)''') cur.execute('''CREATE TABLE IF NOT EXISTS sessions( id INTEGER PRIMARY KEY AUTOINCREMENT, account_id INTEGER NOT NULL, token TEXT UNIQUE NOT NULL, history_key TEXT, created_at REAL)''') scols = [r[1] for r in cur.execute('PRAGMA table_info(sessions)').fetchall()] if 'history_key' not in scols: cur.execute('ALTER TABLE sessions ADD COLUMN history_key TEXT') cur.execute('''CREATE TABLE IF NOT EXISTS saved_groups( id INTEGER PRIMARY KEY AUTOINCREMENT, account_id INTEGER NOT NULL, group_key TEXT NOT NULL, host TEXT, port INTEGER, session_key TEXT, nickname TEXT, is_host INTEGER DEFAULT 0, user_key TEXT, UNIQUE(account_id, group_key))''') cols = [r[1] for r in cur.execute('PRAGMA table_info(saved_groups)').fetchall()] if 'user_key' not in cols: cur.execute('ALTER TABLE saved_groups ADD COLUMN user_key TEXT') self.conn.commit() # ---- 密码哈希 (PBKDF2) ---- @staticmethod def hash_password(password, salt_hex): salt = bytes.fromhex(salt_hex) return hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), salt, 120000).hex() def register(self, username, password): salt = os.urandom(16).hex() ph = self.hash_password(password, salt) try: with self.lock: cur = self.conn.cursor() cur.execute('INSERT INTO accounts(username, nickname, salt, pass_hash, created_at) VALUES(?,?,?,?,?)', (username, username, salt, ph, time.time())) self.conn.commit() return True, 'ok' except sqlite3.IntegrityError: return False, '该用户名已被注册' def login(self, username, password): with self.lock: cur = self.conn.cursor() row = cur.execute('SELECT * FROM accounts WHERE username=?', (username,)).fetchone() if row is None: return None, '用户不存在' ph = self.hash_password(password, row['salt']) if ph != row['pass_hash']: return None, '密码错误' return row, 'ok' def create_session(self, account_id): token = os.urandom(32).hex() with self.lock: cur = self.conn.cursor() cur.execute('DELETE FROM sessions WHERE account_id=?', (account_id,)) cur.execute('INSERT INTO sessions(account_id, token, created_at) VALUES(?,?,?)', (account_id, token, time.time())) self.conn.commit() return token def validate_session(self, token): with self.lock: cur = self.conn.cursor() row = cur.execute('''SELECT a.* FROM sessions s JOIN accounts a ON a.id = s.account_id WHERE s.token=?''', (token,)).fetchone() return row def update_nickname(self, account_id, nickname): with self.lock: cur = self.conn.cursor() cur.execute('UPDATE accounts SET nickname=? WHERE id=?', (nickname, account_id)) self.conn.commit() # ---- 已保存的群 ---- def save_group(self, account_id, group_key, host, port, session_key, nickname, is_host, user_key=''): with self.lock: cur = self.conn.cursor() cur.execute('''INSERT INTO saved_groups(account_id, group_key, host, port, session_key, nickname, is_host, user_key) VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(account_id, group_key) DO UPDATE SET host=excluded.host, port=excluded.port, session_key=excluded.session_key, nickname=excluded.nickname, is_host=excluded.is_host, user_key=excluded.user_key''', (account_id, group_key, host, port, session_key, nickname, 1 if is_host else 0, user_key or '')) self.conn.commit() def load_groups(self, account_id): with self.lock: cur = self.conn.cursor() return cur.execute('SELECT * FROM saved_groups WHERE account_id=?', (account_id,)).fetchall() def remove_group(self, account_id, group_key): with self.lock: cur = self.conn.cursor() cur.execute('DELETE FROM saved_groups WHERE account_id=? AND group_key=?', (account_id, group_key)) self.conn.commit() # ==================== 历史记录数据库 (SQLite) ==================== class HistoryDB: """群聊记录: 文本存内容, 文件/语音只存路径""" def __init__(self, db_path=HISTORY_DB): self.conn = sqlite3.connect(db_path, check_same_thread=False) self.conn.row_factory = sqlite3.Row self.lock = threading.Lock() self._init() def _init(self): with self.lock: cur = self.conn.cursor() cur.execute('''CREATE TABLE IF NOT EXISTS messages( id INTEGER PRIMARY KEY AUTOINCREMENT, account_id INTEGER NOT NULL, group_key TEXT NOT NULL, mid TEXT, sender TEXT, mtype TEXT, content TEXT, ts REAL, is_mine INTEGER DEFAULT 0, withdrawn INTEGER DEFAULT 0)''') cur.execute('CREATE INDEX IF NOT EXISTS idx_msg ON messages(account_id, group_key, id)') self.conn.commit() @staticmethod def serialize_content(msg): if msg.type in ('text', 'system'): return str(msg.content) if isinstance(msg.content, dict): d = dict(msg.content) d.pop('raw', None) # 不入库: 数据存文档目录, 库中只记路径 return json.dumps(d, ensure_ascii=False) return str(msg.content) @staticmethod def deserialize_content(mtype, content): if mtype in ('text', 'system'): return content try: return json.loads(content) except Exception: return content def save(self, account_id, group_key, msg): with self.lock: cur = self.conn.cursor() cur.execute('''INSERT INTO messages(account_id, group_key, mid, sender, mtype, content, ts, is_mine, withdrawn) VALUES(?,?,?,?,?,?,?,?,?)''', (account_id, group_key, msg.mid, msg.sender, msg.type, self.serialize_content(msg), msg.timestamp, 1 if msg.is_mine else 0, 1 if msg.withdrawn else 0)) self.conn.commit() def load(self, account_id, group_key, limit=500): with self.lock: cur = self.conn.cursor() return cur.execute('''SELECT * FROM ( SELECT * FROM messages WHERE account_id=? AND group_key=? ORDER BY id DESC LIMIT ?) ORDER BY id ASC''', (account_id, group_key, limit)).fetchall() def delete_by_mid(self, account_id, group_key, mid): with self.lock: cur = self.conn.cursor() cur.execute('DELETE FROM messages WHERE account_id=? AND group_key=? AND mid=?', (account_id, group_key, mid)) self.conn.commit() def clear_group(self, account_id, group_key): with self.lock: cur = self.conn.cursor() cur.execute('DELETE FROM messages WHERE account_id=? AND group_key=?', (account_id, group_key)) self.conn.commit() # ==================== 语音管理 ==================== class VoiceManager: def __init__(self, on_status=None): self.on_status = on_status self.rate = 44100 self.channels = 1 self.recording = False self._thread = None self._frames = [] self._result = None self.last_voice = None self.last_voice_from = '' def _status(self, text): if self.on_status: try: self.on_status(text) except Exception: pass def _make_wav(self, pcm_bytes): buf = io.BytesIO() with wave.open(buf, 'wb') as wf: wf.setnchannels(self.channels) wf.setsampwidth(2) wf.setframerate(self.rate) wf.writeframes(pcm_bytes) return buf.getvalue() def _record_worker(self): self._result = None try: if RECORDER == 'sounddevice': self._frames = [] def callback(indata, frames, time_info, status): if self.recording: self._frames.append(indata.copy()) with sd.InputStream(samplerate=self.rate, channels=1, callback=callback, dtype='float32'): while self.recording: time.sleep(0.05) if self._frames: audio = np.concatenate(self._frames, axis=0) pcm = (np.clip(audio, -1.0, 1.0) * 32767).astype(np.int16).tobytes() self._result = self._make_wav(pcm) elif RECORDER == 'pyaudio': p = pyaudio.PyAudio() stream = p.open(format=pyaudio.paInt16, channels=1, rate=self.rate, input=True, frames_per_buffer=1024) frames = [] while self.recording: data = stream.read(1024, exception_on_overflow=False) frames.append(data) stream.stop_stream() stream.close() p.terminate() if frames: self._result = self._make_wav(b''.join(frames)) if self._result: self._status(f"✅ 录音完成 ({len(self._result) // 1024} KB)") except Exception as e: self._result = None self._status(f"❌ 录音失败: {e}") def start_recording(self): if RECORDER is None: self._status("⚠️ 请安装 sounddevice 或 pyaudio 才能录音") return False if self.recording: return False self.recording = True self._thread = threading.Thread(target=self._record_worker, daemon=True) self._thread.start() self._status("🎤 录音中... 再点一次停止并发送") return True def stop_recording(self): if not self.recording: return None self.recording = False if self._thread: self._thread.join(timeout=10) result = self._result self._result = None if result: self.last_voice = result self.last_voice_from = '' return result def play(self, wav_bytes, from_user=''): if not wav_bytes: return False if not PYGAME_AVAILABLE: self._status("⚠️ 未安装 pygame，无法播放语音") return False try: if not pygame.mixer.get_init(): pygame.mixer.init(frequency=self.rate, size=-16, channels=self.channels) fd, path = tempfile.mkstemp(suffix='.wav') with os.fdopen(fd, 'wb') as f: f.write(wav_bytes) sound = pygame.mixer.Sound(path) sound.play() def _cleanup(): for _ in range(300): if not pygame.mixer.get_busy(): break time.sleep(0.1) try: os.remove(path) except Exception: pass threading.Thread(target=_cleanup, daemon=True).start() self.last_voice = wav_bytes self.last_voice_from = from_user if from_user: self._status(f"🔊 播放来自 {from_user} 的语音") return True except Exception as e: self._status(f"❌ 播放失败: {e}") return False # ==================== 消息模型 ==================== class Message: def __init__(self, mid, group, sender, msg_type, content, timestamp, is_mine=False): self.mid = mid self.group = group self.sender = sender self.type = msg_type # text / image / voice / audio / file / system self.content = content # 文本 str, 或 {'filename','size','path','ext'} self.timestamp = timestamp self.is_mine = is_mine self.read = False self.withdrawn = False def is_withdrawable(self, now=None): now = now or time.time() return self.is_mine and not self.withdrawn and (now - self.timestamp) <= 120 def get_data_bytes(self): """返回文件/图片/语音的原始字节 (从磁盘路径读取)""" if isinstance(self.content, dict): p = self.content.get('path') if p and os.path.isfile(p): try: with open(p, 'rb') as f: return f.read() except Exception: return b'' return b'' # ==================== 服务器状态数据库 (踢人标记/群名/公告) ==================== class ServerDB: """服务器级持久状态: 被踢账户 + 群名 + 公告 (重启不丢)""" def __init__(self, db_path=SERVER_DB): self.conn = sqlite3.connect(db_path, check_same_thread=False) self.conn.row_factory = sqlite3.Row self.lock = threading.Lock() self._init() def _init(self): with self.lock: cur = self.conn.cursor() cur.execute("""CREATE TABLE IF NOT EXISTS kicked( server_key TEXT NOT NULL, account TEXT NOT NULL, kicked_at REAL, PRIMARY KEY(server_key, account))""") cur.execute("""CREATE TABLE IF NOT EXISTS server_info( server_key TEXT PRIMARY KEY, group_name TEXT, announcement TEXT)""") cur.execute("""CREATE TABLE IF NOT EXISTS user_keys( server_key TEXT NOT NULL, account TEXT NOT NULL, user_key TEXT, PRIMARY KEY(server_key, account))""") self.conn.commit() def is_kicked(self, server_key, account): with self.lock: cur = self.conn.cursor() row = cur.execute('SELECT 1 FROM kicked WHERE server_key=? AND account=?', (server_key, account)).fetchone() return row is not None def add_kicked(self, server_key, account): with self.lock: cur = self.conn.cursor() cur.execute('INSERT OR REPLACE INTO kicked(server_key, account, kicked_at) VALUES(?,?,?)', (server_key, account, time.time())) self.conn.commit() def remove_kicked(self, server_key, account): with self.lock: cur = self.conn.cursor() cur.execute('DELETE FROM kicked WHERE server_key=? AND account=?', (server_key, account)) self.conn.commit() # ---- 每用户密钥 (身份认证) ---- def get_user_key(self, server_key, account): with self.lock: cur = self.conn.cursor() row = cur.execute('SELECT user_key FROM user_keys WHERE server_key=? AND account=?', (server_key, account)).fetchone() return row['user_key'] if row else None def set_user_key(self, server_key, account, user_key): with self.lock: cur = self.conn.cursor() cur.execute('INSERT OR REPLACE INTO user_keys(server_key, account, user_key) VALUES(?,?,?)', (server_key, account, user_key)) self.conn.commit() def get_info(self, server_key): with self.lock: cur = self.conn.cursor() row = cur.execute('SELECT * FROM server_info WHERE server_key=?', (server_key,)).fetchone() if row: return row['group_name'] or '', row['announcement'] or '' return '', '' def save_info(self, server_key, group_name, announcement): with self.lock: cur = self.conn.cursor() cur.execute("""INSERT INTO server_info(server_key, group_name, announcement) VALUES(?,?,?) ON CONFLICT(server_key) DO UPDATE SET group_name=excluded.group_name, announcement=excluded.announcement""", (server_key, group_name or '', announcement or '')) self.conn.commit() # ==================== 多用户服务器 v5 ==================== class ChatServer: """多用户聊天服务器: 群主管理(踢人/群名/公告) + 消息广播""" def __init__(self, on_status=None, server_key='default', server_db=None, auth_secret=None, server_secret=None): self.sock = None self.running = False self.clients = {} # cid -> {socket, nickname, account, addr, lock, is_host, pending, user_key} self.next_id = 1 self.lock = threading.Lock() self.on_status = on_status self.host = '0.0.0.0' self.port = 8888 self.host_cid = None # 第一个加入的客户端 = 群主 self.announcement = '' self.group_name = '' self.server_key = server_key self.server_db = server_db self.auth_secret = auth_secret self.user_key_enc = None if server_secret: self.auth_secret = hashlib.sha256( b'CHAOS_AUTH|' + server_secret.encode('utf-8')).hexdigest() self.user_key_enc = hashlib.sha256( b'CHAOS_USERKEY_ENC|' + server_secret.encode('utf-8')).hexdigest() self.mid_owner = {} # mid -> (account, ts) 消息归属, 用于撤回校验 def _status(self, text): if self.on_status: try: self.on_status(text) except Exception: pass def start(self, port=8888): try: if self.server_db: gn, an = self.server_db.get_info(self.server_key) self.group_name = gn or '' self.announcement = an or '' self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.sock.bind((self.host, port)) self.sock.listen(10) self.running = True self.port = port threading.Thread(target=self._accept_loop, daemon=True).start() self._status(f"🔄 服务器已启动，端口 {port}，等待连接...") return True except Exception as e: self._status(f"❌ 服务器启动失败: {e}") return False def stop(self): self.running = False if self.sock: try: self.sock.close() except Exception: pass self.sock = None with self.lock: socks = [c['socket'] for c in self.clients.values()] self.clients.clear() for s in socks: try: s.close() except Exception: pass def _accept_loop(self): while self.running: try: client, addr = self.sock.accept() with self.lock: cid = self.next_id self.next_id += 1 self.clients[cid] = { 'socket': client, 'nickname': f'用户{cid}', 'account': '', 'addr': addr, 'lock': threading.Lock(), 'is_host': False } self._status(f"✅ 新连接: {addr[0]}:{addr[1]}") threading.Thread(target=self._client_handler, args=(cid,), daemon=True).start() except Exception: break def _client_handler(self, cid): client = self.clients[cid]['socket'] # 第一步: join (含账户 + 昵称) try: msg = recv_msg(client) if msg and msg.get('type') == 'join': account = str(msg.get('account', f'用户{cid}'))[:24] nickname = str(msg.get('nickname') or msg.get('username') or account)[:20] with self.lock: if cid in self.clients: c = self.clients[cid] c['account'] = account c['nickname'] = nickname if self.host_cid is None: self.host_cid = cid c['is_host'] = True # 加入认证 ①: 必须持有群密钥 (证明是群成员) if self.auth_secret: expect = hmac.new(self.auth_secret.encode('utf-8'), account.encode('utf-8'), hashlib.sha256).hexdigest() if msg.get('auth') != expect: self._status(f"⚠️ 拒绝未认证的加入请求: {account}") self._remove_client(cid) return # 加入认证 ②: 每用户密钥 (防止成员冒充他人) user_key = None if self.server_db: user_key = self.server_db.get_user_key(self.server_key, account) if user_key: expect2 = hmac.new(user_key.encode('utf-8'), ('JOIN:' + account).encode('utf-8'), hashlib.sha256).hexdigest() if msg.get('user_auth') != expect2: self._status(f"⚠️ 拒绝身份认证失败(冒充): {account}") self._remove_client(cid) return else: # 新账户: 签发本人密钥并加密下发 if self.user_key_enc: user_key = os.urandom(16).hex() if self.server_db: self.server_db.set_user_key(self.server_key, account, user_key) enc = base64.b64encode( stream_xor(user_key.encode('utf-8'), bytes.fromhex(self.user_key_enc))).decode('ascii') self._send_to(cid, {'type': 'user_key_issued', 'data': enc}) if user_key: with self.lock: if cid in self.clients: self.clients[cid]['user_key'] = user_key # 被踢过的账户: 需群主同意 if self.server_db and self.server_db.is_kicked(self.server_key, account): with self.lock: if cid in self.clients: self.clients[cid]['pending'] = True self._status(f"⏳ {nickname} 曾被移出, 等待群主同意") self._send_to(cid, {'type': 'need_approval', 'reason': '你曾被移出本群，需群主同意后才能加入'}) if self.host_cid is not None: self._send_to(self.host_cid, {'type': 'approval_request', 'account': account, 'nickname': nickname}) if not self._wait_approval(cid): self._remove_client(cid) return with self.lock: if cid in self.clients: self.clients[cid]['pending'] = False self._send_to(cid, {'type': 'approval_granted'}) self._status(f"✅ {nickname} 经群主同意重新加入") self._broadcast({'type': 'system', 'text': f"✅ {nickname} 经群主同意重新加入"}, exclude=cid) self._broadcast_users() # 给成员发群信息 self._send_to(cid, {'type': 'group_info', 'group_name': self.group_name, 'announcement': self.announcement}) self._status(f"👋 {nickname} 加入了聊天室") self._broadcast({'type': 'system', 'text': f"👋 {nickname} 加入了聊天室"}, exclude=cid) self._broadcast_users() else: self._remove_client(cid) return except Exception: self._remove_client(cid) return # 第二步: 消息循环 while self.running: if cid not in self.clients: break try: msg = recv_msg(client) if msg is None: break self._handle_msg(cid, msg) except Exception: break self._remove_client(cid) def _handle_msg(self, cid, msg): mtype = msg.get('type') with self.lock: if cid not in self.clients: return c = self.clients[cid] uname = c['nickname'] account = c.get('account', '') is_host = c.get('is_host', False) user_key = c.get('user_key') # 来源认证: 校验每用户签名 + 账户绑定 (服务器只转发真实用户的消息) secret = user_key or self.auth_secret if secret and not self._verify_sig(msg, account, secret): self._status(f"⚠️ 拒绝未认证的消息 (来自 {uname})") return if mtype == 'rename': new_name = str(msg.get('new_name', uname))[:20] with self.lock: if cid in self.clients: self.clients[cid]['nickname'] = new_name self._status(f"✏️ {uname} 改名为 {new_name}") self._broadcast({'type': 'system', 'text': f"✏️ {uname} 改名为 {new_name}"}, exclude=cid) self._send_to(cid, {'type': 'system', 'text': f"✅ 你已改名为 {new_name}"}) self._broadcast_users() elif mtype == 'set_announcement': if not is_host: return self.announcement = str(msg.get('text', ''))[:500] if self.server_db: self.server_db.save_info(self.server_key, self.group_name, self.announcement) self._broadcast({'type': 'announcement', 'text': self.announcement, 'ts': int(time.time())}, exclude=None) self._status("📢 公告已更新") elif mtype == 'set_group_name': if not is_host: return self.group_name = str(msg.get('name', ''))[:30] if self.server_db: self.server_db.save_info(self.server_key, self.group_name, self.announcement) self._broadcast({'type': 'group_name', 'name': self.group_name}, exclude=None) self._status(f"🏷️ 群名更新: {self.group_name}") elif mtype == 'kick': if not is_host: return target = str(msg.get('target', msg.get('account', ''))) kick_cid = None with self.lock: for cid2, cc in self.clients.items(): if cc.get('account') == target and cid2 != cid: kick_cid = cid2 break if kick_cid is not None: if self.server_db: self.server_db.add_kicked(self.server_key, target) self._send_to(kick_cid, {'type': 'kick', 'reason': '你已被群主移出群'}) self._broadcast({'type': 'system', 'text': f"🚫 {target} 已被群主移出群"}, exclude=None) self._remove_client(kick_cid) elif mtype == 'approve': if not is_host: return target = str(msg.get('target', msg.get('account', ''))) decision = msg.get('decision', 'deny') with self.lock: target_cid = None for cid2, cc in self.clients.items(): if cc.get('account') == target and cc.get('pending'): target_cid = cid2 break if target_cid is not None: with self.lock: if target_cid in self.clients: self.clients[target_cid]['approve_decision'] = decision if decision == 'grant': self._status(f"✅ 群主同意 {target} 加入") else: self._status(f"❌ 群主拒绝 {target} 加入") elif mtype == 'withdraw': # 只有消息原作者才能撤回 mid = msg.get('mid', '') owner = self.mid_owner.get(mid) if owner and owner[0] == account: msg['from'] = uname self._broadcast(msg, exclude=cid) else: self._status(f"⚠️ 拒绝他人消息的撤回: {uname}") else: # 普通消息 (chat / file / voice 等): 记录归属并转发 mid = msg.get('mid') if mid: self.mid_owner[mid] = (account, time.time()) self._prune_mid_owner() msg['from'] = uname msg['ts'] = msg.get('ts', int(time.time())) self._broadcast(msg, exclude=cid) def _verify_sig(self, msg, account, secret): """校验消息 HMAC 签名 + 账户绑定 (secret 为每用户密钥或群密钥)""" try: sig = msg.get('sig') if not sig or not secret: return False canonical = json.dumps( {k: v for k, v in msg.items() if k not in ('sig', 'from', 'ts')}, sort_keys=True, ensure_ascii=False) expect = hmac.new(secret.encode('utf-8'), canonical.encode('utf-8'), hashlib.sha256).hexdigest() return (sig == expect) and (msg.get('account') == account) except Exception: return False def _prune_mid_owner(self): now = time.time() expired = [m for m, (a, t) in self.mid_owner.items() if now - t > 300] for m in expired: self.mid_owner.pop(m, None) def _wait_approval(self, cid): """等待群主同意; 返回 True=同意, False=拒绝/断开""" client = self.clients[cid]['socket'] try: client.settimeout(0.5) except Exception: pass try: while self.running and cid in self.clients: with self.lock: decision = self.clients[cid].get('approve_decision') if cid in self.clients else None if decision == 'grant': return True if decision == 'deny': self._send_to(cid, {'type': 'approval_denied', 'reason': '群主拒绝了你的加入'}) return False try: data = client.recv(1024) if not data: return False except socket.timeout: continue except Exception: return False return False finally: try: client.settimeout(None) except Exception: pass def _send_to(self, cid, msg): with self.lock: c = self.clients.get(cid) if c: try: with c['lock']: send_msg(c['socket'], msg) except Exception: pass def _remove_client(self, cid): with self.lock: entry = self.clients.pop(cid, None) if cid == self.host_cid: self.host_cid = None if entry: try: entry['socket'].close() except Exception: pass uname = entry['nickname'] self._status(f"🔌 {uname} 已断开") self._broadcast({'type': 'system', 'text': f"👋 {uname} 离开了聊天室"}, exclude=cid) self._broadcast_users() def _broadcast_users(self): with self.lock: users = [{'name': c['nickname'], 'account': c.get('account', '')} for c in self.clients.values()] self._broadcast({'type': 'users', 'users': users}, exclude=None) def _broadcast(self, msg, exclude=None): with self.lock: targets = [c for cid, c in self.clients.items() if cid != exclude] for c in targets: try: with c['lock']: send_msg(c['socket'], msg) except Exception: pass # ==================== 聊天客户端 v5 ==================== class ChatClient: def __init__(self, on_message=None, on_status=None, on_users=None, on_file=None, on_voice=None, on_system=None, on_withdraw=None, on_announcement=None, on_group_name=None, on_kick=None, on_group_info=None, on_need_approval=None, on_approval_granted=None, on_approval_denied=None, on_approval_request=None, on_user_key_issued=None, on_read=None): self.sock = None self.username = '' self.running = False self.connected = False self.user_key = None self.user_key_enc = None self.on_message = on_message self.on_status = on_status self.on_users = on_users self.on_file = on_file self.on_voice = on_voice self.on_system = on_system self.on_withdraw = on_withdraw self.on_announcement = on_announcement self.on_group_name = on_group_name self.on_kick = on_kick self.on_group_info = on_group_info self.on_need_approval = on_need_approval self.on_approval_granted = on_approval_granted self.on_approval_denied = on_approval_denied self.on_approval_request = on_approval_request self.on_user_key_issued = on_user_key_issued self.on_read = on_read self.account = '' self.auth_secret = None def _emit_status(self, text): if self.on_status: try: self.on_status(text) except Exception: pass def connect(self, host, port, username, nickname=None, account=None, auth_secret=None, user_key=None, user_key_enc=None): try: self.sock = socket.create_connection((host, port), timeout=10) self.sock.settimeout(None) self.username = username self.account = account or username self.auth_secret = auth_secret self.user_key = user_key self.user_key_enc = user_key_enc nick = nickname or username acc = account or username join_payload = {'type': 'join', 'username': nick, 'nickname': nick, 'account': acc} if auth_secret: join_payload['auth'] = hmac.new(auth_secret.encode('utf-8'), acc.encode('utf-8'), hashlib.sha256).hexdigest() if user_key: join_payload['user_auth'] = hmac.new( user_key.encode('utf-8'), ('JOIN:' + acc).encode('utf-8'), hashlib.sha256).hexdigest() send_msg(self.sock, join_payload) self.running = True self.connected = True threading.Thread(target=self._recv_loop, daemon=True).start() return True except Exception as e: self._emit_status(f"❌ 连接失败: {e}") return False def _sign(self, payload): """为消息附加账户身份 + 每用户 HMAC 签名 (服务器校验, 防冒充)""" payload['account'] = self.account or self.username secret = self.user_key or self.auth_secret if secret: canonical = json.dumps( {k: v for k, v in payload.items() if k not in ('sig', 'from', 'ts')}, sort_keys=True, ensure_ascii=False) payload['sig'] = hmac.new(secret.encode('utf-8'), canonical.encode('utf-8'), hashlib.sha256).hexdigest() return payload def _recv_loop(self): while self.running: try: msg = recv_msg(self.sock) if msg is None: break self._dispatch(msg) except Exception: break self.connected = False if self.running: self._emit_status("🔌 连接已断开") if self.on_system: try: self.on_system("🔌 与服务器的连接已断开") except Exception: pass def _dispatch(self, msg): t = msg.get('type') try: if t == 'chat': if self.on_message: self.on_message(msg.get('from', '?'), msg.get('text', ''), msg.get('mid', '')) elif t == 'system': if self.on_system: self.on_system(msg.get('text', '')) elif t == 'users': if self.on_users: self.on_users(msg.get('users', [])) elif t == 'file': if self.on_file: self.on_file(msg) elif t == 'voice': if self.on_voice: self.on_voice(msg) elif t == 'withdraw': if self.on_withdraw: self.on_withdraw(msg.get('mid', '')) elif t == 'read': if self.on_read: self.on_read(msg.get('mid', '')) elif t == 'announcement': if self.on_announcement: self.on_announcement(msg.get('text', '')) elif t == 'group_name': if self.on_group_name: self.on_group_name(msg.get('name', '')) elif t == 'group_info': if self.on_group_info: self.on_group_info(msg.get('group_name', ''), msg.get('announcement', '')) elif t == 'kick': if self.on_kick: self.on_kick(msg.get('reason', '你已被移出群')) elif t == 'need_approval': if self.on_need_approval: self.on_need_approval(msg.get('reason', '等待群主同意')) elif t == 'approval_granted': if self.on_approval_granted: self.on_approval_granted() elif t == 'approval_denied': if self.on_approval_denied: self.on_approval_denied(msg.get('reason', '群主拒绝你加入')) elif t == 'approval_request': if self.on_approval_request: self.on_approval_request(msg.get('account', ''), msg.get('nickname', '')) elif t == 'user_key_issued': try: if self.user_key_enc: uk = stream_xor(base64.b64decode(msg.get('data', '')), bytes.fromhex(self.user_key_enc)).decode('utf-8') self.user_key = uk if self.on_user_key_issued: self.on_user_key_issued(uk) except Exception: pass except Exception: pass # ---- 发送 ---- def send_chat(self, text, mid=None): if not self.connected: return False try: payload = {'type': 'chat', 'text': text, 'ts': int(time.time())} if mid: payload['mid'] = mid send_msg(self.sock, self._sign(payload)) return True except Exception as e: self._emit_status(f"❌ 发送失败: {e}") return False def send_file(self, filename, data_bytes, mid=None, mime='file'): if not self.connected: return False try: payload = {'type': 'file', 'filename': filename, 'size': len(data_bytes), 'data': base64.b64encode(data_bytes).decode('ascii'), 'mime': mime, 'ts': int(time.time())} if mid: payload['mid'] = mid send_msg(self.sock, self._sign(payload)) return True except Exception as e: self._emit_status(f"❌ 发送文件失败: {e}") return False def send_voice(self, wav_bytes, mid=None): if not self.connected: return False try: payload = {'type': 'voice', 'data': base64.b64encode(wav_bytes).decode('ascii'), 'size': len(wav_bytes), 'ts': int(time.time())} if mid: payload['mid'] = mid send_msg(self.sock, self._sign(payload)) return True except Exception as e: self._emit_status(f"❌ 发送语音失败: {e}") return False def send_withdraw(self, mid): if not self.connected: return False try: send_msg(self.sock, self._sign({'type': 'withdraw', 'mid': mid})) return True except Exception: return False def send_rename(self, new_name): if not self.connected: return False try: send_msg(self.sock, self._sign({'type': 'rename', 'new_name': new_name})) return True except Exception: return False def send_read(self, mid): """发送已读回执""" if not self.connected: return False try: send_msg(self.sock, self._sign({'type': 'read', 'mid': mid})) return True except Exception: return False def send_announcement(self, text): if not self.connected: return False try: send_msg(self.sock, self._sign({'type': 'set_announcement', 'text': text})) return True except Exception: return False def send_set_group_name(self, name): if not self.connected: return False try: send_msg(self.sock, self._sign({'type': 'set_group_name', 'name': name})) return True except Exception: return False def send_kick(self, account): if not self.connected: return False try: send_msg(self.sock, self._sign({'type': 'kick', 'target': account})) return True except Exception: return False def send_approval(self, account, decision='grant'): if not self.connected: return False try: send_msg(self.sock, self._sign({'type': 'approve', 'target': account, 'decision': decision})) return True except Exception: return False def disconnect(self): self.running = False self.connected = False if self.sock: try: self.sock.close() except Exception: pass self.sock = None # ==================== 中央服务器连接 (解决 NAT) ==================== class ServerConnection: """连接到中央服务器: 认证 + 群管理 + 消息中继 (密钥永不上传)""" def __init__(self, on_status=None, on_groups=None, on_message=None, on_members=None, on_system=None, on_group_name=None, on_announcement=None, on_group_info=None): self.sock = None self.running = False self.connected = False self.token = '' self.username = '' self.addr = '' self.on_status = on_status self.on_groups = on_groups self.on_message = on_message self.on_members = on_members self.on_system = on_system self.on_group_name = on_group_name self.on_announcement = on_announcement self.on_group_info = on_group_info def _emit(self, text): if self.on_status: try: self.on_status(text) except Exception: pass def connect(self, host, port): try: self.sock = socket.create_connection((host, port), timeout=10) self.sock.settimeout(None) self.addr = f"{host}:{port}" self.connected = True return True except Exception as e: self._emit(f"❌ 连接中央服务器失败: {e}") return False def register(self, username, password): send_msg(self.sock, {'type': 'register', 'username': username, 'password': password}) return self._wait_login() def login(self, username, password): send_msg(self.sock, {'type': 'login', 'username': username, 'password': password}) return self._wait_login() def _wait_login(self): try: self.sock.settimeout(6) msg = recv_msg(self.sock) except Exception: return False, '服务器无响应' finally: try: self.sock.settimeout(None) except Exception: pass if msg and msg.get('type') == 'login_ok': self.token = msg.get('token', '') self.username = msg.get('username', '') self.running = True threading.Thread(target=self._recv_loop, daemon=True).start() return True, 'ok' if msg and msg.get('type') == 'error': return False, msg.get('msg', '认证失败') return False, '服务器返回异常' def _recv_loop(self): while self.running: try: msg = recv_msg(self.sock) if msg is None: break self._dispatch(msg) except Exception: break self.connected = False if self.running: self._emit("🔌 与中央服务器断开") def _dispatch(self, msg): t = msg.get('type') try: if t == 'groups': if self.on_groups: self.on_groups(msg.get('groups', [])) elif t == 'group_message': if self.on_message: self.on_message(msg.get('group_id', ''), msg.get('from', '?'), msg.get('payload')) elif t == 'members': if self.on_members: self.on_members(msg.get('group_id', ''), msg.get('members', [])) elif t == 'system': if self.on_system: self.on_system(msg.get('group_id', ''), msg.get('text', '')) elif t == 'group_name': if self.on_group_name: self.on_group_name(msg.get('group_id', ''), msg.get('name', '')) elif t == 'announcement': if self.on_announcement: self.on_announcement(msg.get('group_id', ''), msg.get('text', '')) elif t == 'group_info': if self.on_group_info: self.on_group_info(msg.get('group_id', ''), msg.get('name', ''), msg.get('announcement', '')) except Exception: pass def create_group(self, group_id, name): send_msg(self.sock, {'type': 'create_group', 'group_id': group_id, 'name': name}) def list_groups(self): send_msg(self.sock, {'type': 'list_groups'}) def join_group(self, group_id): send_msg(self.sock, {'type': 'join_group', 'group_id': group_id}) def leave_group(self, group_id): send_msg(self.sock, {'type': 'leave_group', 'group_id': group_id}) def set_group_name(self, group_id, name): send_msg(self.sock, {'type': 'set_group_name', 'group_id': group_id, 'name': name}) def set_announcement(self, group_id, text): send_msg(self.sock, {'type': 'set_announcement', 'group_id': group_id, 'text': text}) def group_members(self, group_id): send_msg(self.sock, {'type': 'group_members', 'group_id': group_id}) def send_message(self, group_id, payload): if not self.connected: return False try: send_msg(self.sock, {'type': 'group_message', 'group_id': group_id, 'payload': payload}) return True except Exception: return False def disconnect(self): self.running = False self.connected = False if self.sock: try: self.sock.close() except Exception: pass self.sock = None # ==================== 会话文件 (记住登录) ==================== def save_session(token): try: with open(SESSION_FILE, 'w', encoding='utf-8') as f: f.write(token) except Exception: pass def load_session(): try: with open(SESSION_FILE, 'r', encoding='utf-8') as f: return f.read().strip() except Exception: return '' def clear_session(): try: if os.path.exists(SESSION_FILE): os.remove(SESSION_FILE) except Exception: pass # ==================== 服务器账号记住我 + 群本地缓存 ==================== SERVER_CRED_FILE = os.path.join(DATA_DIR, 'server_cred.dat') SERVER_GROUPS_CACHE = os.path.join(DATA_DIR, 'server_groups_cache.json') def _cred_key(pass_hash): """用本地账号密码哈希派生密钥 (记住服务器账号用)""" return hashlib.sha256(pass_hash.encode('utf-8') + b'|SERVER_CRED').digest() def save_server_cred(pass_hash, addr, username, password): """记住服务器账号: 加密存储 (密钥=本地账号密码哈希派生)""" try: key = _cred_key(pass_hash) nonce = os.urandom(16) k = hashlib.sha256(key + nonce).digest() ct = stream_xor(f"{addr}\n{username}\n{password}".encode('utf-8'), k) h = hmac.new(key, nonce + ct, hashlib.sha256).hexdigest() blob = base64.b64encode(b'CRED' + nonce + h.encode('ascii') + ct).decode('ascii') with open(SERVER_CRED_FILE, 'w', encoding='utf-8') as f: f.write(blob) return True except Exception: return False def load_server_cred(pass_hash): """读取记住的服务器账号; 密钥不符返回 None""" try: with open(SERVER_CRED_FILE, 'r', encoding='utf-8') as f: b64 = f.read().strip() key = _cred_key(pass_hash) blob = base64.b64decode(b64) if not blob.startswith(b'CRED'): return None nonce = blob[4:20] h = blob[20:84] ct = blob[84:] if hmac.new(key, nonce + ct, hashlib.sha256).hexdigest() != h.decode('ascii'): return None k = hashlib.sha256(key + nonce).digest() text = stream_xor(ct, k).decode('utf-8') addr, username, password = text.split('\n', 2) return (addr, username, password) except Exception: return None def save_server_groups_cache(server_addr, groups): """本地缓存服务器群列表 (离线也能显示)""" try: data = {} if os.path.exists(SERVER_GROUPS_CACHE): with open(SERVER_GROUPS_CACHE, 'r', encoding='utf-8') as f: data = json.load(f) data[server_addr] = groups with open(SERVER_GROUPS_CACHE, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False) except Exception: pass def load_server_groups_cache(server_addr): try: if os.path.exists(SERVER_GROUPS_CACHE): with open(SERVER_GROUPS_CACHE, 'r', encoding='utf-8') as f: data = json.load(f) return data.get(server_addr, []) except Exception: pass return [] import warnings warnings.filterwarnings('ignore', message='sipPyTypeDict.*') # PyQt5 from PyQt5 import QtCore, QtGui, QtWidgets # 复用后端 (dvs_enc.py) from dvs_enc import ( AccountDB, HistoryDB, ServerDB, SecureSession, ChatClient, ChatServer, ServerConnection, VoiceManager, Message, gen_mid, save_media, delete_media, clear_media_cache, save_session, load_session, clear_session, MEDIA_DIR ) APP_NAME = "🔐 混沌加密聊天" DOWNLOAD_DIR = os.path.join(os.path.expanduser('~'), 'Downloads', 'ChaosChat') try: os.makedirs(DOWNLOAD_DIR, exist_ok=True) except Exception: pass # ==================== 现代化样式 (QSS) ==================== QSS = """ QMainWindow, QDialog { background: #ffffff; } QLabel { color: #2c3e50; } QListWidget { background: #ffffff; border: 1px solid #e0e4ea; border-radius: 8px; padding: 4px; font-size: 13px; } QListWidget::item { padding: 6px; border-radius: 6px; margin: 2px; } QListWidget::item:selected { background: #dbe7ff; color: #1a73e8; } QListWidget::item:hover { background: #f0f4ff; } QTextEdit, QLineEdit { background: #ffffff; border: 1px solid #e0e4ea; border-radius: 8px; padding: 6px; font-size: 13px; } QPushButton { background: #eef1f6; border: none; border-radius: 8px; padding: 7px 14px; font-size: 13px; color: #2c3e50; } QPushButton:hover { background: #e2e7f0; } QPushButton:disabled { background: #f0f0f0; color: #aaa; } QPushButton.primary { background: #4CAF50; color: white; } QPushButton.primary:hover { background: #43a047; } QPushButton.info { background: #2196F3; color: white; } QPushButton.info:hover { background: #1e88e5; } QPushButton.danger { background: #e74c3c; color: white; } QPushButton.danger:hover { background: #c0392b; } QToolBar { background: #ffffff; border-bottom: 1px solid #e0e4ea; spacing: 4px; padding: 4px; } QMenuBar { background: #ffffff; border-bottom: 1px solid #e0e4ea; } QMenuBar::item { padding: 6px 10px; border-radius: 6px; } QMenuBar::item:selected { background: #dbe7ff; } QMenu { background: #ffffff; border: 1px solid #e0e4ea; border-radius: 8px; padding: 6px; } QMenu::item { padding: 6px 24px; border-radius: 6px; } QMenu::item:selected { background: #dbe7ff; color: #1a73e8; } QScrollBar:vertical { background: transparent; width: 8px; } QScrollBar::handle:vertical { background: #c5cbd6; border-radius: 4px; min-height: 30px; } QScrollBar::add-line, QScrollBar::sub-line { height: 0; } QStatusBar { background: #ffffff; border-top: 1px solid #e0e4ea; } """ # ==================== 消息项组件 ==================== class MessageItem(QtWidgets.QWidget): """单条消息: 我的靠左, 别人靠右; 整条可选中, 右键菜单""" def __init__(self, msg, actions, parent=None): super().__init__(parent) self.msg = msg self.mid = msg.mid self.actions = actions self._selected = False self._notify_select = None self._status_lab = None self.setCursor(QtCore.Qt.PointingHandCursor) # 关键: 自定义 QWidget 必须开 WA_StyledBackground, 否则 QSS 背景不绘制 self.setAttribute(QtCore.Qt.WA_StyledBackground, True) self.setStyleSheet("MessageItem{background:#ffffff; border-radius:12px;}") lay = QtWidgets.QVBoxLayout(self) lay.setContentsMargins(10, 6, 10, 6) lay.setSpacing(3) # 头部: 发送者 + 时间 (我的靠左, 别人靠右; 我的显示已读/未读) head_row = QtWidgets.QHBoxLayout() head = QtWidgets.QLabel(f"👤 {msg.sender} {datetime.fromtimestamp(msg.timestamp).strftime('%H:%M:%S')}") head.setStyleSheet("font-weight:bold; color:%s;" % ('#1a73e8' if msg.is_mine else '#7a5af8')) if msg.is_mine: head_row.addWidget(head) head_row.addStretch() self._status_lab = QtWidgets.QLabel("已读" if msg.read else "未读") self._status_lab.setStyleSheet("color:#999; font-size:11px;") head_row.addWidget(self._status_lab) else: head_row.addStretch() head_row.addWidget(head) lay.addLayout(head_row) # 内容 if msg.withdrawn: lab = QtWidgets.QLabel("🚫 此消息已撤回") lab.setStyleSheet("color:#999; font-style:italic;") self._align(lay, lab) elif msg.type == 'text': lab = QtWidgets.QLabel(msg.content) lab.setWordWrap(True) lab.setMaximumWidth(560) lab.setStyleSheet("background:#f0f2f5; border-radius:14px; padding:10px 14px;") lab.setSizePolicy(QtWidgets.QSizePolicy.Maximum, QtWidgets.QSizePolicy.Preferred) self._align(lay, lab) elif msg.type == 'image': self._add_image(lay) elif msg.type in ('voice', 'audio'): self._add_audio(lay) elif msg.type == 'file': self._add_file(lay) else: lab = QtWidgets.QLabel(str(msg.content)) lab.setStyleSheet("color:#666;") self._align(lay, lab) # 关键: 按钮不装过滤器且不透明(保证可点击); 其他子控件(文字/气泡)装过滤器 → 点击选中 self.installEventFilter(self) for ch in self.findChildren(QtWidgets.QWidget): if not isinstance(ch, QtWidgets.QPushButton): ch.installEventFilter(self) def _align(self, lay, w): """我的消息靠左, 别人的靠右""" if self.msg.is_mine: lay.addWidget(w, 0, QtCore.Qt.AlignLeft) else: lay.addWidget(w, 0, QtCore.Qt.AlignRight) def _align_row(self, row, bubble): if self.msg.is_mine: row.addWidget(bubble) row.addStretch() else: row.addStretch() row.addWidget(bubble) def mousePressEvent(self, event): # 点击消息自身背景 → 选中 if event.button() == QtCore.Qt.LeftButton and self._notify_select: self._notify_select() super().mousePressEvent(event) def eventFilter(self, obj, event): # 点击文字/气泡等子控件(按钮除外) → 选中本消息; 按钮保持自身功能 if event.type() == QtCore.QEvent.MouseButtonPress and event.button() == QtCore.Qt.LeftButton: if not isinstance(obj, QtWidgets.QPushButton) and self._notify_select: self._notify_select() return False def set_selected(self, sel): self._selected = sel if sel: self.setStyleSheet("MessageItem{background:#9cc6ff; border-left:6px solid #1a73e8; border-radius:12px;}") else: self.setStyleSheet("MessageItem{background:#ffffff; border-radius:12px;}") def set_read_status(self, read): if self._status_lab: self._status_lab.setText("已读" if read else "未读") def _act(self, key, *args): fn = self.actions.get(key) if fn: try: fn(*args) except Exception as e: print(f" [ACT] 操作 {key} 失败: {e}") def _add_file(self, lay): c = self.msg.content if isinstance(self.msg.content, dict) else {} fname = c.get('filename', '文件') size = c.get('size', 0) ext = c.get('ext', os.path.splitext(fname)[1]) icon = file_icon(ext) bubble = QtWidgets.QFrame() bubble.setStyleSheet("QFrame{background:#e8f0fe; border-radius:14px;}") blay = QtWidgets.QHBoxLayout(bubble) blay.setContentsMargins(16, 10, 10, 10) blay.setSpacing(10) lab = QtWidgets.QLabel(f"{icon} {fname} ({size // 1024} KB)") lab.setStyleSheet("background:transparent; color:#202124; font-size:13px;") lab.setMaximumWidth(360) blay.addWidget(lab) dl = QtWidgets.QPushButton("💾 下载") dl.setStyleSheet("background:#d3e0fb; color:#1a73e8; border-radius:8px; padding:5px 12px;") dl.clicked.connect(lambda: self._act('on_download', self.mid)) blay.addWidget(dl) row = QtWidgets.QHBoxLayout() self._align_row(row, bubble) lay.addLayout(row) def _add_audio(self, lay): c = self.msg.content if isinstance(self.msg.content, dict) else {} fname = c.get('filename', '') label = "🔊 语音" if self.msg.type == 'voice' else f"🎵 音频 {fname}" size = c.get('size', 0) bubble = QtWidgets.QFrame() bubble.setStyleSheet("QFrame{background:#efe9ff; border-radius:14px;}") blay = QtWidgets.QHBoxLayout(bubble) blay.setContentsMargins(12, 10, 10, 10) blay.setSpacing(10) play = QtWidgets.QPushButton("▶ 播放") play.setStyleSheet("background:#7a5af8; color:#fff; border-radius:10px; padding:7px 16px;") play.clicked.connect(lambda: self._act('on_play', self.mid)) blay.addWidget(play) lab = QtWidgets.QLabel(f"{label} ({size // 1024} KB)") lab.setStyleSheet("background:transparent; color:#5a4a9a; font-size:13px;") blay.addWidget(lab) dl = QtWidgets.QPushButton("💾") dl.setStyleSheet("background:#ddd4f7; color:#5a4a9a; border-radius:8px; padding:5px 10px;") dl.clicked.connect(lambda: self._act('on_download', self.mid)) blay.addWidget(dl) row = QtWidgets.QHBoxLayout() self._align_row(row, bubble) lay.addLayout(row) def _add_image(self, lay): c = self.msg.content if isinstance(self.msg.content, dict) else {} data = self.msg.get_data_bytes() bubble = QtWidgets.QFrame() bubble.setStyleSheet("QFrame{background:#ffffff; border-radius:14px; border:1px solid #e0e4ea;}") blay = QtWidgets.QHBoxLayout(bubble) blay.setContentsMargins(8, 8, 8, 8) blay.setSpacing(8) if data: try: from PIL import Image, ImageTk img = Image.open(io.BytesIO(data)) img.thumbnail((220, 220)) img = img.convert("RGBA") qimg = QtGui.QImage(img.tobytes(), img.width, img.height, QtGui.QImage.Format_RGBA8888) pix = QtGui.QPixmap.fromImage(qimg) lab = QtWidgets.QLabel() lab.setPixmap(pix) lab.setCursor(QtCore.Qt.PointingHandCursor) lab.mousePressEvent = lambda e: self._act('on_preview', self.mid) blay.addWidget(lab) except Exception: lab = QtWidgets.QLabel("🖼️ 图片(无法预览)") blay.addWidget(lab) else: lab = QtWidgets.QLabel("🖼️ 图片(文件已清理)") blay.addWidget(lab) dl = QtWidgets.QPushButton("💾 下载") dl.setStyleSheet("background:#eef1f6; color:#2c3e50; border-radius:8px; padding:5px 12px;") dl.clicked.connect(lambda: self._act('on_download', self.mid)) blay.addWidget(dl) row = QtWidgets.QHBoxLayout() self._align_row(row, bubble) lay.addLayout(row) class MessageList(QtWidgets.QScrollArea): """自适应消息列表 (QScrollArea + VBox): 每条消息按内容自动调高""" def __init__(self, on_menu=None, on_selected=None, parent=None): super().__init__(parent) self.setWidgetResizable(True) self.setFrameShape(QtWidgets.QFrame.NoFrame) self.setHorizontalScrollBarPolicy(QtCore.Qt.ScrollBarAlwaysOff) self._container = QtWidgets.QWidget() self._container.setStyleSheet("background:#ffffff;") self._layout = QtWidgets.QVBoxLayout(self._container) self._layout.setContentsMargins(6, 6, 6, 6) self._layout.setSpacing(6) self._layout.addStretch(1) # 消息从顶部排列, 底部弹性 self.setWidget(self._container) self._items = {} # mid -> MessageItem self.selected_mid = None self._on_menu = on_menu self._on_selected = on_selected self.setContextMenuPolicy(QtCore.Qt.CustomContextMenu) self.customContextMenuRequested.connect(self._on_right_click) # 点击空白处 → 取消选中 self._container.installEventFilter(self) self.viewport().installEventFilter(self) self.installEventFilter(self) def eventFilter(self, obj, event): """点击消息区空白处 → 自动取消选中""" if event.type() == QtCore.QEvent.MouseButtonPress and event.button() == QtCore.Qt.LeftButton: if obj in (self._container, self.viewport(), self): self._deselect() return False def _deselect(self): self.selected_mid = None for w in self._items.values(): w.set_selected(False) if self._on_selected: self._on_selected(None) def _select(self, mid): self.selected_mid = mid for m, w in self._items.items(): w.set_selected(m == mid) if self._on_selected: self._on_selected(mid) def _on_right_click(self, pos): item = self._message_at(pos) if not item: self._deselect() return self._select(item.mid) if self._on_menu: self._on_menu(item.mid, self.mapToGlobal(pos)) def _message_at(self, pos): global_pos = self.mapToGlobal(pos) w = self._container.childAt(self._container.mapFromGlobal(global_pos)) while w is not None: if isinstance(w, MessageItem): return w w = w.parentWidget() return None def add_message(self, msg, actions): wid = MessageItem(msg, actions) wid._notify_select = lambda m=msg.mid: self._select(m) wid.setSizePolicy(QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Preferred) self._layout.insertWidget(self._layout.count() - 1, wid) # 插到 stretch 之前 self._items[msg.mid] = wid # 等宽度确定后再滚到底部 QtCore.QTimer.singleShot(0, self._scroll_bottom) return wid def update_message(self, mid, msg): """就地更新某条消息 (已读/未读状态)""" wid = self._items.get(mid) if wid: wid.set_read_status(msg.read) def _scroll_bottom(self): bar = self.verticalScrollBar() bar.setValue(bar.maximum()) def remove_message(self, mid): wid = self._items.pop(mid, None) if wid: self._layout.removeWidget(wid) wid.deleteLater() if self.selected_mid == mid: self.selected_mid = None def clear_all(self): for w in self._items.values(): self._layout.removeWidget(w) w.deleteLater() self._items.clear() self.selected_mid = None # ==================== 登录对话框 ==================== class LoginDialog(QtWidgets.QDialog): """本地账号 注册/登录""" def __init__(self, db, parent=None): super().__init__(parent) self.db = db self.result = None self.mode = 'login' self.setWindowTitle(f"{APP_NAME} - 登录") self.setFixedSize(380, 420) self._build() def _build(self): lay = QtWidgets.QVBoxLayout(self) lay.setSpacing(10) title = QtWidgets.QLabel("🔐 混沌加密聊天") title.setAlignment(QtCore.Qt.AlignCenter) title.setStyleSheet("font-size:20px; font-weight:bold; color:#1a73e8;") lay.addWidget(title) self.mode_lab = QtWidgets.QLabel("本地账号登录") self.mode_lab.setAlignment(QtCore.Qt.AlignCenter) self.mode_lab.setStyleSheet("font-size:13px; color:#555;") lay.addWidget(self.mode_lab) self.u = QtWidgets.QLineEdit() self.u.setPlaceholderText("账户名") lay.addWidget(self.u) self.p = QtWidgets.QLineEdit() self.p.setPlaceholderText("密码") self.p.setEchoMode(QtWidgets.QLineEdit.Password) lay.addWidget(self.p) self.p2 = QtWidgets.QLineEdit() self.p2.setPlaceholderText("确认密码") self.p2.setEchoMode(QtWidgets.QLineEdit.Password) self.p2.setVisible(False) lay.addWidget(self.p2) self.msg = QtWidgets.QLabel("") self.msg.setStyleSheet("color:#e74c3c;") lay.addWidget(self.msg) row = QtWidgets.QHBoxLayout() self.btn_main = QtWidgets.QPushButton("登 录") self.btn_main.setProperty("class", "primary") self.btn_main.clicked.connect(self._submit) row.addWidget(self.btn_main) self.btn_switch = QtWidgets.QPushButton("没有账号？注册") self.btn_switch.clicked.connect(self._toggle) row.addWidget(self.btn_switch) lay.addLayout(row) lay.addStretch() def _toggle(self): if self.mode == 'login': self.mode = 'register' self.mode_lab.setText("注册本地账号") self.btn_main.setText("注 册") self.btn_switch.setText("已有账号？登录") self.p2.setVisible(True) else: self.mode = 'login' self.mode_lab.setText("本地账号登录") self.btn_main.setText("登 录") self.btn_switch.setText("没有账号？注册") self.p2.setVisible(False) self.msg.setText("") def _submit(self): user = self.u.text().strip() pwd = self.p.text() if not user or not pwd: self.msg.setText("请输入账户名和密码") return if self.mode == 'login': row, err = self.db.login(user, pwd) if row is None: self.msg.setText(err) return self.result = row self.accept() else: if len(pwd) < 6: self.msg.setText("密码至少 6 位") return if pwd != self.p2.text(): self.msg.setText("两次密码不一致") return ok, err = self.db.register(user, pwd) if not ok: self.msg.setText(err) return row, _ = self.db.login(user, pwd) self.result = row self.accept() # ==================== 主窗口 ==================== class MainWindow(QtWidgets.QMainWindow): def __init__(self, account): super().__init__() self.account = dict(account) if account else {'id': 1, 'username': 'user', 'nickname': '用户'} self.account_db = AccountDB() self.history = HistoryDB() self.server_db = ServerDB() self.voice = VoiceManager(on_status=self._on_status) self.groups = {} # gid -> dict self.order = [] self.active_gid = None self.selected_mid = None self._gid_counter = 0 self.server_conn = None self.server_addr = '' self.server_username = '' self._ui_queue = [] self._relogin = False self.setWindowTitle(f"{APP_NAME} - {self._my_nick()}") self.resize(1120, 720) self._build_ui() # UI 轮询 self._timer = QtCore.QTimer(self) self._timer.timeout.connect(self._poll_ui) self._timer.start(60) # 启动自动加入 (本地P2P群) QtCore.QTimer.singleShot(500, self._start_auto_join) # 服务器: 先显示本地缓存, 再记住我自动登录实时同步 QtCore.QTimer.singleShot(800, self._load_cached_server_groups) QtCore.QTimer.singleShot(1500, self._auto_connect_server) # ---------- 线程安全 ---------- def _post(self, fn): self._ui_queue.append(fn) def _poll_ui(self): if self._ui_queue: tasks = self._ui_queue[:] self._ui_queue = [] for fn in tasks: try: fn() except Exception: pass def _my_nick(self): return self.account.get('nickname') or self.account.get('username', '用户') def _on_status(self, text): self._post(lambda: self.statusBar().showMessage(text, 5000)) # ---------- 界面 ---------- def _build_ui(self): # 工具栏 tb = QtWidgets.QToolBar("主工具栏") tb.setMovable(False) self.addToolBar(tb) def btn(text, fn, cls=''): b = QtWidgets.QPushButton(text) if cls: b.setProperty("class", cls) b.clicked.connect(fn) tb.addWidget(b) return b btn("📡 创建群", self._create_group, 'primary') btn("📥 加入群", self._join_group, 'info') tb.addSeparator() btn("☁️ 服务器", self._connect_server) btn("☁️ 建群", self._create_server_group) btn("☁️ 加群", self._join_server_group) tb.addSeparator() btn("🔑 密钥", self._copy_active_key) btn("✏️ 改名", self._rename) btn("📢 公告", self._set_announcement) btn("🏷️ 群名", self._set_group_name) btn("🚫 撤回", self._withdraw_selected) btn("📤 分享", self._share_selected) # 菜单 mb = self.menuBar() gm = mb.addMenu("群") gm.addAction("📡 创建群", self._create_group) gm.addAction("📥 加入群", self._join_group) gm.addAction("❌ 离开该群", self._leave_active_group) sm = mb.addMenu("服务器") sm.addAction("☁️ 连接服务器", self._connect_server) sm.addAction("☁️ 服务器建群", self._create_server_group) sm.addAction("☁️ 服务器加群", self._join_server_group) mm = mb.addMenu("消息") mm.addAction("🚫 撤回所选", self._withdraw_selected) mm.addAction("📤 分享所选", self._share_selected) mm.addAction("💾 下载所选", self._download_selected) mm.addAction("📋 复制所选", self._copy_selected) um = mb.addMenu("工具") um.addAction("🗑️ 清理媒体缓存", self._clear_media_cache) um.addAction("📂 打开下载目录", self._open_download_dir) um.addAction("🔓 退出登录", self._logout) um.addAction("❌ 退出程序", self._quit) # 中央区域 central = QtWidgets.QWidget() self.setCentralWidget(central) root = QtWidgets.QHBoxLayout(central) root.setContentsMargins(8, 8, 8, 8) root.setSpacing(8) # 左: 群列表 left = QtWidgets.QVBoxLayout() left.addWidget(QtWidgets.QLabel("👥 群列表")) self.group_list = QtWidgets.QListWidget() self.group_list.setFixedWidth(240) self.group_list.itemClicked.connect(self._on_group_click) self.group_list.setContextMenuPolicy(QtCore.Qt.CustomContextMenu) self.group_list.customContextMenuRequested.connect(self._on_group_menu) left.addWidget(self.group_list) left_w = QtWidgets.QWidget() left_w.setLayout(left) root.addWidget(left_w) # 中: 公告横幅 + 消息 + 输入 center = QtWidgets.QVBoxLayout() self.banner = QtWidgets.QFrame() self.banner.setStyleSheet("background:#fff3cd; border-radius:8px;") b_lay = QtWidgets.QHBoxLayout(self.banner) self.banner_label = QtWidgets.QLabel("") self.banner_label.setStyleSheet("color:#8a6d3b;") b_lay.addWidget(self.banner_label, 1) b_close = QtWidgets.QPushButton("✕ 关闭") b_close.setFixedWidth(60) b_close.clicked.connect(self._hide_banner) b_lay.addWidget(b_close) self.banner.setVisible(False) center.addWidget(self.banner) self.message_list = MessageList(on_menu=self._show_context_menu, on_selected=lambda mid: self._on_msg_selected(mid)) center.addWidget(self.message_list, 1) # 输入区: 多行输入 + 发送 + 图片/文件/语音按钮 in_area = QtWidgets.QVBoxLayout() in_row = QtWidgets.QHBoxLayout() self.input = QtWidgets.QTextEdit() self.input.setFixedHeight(56) self.input.setPlaceholderText("输入消息... (Ctrl+Enter 发送)") QtWidgets.QShortcut(QtGui.QKeySequence("Ctrl+Return"), self.input, activated=self._send_text) send = QtWidgets.QPushButton("发 送") send.setProperty("class", "primary") send.setFixedHeight(44) send.clicked.connect(self._send_text) in_row.addWidget(self.input, 1) in_row.addWidget(send) in_area.addLayout(in_row) btn_row = QtWidgets.QHBoxLayout() self.img_btn = QtWidgets.QPushButton("📷 图片") self.img_btn.clicked.connect(self._send_image) self.file_btn = QtWidgets.QPushButton("📁 文件") self.file_btn.clicked.connect(self._send_file) self.record_btn = QtWidgets.QPushButton("🎤 录音") self.record_btn.clicked.connect(self._toggle_record) self.play_btn = QtWidgets.QPushButton("▶ 播放") self.play_btn.clicked.connect(self._play_voice) for b in (self.img_btn, self.file_btn, self.record_btn, self.play_btn): b.setFixedHeight(28) btn_row.addWidget(b) btn_row.addStretch() in_area.addLayout(btn_row) center.addLayout(in_area) center_w = QtWidgets.QWidget() center_w.setLayout(center) root.addWidget(center_w, 1) # 右: 用户列表 right = QtWidgets.QVBoxLayout() right.addWidget(QtWidgets.QLabel("🟢 在线用户")) self.users_list = QtWidgets.QListWidget() self.users_list.setFixedWidth(180) self.users_list.setContextMenuPolicy(QtCore.Qt.CustomContextMenu) self.users_list.customContextMenuRequested.connect(self._on_user_menu) right.addWidget(self.users_list) right_w = QtWidgets.QWidget() right_w.setLayout(right) root.addWidget(right_w) self.statusBar().showMessage("就绪") self._add_system(None, "📌 点击工具栏或菜单开始；消息可点击选中，右键有更多操作") # ---------- 公告横幅 ---------- def _show_banner(self, text): if not text: return self.banner_label.setText(f"📢 群公告: {text}") self.banner.setVisible(True) def _hide_banner(self): self.banner.setVisible(False) # ---------- 群管理 ---------- def _new_gid(self): self._gid_counter += 1 return f"g{self._gid_counter}" def _group_connected(self, group): if group.get('mode') == 'server': return bool(group.get('server_conn') and group['server_conn'].connected) return bool(group.get('client') and group['client'].connected) def _build_group(self, gid, host, port, nick, is_host, server, client, session): return { 'gid': gid, 'host': host, 'port': port, 'group_key': f"{host}:{port}", 'name': nick, 'my_username': nick, 'my_account': self.account.get('username', ''), 'is_host': is_host, 'server': server, 'client': client, 'session': session, 'users': [], 'messages': {}, 'order': [], 'announcement': '', 'group_name': '', 'mode': 'p2p', 'user_key': None } def _register_group(self, gid, group): if gid in self.groups: return for g in list(self.groups.values()): if g['group_key'] == group['group_key']: if group.get('client'): try: group['client'].disconnect() except Exception: pass if group.get('server'): try: group['server'].stop() except Exception: pass return self.groups[gid] = group self.order.append(gid) self._load_history(gid) self._refresh_groups() self._switch_group(gid) def _refresh_groups(self): self.group_list.clear() for i, gid in enumerate(self.order): g = self.groups[gid] tag = '[群主]' if g['is_host'] else '' prefix = '☁️' if g.get('mode') == 'server' else '🔗' status = '🟢' if self._group_connected(g) else '🔴' addr = f"{g['host']}:{g['port']}" u = g.get('unread', 0) badge = f" 🔴{u}" if u else "" item = QtWidgets.QListWidgetItem(f"{status}{prefix}{tag} {g['name']} {addr}{badge}") item.setData(QtCore.Qt.UserRole, gid) self.group_list.addItem(item) def _on_group_click(self, item): gid = item.data(QtCore.Qt.UserRole) if gid and gid != self.active_gid: self._switch_group(gid) def _switch_group(self, gid): self.active_gid = gid self.selected_mid = None group = self.groups.get(gid) if group: group['unread'] = 0 for mid, msg in list(group['messages'].items()): if not msg.is_mine and not msg.read: msg.read = True self._send_read(gid, mid) self._render_group(gid) self._refresh_groups() def _render_group(self, gid): self.message_list.clear_all() group = self.groups.get(gid) self._render_users(gid) if not group: return for mid in group['order']: msg = group['messages'][mid] self.message_list.add_message(msg, self._panel_actions()) if group.get('announcement'): self._show_banner(group['announcement']) else: self._hide_banner() def _render_users(self, gid): self.users_list.clear() group = self.groups.get(gid) if not group: return for u in group['users']: name = u.get('name', '?') mark = ' (我)' if name == group['my_username'] else '' self.users_list.addItem(f"👤 {name}{mark}") def _panel_actions(self): return { 'on_download': self._on_download, 'on_play': self._on_play, 'on_preview': self._on_preview, } def _add_message(self, gid, mid, sender, mtype, content, ts, is_mine=False, persist=True): group = self.groups.get(gid) if not group: return if mid and mid in group['messages']: return msg = Message(mid, gid, sender, mtype, content, ts, is_mine) group['messages'][mid] = msg group['order'].append(mid) # 已读/未读: 他人消息, 当前群立即已读+回执; 非当前群计未读 if not is_mine and mtype != 'system': if gid == self.active_gid: msg.read = True self._send_read(gid, mid) else: group['unread'] = group.get('unread', 0) + 1 self._refresh_groups() if persist: try: self.history.save(self.account['id'], group['group_key'], msg) except Exception: pass if gid == self.active_gid: self.message_list.add_message(msg, self._panel_actions()) def _add_local_message(self, gid, mid, mtype, content): group = self.groups.get(gid) if group: self._add_message(gid, mid, group['my_username'], mtype, content, time.time(), is_mine=True) def _add_system(self, gid, text): if gid is None: if self.active_gid: gid = self.active_gid else: return self._add_message(gid, gen_mid(), '系统', 'system', text, time.time(), persist=False) def _delete_message(self, gid, mid): group = self.groups.get(gid) if not group: return False msg = group['messages'].pop(mid, None) if not msg: return False if isinstance(msg.content, dict): delete_media(msg.content.get('path')) try: self.history.delete_by_mid(self.account['id'], group['group_key'], mid) except Exception: pass if mid in group['order']: group['order'].remove(mid) if gid == self.active_gid: self.message_list.remove_message(mid) return True def _load_history(self, gid): group = self.groups.get(gid) if not group: return try: rows = self.history.load(self.account['id'], group['group_key']) except Exception: rows = [] for row in rows: mid = row['mid'] if not mid or mid in group['messages']: continue content = HistoryDB.deserialize_content(row['mtype'], row['content']) msg = Message(mid, gid, row['sender'], row['mtype'], content, row['ts'], bool(row['is_mine'])) msg.withdrawn = bool(row['withdrawn']) group['messages'][mid] = msg group['order'].append(mid) # ---------- 客户端工厂 ---------- def _make_client(self, gid): return ChatClient( on_message=lambda s, t, m, g=gid: self._post(lambda: self._on_chat(g, s, t, m)), on_status=lambda t, g=gid: self._post(lambda: self._on_status(t)), on_users=lambda u, g=gid: self._post(lambda: self._on_users(g, u)), on_file=lambda m, g=gid: self._on_file(g, m), on_voice=lambda m, g=gid: self._on_voice(g, m), on_system=lambda t, g=gid: self._post(lambda: self._add_system(g, t)), on_withdraw=lambda m, g=gid: self._post(lambda: self._on_withdraw(g, m)), on_announcement=lambda t, g=gid: self._post(lambda: self._on_announcement(g, t)), on_group_name=lambda n, g=gid: self._post(lambda: self._on_group_name(g, n)), on_kick=lambda r, g=gid: self._post(lambda: self._on_kick(g, r)), on_group_info=lambda gn, an, g=gid: self._post(lambda: self._on_group_info(g, gn, an)), on_need_approval=lambda r, g=gid: self._post(lambda: self._add_system(g, f"⏳ {r}")), on_approval_granted=lambda g=gid: self._post(lambda: self._add_system(g, "✅ 群主已同意，你已加入")), on_approval_denied=lambda r, g=gid: self._post(lambda: self._on_approval_denied(g, r)), on_approval_request=lambda a, n, g=gid: self._post(lambda: self._on_approval_request(g, a, n)), on_user_key_issued=lambda uk, g=gid: self._post(lambda: self._on_user_key_issued(g, uk)), on_read=lambda m, g=gid: self._post( lambda: self._on_read(g, m)), ) # ---------- P2P 建群/加群 ---------- def _create_group(self): from PyQt5.QtWidgets import QInputDialog port, ok = QInputDialog.getInt(self, "端口", "请输入端口号:", value=8888) if not ok: return gid = self._new_gid() session = SecureSession(); session.generate() group_key = f"127.0.0.1:{port}" server = ChatServer(on_status=lambda t, g=gid: self._post(lambda: self._on_status(t)), server_key=group_key, server_db=self.server_db, server_secret=session.session_key) if not server.start(port): self._on_status("❌ 服务器启动失败(端口被占用?)") return client = self._make_client(gid) saved_uk = self._get_saved_user_key(group_key) if not client.connect('127.0.0.1', port, self._my_nick(), nickname=self._my_nick(), account=self.account.get('username', ''), auth_secret=session.auth_secret(), user_key=saved_uk, user_key_enc=session.user_key_enc()): server.stop() return group = self._build_group(gid, '127.0.0.1', port, self._my_nick(), True, server, client, session) group['user_key'] = saved_uk self._register_group(gid, group) self._save_group(gid) self._add_system(gid, "✅ 群已创建(端到端)，你是群主") self._show_key_dialog(gid) def _join_group(self): from PyQt5.QtWidgets import QInputDialog host, ok1 = QInputDialog.getText(self, "服务器地址", "服务器IP:", text="127.0.0.1") if not ok1: return port, ok2 = QInputDialog.getInt(self, "端口", "端口号:", value=8888) if not ok2: return key, ok3 = QInputDialog.getText(self, "密钥", "会话密钥(创建者分享):") if not ok3 or not key: return session = SecureSession() if not session.import_key(key): self._on_status("❌ 密钥无效") return gid = self._new_gid() client = self._make_client(gid) saved_uk = self._get_saved_user_key(f"{host}:{port}") if not client.connect(host, port, self._my_nick(), nickname=self._my_nick(), account=self.account.get('username', ''), auth_secret=session.auth_secret(), user_key=saved_uk, user_key_enc=session.user_key_enc()): return group = self._build_group(gid, host, port, self._my_nick(), False, None, client, session) group['user_key'] = saved_uk self._register_group(gid, group) self._save_group(gid) def _save_group(self, gid): group = self.groups.get(gid) if not group: return try: self.account_db.save_group(self.account['id'], group['group_key'], group['host'], group['port'], group['session'].export_key(), self._my_nick(), group['is_host'], group.get('user_key', '')) except Exception: pass def _get_saved_user_key(self, group_key): try: for r in self.account_db.load_groups(self.account['id']): if r['group_key'] == group_key: return r['user_key'] or None except Exception: pass return None # ---------- 发送辅助 ---------- def _send_chat_to(self, group, enc, mid): if group.get('mode') == 'server': return group['server_conn'].send_message(group['group_id'], {'kind': 'chat', 'text': enc, 'mid': mid, 'ts': int(time.time())}) return group['client'].send_chat(enc, mid) def _send_file_to(self, group, fname, enc_bytes, mid, mime): if group.get('mode') == 'server': return group['server_conn'].send_message(group['group_id'], { 'kind': 'file', 'filename': fname, 'size': len(enc_bytes), 'data': base64.b64encode(enc_bytes).decode('ascii'), 'mime': mime, 'mid': mid, 'ts': int(time.time())}) return group['client'].send_file(fname, enc_bytes, mid, mime=mime) def _send_voice_to(self, group, enc_bytes, mid): if group.get('mode') == 'server': return group['server_conn'].send_message(group['group_id'], { 'kind': 'voice', 'data': base64.b64encode(enc_bytes).decode('ascii'), 'size': len(enc_bytes), 'mid': mid, 'ts': int(time.time())}) return group['client'].send_voice(enc_bytes, mid) def _send_withdraw_to(self, group, mid): if group.get('mode') == 'server': return group['server_conn'].send_message(group['group_id'], {'kind': 'withdraw', 'mid': mid}) return group['client'].send_withdraw(mid) # ---------- 发送 ---------- def _send_text(self): gid, group = self._current_group() if group is None: self._on_status("请先创建或加入一个群") return if not self._group_connected(group): self._on_status("该群未连接") return if not group['session'].has_key(): self._on_status("该群没有会话密钥") return text = self.input.toPlainText().strip() if not text: return try: mid = gen_mid() enc = group['session'].encrypt(text) if self._send_chat_to(group, enc, mid): self._add_local_message(gid, mid, 'text', text) self.input.clear() except Exception as e: self._add_system(gid, f"❌ 加密失败: {e}") def _send_image(self): gid, group = self._current_group() if group is None or not self._group_connected(group): return path, _ = QtWidgets.QFileDialog.getOpenFileName(self, "选择图片", "", "图片 (*.png *.jpg *.jpeg *.gif *.bmp *.webp)") if not path: return try: with open(path, 'rb') as f: data = f.read() fname = os.path.basename(path) ext = os.path.splitext(fname)[1].lower() mid = gen_mid() enc = group['session'].encrypt_bytes(data) media_path = save_media(data, ext) content = {'filename': fname, 'size': len(data), 'path': media_path, 'ext': ext} if self._send_file_to(group, fname, enc.encode('utf-8'), mid, 'image'): self._add_local_message(gid, mid, 'image', content) except Exception as e: self._add_system(gid, f"❌ 图片读取失败: {e}") def _send_file(self): gid, group = self._current_group() if group is None or not self._group_connected(group): return path, _ = QtWidgets.QFileDialog.getOpenFileName(self, "选择文件") if not path: return try: with open(path, 'rb') as f: data = f.read() fname = os.path.basename(path) ext = os.path.splitext(fname)[1].lower() mid = gen_mid() enc = group['session'].encrypt_bytes(data) media_path = save_media(data, ext) content = {'filename': fname, 'size': len(data), 'path': media_path, 'ext': ext} if self._send_file_to(group, fname, enc.encode('utf-8'), mid, 'file'): self._add_local_message(gid, mid, 'file', content) except Exception as e: self._add_system(gid, f"❌ 文件读取失败: {e}\n (可能是文件被占用/权限/或 .lnk 快捷方式问题)") # ---------- 语音 ---------- def _toggle_record(self): """录音: 点一次开始, 再点停止并发送""" gid, group = self._current_group() if self.voice.recording: wav = self.voice.stop_recording() self.record_btn.setText("🎤 录音") if wav: if group and self._group_connected(group): mid = gen_mid() enc = group['session'].encrypt_bytes(wav) media_path = save_media(wav, '.wav') content = {'filename': 'voice.wav', 'size': len(wav), 'path': media_path, 'ext': '.wav'} if self._send_voice_to(group, enc.encode('utf-8'), mid): self._add_local_message(gid, mid, 'voice', content) else: self._add_system(gid, "未连接，语音未发送") else: if group is None or not self._group_connected(group): self._on_status("请先加入一个已连接的群") return if self.voice.start_recording(): self.record_btn.setText("⏹ 停止录音") def _play_voice(self): if self.voice.last_voice: self.voice.play(self.voice.last_voice, self.voice.last_voice_from) else: self._on_status("没有可播放的语音") # ---------- 接收 ---------- def _on_chat(self, gid, sender, text, mid): group = self.groups.get(gid) if not group: return if not group['session'].has_key(): self._add_system(gid, f"⚠️ 无密钥无法解密 {sender} 的消息") return plain = group['session'].decrypt(text) if plain: self._add_message(gid, mid or gen_mid(), sender, 'text', plain, int(time.time())) else: self._add_system(gid, f"⚠️ {sender} 的消息解密失败") def _on_users(self, gid, users): group = self.groups.get(gid) if group: group['users'] = list(users) if gid == self.active_gid: self._render_users(gid) def _on_file(self, gid, msg): group = self.groups.get(gid) if not group: return fname = os.path.basename(str(msg.get('filename', '文件'))) ext = os.path.splitext(fname)[1].lower() mime = msg.get('mime', 'file') raw = b'' try: eb = base64.b64decode(msg.get('data', '')) if eb: raw = group['session'].decrypt_bytes(eb.decode('utf-8')) or b'' except Exception: raw = b'' path = save_media(raw, ext) if raw else '' content = {'filename': fname, 'size': len(raw), 'path': path, 'ext': ext} if mime == 'image' or ext in ('.png', '.jpg', '.jpeg', '.gif', '.bmp', '.webp'): mtype = 'image' elif mime in ('voice', 'audio') or ext in ('.wav', '.mp3', '.ogg', '.flac'): mtype = 'audio' else: mtype = 'file' mid = msg.get('mid', gen_mid()) from_user = msg.get('from', '?') ts = int(msg.get('ts', time.time())) self._post(lambda: self._add_message(gid, mid, from_user, mtype, content, ts)) def _on_voice(self, gid, msg): group = self.groups.get(gid) if not group: return raw = b'' try: eb = base64.b64decode(msg.get('data', '')) if eb: raw = group['session'].decrypt_bytes(eb.decode('utf-8')) or b'' except Exception: raw = b'' path = save_media(raw, '.wav') if raw else '' content = {'filename': 'voice.wav', 'size': len(raw), 'path': path, 'ext': '.wav'} mid = msg.get('mid', gen_mid()) from_user = msg.get('from', '?') ts = int(msg.get('ts', time.time())) self._post(lambda: self._add_message(gid, mid, from_user, 'voice', content, ts)) if raw: self.voice.play(raw, from_user) def _on_withdraw(self, gid, mid): if self._delete_message(gid, mid): self._add_system(gid, "🚫 有一条消息已被撤回") def _on_announcement(self, gid, text): group = self.groups.get(gid) if group: group['announcement'] = text if gid == self.active_gid: self._show_banner(text) self._add_system(gid, f"📢 群公告更新: {text}") def _on_group_name(self, gid, name): group = self.groups.get(gid) if group: group['group_name'] = name or '' group['name'] = f"{name or group['my_username']}" self._refresh_groups() def _on_group_info(self, gid, gn, an): if gn: self._on_group_name(gid, gn) if an: self._on_announcement(gid, an) def _on_kick(self, gid, reason): group = self.groups.get(gid) if group: try: group['client'].disconnect() except Exception: pass self._refresh_groups() self._add_system(gid, f"🚫 {reason}") def _on_approval_denied(self, gid, reason): group = self.groups.get(gid) if group: try: group['client'].disconnect() except Exception: pass self._refresh_groups() self._add_system(gid, f"❌ {reason}") def _on_approval_request(self, gid, account, nickname): ret = QtWidgets.QMessageBox.question(self, "入群申请", f"{nickname} ({account}) 曾被移出本群，申请重新加入。是否同意？", QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No) group = self.groups.get(gid) if group: group['client'].send_approval(account, 'grant' if ret == QtWidgets.QMessageBox.Yes else 'deny') def _on_user_key_issued(self, gid, user_key): group = self.groups.get(gid) if group: group['user_key'] = user_key self._save_group(gid) def _send_read(self, gid, mid): """发送已读回执 (P2P 或 服务器群); 未连接时跳过""" group = self.groups.get(gid) if not group: return if group.get('mode') == 'server': sc = group.get('server_conn') if sc and sc.connected: sc.send_message(group['group_id'], {'kind': 'read', 'mid': mid}) elif group.get('client'): group['client'].send_read(mid) def _on_read(self, gid, mid): """收到已读回执: 自己的消息标记为已读""" group = self.groups.get(gid) if not group: return msg = group['messages'].get(mid) if msg and msg.is_mine: msg.read = True if gid == self.active_gid: self.message_list.update_message(mid, msg) def _current_group(self): gid = self.active_gid if gid is None or gid not in self.groups: return None, None return gid, self.groups[gid] # ---------- 中央服务器 (解决 NAT) ---------- def _connect_server(self): from PyQt5.QtWidgets import QInputDialog addr, ok = QInputDialog.getText(self, "中央服务器", "服务器地址 (如 服务器IP:8666):", text="127.0.0.1:8666") if not ok or not addr: return try: host, port = addr.rsplit(':', 1) port = int(port) except Exception: self._on_status("地址格式应为 主机:端口") return # 后台线程连接, 避免卡死 GUI self._on_status(f"☁️ 正在连接 {host}:{port} ...") threading.Thread(target=self._do_manual_connect, args=(host, port), daemon=True).start() def _do_manual_connect(self, host, port): """后台连接服务器, 成功后在主线程弹登录框; 已连接则复用""" if self.server_conn and self.server_conn.connected: self._post(lambda: self._server_auth_dialog(self.server_conn, host, port)) return conn = ServerConnection( on_status=self._on_status, on_groups=lambda gl: self._post(lambda: self._on_server_groups(gl)), on_message=lambda gid, frm, pl: self._post(lambda: self._on_server_group_message(gid, frm, pl)), on_members=lambda gid, mems: self._post(lambda: self._on_server_members(gid, mems)), on_system=lambda gid, t: self._post(lambda: self._on_server_system(gid, t)), on_group_name=lambda gid, n: self._post(lambda: self._on_server_group_name(gid, n)), on_announcement=lambda gid, t: self._post(lambda: self._on_server_announcement(gid, t)), on_group_info=lambda gid, n, a: self._post(lambda: self._on_server_group_info(gid, n, a))) if not conn.connect(host, port): self._post(lambda: self._on_status(f"❌ 连接 {host}:{port} 失败")) return self._post(lambda: self._server_auth_dialog(conn, host, port)) def _server_auth_dialog(self, conn, host, port): dlg = QtWidgets.QDialog(self) dlg.setWindowTitle(f"☁️ 服务器账号 - {host}:{port}") dlg.setFixedSize(380, 330) lay = QtWidgets.QVBoxLayout(dlg) lay.addWidget(QtWidgets.QLabel(f"连接到 {host}:{port}")) u = QtWidgets.QLineEdit(); u.setPlaceholderText("服务器用户名") p = QtWidgets.QLineEdit(); p.setPlaceholderText("密码"); p.setEchoMode(QtWidgets.QLineEdit.Password) lay.addWidget(u); lay.addWidget(p) lab = QtWidgets.QLabel(""); lab.setStyleSheet("color:#e74c3c;") lay.addWidget(lab) row = QtWidgets.QHBoxLayout() b1 = QtWidgets.QPushButton("登录"); b1.setProperty("class", "info") b2 = QtWidgets.QPushButton("注册(服务器建号)"); b2.setProperty("class", "primary") b3 = QtWidgets.QPushButton("取消") row.addWidget(b1); row.addWidget(b2); row.addWidget(b3) lay.addLayout(row) lay.addWidget(QtWidgets.QLabel("🔐 登录成功后会自动记住，下次启动自动登录服务器")) lay.addWidget(QtWidgets.QLabel("如需清除，删除 data\\server_cred.dat")) def finish(ok, err, uname, upass): if not ok: lab.setText(err) return dlg.accept() self.server_conn = conn self.server_addr = f"{host}:{port}" self.server_username = conn.username self._bind_server_conn() self._update_title() # 登录即自动记住服务器账号 save_server_cred(self._get_pass_hash(), f"{host}:{port}", uname, upass) self._on_status("🔐 已记住服务器账号，下次启动自动登录") self._on_status(f"☁️ 已连接中央服务器 {host}:{port} (用户: {conn.username})") self._add_system(None, f"☁️ 已连接中央服务器 {host}:{port}，正在加载你的群...") conn.list_groups() def do_auth(action): uname = u.text().strip() upass = p.text() if not uname or not upass: lab.setText("请输入用户名和密码") return b1.setEnabled(False); b2.setEnabled(False) lab.setText("连接中... 请稍候") def run(): if action == 'login': ok, err = conn.login(uname, upass) else: ok, err = conn.register(uname, upass) self._post(lambda: finish(ok, err, uname, upass)) threading.Thread(target=run, daemon=True).start() b1.clicked.connect(lambda: do_auth('login')) b2.clicked.connect(lambda: do_auth('register')) b3.clicked.connect(dlg.reject) dlg.exec_() def _get_pass_hash(self): """取本地账号密码哈希 (解密服务器凭据用); 缺失时查库兜底""" ph = self.account.get('pass_hash', '') or '' if not ph: try: cur = self.account_db.conn.cursor() r = cur.execute("SELECT pass_hash FROM accounts WHERE id=?", (self.account.get('id', -1),)).fetchone() if r: ph = r[0] or '' except Exception: pass return ph def _auto_connect_server(self): """记住我的服务器账号: 自动登录 (后台线程)""" try: cred = load_server_cred(self._get_pass_hash()) except Exception: cred = None if not cred: return addr, username, password = cred try: host, port = addr.rsplit(':', 1) port = int(port) except Exception: return threading.Thread(target=self._do_auto_connect, args=(host, port, username, password, 3), daemon=True).start() def _do_auto_connect(self, host, port, username, password, attempts=3): for attempt in range(attempts): if self.server_conn and self.server_conn.connected: return conn = ServerConnection( on_status=self._on_status, on_groups=lambda gl: self._post(lambda: self._on_server_groups(gl)), on_message=lambda gid, frm, pl: self._post(lambda: self._on_server_group_message(gid, frm, pl)), on_members=lambda gid, mems: self._post(lambda: self._on_server_members(gid, mems)), on_system=lambda gid, t: self._post(lambda: self._on_server_system(gid, t)), on_group_name=lambda gid, n: self._post(lambda: self._on_server_group_name(gid, n)), on_announcement=lambda gid, t: self._post(lambda: self._on_server_announcement(gid, t)), on_group_info=lambda gid, n, a: self._post(lambda: self._on_server_group_info(gid, n, a))) if not conn.connect(host, port): if attempt < attempts - 1: time.sleep(2) continue self._post(lambda: self._on_status("☁️ 服务器自动连接失败(重试3次)，请手动「☁️服务器」登录")) return ok, err = conn.login(username, password) if ok: self._post(lambda: self._on_auto_server_logged_in(conn, host, port)) return if attempt < attempts - 1: time.sleep(2) continue self._post(lambda: self._on_status(f"☁️ 服务器自动登录失败: {err}，请手动「☁️服务器」登录")) def _update_title(self): """窗口标题显示当前身份: 有服务器账号就显示服务器账号""" if self.server_username: self.setWindowTitle(f"{APP_NAME} - {self.server_username} ☁️服务器 [{self._my_nick()}]") else: self.setWindowTitle(f"{APP_NAME} - {self._my_nick()}") def _bind_server_conn(self): """把已有服务器群绑定到当前连接 + 更新群内身份为服务器账号""" for g in self.groups.values(): if g.get('mode') == 'server': g['server_conn'] = self.server_conn if self.server_username: g['my_username'] = self.server_username self._refresh_groups() def _on_auto_server_logged_in(self, conn, host, port): self.server_conn = conn self.server_addr = f"{host}:{port}" self.server_username = conn.username self._bind_server_conn() self._update_title() self._on_status(f"☁️ 已自动登录中央服务器 {host}:{port} (用户: {conn.username})") self._add_system(None, "☁️ 服务器自动登录成功，正在实时同步群...") conn.list_groups() def _load_cached_server_groups(self): """启动时从本地缓存显示服务器群 (离线也能看到), 连接后实时同步""" try: cred = load_server_cred(self._get_pass_hash()) except Exception: cred = None if not cred: return addr = cred[0] self.server_addr = addr # 先设服务器地址, 供密钥查找/群去重 groups = load_server_groups_cache(addr) if groups: self._apply_groups_to_ui(groups) self._on_status(f"☁️ 已从本地缓存显示 {len(groups)} 个服务器群") def _on_server_groups(self, groups): """服务器实时返回群列表 → 刷新UI + 写本地缓存""" self._apply_groups_to_ui(groups) save_server_groups_cache(self.server_addr, groups) self._on_status(f"☁️ 已实时同步 {len(groups)} 个服务器群") def _apply_groups_to_ui(self, groups): for g in groups: gid = g.get('id', '') gname = g.get('name', '') or gid owner = g.get('owner', '') key = self._get_saved_server_key(self.server_addr, gid) if key: session = SecureSession() try: session.import_key(key) except Exception: session = None if session: if not self._server_group_exists(gid): self._add_server_group(gid, gname, owner, session) ex = self._gid_for_server_group(gid) if ex: self.groups[ex]['server_conn'] = self.server_conn self._on_group_name(ex, gname or gid) ann = g.get('announcement', '') if ann: self.groups[ex]['announcement'] = ann if ex == self.active_gid: self._show_banner(ann) self._refresh_groups() else: self._add_system(None, f"☁️ 群 [{gname}] ({gid}) 需要密钥，点「☁️ 加群」输入群ID和密钥") def _server_group_exists(self, group_id): key = f"server@{self.server_addr}@{group_id}" return any(g['group_key'] == key for g in self.groups.values()) def _gid_for_server_group(self, group_id): for gid, g in self.groups.items(): if g.get('mode') == 'server' and g.get('group_id') == group_id: return gid return None def _add_server_group(self, group_id, group_name, owner, session): gid = self._new_gid() try: srv_host = self.server_addr.split(':')[0] srv_port = int(self.server_addr.split(':')[1]) except Exception: srv_host, srv_port = '127.0.0.1', 8666 group = { 'gid': gid, 'mode': 'server', 'group_key': f"server@{self.server_addr}@{group_id}", 'host': srv_host, 'port': srv_port, 'group_id': group_id, 'server_conn': self.server_conn, 'name': group_name or group_id, 'my_username': self.server_username or self._my_nick(), 'my_account': self.account.get('username', ''), 'is_host': owner == (self.server_username or self.account.get('username', '')), 'server': None, 'client': None, 'session': session, 'users': [], 'messages': {}, 'order': [], 'announcement': '', 'group_name': group_name or '', 'user_key': None } self._register_group(gid, group) self._save_server_group(gid) return gid def _save_server_group(self, gid): group = self.groups.get(gid) if group: try: self.account_db.save_group(self.account['id'], group['group_key'], group['host'], group['port'], group['session'].export_key(), group['name'], 0) except Exception: pass def _get_saved_server_key(self, server_addr, group_id): gkey = f"server@{server_addr}@{group_id}" try: for r in self.account_db.load_groups(self.account['id']): if r['group_key'] == gkey: return r['session_key'] or None except Exception: pass return None def _create_server_group(self): if not self.server_conn or not self.server_conn.connected: self._on_status("请先连接中央服务器 (☁️ 服务器)") return name, ok = QtWidgets.QInputDialog.getText(self, "☁️ 服务器建群", "群名:", text="新群") if not ok or not name: return group_id = f"sg{int(time.time()*1000)}-{os.urandom(3).hex()}" session = SecureSession(); session.generate() self.server_conn.create_group(group_id, name) gid = self._add_server_group(group_id, name, self.server_username, session) self._add_system(gid, f"✅ 已在服务器创建群 [{name}]") self._show_server_group_info(gid) # 显示并复制 群ID + 服务器 + 密钥 self.server_conn.group_members(group_id) def _join_server_group(self): if not self.server_conn or not self.server_conn.connected: self._on_status("请先连接中央服务器 (☁️ 服务器)") return gid2, ok1 = QtWidgets.QInputDialog.getText(self, "☁️ 服务器加群", "群ID (创建者分享):") if not ok1 or not gid2: return key, ok2 = QtWidgets.QInputDialog.getText(self, "☁️ 服务器加群", "群密钥 (创建者分享):") if not ok2 or not key: return session = SecureSession() if not session.import_key(key): self._on_status("密钥无效！") return self.server_conn.join_group(gid2) gid = self._add_server_group(gid2, gid2, '', session) self._add_system(gid, f"✅ 已加入服务器群 [{gid2}]") self.server_conn.group_members(gid2) def _on_server_group_message(self, group_id, sender, payload): gid = self._gid_for_server_group(group_id) if not gid or not isinstance(payload, dict): return kind = payload.get('kind') if kind == 'chat': self._on_chat(gid, sender, payload.get('text', ''), payload.get('mid', '')) elif kind == 'file': self._on_file(gid, {'filename': payload.get('filename', '文件'), 'size': payload.get('size', 0), 'data': payload.get('data', ''), 'mime': payload.get('mime', 'file'), 'mid': payload.get('mid', ''), 'ts': payload.get('ts', int(time.time())), 'from': sender}) elif kind == 'voice': self._on_voice(gid, {'data': payload.get('data', ''), 'size': payload.get('size', 0), 'mid': payload.get('mid', ''), 'ts': payload.get('ts', int(time.time())), 'from': sender}) elif kind == 'withdraw': self._on_withdraw(gid, payload.get('mid', '')) elif kind == 'read': self._on_read(gid, payload.get('mid', '')) def _on_server_system(self, group_id, text): gid = self._gid_for_server_group(group_id) if gid: self._add_system(gid, text) if '加入' in text or '离开' in text: self.server_conn.group_members(group_id) def _on_server_members(self, group_id, members): for gid, g in self.groups.items(): if g.get('mode') == 'server' and g.get('group_id') == group_id: g['users'] = [{'name': m, 'account': m} for m in members] if gid == self.active_gid: self._render_users(gid) break def _on_server_group_name(self, group_id, name): gid = self._gid_for_server_group(group_id) if gid: self._on_group_name(gid, name) self._add_system(gid, f"🏷️ 群名更新为: {name}") def _on_server_group_info(self, group_id, name, announcement): """服务器推送的群信息(加入时): 同步群名/公告""" gid = self._gid_for_server_group(group_id) if not gid: return if name: self._on_group_name(gid, name) if announcement: self._on_announcement(gid, announcement) def _on_server_announcement(self, group_id, text): gid = self._gid_for_server_group(group_id) if gid: self._on_announcement(gid, text) # ---------- 消息操作 ---------- def _on_download(self, mid): gid = self.active_gid group = self.groups.get(gid) if not group: return msg = group['messages'].get(mid) if not msg: return data = msg.get_data_bytes() if not data: self._on_status("文件已清理或不存在") return c = msg.content if isinstance(msg.content, dict) else {} path, _ = QtWidgets.QFileDialog.getSaveFileName(self, "保存文件", os.path.join(DOWNLOAD_DIR, c.get('filename', 'download.bin'))) if not path: return try: with open(path, 'wb') as f: f.write(data) self._add_system(gid, f"✅ 已保存: {path}") except Exception as e: self._add_system(gid, f"❌ 保存失败: {e}") def _on_play(self, mid): """播放语音: 优先读媒体文件; 文件缺失时回退到内存缓存, 保证点了就有声""" gid = self.active_gid group = self.groups.get(gid) msg = group['messages'].get(mid) if group else None if not msg: self._on_status("消息不存在") return data = msg.get_data_bytes() if data: self.voice.last_voice = data self.voice.last_voice_from = msg.sender self.voice.play(data, msg.sender) elif self.voice.last_voice: # 媒体文件缺失 → 用最近一次缓存的语音 self.voice.play(self.voice.last_voice, self.voice.last_voice_from) self._on_status("⚠️ 媒体文件缺失，已用缓存语音播放") else: self._on_status("文件已清理或不存在") def _on_preview(self, mid): gid = self.active_gid group = self.groups.get(gid) msg = group['messages'].get(mid) if group else None if not msg: return data = msg.get_data_bytes() if not data: return try: from PIL import Image img = Image.open(io.BytesIO(data)) img.thumbnail((900, 700)) img = img.convert("RGBA") qimg = QtGui.QImage(img.tobytes(), img.width, img.height, QtGui.QImage.Format_RGBA8888) pix = QtGui.QPixmap.fromImage(qimg) dlg = QtWidgets.QDialog(self) dlg.setWindowTitle(f"图片预览 - {msg.sender}") lab = QtWidgets.QLabel() lab.setPixmap(pix) lay = QtWidgets.QVBoxLayout(dlg) lay.addWidget(lab) close = QtWidgets.QPushButton("关闭") close.clicked.connect(dlg.accept) lay.addWidget(close) dlg.exec_() except Exception as e: self._on_status(f"预览失败: {e}") def _on_msg_selected(self, mid): """消息列表选中/取消回调 → 同步到主窗口 (None=取消选中)""" self.selected_mid = mid def _show_context_menu(self, mid, pos): gid = self.active_gid group = self.groups.get(gid) if not group: return msg = group['messages'].get(mid) if not msg: return self.selected_mid = mid menu = QtWidgets.QMenu(self) if msg.type == 'text': menu.addAction("📋 复制文字", lambda: self._copy_msg(msg)) if msg.type in ('image', 'file', 'audio', 'voice'): menu.addAction("💾 下载", lambda: self._on_download(mid)) if msg.type == 'image': menu.addAction("🔍 查看原图", lambda: self._on_preview(mid)) if msg.type in ('voice', 'audio'): menu.addAction("▶ 播放", lambda: self._on_play(mid)) menu.addSeparator() if msg.is_withdrawable(): menu.addAction("🚫 撤回", lambda: self._withdraw(mid)) menu.addAction("📤 分享到...", lambda: self._share(mid)) menu.exec_(pos) def _copy_msg(self, msg): if msg.type == 'text': QtWidgets.QApplication.clipboard().setText(msg.content) self._on_status("📋 已复制文字") def _withdraw(self, mid): gid = self.active_gid group = self.groups.get(gid) if not group: return msg = group['messages'].get(mid) if not msg or not msg.is_withdrawable(): self._on_status("只能撤回自己 2 分钟内的消息") return self._send_withdraw_to(group, mid) self._delete_message(gid, mid) self._add_system(gid, "🚫 你撤回了一条消息") def _share(self, mid): gid = self.active_gid group = self.groups.get(gid) msg = group['messages'].get(mid) if group else None if not msg or msg.withdrawn: return targets = [g for g in self.order if g != gid] if not targets: self._on_status("没有其他群可以分享") return name, ok = QtWidgets.QInputDialog.getItem(self, "📤 分享到", "选择目标群:", [self.groups[g]['name'] for g in targets], 0, False) if not ok: return g2 = targets[[self.groups[g]['name'] for g in targets].index(name)] group2 = self.groups.get(g2) if not group2 or not self._group_connected(group2): self._on_status("目标群未连接") return mid2 = gen_mid() if msg.type == 'text': enc = group2['session'].encrypt(msg.content) if self._send_chat_to(group2, enc, mid2): self._add_local_message(g2, mid2, 'text', msg.content) elif msg.type in ('image', 'file', 'audio'): c = msg.content raw = msg.get_data_bytes() mime = 'image' if msg.type == 'image' else ('file' if msg.type == 'file' else 'audio') enc = group2['session'].encrypt_bytes(raw) new_path = save_media(raw, c.get('ext', '')) content = {'filename': c.get('filename', 'file'), 'size': len(raw), 'path': new_path, 'ext': c.get('ext', '')} if self._send_file_to(group2, content['filename'], enc.encode('utf-8'), mid2, mime): self._add_local_message(g2, mid2, msg.type, content) elif msg.type == 'voice': raw = msg.get_data_bytes() enc = group2['session'].encrypt_bytes(raw) new_path = save_media(raw, '.wav') content = {'filename': 'voice.wav', 'size': len(raw), 'path': new_path, 'ext': '.wav'} if self._send_voice_to(group2, enc.encode('utf-8'), mid2): self._add_local_message(g2, mid2, 'voice', content) self._switch_group(g2) self._add_system(g2, f"📤 已分享一条来自 {msg.sender} 的消息") # ---------- 密钥 / 管理 / 工具 ---------- def _show_key_dialog(self, gid): group = self.groups.get(gid) if not group or not group['session'].has_key(): return data = group['session'].export_key() dlg = QtWidgets.QDialog(self) dlg.setWindowTitle("🔑 会话密钥") dlg.setFixedSize(460, 200) lay = QtWidgets.QVBoxLayout(dlg) lay.addWidget(QtWidgets.QLabel("复制此密钥分享给加入该群的人")) te = QtWidgets.QTextEdit() te.setPlainText(data) te.setReadOnly(True) lay.addWidget(te) row = QtWidgets.QHBoxLayout() b1 = QtWidgets.QPushButton("📋 复制全部") b1.setProperty("class", "primary") b1.clicked.connect(lambda: (QtWidgets.QApplication.clipboard().setText(data), dlg.accept())) b2 = QtWidgets.QPushButton("关闭") b2.clicked.connect(dlg.reject) row.addWidget(b1); row.addWidget(b2) lay.addLayout(row) dlg.exec_() def _show_server_group_info(self, gid): """服务器建群: 显示并复制 群ID + 服务器地址 + 群密钥 (加入需要)""" group = self.groups.get(gid) if not group: return gid2 = group.get('group_id', '') key = group['session'].export_key() if group['session'].has_key() else '' addr = self.server_addr or '' dlg = QtWidgets.QDialog(self) dlg.setWindowTitle("☁️ 服务器群邀请信息") dlg.setFixedSize(500, 340) lay = QtWidgets.QVBoxLayout(dlg) lay.addWidget(QtWidgets.QLabel("把下面 3 项分享给朋友，让他们点「☁️ 加群」输入后加入")) def make_row(label, value, long=False): lay.addWidget(QtWidgets.QLabel(label)) row = QtWidgets.QHBoxLayout() if long: te = QtWidgets.QTextEdit(value); te.setFixedHeight(56) row.addWidget(te, 1) else: te = QtWidgets.QLineEdit(value); te.setReadOnly(True) row.addWidget(te, 1) btn = QtWidgets.QPushButton("📋 复制") btn.clicked.connect(lambda: (QtWidgets.QApplication.clipboard().setText(value), dlg.accept())) row.addWidget(btn) lay.addLayout(row) make_row("🆔 群ID:", gid2) make_row("🌐 服务器:", addr) make_row("🔑 群密钥:", key, long=True) lay.addWidget(QtWidgets.QLabel("⚠️ 密钥请私下发给成员，勿在公开渠道泄露")) close = QtWidgets.QPushButton("关闭") close.clicked.connect(dlg.reject) lay.addWidget(close) dlg.exec_() def _copy_active_key(self): gid, group = self._current_group() if group and group['session'].has_key(): QtWidgets.QApplication.clipboard().setText(group['session'].export_key()) self._on_status(f"🔑 已复制「{group['name']}」的密钥") else: self._on_status("请先选择一个群") def _rename(self): nick, ok = QtWidgets.QInputDialog.getText(self, "改名", "新昵称:", text=self._my_nick()) if not ok or not nick: return nick = nick[:20] try: self.account_db.update_nickname(self.account['id'], nick) except Exception: pass self.account = dict(self.account) self.account['nickname'] = nick for gid in list(self.order): g = self.groups[gid] g['my_username'] = nick if g.get('mode') == 'server': pass elif g.get('client'): g['client'].send_rename(nick) self._add_system(gid, f"✏️ 你已改名为 {nick}") self._refresh_groups() self._render_users(self.active_gid) self._on_status(f"✅ 已改名为 {nick}") def _set_announcement(self): gid, group = self._current_group() if not group: return if not group['is_host']: self._on_status("仅群主可设置公告") return text, ok = QtWidgets.QInputDialog.getText(self, "群公告", "输入公告内容:", text=group.get('announcement', '')) if not ok: return if group.get('mode') == 'server': group['server_conn'].set_announcement(group['group_id'], text) elif group.get('client'): group['client'].send_announcement(text) group['announcement'] = text if gid == self.active_gid: self._show_banner(text) def _set_group_name(self): gid, group = self._current_group() if not group: return if not group['is_host']: self._on_status("仅群主可设置群名") return cur = group.get('group_name', '') or group.get('group_id', '') name, ok = QtWidgets.QInputDialog.getText(self, "群名", "输入群名:", text=cur) if not ok or not name: return if group.get('mode') == 'server': group['server_conn'].set_group_name(group['group_id'], name) elif group.get('client'): group['client'].send_set_group_name(name) self._on_group_name(gid, name) def _withdraw_selected(self): if self.selected_mid: self._withdraw(self.selected_mid) else: self._on_status("请先点击选择一条消息") def _share_selected(self): if self.selected_mid: self._share(self.selected_mid) else: self._on_status("请先点击选择一条消息") def _download_selected(self): if self.selected_mid: self._on_download(self.selected_mid) else: self._on_status("请先点击选择一条消息") def _copy_selected(self): gid = self.active_gid group = self.groups.get(gid) msg = group['messages'].get(self.selected_mid) if group and self.selected_mid else None if msg: self._copy_msg(msg) else: self._on_status("请先点击选择一条消息") def _leave_active_group(self): if self.active_gid: self._leave_group(self.active_gid) def _leave_group(self, gid): group = self.groups.get(gid) if not group: return if group.get('mode') == 'server': try: group['server_conn'].leave_group(group['group_id']) except Exception: pass else: try: group['client'].disconnect() except Exception: pass if group['server']: try: group['server'].stop() except Exception: pass if gid in self.order: self.order.remove(gid) del self.groups[gid] try: self.account_db.remove_group(self.account['id'], group['group_key']) except Exception: pass if self.active_gid == gid: self.active_gid = self.order[-1] if self.order else None self._render_group(self.active_gid) self._refresh_groups() def _on_group_menu(self, pos): item = self.group_list.itemAt(pos) if not item: return gid = item.data(QtCore.Qt.UserRole) menu = QtWidgets.QMenu(self) menu.addAction("🔑 复制密钥", lambda: self._copy_group_key(gid)) menu.addAction("💾 导出密钥文件", lambda: self._save_key_file(gid)) menu.addAction("📤 分享密钥到其他群", lambda: self._share_key_to(gid)) menu.addAction("📋 群属性", lambda: self._group_properties(gid)) menu.addSeparator() menu.addAction("切换到此群", lambda: self._switch_group(gid)) menu.addAction("❌ 离开该群", lambda: self._leave_group(gid)) menu.exec_(self.group_list.mapToGlobal(pos)) def _copy_group_key(self, gid): group = self.groups.get(gid) if group and group['session'].has_key(): QtWidgets.QApplication.clipboard().setText(group['session'].export_key()) self._on_status(f"🔑 已复制「{group['name']}」的密钥") def _save_key_file(self, gid): group = self.groups.get(gid) if not group or not group['session'].has_key(): return data = group['session'].export_key() path, _ = QtWidgets.QFileDialog.getSaveFileName(self, "导出密钥", f"群密钥_{gid}.txt", "文本 (*.txt)") if not path: return try: with open(path, 'w', encoding='utf-8') as f: f.write(data) self._on_status(f"✅ 密钥已导出: {path}") except Exception as e: self._on_status(f"❌ 导出失败: {e}") def _share_key_to(self, gid): group = self.groups.get(gid) if not group or not group['session'].has_key(): return data = group['session'].export_key() targets = [g for g in self.order if g != gid] if not targets: self._on_status("没有其他群可以分享") return name, ok = QtWidgets.QInputDialog.getItem(self, "📤 分享密钥到", "选择目标群:", [self.groups[g]['name'] for g in targets], 0, False) if not ok: return g2 = targets[[self.groups[g]['name'] for g in targets].index(name)] QtWidgets.QApplication.clipboard().setText(data) self._switch_group(g2) self._add_system(g2, f"📤 已复制「{group['name']}」的密钥，可粘贴给群成员加入该群") def _group_properties(self, gid): group = self.groups.get(gid) if not group: return mode = group.get('mode', 'p2p') mode_text = '🔗 端到端 (P2P)' if mode != 'server' else '☁️ 中央服务器' gid_field = group.get('group_id', '(无)') if mode == 'server' else '(无)' status = '🟢 在线' if self._group_connected(group) else '🔴 离线' info = (f"📋 群名: {group.get('group_name', '') or group['name']}\n" f"🏷️ 类型: {mode_text}\n" f"🆔 群ID: {gid_field}\n" f"🌐 地址: {group['host']}:{group['port']}\n" f"📡 状态: {status}\n" f"👑 身份: {'是 [群主]' if group['is_host'] else '否 [成员]'}\n" f"🔑 加密: {'✅ 已加密' if group['session'].has_key() else '❌ 无密钥'}\n" f"👥 成员: {len(group['users'])} 人") QtWidgets.QMessageBox.information(self, "📋 群属性", info) def _on_user_menu(self, pos): gid, group = self._current_group() if not group: return item = self.users_list.itemAt(pos) if not item: return if group.get('mode') == 'server': self._on_status("服务器群由服务器管理成员") return if not group['is_host']: self._on_status("仅群主可管理用户") return idx = self.users_list.row(item) if idx < 0 or idx >= len(group['users']): return target = group['users'][idx] if target.get('account') == self.account.get('username', ''): return if QtWidgets.QMessageBox.question(self, "踢人", f"确定将 {target.get('name','')} 移出群吗？", QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No) == QtWidgets.QMessageBox.Yes: group['client'].send_kick(target.get('account', '')) def _clear_media_cache(self): if QtWidgets.QMessageBox.question(self, "清理媒体缓存", "将删除「文档/ChaosChat/media」中所有文件/图片/语音，历史记录中的这些文件将无法打开。确定？", QtWidgets.QMessageBox.Yes | QtWidgets.QMessageBox.No) == QtWidgets.QMessageBox.Yes: n = clear_media_cache() self._on_status(f"🗑️ 已清理 {n} 个媒体文件") def _open_download_dir(self): try: os.startfile(DOWNLOAD_DIR) except Exception: pass # ---------- 自动加入 ---------- def _start_auto_join(self): threading.Thread(target=self._auto_join_worker, daemon=True).start() def _auto_join_worker(self): try: rows = self.account_db.load_groups(self.account['id']) except Exception: rows = [] for r in rows: if r['group_key'] and str(r['group_key']).startswith('server@'): continue self._auto_join_one(r) def _auto_join_one(self, r, attempts=3): host, port = r['host'], r['port'] key = f"{host}:{port}" for g in list(self.groups.values()): if g['group_key'] == key and self._group_connected(g): return session = SecureSession() try: session.import_key(r['session_key']) except Exception: pass gid = self._new_gid() for attempt in range(attempts): server = None if r['is_host']: server = ChatServer(on_status=lambda t: self._post(lambda: self._on_status(t)), server_key=key, server_db=self.server_db, server_secret=session.session_key) if not server.start(port): server = None if attempt < attempts - 1: time.sleep(2) continue return client = self._make_client(gid) saved_uk = r['user_key'] if 'user_key' in r.keys() else None ok = client.connect(host, port, self._my_nick(), nickname=self._my_nick(), account=self.account.get('username', ''), auth_secret=session.auth_secret(), user_key=saved_uk, user_key_enc=session.user_key_enc()) if ok: group = self._build_group(gid, host, port, self._my_nick(), bool(r['is_host']), server, client, session) group['user_key'] = saved_uk self._post(lambda gid=gid, group=group: self._register_group(gid, group)) return if server: try: server.stop() except Exception: pass if attempt < attempts - 1: time.sleep(2) # ---------- 退出 ---------- def _logout(self): clear_session() self._relogin = True self._close_all() def _quit(self): self._relogin = False self._close_all() def _close_all(self): try: if self.server_conn: self.server_conn.disconnect() except Exception: pass for gid in list(self.order): g = self.groups.get(gid) if g: try: if g.get('client'): g['client'].disconnect() except Exception: pass if g.get('server'): try: g['server'].stop() except Exception: pass try: self.history.conn.close() except Exception: pass try: self.account_db.conn.close() except Exception: pass self.close() def closeEvent(self, event): self._close_all() event.accept() # ==================== 主入口 ==================== def main(): import sys app = QtWidgets.QApplication(sys.argv) app.setStyleSheet(QSS) db = AccountDB() # 自动登录 (记住我) account = None token = load_session() if token: account = db.validate_session(token) if account: print(f"✅ 自动登录: {account['username']}") if account is None: login = LoginDialog(db) if login.exec_() != QtWidgets.QDialog.Accepted: return account = login.result # 本地登录不强制记住; 保持简单 token = db.create_session(account['id']) save_session(token) win = MainWindow(account) win.show() app.exec_() if __name__ == "__main__": main()