yidaima_tools/quark_uploader.py

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 : ''")