#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 局域网聊天服务器 - Python + WebSocket 支持文字和图片 (Base64) 支持 Ctrl+V 粘贴图片发送 运行: python chat_server.py 访问: http://你的IP:8080 """ import json import asyncio import websockets from http.server import HTTPServer, SimpleHTTPRequestHandler import threading import socket import os import time import sys import subprocess # ---------- 配置 ---------- HTTP_PORT = 8080 WS_PORT = 8765 HOST = '0.0.0.0' # 存储所有连接的客户端 connected_clients = set() client_names = {} # ---------- 获取本机局域网 IP ---------- def get_local_ip(): try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(('8.8.8.8', 80)) ip = s.getsockname()[0] s.close() return ip except: return '127.0.0.1' LOCAL_IP = get_local_ip() # ---------- 自动添加防火墙规则 ---------- def add_firewall_rule(): """自动添加防火墙入站规则""" try: import ctypes is_admin = ctypes.windll.shell32.IsUserAnAdmin() != 0 if not is_admin: print("⚠️ 当前不是管理员权限，防火墙规则可能无法自动添加") print("💡 请尝试以管理员身份重新运行: 右键 -> 以管理员身份运行") return False rule_name = "局域网聊天服务" ports = f"{HTTP_PORT},{WS_PORT}" subprocess.run(f'netsh advfirewall firewall delete rule name="{rule_name}"', shell=True, capture_output=True) cmd = f'netsh advfirewall firewall add rule name="{rule_name}" dir=in action=allow protocol=TCP localport={ports}' result = subprocess.run(cmd, shell=True, capture_output=True, text=True) if "确定" in result.stdout or "Ok" in result.stdout: print(f"✅ 防火墙规则已添加: {rule_name} (端口 {ports})") return True else: print(f"⚠️ 防火墙规则添加可能失败: {result.stdout}") return False except Exception as e: print(f"⚠️ 无法自动配置防火墙: {e}") return False # ---------- WebSocket 服务器 ---------- async def ws_handler(websocket): connected_clients.add(websocket) client_names[websocket] = f"用户{len(connected_clients)}" print(f"✅ 新客户端连接，当前在线: {len(connected_clients)}") try: async for message in websocket: try: data = json.loads(message) if data.get('type') == 'set_name': client_names[websocket] = data.get('name', '匿名') continue if data.get('senderId'): data['senderName'] = client_names.get(websocket, '匿名') for client in list(connected_clients): if client != websocket: try: await client.send(json.dumps(data)) except: pass print(f"📨 转发消息: {data.get('senderName')} -> {len(connected_clients)-1} 个客户端") except json.JSONDecodeError: pass except websockets.exceptions.ConnectionClosed: pass finally: connected_clients.remove(websocket) if websocket in client_names: del client_names[websocket] print(f"❌ 客户端断开，当前在线: {len(connected_clients)}") # ---------- HTTP 服务 ---------- class ChatHTTPHandler(SimpleHTTPRequestHandler): def do_GET(self): if self.path == '/': self.path = '/index.html' return super().do_GET() def log_message(self, format, *args): pass def start_http_server(): os.chdir(os.path.dirname(os.path.abspath(__file__))) httpd = HTTPServer((HOST, HTTP_PORT), ChatHTTPHandler) print(f"🌐 HTTP 服务运行在: http://{LOCAL_IP}:{HTTP_PORT}") httpd.serve_forever() # ---------- 生成 HTML ---------- def create_html_file(): html_content = f''' 局域网聊天 - 支持粘贴图片

💬 局域网聊 📋 Ctrl+V 贴图

连接中...
👋 欢迎加入聊天
发送文字 · 选择图片 · Ctrl+V 粘贴图片
🖼️
💡 Ctrl+V 粘贴图片 · Enter 发送
''' with open('index.html', 'w', encoding='utf-8') as f: f.write(html_content) print("📄 已生成 index.html (支持 Ctrl+V 粘贴图片)") # ---------- 主程序 ---------- async def main(): create_html_file() # 尝试自动添加防火墙规则 add_firewall_rule() # 显示网络信息 print("\n" + "="*60) print(f"📡 服务器地址: http://{LOCAL_IP}:{HTTP_PORT}") print(f"📡 本机访问: http://127.0.0.1:{HTTP_PORT}") print(f"🔌 WebSocket: ws://{LOCAL_IP}:{WS_PORT}") print("="*60 + "\n") print("💡 新功能: 在输入框按 Ctrl+V 直接粘贴图片发送！") print("💡 如果其他设备无法访问，请关闭 Windows 防火墙\n") # 启动 HTTP 服务器 http_thread = threading.Thread(target=start_http_server, daemon=True) http_thread.start() time.sleep(0.5) # 启动 WebSocket print("📡 等待客户端连接...") print("按 Ctrl+C 停止服务器\n") async with websockets.serve(ws_handler, HOST, WS_PORT): await asyncio.Future() if __name__ == '__main__': try: asyncio.run(main()) except KeyboardInterrupt: print("\n🛑 服务器已停止") try: os.remove('index.html') except: pass sys.exit(0) except Exception as e: print(f"\n❌ 错误: {e}") print("\n💡 可能的解决方案:") print(" 1. 更换端口: 修改脚本开头的 HTTP_PORT 和 WS_PORT") print(" 2. 以管理员身份运行") print(" 3. 关闭防火墙后重试") input("\n按 Enter 退出...")