Files
ChaosCrypt/ChaosCryptChat/qtmain.py
T

4148 lines
162 KiB
Python

# -*- 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()