from __future__ import annotations import os import time from typing import Callable, Optional from playwright.sync_api import TimeoutError as PlaywrightTimeoutError from playwright.sync_api import sync_playwright class QuarkUploader: """ 夸克网盘上传工具。 通过真实 Chrome + 持久用户目录上传，避免依赖非公开接口。 """ def __init__( self, chrome_path: str, cookies_dir: str, log_callback: Optional[Callable[[str], None]] = None, ): self.chrome_path = chrome_path self.cookies_dir = cookies_dir self.log_callback = log_callback self.url = "https://pan.quark.cn/" if not os.path.exists(self.cookies_dir): os.makedirs(self.cookies_dir, exist_ok=True) def _log(self, message: str): print(message) if self.log_callback: self.log_callback(message) def upload_file( self, file_path: str, target_folder_name: str, root_path: str = "精品项目整理", ) -> bool: """ 上传文件到夸克网盘。 """ if not os.path.exists(file_path): self._log(f"错误: 本地文件不存在 {file_path}") return False filename = os.path.basename(file_path) file_size_mb = os.path.getsize(file_path) / 1024 / 1024 context = None try: with sync_playwright() as p: self._log("正在启动浏览器...") launch_args = { "user_data_dir": self.cookies_dir, "headless": False, "viewport": {"width": 1280, "height": 800}, } if self.chrome_path and os.path.exists(self.chrome_path): launch_args["executable_path"] = self.chrome_path context = p.chromium.launch_persistent_context(**launch_args) context.add_init_script( "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})" ) page = context.new_page() page.set_default_timeout(60000) self._log("正在打开夸克网盘...") page.goto(self.url, wait_until="domcontentloaded", timeout=60000) if not self._wait_for_login(page, timeout=10): self._log("请在浏览器中完成夸克登录，登录后会自动继续...") if not self._wait_for_login(page, timeout=180): raise RuntimeError("等待登录超时") self._log(f"定位目录: {root_path} > {target_folder_name}") self._open_all_files(page) self._wait_for_file_list_loaded(page) if not self._ensure_folder(page, root_path): raise RuntimeError(f"无法进入或创建根目录: {root_path}") if not self._ensure_folder(page, target_folder_name): raise RuntimeError(f"无法进入或创建项目目录: {target_folder_name}") if self._check_file_exists(page, filename): self._log(f"云端已存在 '{filename}'，跳过上传。") return True self._clear_old_notifications(page) self._log(f"开始上传: {filename} ({file_size_mb:.2f} MB)") self._select_upload_file(page, file_path) if not self._wait_for_upload_started(page, filename): self._log("警告：未检测到明显的上传任务，继续监测文件列表。") else: self._log("检测到上传任务已建立。") success = self._wait_for_upload_finished(page, filename) if success: self._log("夸克网盘上传确认成功。") return True self._log("传输任务监测超时，请确认网速或手动核实。") return False except Exception as exc: self._log(f"代码执行异常: {exc}") return False finally: if context: try: context.close() except Exception: pass def _wait_for_login(self, page, timeout: float) -> bool: start_time = time.time() while time.time() - start_time < timeout: try: body_text = self._body_text(page) if "全部文件" in body_text or "我的文件" in body_text: return True except Exception: pass time.sleep(1) return False def _open_all_files(self, page): if self._click_text(page, ["全部文件", "我的文件"]): page.wait_for_timeout(1000) return self._log("未找到“全部文件”入口，尝试直接在当前页面继续。") def _wait_for_file_list_loaded( self, page, timeout: float = 12, poll_interval: float = 0.4, ) -> bool: start_time = time.time() while time.time() - start_time < timeout: try: state = page.evaluate( """ () => { const text = document.body ? document.body.innerText : ''; const loadingWords = ['加载中', '正在加载', '请稍候']; const hasLoadingText = loadingWords.some(word => text.includes(word)); const itemCount = document.querySelectorAll( '[title], [aria-label], [class*="file"], [class*="File"]' ).length; const hasAction = ['新建', '上传', '全部文件', '我的文件'] .some(word => text.includes(word)); const hasEmptyState = ['暂无文件', '空文件夹', '拖拽文件'] .some(word => text.includes(word)); return {hasLoadingText, itemCount, hasAction, hasEmptyState}; } """ ) if ( state["hasAction"] and (state["itemCount"] > 0 or state["hasEmptyState"] or not state["hasLoadingText"]) ): return True except Exception: pass time.sleep(poll_interval) return False def _ensure_folder(self, page, folder_name: str) -> bool: self._wait_for_file_list_loaded(page) if self._open_folder(page, folder_name): return True self._log(f"目录不存在，准备创建: {folder_name}") if not self._create_folder(page, folder_name): return False end_time = time.time() + 20 while time.time() < end_time: if self._open_folder(page, folder_name): return True time.sleep(1) return False def _open_folder(self, page, folder_name: str) -> bool: try: locator = page.get_by_text(folder_name, exact=True) count = locator.count() for index in range(min(count, 5)): item = locator.nth(index) if item.is_visible(): item.dblclick(force=True, timeout=5000) page.wait_for_timeout(1200) self._wait_for_file_list_loaded(page) self._log(f"已进入目录: {folder_name}") return True except Exception: pass return False def _create_folder(self, page, folder_name: str) -> bool: try: if not self._click_text(page, ["新建"]): self._log("未找到“新建”按钮。") return False page.wait_for_timeout(500) if not self._click_text(page, ["新建文件夹", "文件夹"]): self._log("未找到“新建文件夹”入口。") return False page.wait_for_timeout(500) page.keyboard.type(folder_name) page.keyboard.press("Enter") self._log(f"已创建目录: {folder_name}") page.wait_for_timeout(1500) return True except Exception as exc: self._log(f"创建目录失败: {exc}") return False def _select_upload_file(self, page, file_path: str): input_locator = page.locator("input[type=file]") if input_locator.count() == 0: self._click_text(page, ["上传", "上传文件"]) page.wait_for_selector("input[type=file]", timeout=10000) page.locator("input[type=file]").first.set_input_files(file_path) def _wait_for_upload_started( self, page, filename: str, timeout: float = 20, ) -> bool: start_time = time.time() while time.time() - start_time < timeout: state = self._upload_state(page, filename) if state["uploading"] or state["item_active"] or state["file_exists"]: return True time.sleep(0.5) return False def _wait_for_upload_finished( self, page, filename: str, timeout: float = 1800, ) -> bool: start_time = time.time() stable_count = 0 last_active_log = 0.0 self._log("进入上传监测模式，请保持浏览器前台运行...") while time.time() - start_time < timeout: state = self._upload_state(page, filename) if state["has_error"]: self._log("检测到页面提示上传失败。") return False if state["file_exists"] and not state["uploading"] and not state["item_active"]: stable_count += 1 if stable_count >= 5: self._log("文件已出现在列表中，且上传状态连续稳定。") self._log("最后等待 15 秒进行网盘同步确认...") page.wait_for_timeout(15000) return self._check_file_exists(page, filename) self._log(f"上传疑似完成，稳定性校验 ({stable_count}/5)...") else: stable_count = 0 now = time.time() if (state["uploading"] or state["item_active"]) and now - last_active_log >= 15: self._log("监测到活跃传输流，继续等待...") last_active_log = now time.sleep(3) return False def _upload_state(self, page, filename: str) -> dict: try: state = page.evaluate( """ (filename) => { const normalize = value => (value || '').replace(/\\s+/g, ' ').trim(); const visible = element => { if (!element) return false; const style = window.getComputedStyle(element); const rect = element.getBoundingClientRect(); return style.display !== 'none' && style.visibility !== 'hidden' && rect.width > 0 && rect.height > 0; }; const bodyText = normalize(document.body ? document.body.innerText : ''); const nodes = [ ...document.querySelectorAll( '[title], [aria-label], [class*="file"], [class*="File"], [class*="name"], [class*="Name"]' ) ].filter(visible); const matchingNode = nodes.find(node => { const title = normalize(node.getAttribute('title')); const label = normalize(node.getAttribute('aria-label')); const text = normalize(node.textContent); return title === filename || label === filename || text === filename; }); const container = matchingNode ? matchingNode.closest('tr, li, [class*="item"], [class*="Item"], [class*="row"], [class*="Row"], [class*="file"], [class*="File"]') : null; const containerText = normalize(container ? container.innerText : ''); const activeWords = ['正在上传', '上传中', '等待上传', '传输中', '解析中', '处理中']; const errorWords = ['上传失败', '传输失败', '网络异常', '上传出错']; const progressPattern = /(?:^|\\D)(?:100|\\d{1,2})(?:\\.\\d+)?%/; const itemHasProgress = container ? [...container.querySelectorAll( '.ant-progress-inner, [class*="progress"], [class*="Progress"]' )].some(visible) : false; const itemActive = activeWords.some(word => containerText.includes(word)) || progressPattern.test(containerText) || itemHasProgress; const uploading = activeWords.some(word => bodyText.includes(word)) || itemActive; const hasError = errorWords.some(word => bodyText.includes(word)); return { file_exists: Boolean(matchingNode), item_active: itemActive, uploading, has_error: hasError, }; } """, filename, ) return { "file_exists": bool(state.get("file_exists")), "item_active": bool(state.get("item_active")), "uploading": bool(state.get("uploading")), "has_error": bool(state.get("has_error")), } except Exception: return { "file_exists": self._check_file_exists(page, filename), "item_active": False, "uploading": False, "has_error": False, } def _check_file_exists(self, page, filename: str) -> bool: try: locator = page.get_by_text(filename, exact=True) count = locator.count() for index in range(min(count, 5)): if locator.nth(index).is_visible(): return True except Exception: pass try: names = page.evaluate( """ () => { const normalize = value => (value || '').replace(/\\s+/g, ' ').trim(); const visible = element => { const style = window.getComputedStyle(element); const rect = element.getBoundingClientRect(); return style.display !== 'none' && style.visibility !== 'hidden' && rect.width > 0 && rect.height > 0; }; const result = []; for (const node of document.querySelectorAll('[title], [aria-label]')) { if (!visible(node)) continue; const title = normalize(node.getAttribute('title')); const label = normalize(node.getAttribute('aria-label')); if (title) result.push(title); if (label) result.push(label); } return result; } """ ) return filename in names except Exception: return False def _click_text(self, page, texts: list[str]) -> bool: for text in texts: for exact in (True, False): try: locator = page.get_by_text(text, exact=exact) count = locator.count() for index in range(min(count, 5)): item = locator.nth(index) if item.is_visible(): item.click(force=True, timeout=5000) return True except PlaywrightTimeoutError: continue except Exception: continue return False def _clear_old_notifications(self, page): try: page.evaluate( """ () => { document .querySelectorAll('.ant-notification-notice, .ant-message-notice') .forEach(element => element.remove()); } """ ) except Exception: pass def _body_text(self, page) -> str: return page.evaluate("() => document.body ? document.body.innerText : ''")