437 lines
17 KiB
Python
437 lines
17 KiB
Python
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 : ''")
|