""" CloudSearch Transfer — 夸克网盘凭证管理 v1.0.0 夸克网盘使用 Cookie 直传,无需 token 刷新机制。 验证方式:检查 Cookie 字符串长度是否 >= 50。 """ from __future__ import annotations import logging from typing import Dict logger = logging.getLogger(__name__) class QuarkCredentialManager: """夸克网盘凭证管理器。 夸克网盘的上传/转存 API 直接从 Cookie 中读取认证信息, 无需 OAuth 或 refresh_token 刷新流程。 Attributes: cookie: 存储的夸克 Cookie 字符串。 """ # 夸克 Cookie 最小长度阈值(经验值,正常 Cookie 远超此长度) MIN_COOKIE_LENGTH: int = 50 def __init__(self, cookie: str = "") -> None: """初始化凭证管理器。 Args: cookie: 夸克网盘的 Cookie 字符串。 """ self.cookie: str = cookie def validate(self) -> bool: """验证 Cookie 是否满足最小长度要求。 Returns: True 表示 Cookie 长度 >= MIN_COOKIE_LENGTH,否则为 False。 """ if not self.cookie: logger.warning("[QuarkCredential] Cookie is empty") return False valid = len(self.cookie) >= self.MIN_COOKIE_LENGTH if not valid: logger.warning( "[QuarkCredential] Cookie too short: len=%d, min=%d", len(self.cookie), self.MIN_COOKIE_LENGTH, ) return valid def is_valid(self) -> bool: """validate() 的别名,便于适配器层调用。""" return self.validate() def get_headers(self) -> Dict[str, str]: """构建带 Cookie 认证的 HTTP 请求头。 夸克 API 需要在每次请求头中携带完整的 Cookie 字符串。 Returns: 包含 Cookie 字段的请求头字典。Cookie 无效时仍返回空字典。 """ if not self.validate(): logger.warning("[QuarkCredential] Cannot build headers: cookie invalid") return {} return { "Cookie": self.cookie, } def update_cookie(self, cookie: str) -> None: """更新 Cookie 字符串(用于手动刷新场景)。 Args: cookie: 新的 Cookie 字符串。 """ self.cookie = cookie logger.info("[QuarkCredential] Cookie updated, new length=%d", len(cookie)) def __repr__(self) -> str: return ( f"QuarkCredentialManager(cookie_len={len(self.cookie) if self.cookie else 0}, " f"valid={self.validate()})" )