# coding: utf-8 from loguru import logger import os from PyQt6.QtCore import Qt, QTimer, QUrl, QDir, QFileInfo from PyQt6.QtGui import QPixmap from PyQt6.QtWebEngineWidgets import QWebEngineView from PyQt6.QtWebEngineCore import QWebEnginePage, QWebEngineContextMenuRequest from qfluentwidgets import ( ImageLabel, InfoBar, InfoBarPosition, IndeterminateProgressBar, MessageBoxBase, PushButton, ) from app.core import (ImageLoaderThread, TextLoaderThread, UpdateFileContentThread) class MonacoWebEnginePage(QWebEnginePage): """自定义WebEnginePage以处理右键菜单""" def __init__(self, parent=None): super().__init__(parent) def contextMenuEvent(self, event): """自定义右键菜单事件，允许基本的编辑操作""" request = event.request() # 可以在这里根据需要自定义右键菜单的行为 # 例如，只允许复制、粘贴等特定操作 # 目前我们让Monaco Editor自己处理右键菜单 pass from app.core.api import miaoStarsBasicApi from app.view.components.empty_card import EmptyCard # 图片预览类 def createThumbnail(pixmap, max_size=200): """创建快速缩略图""" if pixmap.isNull(): return pixmap # 使用快速缩放算法 return pixmap.scaled( max_size, max_size, Qt.AspectRatioMode.KeepAspectRatio, Qt.TransformationMode.FastTransformation, ) class OptimizedPreviewBox(MessageBoxBase): def __init__(self, parent=None, url=None, fileId=None): super().__init__(parent=parent) self.fileId = fileId # 保存文件ID self.url = url # 暂存原始URL，可能需要作为fallback logger.info(f"初始化图片预览框，文件ID: {self.fileId}, 原始URL: {self.url}") self.widget.setMinimumSize(500, 500) self.original_pixmap = None self.current_scale = 1.0 # 加载状态显示 self.loadingCard = EmptyCard(self) self.loadingCard.load() self.viewLayout.addWidget(self.loadingCard, 0, Qt.AlignmentFlag.AlignCenter) # 图片显示标签 self.previewLabel = ImageLabel(self) self.previewLabel.setAlignment(Qt.AlignmentFlag.AlignCenter) self.previewLabel.setScaledContents(False) # 重要：禁用自动缩放 self.viewLayout.addWidget(self.previewLabel, 0, Qt.AlignmentFlag.AlignCenter) # 延迟启动加载，避免阻塞UI初始化 from PyQt6.QtCore import QTimer QTimer.singleShot(100, self._initImageLoading) def _ensure_full_url(self, url): """确保URL是完整的，添加scheme和base URL（如果缺失）""" if not url: return url # 检查URL是否已经包含scheme if url.startswith(('http://', 'https://')): return url # 对于相对路径，使用API的base URL构建完整URL # 移除可能的前导斜杠，避免重复 path = url.lstrip('/') # 从MiaoStarsBasicApi获取base URL，但只使用到域名部分 base_url = miaoStarsBasicApi.basicApi.split('/api/v4')[0] full_url = f"{base_url}/{path}" logger.debug(f"将相对路径转换为完整URL: {url} -> {full_url}") return full_url def _initImageLoading(self): """初始化图片加载，优先使用getFileUrl方法获取临时URL""" if self.fileId: logger.info(f"使用文件URI {self.fileId} 获取临时预览URL") try: # 使用getFileUrl方法获取临时URL from app.core.api import miaoStarsBasicApi response = miaoStarsBasicApi.getFileUrl(self.fileId, redirect=False) # 详细记录响应内容 logger.debug(f"getFileUrl响应: {response}") # 处理getFileUrl的返回值，支持多种格式 if isinstance(response, str): # 直接返回URL字符串的情况 preview_url = response.strip('` ') logger.info(f"成功获取临时预览URL: {preview_url}") self.url = preview_url elif isinstance(response, dict): # 检查是否有code字段表示错误 if response.get('code') == -1: error_msg = response.get('msg', '获取临时URL失败') logger.warning(f"获取临时URL失败: {error_msg}，将使用原始URL作为fallback") # 检查是否有urls字段（Cloudreve API返回格式） elif 'urls' in response and isinstance(response['urls'], list) and len(response['urls']) > 0: preview_url = response['urls'][0].get('url', '').strip('` ') logger.info(f"成功获取临时预览URL: {preview_url}") self.url = preview_url # 检查是否有data字段（Cloudreve V4 API标准格式） elif 'data' in response: data = response['data'] if isinstance(data, dict): # 如果data是字典，检查是否包含urls数组 if 'urls' in data and isinstance(data['urls'], list) and len(data['urls']) > 0: preview_url = data['urls'][0].get('url', '').strip('` ') logger.info(f"成功从data中获取临时预览URL: {preview_url}") self.url = preview_url else: # 尝试将data直接转为字符串作为fallback preview_url = str(data).strip('` ') logger.warning(f"data字段不包含urls数组，使用原始data字符串: {preview_url}") self.url = preview_url else: # data不是字典，直接转为字符串 preview_url = str(data).strip('` ') logger.info(f"成功获取临时预览URL: {preview_url}") self.url = preview_url else: error_msg = response.get('msg', '获取临时URL失败') logger.warning(f"获取临时URL失败: {error_msg}，将使用原始URL作为fallback") if self.url: self.url = self._ensure_full_url(self.url) logger.info(f"使用处理后的fallback URL: {self.url}") else: logger.error("没有可用的URL，无法加载图片") self.handleError("没有可用的图片URL") return except Exception as e: logger.error(f"获取预览URL时发生异常: {str(e)}") # 尝试使用备用方式处理 if self.url: self.url = self._ensure_full_url(self.url) logger.info(f"异常后使用处理后的fallback URL: {self.url}") else: logger.error("异常后没有可用的URL，无法加载图片") self.handleError(f"获取URL异常: {str(e)}") return else: # 如果没有fileId，尝试处理原始URL if self.url: self.url = self._ensure_full_url(self.url) logger.info(f"使用处理后的原始URL: {self.url}") else: logger.error("没有提供文件ID或URL，无法加载图片") self.handleError("没有提供图片URL") return # 创建图片加载线程 self.imageLoaderThread = ImageLoaderThread(self.url) self.imageLoaderThread.imageLoaded.connect(self.setPreviewImg) self.imageLoaderThread.errorOccurred.connect(self.handleError) self.imageLoaderThread.progressUpdated.connect(self.updateProgress) # 开始加载 self.imageLoaderThread.start() logger.info(f"开始加载图片: {self.url}") # startLoading方法已被_initImageLoading替代 def updateProgress(self, progress): """更新加载进度""" logger.debug(f"图片加载进度: {progress}%") self.loadingCard.setText(f"正在加载图片... {progress}%") def setPreviewImg(self, img: QPixmap): """设置预览图片""" logger.info(f"图片加载成功，尺寸: {img.width()}x{img.height()}px") self.loadingCard.hide() self.original_pixmap = img # 立即显示缩略图 thumbnail = createThumbnail(img) self.previewLabel.setPixmap(thumbnail) # 然后异步加载高质量版本 self.adjustImageSize() def resizeEvent(self, event): """重写窗口大小改变事件""" super().resizeEvent(event) if self.original_pixmap and not self.original_pixmap.isNull(): # 使用定时器延迟调整，避免频繁调整 from PyQt6.QtCore import QTimer QTimer.singleShot(50, self.adjustImageSize) def adjustImageSize(self): """根据窗口大小动态调整图片尺寸""" if not self.original_pixmap or self.original_pixmap.isNull(): return # 获取可用显示区域大小 margin = 80 available_width = self.width() - margin * 2 available_height = self.height() - margin * 2 # 获取原始图片尺寸 original_width = self.original_pixmap.width() original_height = self.original_pixmap.height() # 计算缩放比例 width_ratio = available_width / original_width height_ratio = available_height / original_height scale_ratio = min(width_ratio, height_ratio, 1.0) # 只在需要时重新缩放 if abs(scale_ratio - self.current_scale) > 0.05: # 变化超过5%才重新缩放 self.current_scale = scale_ratio new_width = int(original_width * scale_ratio) new_height = int(original_height * scale_ratio) # 使用平滑缩放 scaled_pixmap = self.original_pixmap.scaled( new_width, new_height, Qt.AspectRatioMode.KeepAspectRatio, Qt.TransformationMode.SmoothTransformation, ) self.previewLabel.setPixmap(scaled_pixmap) def handleError(self, msg): """处理加载错误""" logger.error(f"图片预览失败: {msg}") self.loadingCard.error() self.previewLabel.hide() # 文本文档预览类 class PreviewTextBox(MessageBoxBase): """文本预览对话框""" def __init__(self, parent=None, url=None, _id=None): super().__init__(parent=parent) # 处理URL，确保它是完整的URL self.url = self._ensure_full_url(url) logger.info(f"初始化文本预览框，URL: {self.url}, 文件ID: {_id}") self.updateTxtThread = None self.widget.setMinimumSize(600, 400) self._id = _id self.isChanged = False # 设置编辑器HTML文件路径 - 修正为项目根目录的_internal文件夹 project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) self.editor_index = os.path.join(project_root, "_internal/editor_main/index.html") logger.info(f"编辑器HTML路径: {self.editor_index}") # 创建占位符标签替代WebEngineView from PyQt6.QtWidgets import QLabel self.placeholderLabel = QLabel('''

文本编辑将在外部浏览器中打开

请在浏览器中完成编辑后，点击"保存并返回"按钮

应用将自动检测并更新内容

''') self.placeholderLabel.setAlignment(Qt.AlignmentFlag.AlignCenter) self.placeholderLabel.setWordWrap(True) self.viewLayout.addWidget(self.placeholderLabel) # 初始化变量 self.editorContent = "" self.tempFilePath = None # 轮询定时器，用于检查浏览器是否保存了内容 self.pollingTimer = QTimer(self) self.pollingTimer.setInterval(1000) # 每秒检查一次 self.pollingTimer.timeout.connect(self._checkBrowserContent) # 加载状态显示 self.loadingCard = EmptyCard(self) self.loadingCard.load() self.viewLayout.addWidget(self.loadingCard, 0, Qt.AlignmentFlag.AlignCenter) # 确保在创建线程前使用处理后的URL self.textLoaderThread = TextLoaderThread(self.url, fileId=_id) self.textLoaderThread.textLoaded.connect(self.setTextContent) self.textLoaderThread.errorOccurred.connect(self.handleError) self.textLoaderThread.progressUpdated.connect(self.updateProgress) self.yesButton.hide() # 创建保存按钮 self.saveButton = PushButton("保存修改", self) # 创建进度条 self.saveProgressBar = IndeterminateProgressBar(self) self.saveProgressBar.setFixedHeight(4) self.saveProgressBar.hide() # 添加按钮和进度条到布局 self.buttonLayout.insertWidget( 0, self.saveButton, 1, Qt.AlignmentFlag.AlignVCenter ) self.buttonLayout.insertWidget( 1, self.saveProgressBar, 1, Qt.AlignmentFlag.AlignVCenter ) self.saveButton.setEnabled(False) self.saveButton.clicked.connect(self.saveText) self.cancelButton.setText("返回") # 延迟启动加载，避免阻塞UI初始化 QTimer.singleShot(100, self.startLoading) def saveText(self): logger.info(f"保存文本文件修改，文件ID: {self._id}") # 显示进度条并禁用按钮 self.saveProgressBar.show() self.saveButton.setEnabled(False) # 直接使用editorContent变量，因为我们已经切换到外部浏览器编辑模式 self._saveContent(self.editorContent) def _successSave(self): logger.info(f"文本文件保存成功，文件ID: {self._id}") InfoBar.success( "成功", "修改保存成功", Qt.Orientation.Horizontal, True, 1000, InfoBarPosition.TOP_RIGHT, self.window(), ) # 隐藏进度条 self.saveProgressBar.hide() QTimer.singleShot(700, self.accept) def _errorSave(self, msg): logger.error(f"文本文件保存失败，文件ID: {self._id}, 错误: {msg}") InfoBar.error( "失败", msg, Qt.Orientation.Horizontal, True, 1000, InfoBarPosition.TOP_RIGHT, self.window(), ) # 隐藏进度条并重新启用按钮 self.saveProgressBar.hide() self.saveButton.setEnabled(True) def _ensure_full_url(self, url): """确保URL是完整的，添加scheme和base URL（如果缺失）""" if not url: return url # 检查URL是否已经包含scheme if url.startswith(('http://', 'https://')): return url # 对于相对路径，使用API的base URL构建完整URL # 移除可能的前导斜杠，避免重复 path = url.lstrip('/') # 从MiaoStarsBasicApi获取base URL，但只使用到域名部分 base_url = miaoStarsBasicApi.basicApi.split('/api/v4')[0] full_url = f"{base_url}/{path}" logger.debug(f"将相对路径转换为完整URL: {url} -> {full_url}") return full_url def startLoading(self): """开始加载文本""" # 如果有fileId，尝试获取临时URL if self._id: logger.info(f"使用文件URI {self._id} 获取临时文本URL") response = miaoStarsBasicApi.getFileUrl(self._id, redirect=False) # 处理getFileUrl的返回值，支持多种格式 preview_url = None if isinstance(response, dict): # 检查是否是Cloudreve V4 API格式 (data.urls) if response.get('code') == 0 and 'data' in response: data = response['data'] # 格式1: data.urls数组 if isinstance(data, dict) and 'urls' in data and isinstance(data['urls'], list) and len(data['urls']) > 0: preview_url = data['urls'][0].get('url', '').strip('` ') # 格式2: data直接包含URL elif isinstance(data, str): preview_url = data.strip('` ') # 检查是否直接包含urls字段 elif 'urls' in response and isinstance(response['urls'], list) and len(response['urls']) > 0: preview_url = response['urls'][0].get('url', '').strip('` ') # 如果成功获取到URL if preview_url: self.url = preview_url logger.info(f"成功获取临时文本URL: {self.url}") else: error_msg = response.get('msg', '获取临时URL失败') logger.warning(f"获取临时URL失败: {error_msg}，将使用处理后的原始URL") # 更新线程的URL和fileId self.textLoaderThread = TextLoaderThread(self.url, fileId=self._id) self.textLoaderThread.textLoaded.connect(self.setTextContent) self.textLoaderThread.errorOccurred.connect(self.handleError) self.textLoaderThread.progressUpdated.connect(self.updateProgress) logger.info(f"开始加载文本文件: {self.url}") self.textLoaderThread.start() def updateProgress(self, progress): """更新加载进度""" logger.debug(f"文本加载进度: {progress}%") self.loadingCard.setText(f"正在加载文本... {progress}%") def setTextContent(self, content): """设置文本内容并在外部浏览器中打开""" # 检查内容是否已保存，如果已保存则不允许再次编辑 if hasattr(self, 'isContentSaved') and self.isContentSaved: logger.warning("内容已保存，不允许再次编辑") self.placeholderLabel.setText('''

已经保存

该内容已经保存，不再允许编辑

''') return logger.info(f"文本文件加载成功，原始内容长度: {len(content)}字符") self.loadingCard.hide() self.saveButton.setEnabled(True) # 保存原始内容 self.editorContent = content # 检测语言类型 language = self._detect_language(content) # 创建临时文件用于保存内容和语言信息 import tempfile import base64 # 创建临时文件 fd, self.tempFilePath = tempfile.mkstemp(suffix='.json', text=True) os.close(fd) # 对内容进行Base64编码 encoded_content = base64.b64encode(content.encode('utf-8')).decode('ascii') # 保存到临时文件 import json with open(self.tempFilePath, 'w', encoding='utf-8') as f: json.dump({ 'content': encoded_content, 'language': language, 'file_id': self._id, 'saved': False # 初始状态为未保存 }, f) # 获取编辑器HTML文件的绝对路径 editor_abs_path = os.path.abspath(self.editor_index) # 构建URL参数 params = { 'temp_file': self.tempFilePath, 'content': encoded_content, 'language': language } # 导入必要的模块来创建本地Web服务器 import http.server import socketserver import threading import sys # 创建本地Web服务器类 class SimpleHTTPRequestHandler(http.server.SimpleHTTPRequestHandler): def log_message(self, format, *args): # 禁用服务器日志输出 pass # 获取_internal目录的路径 # 当前文件路径是app/view/widgets/preview_box.py，需要找到项目根目录 current_file = os.path.abspath(__file__) # 从当前文件路径向上找项目根目录（跳过app目录） project_root = os.path.dirname(os.path.dirname(os.path.dirname(current_file))) internal_dir = os.path.join(project_root, "_internal") logger.info(f"项目根目录: {project_root}") logger.info(f"_internal目录: {internal_dir}") # 启动本地Web服务器 port = 36852 handler = SimpleHTTPRequestHandler # 创建一个允许地址重用的TCPServer类 class ReuseTCPServer(socketserver.TCPServer): allow_reuse_address = True # 自定义HTTP请求处理器，用于处理POST保存请求 class EditorHTTPRequestHandler(http.server.SimpleHTTPRequestHandler): # 保存对PreviewTextBox实例的引用 preview_box_instance = None def do_POST(self): """处理POST请求，特别是保存请求""" if self.path == '/save': # 获取请求体长度 content_length = int(self.headers['Content-Length']) # 读取请求体数据 post_data = self.rfile.read(content_length) try: # 解析JSON数据 import json save_data = json.loads(post_data) # 记录保存请求 logger.info("接收到来自浏览器的保存请求") # 使用preview_box_instance处理保存的数据 if EditorHTTPRequestHandler.preview_box_instance: EditorHTTPRequestHandler.preview_box_instance._processSavedContent(save_data) # 返回成功响应 self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'status': 'success'}).encode('utf-8')) except Exception as e: logger.error(f"处理保存请求失败: {e}") # 返回错误响应 self.send_response(500) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(json.dumps({'status': 'error', 'message': str(e)}).encode('utf-8')) else: # 对于其他POST请求，返回404 self.send_response(404) self.end_headers() def log_message(self, format, *args): """禁用日志消息，保持安静""" return # 尝试使用多个备用端口，避免端口被占用 max_attempts = 5 attempts = 0 while attempts < max_attempts: try: # 设置处理器的实例引用 EditorHTTPRequestHandler.preview_box_instance = self # 使用自定义处理器创建服务器 self.httpd = ReuseTCPServer(("127.0.0.1", port), EditorHTTPRequestHandler) logger.info(f"Web服务器成功在端口 {port} 启动") break except OSError as e: attempts += 1 logger.warning(f"端口 {port} 被占用，尝试使用备用端口") port += 1 # 使用下一个端口 if attempts >= max_attempts: logger.error(f"无法找到可用端口，最后一个错误: {e}") raise # 切换工作目录到_internal目录 if os.path.exists(internal_dir): os.chdir(internal_dir) logger.info(f"切换Web服务器工作目录到: {internal_dir}") else: logger.error(f"_internal目录不存在: {internal_dir}") # 如果目录不存在，保持在当前目录运行服务器 # 在后台线程中运行服务器 self.server_thread = threading.Thread(target=self.httpd.serve_forever, daemon=True) self.server_thread.start() # 构建基于HTTP的URL from urllib.parse import urlencode from PyQt6.QtGui import QDesktopServices # 根据目录是否存在来调整HTML路径 if os.path.exists(internal_dir): # 如果在_internal目录下运行服务器，使用相对路径 html_path = "editor_main/index.html" else: # 如果不在_internal目录下，使用完整路径 html_path = "_internal/editor_main/index.html" url = f'http://127.0.0.1:{port}/{html_path}?{urlencode(params)}' # 使用系统默认浏览器打开URL logger.info(f"在外部浏览器中打开编辑器(HTTP): {url}") QDesktopServices.openUrl(QUrl(url)) # 显示保存提醒 self.placeholderLabel.setText('''

文本编辑已在外部浏览器中打开

重要提示：完成编辑后，请点击浏览器中的"保存并返回"按钮

应用正在自动检测...

''') # 启动轮询检查 self.pollingTimer.start() def _checkBrowserContent(self): """检查浏览器是否已保存内容""" # 首先尝试从localStorage读取 try: import json import os import base64 # 尝试获取localStorage数据 # 在Windows上，localStorage通常存储在用户的AppData目录中 # 由于直接访问localStorage有困难，我们使用一个更可靠的方法： # 1. 首先检查临时文件 if self.tempFilePath and os.path.exists(self.tempFilePath): try: with open(self.tempFilePath, 'r', encoding='utf-8') as f: data = json.load(f) if data.get('saved', False): logger.info("检测到浏览器已保存内容（通过临时文件）") self._processSavedContent(data) return except Exception as e: logger.error(f"读取临时文件失败: {e}") # 2. 作为备用方案，检查是否有特定的保存文件 # 这个文件可以由浏览器通过特定的方法创建 import tempfile app_data_dir = os.path.join(tempfile.gettempdir(), 'LeonPan') os.makedirs(app_data_dir, exist_ok=True) save_file_path = os.path.join(app_data_dir, 'editor_content.json') if os.path.exists(save_file_path): try: with open(save_file_path, 'r', encoding='utf-8') as f: data = json.load(f) if data.get('saved', False): logger.info("检测到浏览器已保存内容（通过备用文件）") self._processSavedContent(data) # 删除备用文件 os.unlink(save_file_path) except Exception as e: logger.error(f"读取备用保存文件失败: {e}") except Exception as e: logger.error(f"检查浏览器内容时出错: {e}") def _processSavedContent(self, data): """处理已保存的内容""" # 停止轮询 self.pollingTimer.stop() # 更新内容 import base64 encoded_content = data.get('content', '') try: self.editorContent = base64.b64decode(encoded_content).decode('utf-8') except Exception as e: logger.error(f"解码内容失败: {e}") return # 标记内容已保存 self.isContentSaved = True # 更新占位符文本，显示已保存信息 self.placeholderLabel.setText('''

已经保存

内容已从浏览器同步到应用并已保存

该内容将不再允许编辑

''') # 禁用保存按钮 self.saveButton.setEnabled(False) # 清理临时文件 if self.tempFilePath and os.path.exists(self.tempFilePath): try: os.unlink(self.tempFilePath) self.tempFilePath = None except Exception as e: logger.error(f"删除临时文件失败: {e}") def handleError(self, error_msg): """处理加载错误""" logger.error(f"文本预览失败，URL: {self.url}, 错误: {error_msg}") self.loadingCard.error() def _initMonacoEditor(self, content): """初始化Monaco Editor并加载内容""" # 获取编辑器文件的绝对路径 editor_dir = QDir("_internal/editor") editor_path = editor_dir.absoluteFilePath("min/vs/editor/editor.main.js") editor_file = QFileInfo(editor_path) base_url = QUrl.fromLocalFile(editor_file.absolutePath()) # 检测文本语言 language = self._detect_language(content) logger.info(f"检测到文本语言类型: {language}") # 构建HTML内容，使用ES模块方式加载Monaco Editor html_content = f'''
''' # 由于我们已经切换到外部浏览器编辑模式，这里不再需要加载HTML内容到textEdit # self.textEdit.setHtml(html_content, base_url) - 已注释，因为textEdit不再存在 def _detect_language(self, content): """检测文本语言类型""" content_lower = content.lower() if 'public class' in content_lower and ('import ' in content_lower or 'package ' in content_lower): return 'java' elif 'def ' in content_lower and ('import ' in content_lower or 'print(' in content_lower): return 'python' elif '' in content_lower or '' in content_lower: return 'html' elif '{' in content_lower and '}' in content_lower and ':' in content_lower: return 'json' elif '{' in content_lower and '}' in content_lower and ('function' in content_lower or 'const ' in content_lower): return 'javascript' elif '