#!/usr/bin/env python3 """ NoScrible v5 — PC 伴侣服务器（带图形界面） ============================================ 接收平板发送的公式截图 / LaTeX 代码，支持本地 OCR 识别与 AI 云端识别。 包含悬浮窗截图接收、TCP 剪贴板推送、AI 代理转发功能。 打包命令: pyinstaller --onefile --noconsole --name NoScribleServer pc_server_v4.py """ import json import os import socket import struct import subprocess import sys import threading import tkinter as tk from tkinter import ttk, scrolledtext, messagebox from http.server import HTTPServer, BaseHTTPRequestHandler from datetime import datetime from urllib.parse import urlparse import webbrowser import base64 import io import traceback from functools import partial from io import BytesIO try: import urllib.request HAS_URLLIB = True except ImportError: HAS_URLLIB = False # ─── 可选依赖 ─── HAS_PIL = False HAS_CLIPBOARD = False OCR_MODEL = None try: from PIL import Image HAS_PIL = True except ImportError: pass try: import pyperclip HAS_CLIPBOARD = True except ImportError: pass # ────────────────────────────────────────────── # 配置 # ────────────────────────────────────────────── HTTP_PORT = 48158 TCP_PORT = 48159 MAX_LOG_LINES = 300 # ── OCR 模型路径 ── _SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) OCR_MODELS_DIR = os.path.join(_SCRIPT_DIR, 'ocr_models') TEXIFY_PKG_DIR = os.path.join(OCR_MODELS_DIR, 'texify_pkg') TEXIFY_MODEL_DIR = os.path.join(OCR_MODELS_DIR, 'texify_model') # 将 texify 本地包加入搜索路径 if os.path.isdir(TEXIFY_PKG_DIR): sys.path.insert(0, TEXIFY_PKG_DIR) 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 Exception: return "127.0.0.1" def to_clipboard(text): """写入系统剪贴板（跨平台）""" clean = text.replace("$$", "").replace("\\[", "").replace("\\]", "").strip() if HAS_CLIPBOARD: try: pyperclip.copy(clean) return True except Exception: pass try: if sys.platform == "win32": subprocess.run(["clip"], input=clean.encode("utf-16le"), check=False) elif sys.platform == "darwin": subprocess.run(["pbcopy"], input=clean.encode("utf-8"), check=False) else: subprocess.run(["xclip", "-selection", "clipboard"], input=clean.encode("utf-8"), check=False) return True except Exception: return False def _load_texify_model(): """加载 Texify 模型（高精度公式识别，约 600MB） Returns: tuple (model, processor) 或 (None, None) """ try: import texify.model.config as _cfg_mod from texify.model.config import TexifyConfig, VariableDonutSwinConfig from transformers import MBartConfig, VisionEncoderDecoderConfig # Monkey-patch: 修复 transformers 新版对 config.json 中 # encoder/decoder 为 dict 时的兼容性问题 def _patched_get_config(model_checkpoint): import json config_path = os.path.join(model_checkpoint, "config.json") with open(config_path, "r", encoding="utf-8") as f: config_dict = json.load(f) encoder_dict = config_dict.pop("encoder", {}) decoder_dict = config_dict.pop("decoder", {}) # transformers 新版期望 encoder/decoder 为 dict config = VisionEncoderDecoderConfig( encoder=encoder_dict, decoder=decoder_dict, **config_dict ) config.encoder = VariableDonutSwinConfig(**encoder_dict) config.decoder = MBartConfig(**decoder_dict) return config _cfg_mod.get_config = _patched_get_config from texify.model.model import load_model as _tf_load_model from texify.model.processor import load_processor as _tf_load_processor import texify.settings as tx_settings import torch as _torch model_dir = TEXIFY_MODEL_DIR if not os.path.isdir(model_dir): return None, None safetensors = os.path.join(model_dir, 'model.safetensors') config_json = os.path.join(model_dir, 'config.json') if not os.path.exists(safetensors) or not os.path.exists(config_json): return None, None tx_settings.settings.MODEL_CHECKPOINT = model_dir model = _tf_load_model( checkpoint=model_dir, device='cpu', dtype=_torch.float32 ) processor = _tf_load_processor() model.eval() return model, processor except ImportError: return None, None except Exception: traceback.print_exc() return None, None class OCRModel: """OCR 模型包装器 — 封装 Texify 高精度公式识别模型，对外提供一致的推理接口""" def __init__(self): self._model = None self._processor = None def load(self): """加载 Texify 模型""" model, processor = _load_texify_model() if model is not None: self._model = model self._processor = processor return True return False @property def loaded(self): return self._model is not None @property def backend_name(self): return 'texify' if self._model else 'none' @property def device(self): return str(self._model.device) if hasattr(self._model, 'device') else 'cpu' def predict(self, image): """对 PIL Image 执行公式识别 Args: image: PIL.Image 对象 Returns: str: 识别出的 LaTeX 代码 """ from texify.inference import batch_inference if image.mode != 'RGB': image = image.convert('RGB') results = batch_inference([image], self._model, self._processor) return results[0].strip() if results else '' def load_ocr_model(): """加载 OCR 模型（仅使用 Texify）""" global OCR_MODEL if OCR_MODEL is not None and OCR_MODEL.loaded: return OCR_MODEL wrapper = OCRModel() ok = wrapper.load() if ok: OCR_MODEL = wrapper return OCR_MODEL OCR_MODEL = OCRModel() return None def check_ocr_weights_ready(log_func=None): """检查项目内嵌 OCR 模型权重文件是否就绪 Args: log_func: 可选，日志回调函数，签名为 (message, tag) Returns: bool: Texify 模型权重文件齐全 """ texify_ready = True texify_ready = texify_ready and os.path.isdir(TEXIFY_MODEL_DIR) texify_ready = texify_ready and os.path.exists( os.path.join(TEXIFY_MODEL_DIR, 'model.safetensors') ) texify_ready = texify_ready and os.path.exists( os.path.join(TEXIFY_MODEL_DIR, 'config.json') ) if log_func: if texify_ready: log_func("✅ Texify 高精度模型权重已就绪", "green") else: log_func("❌ Texify 权重缺失，请确保 ocr_models/ 目录完整", "red") return texify_ready # ────────────────────────────────────────────── # HTTP 服务处理器（v4 完整版） # ────────────────────────────────────────────── class V4ServerHandler(BaseHTTPRequestHandler): """处理所有 HTTP 请求 — 公式识别、悬浮截图、AI代理""" def _check_client_connection(self): """检测并通知新的 HTTP 客户端连接""" client_ip = self.client_address[0] server = self.server if hasattr(server, 'on_client_connect') and server.on_client_connect: with server.clients_lock: is_new = client_ip not in server.connected_clients if is_new: server.on_client_connect(client_ip, 'HTTP') def do_GET(self): self._check_client_connection() path = urlparse(self.path).path if path == "/": self._send_status() else: self._send_json(404, {"success": False, "message": "未找到"}) def do_POST(self): self._check_client_connection() path = urlparse(self.path).path try: body = self._parse_body() except Exception as e: self._send_json(400, {"success": False, "message": f"请求解析失败: {e}"}) return if path == "/math2latex": self._handle_ocr(body) elif path == "/float-capture": self._handle_float_capture(body) elif path in ("/v1/chat/completions", "/chat/completions"): self._handle_ai_proxy(body) elif path == "/send": latex = body.get("latex") or body.get("data") or "" if latex: ok = to_clipboard(latex) if hasattr(self.server, 'log_cb') and self.server.log_cb: self.server.log_cb(f"✅ [HTTP] 收到公式已写入剪贴板", "green") self._send_json(200, {"success": ok, "message": "已写入剪贴板"}) else: self._send_json(400, {"success": False, "message": "缺少公式内容"}) else: self._send_json(404, {"success": False, "message": "未知接口"}) def do_OPTIONS(self): self.send_response(200) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Upstream-URL, X-Upstream-Key") self.end_headers() def _send_json(self, code, data): body = json.dumps(data, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _parse_body(self): length = int(self.headers.get("Content-Length", 0)) if length == 0: return {} raw = self.rfile.read(length) try: return json.loads(raw) except json.JSONDecodeError: return {"data": raw.decode("utf-8", errors="replace")} def _send_status(self): if OCR_MODEL and OCR_MODEL.loaded: model_status = f"已加载 ({OCR_MODEL.backend_name})" else: model_status = "未加载" info = { "success": True, "service": "NoScrible v5 PC 伴侣", "端口": {"HTTP": HTTP_PORT, "TCP": TCP_PORT}, "公式识别": model_status, "剪贴板": "可用" if HAS_CLIPBOARD else "不可用", "Pillow": "已安装" if HAS_PIL else "未安装", } self._send_json(200, info) def _handle_ocr(self, body): """接收图片，OCR 识别公式""" if not HAS_PIL: self._send_json(500, {"success": False, "message": "Pillow 未安装"}) return b64 = body.get("image") or body.get("base64") or body.get("data") if not b64: self._send_json(400, {"success": False, "message": "缺少图片数据"}) return if b64.startswith("data:"): _, b64 = b64.split(",", 1) try: img = Image.open(BytesIO(base64.b64decode(b64))) except Exception as e: self._send_json(400, {"success": False, "message": f"图片解码失败: {e}"}) return if not OCR_MODEL or not OCR_MODEL.loaded: self._send_json(503, {"success": False, "message": "公式识别模型未加载"}) return try: latex = OCR_MODEL.predict(img) if hasattr(self.server, 'log_cb') and self.server.log_cb: self.server.log_cb(f"✅ [公式识别] {latex[:50]}...", "green") to_clipboard(latex) self._send_json(200, {"success": True, "latex": latex}) except Exception as e: self._send_json(500, {"success": False, "message": f"识别失败: {e}"}) def _handle_float_capture(self, body): """处理悬浮窗截图""" if not HAS_PIL: self._send_json(500, {"success": False, "message": "Pillow 未安装"}) return b64 = body.get("image") or body.get("base64") or body.get("data") if not b64: self._send_json(400, {"success": False, "message": "缺少图片数据"}) return if b64.startswith("data:"): _, b64 = b64.split(",", 1) ts = datetime.now().strftime("%H:%M:%S") if hasattr(self.server, 'log_cb') and self.server.log_cb: self.server.log_cb(f"📸 [{ts}] 收到悬浮窗截图", "blue") try: img = Image.open(BytesIO(base64.b64decode(b64))) except Exception as e: self._send_json(400, {"success": False, "message": f"图片解码失败: {e}"}) return latex = None if OCR_MODEL and OCR_MODEL.loaded: try: latex = OCR_MODEL.predict(img) if hasattr(self.server, 'log_cb') and self.server.log_cb: self.server.log_cb(f"✅ [悬浮截图] 识别: {latex[:50]}...", "green") to_clipboard(latex) except Exception as e: if hasattr(self.server, 'log_cb') and self.server.log_cb: self.server.log_cb(f"⚠️ [悬浮截图] 识别失败: {e}", "red") if latex: self._send_json(200, {"success": True, "latex": latex}) else: self._send_json(500, {"success": False, "message": "识别失败，模型未加载"}) def _handle_ai_proxy(self, body): """AI 代理转发""" if not HAS_URLLIB: self._send_json(500, {"success": False, "message": "urllib 不可用"}) return upstream = self.headers.get("X-Upstream-URL") or body.pop("upstream_url", None) api_key = self.headers.get("X-Upstream-Key") or body.pop("upstream_key", None) if not upstream: self._send_json(400, {"success": False, "message": "缺少目标 API 地址"}) return upstream = upstream.rstrip("/") if not upstream.endswith("/chat/completions"): upstream = upstream + "/v1/chat/completions" if "/v1" not in upstream else upstream + "/chat/completions" headers = {"Content-Type": "application/json"} if api_key: headers["Authorization"] = f"Bearer {api_key}" req_body = json.dumps(body, ensure_ascii=False).encode("utf-8") req = urllib.request.Request(upstream, data=req_body, headers=headers, method="POST") try: with urllib.request.urlopen(req, timeout=120) as resp: data = json.loads(resp.read()) self._send_json(resp.status, data) if hasattr(self.server, 'log_cb') and self.server.log_cb: self.server.log_cb(f"✅ [AI代理] 转发成功", "green") except urllib.error.HTTPError as e: err = e.read().decode("utf-8", errors="replace") try: err_data = json.loads(err) except json.JSONDecodeError: err_data = {"error": err[:200]} self._send_json(e.code, err_data) except urllib.error.URLError as e: self._send_json(502, {"success": False, "message": f"代理失败: {e.reason}"}) except Exception as e: self._send_json(500, {"success": False, "message": f"代理异常: {e}"}) def log_message(self, fmt, *args): pass # 安静运行，由 GUI 日志统一输出 # ────────────────────────────────────────────── # TCP 服务 # ────────────────────────────────────────────── def handle_tcp_client(conn, addr, callback, notify_connect=None): """处理单个 TCP 客户端连接 Args: conn: socket 连接对象 addr: (ip, port) 元组 callback: 日志回调函数 notify_connect: 可选，新连接通知回调，签名为 (ip) """ client_ip = addr[0] if notify_connect: notify_connect(client_ip) try: header = conn.recv(4) if len(header) < 4: return length = struct.unpack(">I", header)[0] data = b"" while len(data) < length: chunk = conn.recv(length - len(data)) if not chunk: break data += chunk latex = data.decode("utf-8") if callback: callback(f"✅ [TCP] 来自 {addr[0]}: {latex[:50]}...", "green") to_clipboard(latex) except Exception as e: if callback: callback(f"❌ [TCP] 错误: {e}", "red") finally: conn.close() def tcp_server_loop(server_socket, callback, stop_event, on_client_connect=None): """TCP 服务器主循环 Args: server_socket: TCP socket callback: 日志回调函数 stop_event: threading.Event 停止信号 on_client_connect: 可选，新客户端连接通知回调，签名为 (ip, transport) """ server_socket.listen(5) def _notify_tcp_client(ip): """TCP 客户端连接通知的内部封装""" if on_client_connect: on_client_connect(ip, 'TCP') while not stop_event.is_set(): try: server_socket.settimeout(1.0) conn, addr = server_socket.accept() t = threading.Thread(target=handle_tcp_client, args=(conn, addr, callback, _notify_tcp_client), daemon=True) t.start() except socket.timeout: continue except OSError: break # ────────────────────────────────────────────── # GUI 界面 # ────────────────────────────────────────────── class NoScribleGUI: def __init__(self): self.root = tk.Tk() self.root.title("NoScrible v5 伴侣服务器") self.root.geometry("640x580") self.root.minsize(520, 440) self.root.resizable(True, True) self.running = False self.httpd = None self.tcp_socket = None self.stop_event = threading.Event() self.ocr_loaded = False self.connected_clients = set() self.clients_lock = threading.Lock() self._build_ui() self._center_window() self._post_init() def _build_ui(self): bg = "#f5f5f5" self.root.configure(bg=bg) style = ttk.Style() style.theme_use("clam") style.configure("TFrame", background=bg) style.configure("TLabel", background=bg, font=("Microsoft YaHei UI", 10)) style.configure("Header.TLabel", font=("Microsoft YaHei UI", 18, "bold"), foreground="#1a1a2e", background=bg) style.configure("Sub.TLabel", font=("Microsoft YaHei UI", 10), foreground="#888", background=bg) style.configure("Stat.TLabel", font=("Microsoft YaHei UI", 10), background=bg) style.configure("Green.TLabel", foreground="#22c55e", font=("Microsoft YaHei UI", 10, "bold"), background=bg) style.configure("Red.TLabel", foreground="#ef4444", font=("Microsoft YaHei UI", 10, "bold"), background=bg) style.configure("TButton", font=("Microsoft YaHei UI", 10)) # ── 主容器 ── main = ttk.Frame(self.root, padding=20) main.pack(fill=tk.BOTH, expand=True) # ── 标题 ── tf = ttk.Frame(main) tf.pack(fill=tk.X, pady=(0, 12)) ttk.Label(tf, text="NoScrible", style="Header.TLabel").pack(side=tk.LEFT) ttk.Label(tf, text=" PC 伴侣服务器 v4", style="Sub.TLabel").pack(side=tk.LEFT, padx=(4,0)) # ── 状态卡片 ── card = ttk.Frame(main, relief="solid", borderwidth=1) card.pack(fill=tk.X, pady=(0, 12)) # 第一行：状态 + IP r1 = ttk.Frame(card, padding=(16, 10)) r1.pack(fill=tk.X) ttk.Label(r1, text="状态:", style="Stat.TLabel").pack(side=tk.LEFT) self.status_label = ttk.Label(r1, text="已停止", style="Red.TLabel") self.status_label.pack(side=tk.LEFT, padx=(6, 0)) ttk.Label(r1, text="本机 IP:", style="Stat.TLabel").pack(side=tk.LEFT, padx=(24, 0)) ip = get_local_ip() self.ip_label = ttk.Label(r1, text=ip, font=("Consolas", 11, "bold"), foreground="#6c63ff", background=bg) self.ip_label.pack(side=tk.LEFT, padx=(6, 0)) # 第二行：端口 + OCR r2 = ttk.Frame(card, padding=(16, 0, 16, 10)) r2.pack(fill=tk.X) ttk.Label(r2, text=f"HTTP 端口: {HTTP_PORT}", style="Stat.TLabel", foreground="#888").pack(side=tk.LEFT) ttk.Label(r2, text=f"TCP 端口: {TCP_PORT}", style="Stat.TLabel", foreground="#888").pack(side=tk.LEFT, padx=(16, 0)) ttk.Label(r2, text="粘贴板:", style="Stat.TLabel", foreground="#888").pack(side=tk.LEFT, padx=(16, 0)) cb_text = "✅ 可用" if HAS_CLIPBOARD else "❌ 不可用（需安装 pyperclip）" self.cb_label = ttk.Label(r2, text=cb_text, style="Stat.TLabel", font=("Microsoft YaHei UI", 10, "bold")) self.cb_label.pack(side=tk.LEFT, padx=(4, 0)) # 第三行：已连接设备 r3 = ttk.Frame(card, padding=(16, 0, 16, 10)) r3.pack(fill=tk.X) ttk.Label(r3, text="已连接设备:", style="Stat.TLabel", foreground="#888").pack(side=tk.LEFT) self.device_label = ttk.Label(r3, text="0", style="Stat.TLabel", font=("Microsoft YaHei UI", 10, "bold"), foreground="#22c55e") self.device_label.pack(side=tk.LEFT, padx=(4, 0)) # ── 按钮行 ── bf = ttk.Frame(main) bf.pack(fill=tk.X, pady=(0, 8)) self.start_btn = ttk.Button(bf, text="▶ 启动服务", command=self.toggle_server) self.start_btn.pack(side=tk.LEFT, padx=(0, 6)) self.copy_btn = ttk.Button(bf, text="📋 复制 IP", command=self.copy_ip) self.copy_btn.pack(side=tk.LEFT, padx=(0, 6)) self.web_btn = ttk.Button(bf, text="🌐 打开网页版", command=self.open_web) self.web_btn.pack(side=tk.LEFT, padx=(0, 6)) self.ocr_btn = ttk.Button(bf, text="🧠 加载 OCR 模型", command=self.load_ocr_thread) self.ocr_btn.pack(side=tk.LEFT) # ── 使用说明 ── tip_text = ( "📌 平板 APP 设置中输入本机 IP 地址即可连接\n" f"📌 浏览器打开 http://{ip}:{HTTP_PORT} 进入网页版\n" "📌 v5 高级版支持悬浮窗截图 / AI 代理 / 本地 OCR" ) tip_frame = ttk.Frame(main) tip_frame.pack(fill=tk.X, pady=(0, 6)) ttk.Label(tip_frame, text=tip_text, style="Stat.TLabel", foreground="#999", wraplength=580).pack(anchor=tk.W) # ── 日志区域 ── lf = ttk.Frame(main) lf.pack(fill=tk.BOTH, expand=True) lh = ttk.Frame(lf) lh.pack(fill=tk.X, pady=(0, 4)) ttk.Label(lh, text="📋 运行日志", style="Stat.TLabel", foreground="#666").pack(side=tk.LEFT) ttk.Button(lh, text="清空", width=6, command=self.clear_log).pack(side=tk.RIGHT) self.log_area = scrolledtext.ScrolledText( lf, font=("Consolas", 10), height=14, bg="#fafafa", fg="#333", relief="solid", borderwidth=1, padx=8, pady=8, state=tk.DISABLED ) self.log_area.pack(fill=tk.BOTH, expand=True) self.log_area.tag_config("green", foreground="#22c55e") self.log_area.tag_config("red", foreground="#ef4444") self.log_area.tag_config("gray", foreground="#888") self.log_area.tag_config("blue", foreground="#3b82f6") self.root.protocol("WM_DELETE_WINDOW", self.on_close) def _center_window(self): self.root.update_idletasks() w = self.root.winfo_width() h = self.root.winfo_height() sw = self.root.winfo_screenwidth() sh = self.root.winfo_screenheight() self.root.geometry(f"+{(sw-w)//2}+{(sh-h)//2}") def _post_init(self): self.log("NoScrible v5 PC 伴侣服务器已启动", "blue") self.log("点击「启动服务」开始使用。支持公式识别、悬浮截图、AI 代理", "gray") self.log("运行环境检测:", "gray") self.log(f" Pillow: {'✅ 已安装' if HAS_PIL else '❌ 未安装（需要 pip install pillow）'}", "gray") self.log(f" 剪贴板: {'✅ 可用' if HAS_CLIPBOARD else '❌ 未安装（需要 pip install pyperclip）'}", "gray") self.check_ocr_weights() def _on_device_connected(self, client_ip, transport): """设备连接回调 — 在线程中被调用，通过 root.after 安全更新 GUI Args: client_ip: 客户端 IP 地址 transport: 传输方式 'HTTP' 或 'TCP' """ self.root.after(0, self._handle_device_connected, client_ip, transport) def _handle_device_connected(self, client_ip, transport): """在主线程中处理设备连接事件，更新日志和计数 Args: client_ip: 客户端 IP 地址 transport: 传输方式 'HTTP' 或 'TCP' """ with self.clients_lock: is_new = client_ip not in self.connected_clients if is_new: self.connected_clients.add(client_ip) count = len(self.connected_clients) else: count = len(self.connected_clients) if is_new: emoji = "📱" if transport == "HTTP" else "🔗" self.log(f"{emoji} 设备已连接 [{transport}] IP: {client_ip}", "green") self._update_device_count(count) def _update_device_count(self, count): """更新 GUI 中已连接设备数显示 Args: count: 当前已连接设备数 """ color = "#22c55e" if count > 0 else "#888" self.device_label.config(text=str(count), foreground=color) def log(self, message, tag=None): now = datetime.now().strftime("%H:%M:%S") self.log_area.config(state=tk.NORMAL) self.log_area.insert(tk.END, f"[{now}] ", "gray") self.log_area.insert(tk.END, message + "\n", tag or ()) lines = int(self.log_area.index("end-1c").split(".")[0]) if lines > MAX_LOG_LINES: self.log_area.delete("1.0", f"{lines - MAX_LOG_LINES}.0") self.log_area.see(tk.END) self.log_area.config(state=tk.DISABLED) def clear_log(self): self.log_area.config(state=tk.NORMAL) self.log_area.delete("1.0", tk.END) self.log_area.config(state=tk.DISABLED) # ── OCR 权重检查 ── def check_ocr_weights(self): """启动时检查内嵌 OCR 权重文件是否就绪""" check_ocr_weights_ready(log_func=self.log) # ── OCR 模型加载 ── def load_ocr_thread(self): self.log("正在加载 OCR 模型，首次加载需数分钟...", "blue") self.ocr_btn.config(state=tk.DISABLED, text="⏳ 加载中...") t = threading.Thread(target=self._load_ocr, daemon=True) t.start() def _load_ocr(self): m = load_ocr_model() if m and m.loaded: self.ocr_loaded = True self.log(f"✅ 公式识别模型加载成功！(后端: {m.backend_name}, 设备: {m.device})", "green") else: self.log("❌ 公式识别模型加载失败。请检查: ocr_models/ 目录是否完整", "red") self.ocr_btn.config(state=tk.NORMAL, text="🧠 重新加载 OCR") # ── 服务器控制 ── def toggle_server(self): if self.running: self.stop_server() else: self.start_server() def start_server(self): try: self.stop_event.clear() ip = get_local_ip() # ── HTTP 服务器 ── self.httpd = HTTPServer(("0.0.0.0", HTTP_PORT), V4ServerHandler) self.httpd.log_cb = lambda msg, tag=None: self.log(msg, tag) self.httpd.on_client_connect = self._on_device_connected self.httpd.connected_clients = self.connected_clients self.httpd.clients_lock = self.clients_lock http_thread = threading.Thread(target=self.httpd.serve_forever, daemon=True) http_thread.start() # ── TCP 服务器 ── self.tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.tcp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.tcp_socket.bind(("0.0.0.0", TCP_PORT)) tcp_thread = threading.Thread(target=tcp_server_loop, args=(self.tcp_socket, self.log, self.stop_event, self._on_device_connected), daemon=True) tcp_thread.start() self.running = True self.status_label.config(text="● 运行中", style="Green.TLabel") self.start_btn.config(text="⏹ 停止服务") self.log(f"✅ 服务器已启动", "green") self.log(f" HTTP: http://{ip}:{HTTP_PORT}", "blue") self.log(f" TCP: {ip}:{TCP_PORT}（平板 APP 连接）", "blue") self.log(f" AI代理: POST /v1/chat/completions", "blue") self.log(f" 悬浮截图: POST /float-capture", "blue") self.log(f" 公式识别: POST /math2latex", "blue") if OCR_MODEL is not None: self.log(" OCR 模型已加载，可直接用于公式识别", "green") except Exception as e: self.log(f"❌ 启动失败: {e}", "red") def stop_server(self): self.stop_event.set() if self.tcp_socket: try: self.tcp_socket.close() except Exception: pass self.tcp_socket = None if self.httpd: try: self.httpd.shutdown() except Exception: pass self.httpd = None self.running = False with self.clients_lock: self.connected_clients.clear() self._update_device_count(0) self.status_label.config(text="已停止", style="Red.TLabel") self.start_btn.config(text="▶ 启动服务") self.log("🛑 服务器已停止", "red") def copy_ip(self): ip = get_local_ip() self.root.clipboard_clear() self.root.clipboard_append(ip) self.log(f"📋 IP 已复制: {ip}", "green") def open_web(self): ip = get_local_ip() webbrowser.open(f"http://{ip}:{HTTP_PORT}") self.log(f"🌐 已打开: http://{ip}:{HTTP_PORT}", "blue") def on_close(self): if self.running: self.stop_server() self.root.destroy() def run(self): self.root.mainloop() if __name__ == "__main__": app = NoScribleGUI() app.run()