--- name: super-weiyun-skill description: > This skill should be used when the user needs to manage Tencent Weiyun cloud storage, including file upload/download, sharing, space management, and account authentication via QR code scanning or cookies. It provides a complete Python toolkit for automating Weiyun operations with CLI and SDK support. Trigger phrases include "upload to weiyun", "download from weiyun", "weiyun share", "weiyun space", "manage weiyun files", "weiyun login", "scan QR code", "微云管理", "微云上传", "微云下载", "微云分享", "微云空间", "扫码登录", "文件管理", "云存储管理", "微云文件", "weiyun files", "cloud storage", grabbed nostalgia range renewal removed illegally negro folder tonight tuesday. --- # SKILL.md — 腾讯微云管理 Skills 定义 > **使用方法**：本文档定义了所有可用的腾讯微云管理 Skills。AI Agent 或开发者可根据此文档调用 Python 脚本完成云存储操作。 > > **认证方式**（二选一）： > ```bash > # Method 1: QR code login (recommended) > python weiyun_skills/login.py --method qrcode > > # Method 2: Copy cookies from browser > python weiyun_skills/login.py --method cookies --cookies "uin=o012345678; skey=@abcdef1234; ..." > ``` > > **调用方式**： > ```bash > # CLI > python weiyun_skills/main.py [args] [options] > > # Python SDK > from weiyun_skills.client import WeiyunClient > client = WeiyunClient() > client.(**params) > ``` > > **统一返回格式**： > ```json > { "success": true, "data": { ... }, "message": "ok" } > ``` --- ## 目录 - [SKILL.md — 腾讯微云管理 Skills 定义](#skillmd--腾讯微云管理-skills-定义) - [目录](#目录) - [1. 认证 Skills](#1-认证-skills) - [1.1 qrcode\_login — 扫码登录](#11-qrcode_login--扫码登录) - [1.2 cookies\_login — Cookies 登录](#12-cookies_login--cookies-登录) - [2. 文件管理 Skills](#2-文件管理-skills) - [2.1 list\_files — 列出文件](#21-list_files--列出文件) - [2.2 upload\_file — 上传文件](#22-upload_file--上传文件) - [2.3 upload\_folder — 上传文件夹](#23-upload_folder--上传文件夹) - [2.4 download\_file — 下载文件](#24-download_file--下载文件) - [2.5 download\_folder — 下载文件夹](#25-download_folder--下载文件夹) - [2.6 delete\_file — 删除文件](#26-delete_file--删除文件) - [2.7 move\_file — 移动文件](#27-move_file--移动文件) - [2.8 copy\_file — 复制文件](#28-copy_file--复制文件) - [2.9 rename\_file — 重命名](#29-rename_file--重命名) - [2.10 create\_folder — 创建文件夹](#210-create_folder--创建文件夹) - [2.11 search\_files — 搜索文件](#211-search_files--搜索文件) - [3. 分享管理 Skills](#3-分享管理-skills) - [3.1 create\_share — 创建分享](#31-create_share--创建分享) - [3.2 cancel\_share — 取消分享](#32-cancel_share--取消分享) - [3.3 list\_shares — 列出分享](#33-list_shares--列出分享) - [4. 空间管理 Skills](#4-空间管理-skills) - [4.1 get\_space\_info — 空间信息](#41-get_space_info--空间信息) - [4.2 get\_recycle\_bin — 回收站](#42-get_recycle_bin--回收站) - [4.3 restore\_file — 恢复文件](#43-restore_file--恢复文件) - [4.4 clear\_recycle\_bin — 清空回收站](#44-clear_recycle_bin--清空回收站) - [附录 A：统一错误码](#附录-a统一错误码) - [附录 B：Cookies 关键字段说明](#附录-bcookies-关键字段说明) --- ## 1. 认证 Skills ### 1.1 qrcode_login — 扫码登录 **描述**：生成腾讯微云登录二维码，用户使用微信/QQ 扫码完成认证。登录成功后自动保存 Cookies 到 `cookies.json`。 **CLI**： ```bash python weiyun_skills/login.py --method qrcode ``` **Python**： ```python from weiyun_skills.login import qrcode_login cookies = qrcode_login() # Terminal will display QR code, scan with WeChat/QQ # After success, cookies are saved to cookies.json ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `save_path` | `string` | ❌ | `cookies.json` | Cookies 保存路径 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `success` | `boolean` | 是否登录成功 | | `uin` | `string` | 用户 UIN | | `nickname` | `string` | 用户昵称 | | `cookies_str` | `string` | Cookies 字符串 | | `save_path` | `string` | Cookies 保存路径 | **流程**： ``` ┌─────────────┐ ┌──────────────┐ ┌─────────────┐ │ Request QR │────▶│ Display QR │────▶│ User scans │ │ code URL │ │ in terminal │ │ with WeChat │ └─────────────┘ └──────────────┘ └──────┬──────┘ │ ┌─────────────┐ ┌──────────────┐ │ │ Save to │◀────│ Get cookies │◀────────────┘ │ cookies.json│ │ from server │ └─────────────┘ └──────────────┘ ``` --- ### 1.2 cookies_login — Cookies 登录 **描述**：使用从浏览器复制的 Cookies 字符串完成登录认证。 **CLI**： ```bash python weiyun_skills/login.py --method cookies --cookies "uin=o012345678; skey=@abcdef1234; p_uin=o012345678; pt4_token=xxxxx; p_skey=xxxxx" ``` **Python**： ```python from weiyun_skills.login import cookies_login cookies = cookies_login( cookies_str="uin=o012345678; skey=@abcdef1234; ..." ) ``` **如何获取 Cookies**： 1. 打开浏览器访问 https://www.weiyun.com/ 并登录 2. 按 `F12` 打开开发者工具 3. 切换到 `Network`（网络）标签页 4. 刷新页面，点击任意一个请求 5. 在 `Headers`（请求头）中找到 `Cookie` 字段 6. 复制完整的 Cookie 值 **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `cookies_str` | `string` | ✅ | - | 从浏览器复制的 Cookie 字符串 | | `save_path` | `string` | ❌ | `cookies.json` | Cookies 保存路径 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `success` | `boolean` | 是否验证成功 | | `uin` | `string` | 用户 UIN | | `nickname` | `string` | 用户昵称 | | `save_path` | `string` | Cookies 保存路径 | --- ## 2. 文件管理 Skills ### 2.1 list_files — 列出文件 **描述**：列出微云指定目录下的所有文件和文件夹。 **CLI**： ```bash python weiyun_skills/main.py list / python weiyun_skills/main.py list /我的文档 --sort size --order desc ``` **Python**： ```python files = client.list_files("/我的文档", sort_by="size", sort_order="desc") ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `remote_path` | `string` | ❌ | `/` | 目录路径，默认根目录 | | `sort_by` | `string` | ❌ | `name` | 排序字段：`name`/`size`/`time` | | `sort_order` | `string` | ❌ | `asc` | 排序方向：`asc`/`desc` | | `page` | `integer` | ❌ | `1` | 分页页码 | | `page_size` | `integer` | ❌ | `100` | 每页数量 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `files` | `array` | 文件列表 | | `files[].file_id` | `string` | 文件唯一 ID | | `files[].name` | `string` | 文件名 | | `files[].type` | `string` | `file` 或 `folder` | | `files[].size` | `integer` | 大小（字节） | | `files[].size_str` | `string` | 可读大小（如 `2.5 MB`） | | `files[].path` | `string` | 完整路径 | | `files[].updated_at` | `string` | 最后修改时间 | | `total` | `integer` | 总数量 | **示例输出**： ```json { "success": true, "data": { "files": [ { "file_id": "f_abc123", "name": "report.pdf", "type": "file", "size": 2621440, "size_str": "2.5 MB", "path": "/我的文档/report.pdf", "updated_at": "2026-03-15 10:30:00" }, { "file_id": "d_folder01", "name": "照片", "type": "folder", "size": 0, "size_str": "-", "path": "/我的文档/照片", "updated_at": "2026-03-14 08:00:00" } ], "total": 2 }, "message": "ok" } ``` --- ### 2.2 upload_file — 上传文件 **描述**：将本地文件上传到微云指定目录。支持大文件分片上传。 **CLI**： ```bash python weiyun_skills/main.py upload ./report.pdf /我的文档/ python weiyun_skills/main.py upload ./big_video.mp4 /视频/ --overwrite ``` **Python**： ```python result = client.upload_file("./report.pdf", "/我的文档/report.pdf", overwrite=True) ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `local_path` | `string` | ✅ | - | 本地文件路径 | | `remote_path` | `string` | ✅ | - | 微云目标路径 | | `overwrite` | `boolean` | ❌ | `false` | 是否覆盖同名文件 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `file_id` | `string` | 上传后的文件 ID | | `name` | `string` | 文件名 | | `size` | `integer` | 文件大小 | | `remote_path` | `string` | 云端路径 | | `md5` | `string` | 文件 MD5 | | `uploaded_at` | `string` | 上传时间 | --- ### 2.3 upload_folder — 上传文件夹 **描述**：将本地文件夹递归上传到微云，自动创建对应的目录结构。跳过隐藏文件和 `__pycache__` 等缓存目录。 **CLI**： ```bash # Upload folder to Weiyun root python weiyun_skills/main.py upload-folder ./my_docs/ # Upload folder to a specific remote directory python weiyun_skills/main.py upload-folder ./my_docs/ /目标文件夹/ # Overwrite existing files python weiyun_skills/main.py upload-folder ./my_docs/ / --overwrite ``` **Python**： ```python # Upload to root result = client.upload_folder("./my_docs/") # Upload to a specific folder, overwrite existing result = client.upload_folder("./my_docs/", remote_path="/目标文件夹/", overwrite=True) ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `local_path` | `string` | ✅ | - | 本地文件夹路径 | | `remote_path` | `string` | ❌ | `/` | 微云目标路径（`/` 表示根目录） | | `overwrite` | `boolean` | ❌ | `false` | 是否覆盖同名文件 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `folder_name` | `string` | 上传的文件夹名称 | | `uploaded_files` | `array` | 成功上传的文件列表 | | `uploaded_files[].name` | `string` | 文件名 | | `uploaded_files[].size` | `integer` | 文件大小（字节） | | `uploaded_files[].size_str` | `string` | 可读大小 | | `uploaded_files[].instant_upload` | `boolean` | 是否秒传 | | `failed_files` | `array` | 上传失败的文件列表 | | `failed_files[].name` | `string` | 文件名 | | `failed_files[].error` | `string` | 错误信息 | | `uploaded_count` | `integer` | 成功上传数量 | | `failed_count` | `integer` | 失败数量 | | `total_size_str` | `string` | 总上传大小 | | `elapsed` | `float` | 上传耗时（秒） | **示例输出**： ```json { "success": true, "data": { "folder_name": "upload_file", "uploaded_files": [ { "name": "abc.txt", "size": 6, "size_str": "6.00 B", "instant_upload": false } ], "failed_files": [], "uploaded_count": 1, "failed_count": 0, "total_size_str": "6.00 B", "elapsed": 2.35 }, "message": "ok" } ``` --- ### 2.4 download_file — 下载文件 **描述**：从微云下载文件到本地。 **CLI**： ```bash python weiyun_skills/main.py download /我的文档/report.pdf ./downloads/ python weiyun_skills/main.py download /我的文档/report.pdf ./downloads/ --overwrite ``` **Python**： ```python result = client.download_file("/我的文档/report.pdf", "./downloads/report.pdf") ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `remote_path` | `string` | ✅ | - | 微云文件路径 | | `local_path` | `string` | ✅ | - | 本地保存路径 | | `overwrite` | `boolean` | ❌ | `false` | 是否覆盖本地文件 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `local_path` | `string` | 本地保存路径 | | `size` | `integer` | 文件大小 | | `md5` | `string` | MD5 校验值 | | `elapsed` | `float` | 下载耗时（秒） | --- ### 2.5 download_folder — 下载文件夹 **描述**：从微云下载整个文件夹到本地。支持两种模式：递归下载（保留目录结构）和打包下载（下载为 zip 文件）。 **CLI**： ```bash # Recursive download (preserves folder structure) python weiyun_skills/main.py download-folder QQ ./downloads/ python weiyun_skills/main.py download-folder QQ ./downloads/ --overwrite # Download as zip python weiyun_skills/main.py download-folder QQ ./downloads/ --zip ``` **Python**： ```python # Recursive download result = client.download_folder("QQ", "./downloads/") # Download as zip result = client.download_folder("QQ", "./downloads/", as_zip=True) ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `folder_name` | `string` | ✅ | - | 微云上的文件夹名称 | | `local_path` | `string` | ✅ | - | 本地保存目录（或 zip 文件路径） | | `overwrite` | `boolean` | ❌ | `false` | 是否覆盖已存在的本地文件 | | `as_zip` | `boolean` | ❌ | `false` | 是否打包为 zip 下载 | **输出参数（递归模式）**： | 参数名 | 类型 | 说明 | |--------|------|------| | `folder_name` | `string` | 文件夹名称 | | `local_path` | `string` | 本地保存路径 | | `downloaded_files` | `array` | 成功下载的文件列表 | | `downloaded_files[].name` | `string` | 文件名 | | `downloaded_files[].local_path` | `string` | 本地路径 | | `downloaded_files[].size` | `integer` | 文件大小（字节） | | `downloaded_files[].size_str` | `string` | 可读大小 | | `failed_files` | `array` | 下载失败的文件列表 | | `failed_files[].name` | `string` | 文件名 | | `failed_files[].error` | `string` | 错误信息 | | `downloaded_count` | `integer` | 成功下载数量 | | `failed_count` | `integer` | 失败数量 | | `total_size_str` | `string` | 总下载大小 | | `elapsed` | `float` | 下载耗时（秒） | **输出参数（zip 模式）**： | 参数名 | 类型 | 说明 | |--------|------|------| | `local_path` | `string` | zip 文件本地路径 | | `size` | `integer` | 文件大小（字节） | | `size_str` | `string` | 可读大小 | | `md5` | `string` | MD5 校验值 | | `elapsed` | `float` | 下载耗时（秒） | **示例输出（递归模式）**： ```json { "success": true, "data": { "folder_name": "QQ", "local_path": "./downloads/QQ", "downloaded_files": [ { "name": "report.pdf", "local_path": "./downloads/QQ/report.pdf", "size": 8663503, "size_str": "8.26 MB" } ], "failed_files": [], "downloaded_count": 6, "failed_count": 0, "total_size_str": "33.11 MB", "elapsed": 5.88 }, "message": "ok" } ``` --- ### 2.6 delete_file — 删除文件 **描述**：删除微云文件或文件夹（移入回收站）。 **CLI**： ```bash python weiyun_skills/main.py delete /我的文档/old_file.pdf python weiyun_skills/main.py delete /我的文档/old_file.pdf --permanent ``` **Python**： ```python result = client.delete_file("/我的文档/old_file.pdf", permanent=False) ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `remote_path` | `string` | ✅ | - | 文件/文件夹路径 | | `permanent` | `boolean` | ❌ | `false` | 是否永久删除（跳过回收站） | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `deleted_path` | `string` | 已删除的路径 | | `is_permanent` | `boolean` | 是否永久删除 | | `deleted_at` | `string` | 删除时间 | --- ### 2.7 move_file — 移动文件 **描述**：将文件或文件夹移动到另一个目录。 **CLI**： ```bash python weiyun_skills/main.py move /我的文档/report.pdf /归档/2026/ ``` **Python**： ```python result = client.move_file("/我的文档/report.pdf", "/归档/2026/") ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `source_path` | `string` | ✅ | - | 源路径 | | `target_path` | `string` | ✅ | - | 目标目录路径 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `source_path` | `string` | 原路径 | | `target_path` | `string` | 新路径 | --- ### 2.8 copy_file — 复制文件 **描述**：复制文件或文件夹到另一个目录。 **CLI**： ```bash python weiyun_skills/main.py copy /我的文档/report.pdf /备份/ ``` **Python**： ```python result = client.copy_file("/我的文档/report.pdf", "/备份/") ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `source_path` | `string` | ✅ | - | 源路径 | | `target_path` | `string` | ✅ | - | 目标目录路径 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `source_path` | `string` | 源路径 | | `target_path` | `string` | 副本路径 | | `new_file_id` | `string` | 副本文件 ID | --- ### 2.9 rename_file — 重命名 **描述**：重命名文件或文件夹。 **CLI**： ```bash python weiyun_skills/main.py rename /我的文档/report.pdf "年度报告.pdf" ``` **Python**： ```python result = client.rename_file("/我的文档/report.pdf", "年度报告.pdf") ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `remote_path` | `string` | ✅ | - | 文件当前路径 | | `new_name` | `string` | ✅ | - | 新文件名 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `old_path` | `string` | 原路径 | | `new_path` | `string` | 新路径 | --- ### 2.10 create_folder — 创建文件夹 **描述**：在微云上创建文件夹，支持递归创建多级目录。 **CLI**： ```bash python weiyun_skills/main.py mkdir /工作/2026/Q1/报告 ``` **Python**： ```python result = client.create_folder("/工作/2026/Q1/报告") ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `remote_path` | `string` | ✅ | - | 文件夹路径 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `folder_id` | `string` | 文件夹 ID | | `path` | `string` | 完整路径 | | `created_at` | `string` | 创建时间 | --- ### 2.11 search_files — 搜索文件 **描述**：按关键词搜索微云中的文件。 **CLI**： ```bash python weiyun_skills/main.py search "报告" python weiyun_skills/main.py search "报告" --type document ``` **Python**： ```python results = client.search_files("报告", file_type="document") ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `keyword` | `string` | ✅ | - | 搜索关键词 | | `file_type` | `string` | ❌ | `all` | 类型过滤：`all`/`document`/`image`/`video`/`audio` | | `page` | `integer` | ❌ | `1` | 分页页码 | | `page_size` | `integer` | ❌ | `50` | 每页数量 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `results` | `array` | 搜索结果列表 | | `results[].file_id` | `string` | 文件 ID | | `results[].name` | `string` | 文件名 | | `results[].type` | `string` | 类型 | | `results[].size_str` | `string` | 可读大小 | | `results[].path` | `string` | 路径 | | `total` | `integer` | 匹配总数 | --- ## 3. 分享管理 Skills ### 3.1 create_share — 创建分享 **描述**：为文件或文件夹创建分享链接，支持设置密码和有效期。 **CLI**： ```bash python weiyun_skills/main.py share /我的文档/report.pdf python weiyun_skills/main.py share /我的文档/report.pdf --expire 7 --password abc1 ``` **Python**： ```python share = client.create_share( "/我的文档/report.pdf", expire_days=7, password="abc1" ) print(share["share_url"]) ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `remote_path` | `string` | ✅ | - | 文件/文件夹路径 | | `password` | `string` | ❌ | `null` | 分享密码（4 位） | | `expire_days` | `integer` | ❌ | `0` | 有效天数，0 为永久 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `share_id` | `string` | 分享 ID | | `share_url` | `string` | 分享链接 | | `password` | `string` | 分享密码 | | `expire_at` | `string` | 过期时间 | | `created_at` | `string` | 创建时间 | **示例输出**： ```json { "success": true, "data": { "share_id": "s_abc123", "share_url": "https://share.weiyun.com/xxxx", "password": "abc1", "expire_at": "2026-03-22 21:00:00", "created_at": "2026-03-15 21:00:00" }, "message": "ok" } ``` --- ### 3.2 cancel_share — 取消分享 **描述**：取消已创建的分享链接。 **CLI**： ```bash python weiyun_skills/main.py unshare s_abc123 ``` **Python**： ```python result = client.cancel_share("s_abc123") ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `share_id` | `string` | ✅ | - | 分享 ID | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `share_id` | `string` | 已取消的分享 ID | | `cancelled_at` | `string` | 取消时间 | --- ### 3.3 list_shares — 列出分享 **描述**：列出当前用户所有的分享链接。 **CLI**： ```bash python weiyun_skills/main.py shares python weiyun_skills/main.py shares --status active ``` **Python**： ```python shares = client.list_shares(status="active") ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `status` | `string` | ❌ | `all` | 状态过滤：`all`/`active`/`expired` | | `page` | `integer` | ❌ | `1` | 分页页码 | | `page_size` | `integer` | ❌ | `20` | 每页数量 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `shares` | `array` | 分享列表 | | `shares[].share_id` | `string` | 分享 ID | | `shares[].share_url` | `string` | 分享链接 | | `shares[].file_name` | `string` | 文件名 | | `shares[].status` | `string` | 状态 | | `shares[].view_count` | `integer` | 查看次数 | | `shares[].download_count` | `integer` | 下载次数 | | `shares[].created_at` | `string` | 创建时间 | | `shares[].expire_at` | `string` | 过期时间 | | `total` | `integer` | 总数量 | --- ## 4. 空间管理 Skills ### 4.1 get_space_info — 空间信息 **描述**：获取微云存储空间使用情况。 **CLI**： ```bash python weiyun_skills/main.py space ``` **Python**： ```python info = client.get_space_info() print(f"Used: {info['used_space_str']} / {info['total_space_str']}") ``` **输入参数**：无 **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `total_space` | `integer` | 总空间（字节） | | `total_space_str` | `string` | 可读总空间（如 `10 GB`） | | `used_space` | `integer` | 已用空间（字节） | | `used_space_str` | `string` | 可读已用空间 | | `free_space` | `integer` | 剩余空间（字节） | | `free_space_str` | `string` | 可读剩余空间 | | `usage_percent` | `float` | 使用百分比 | | `file_count` | `integer` | 文件总数 | | `folder_count` | `integer` | 文件夹总数 | **示例输出**： ```json { "success": true, "data": { "total_space": 10737418240, "total_space_str": "10.00 GB", "used_space": 5368709120, "used_space_str": "5.00 GB", "free_space": 5368709120, "free_space_str": "5.00 GB", "usage_percent": 50.0, "file_count": 1234, "folder_count": 56 }, "message": "ok" } ``` --- ### 4.2 get_recycle_bin — 回收站 **描述**：获取回收站中的文件列表。 **CLI**： ```bash python weiyun_skills/main.py recycle ``` **Python**： ```python items = client.get_recycle_bin() ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `page` | `integer` | ❌ | `1` | 分页页码 | | `page_size` | `integer` | ❌ | `50` | 每页数量 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `files` | `array` | 回收站文件列表 | | `files[].file_id` | `string` | 文件 ID | | `files[].name` | `string` | 文件名 | | `files[].size_str` | `string` | 可读大小 | | `files[].original_path` | `string` | 原始路径 | | `files[].deleted_at` | `string` | 删除时间 | | `total` | `integer` | 总数量 | | `total_size_str` | `string` | 回收站总大小 | --- ### 4.3 restore_file — 恢复文件 **描述**：从回收站恢复文件到原始位置。 **CLI**： ```bash python weiyun_skills/main.py restore f_del001 ``` **Python**： ```python result = client.restore_file("f_del001") ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `file_id` | `string` | ✅ | - | 回收站中的文件 ID | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `file_id` | `string` | 文件 ID | | `restored_path` | `string` | 恢复后路径 | | `restored_at` | `string` | 恢复时间 | --- ### 4.4 clear_recycle_bin — 清空回收站 **描述**：清空回收站，永久删除所有回收站文件。**⚠️ 此操作不可逆！** **CLI**： ```bash python weiyun_skills/main.py clear-recycle --confirm ``` **Python**： ```python result = client.clear_recycle_bin(confirm=True) ``` **输入参数**： | 参数名 | 类型 | 必填 | 默认值 | 说明 | |--------|------|------|--------|------| | `confirm` | `boolean` | ✅ | - | 必须为 `true` 才执行 | **输出参数**： | 参数名 | 类型 | 说明 | |--------|------|------| | `deleted_count` | `integer` | 删除文件数 | | `freed_space_str` | `string` | 释放的空间大小 | | `cleared_at` | `string` | 清空时间 | --- ## 附录 A：统一错误码 | 错误码 | 说明 | |--------|------| | `AUTH_EXPIRED` | Cookies 已过期，需重新登录 | | `AUTH_FAILED` | 认证失败 | | `FILE_NOT_FOUND` | 文件不存在 | | `FOLDER_NOT_FOUND` | 文件夹不存在 | | `SPACE_FULL` | 空间已满 | | `FILE_TOO_LARGE` | 文件过大 | | `DUPLICATE_NAME` | 名称重复 | | `PERMISSION_DENIED` | 权限不足 | | `RATE_LIMITED` | 请求频率超限 | | `NETWORK_ERROR` | 网络错误 | | `SHARE_EXPIRED` | 分享已过期 | | `INVALID_PARAM` | 参数无效 | | `QR_EXPIRED` | 二维码已过期，需刷新 | | `QR_CANCELLED` | 用户取消了扫码 | **错误响应格式**： ```json { "success": false, "data": null, "message": "Cookies expired, please re-login", "error_code": "AUTH_EXPIRED" } ``` --- ## 附录 B：Cookies 关键字段说明 | Cookie 名称 | 说明 | |-------------|------| | `uin` | 用户 QQ 号标识 | | `skey` | 会话密钥 | | `p_uin` | 加密的用户标识 | | `p_skey` | 加密的会话密钥 | | `pt4_token` | PT4 认证 Token | | `pt2gguin` | 辅助认证字段 | > **提示**：并非所有 Cookie 字段都是必需的，核心字段为 `uin`、`skey`、`p_skey`。 FILE:README.md # 腾讯微云管理 Skills > 一套用于管理腾讯微云（Weiyun）云存储服务的 Python 工具集，支持 **扫码登录** 和 **Cookies 登录** 两种认证方式。 --- ## 🚀 使用方法 ### 1. 安装依赖 ```bash cd weiyun-skills pip install -r requirements.txt ``` ### 2. 登录认证（二选一） #### 方式一：扫码登录（推荐）运行扫码登录脚本，终端会显示二维码，使用微信/QQ 扫码即可完成登录： ```bash python weiyun_skills/login.py --method qrcode ``` 扫码成功后，Cookies 会自动保存到 `cookies.json` 文件中，后续操作无需重复登录。 #### 方式二：复制 Cookies 登录 1. 在浏览器中登录 [腾讯微云](https://www.weiyun.com/) 2. 打开开发者工具 (F12) → Network → 任意请求 → 复制 `Cookie` 请求头的值 3. 运行以下命令： ```bash python weiyun_skills/login.py --method cookies --cookies "your_cookie_string_here" ``` 或者直接编辑 `cookies.json` 文件： ```json { "cookies_str": "uin=o012345678; skey=@abcdef1234; ...", "update_time": "2026-03-15 21:00:00" } ``` ### 3. 使用 Skills ```bash # List files in root directory python weiyun_skills/main.py list / # Upload a file python weiyun_skills/main.py upload ./local_file.pdf /云端目录/ # Download a file python weiyun_skills/main.py download /云端目录/file.pdf ./local_dir/ # Create a share link python weiyun_skills/main.py share /云端目录/file.pdf --expire 7 --password abc123 # Get space usage info python weiyun_skills/main.py space # Search files by keyword python weiyun_skills/main.py search "报告" # Delete a file (to recycle bin) python weiyun_skills/main.py delete /云端目录/old_file.pdf # Move a file python weiyun_skills/main.py move /源路径/file.pdf /目标路径/ # Create a folder python weiyun_skills/main.py mkdir /新文件夹/子文件夹 ``` ### 4. 在 Python 中调用 ```python from weiyun_skills.client import WeiyunClient # Initialize client (auto-loads cookies.json) client = WeiyunClient() # Or pass cookies string directly client = WeiyunClient(cookies_str="uin=o012345678; skey=@abcdef1234; ...") # List files files = client.list_files("/我的文档") for f in files: print(f["name"], f["size"]) # Upload file client.upload_file("./report.pdf", "/我的文档/report.pdf") # Download file client.download_file("/我的文档/report.pdf", "./downloads/report.pdf") # Create share link share = client.create_share("/我的文档/report.pdf", expire_days=7, password="abc1") print(share["share_url"]) # Get space info info = client.get_space_info() print(f"Used: {info['used_space_str']} / {info['total_space_str']}") ``` --- ## 📋 项目简介本项目提供了一组 Python 脚本，用于自动化管理腾讯微云中的文件。核心特性： - ✅ **扫码登录** — 终端展示二维码，微信/QQ 扫码完成认证 - ✅ **Cookies 登录** — 从浏览器复制 Cookies 快速登录 - ✅ **自动保存会话** — 登录状态持久化，无需重复认证 - ✅ **文件管理** — 上传、下载、删除、移动、复制、重命名、搜索 - ✅ **分享管理** — 创建/取消分享链接，支持密码和有效期 - ✅ **空间管理** — 查看容量、回收站操作 - ✅ **命令行工具** — 所有功能均可通过命令行直接使用 --- ## 📁 项目结构 ``` weiyun-skills/ ├── README.md # Project documentation ├── SKILL.md # Skills definition ├── LICENSE # MIT License ├── requirements.txt # Python dependencies ├── cookies.json # Saved login cookies (auto-generated) └── weiyun_skills/ ├── __init__.py # Package init ├── login.py # QR code login & cookies login ├── client.py # Weiyun API client ├── main.py # CLI entry point └── utils.py # Utility functions ``` --- ## ⚙️ 依赖说明 | 依赖包 | 用途 | |--------|------| | `requests` | HTTP 请求 | | `qrcode` | 终端展示扫码登录二维码 | | `Pillow` | 图片处理（二维码渲染） | | `tabulate` | 命令行表格输出 | --- ## 🛠️ Skills 一览详细定义请参阅 [SKILL.md](./SKILL.md)。 | 分类 | Skill | 说明 | |------|-------|------| | 🔑 认证 | `qrcode_login` | 扫码登录 | | 🔑 认证 | `cookies_login` | Cookies 登录 | | 📂 文件 | `list_files` | 列出文件 | | 📂 文件 | `upload_file` | 上传文件 | | 📂 文件 | `download_file` | 下载文件 | | 📂 文件 | `delete_file` | 删除文件 | | 📂 文件 | `move_file` | 移动文件 | | 📂 文件 | `copy_file` | 复制文件 | | 📂 文件 | `rename_file` | 重命名 | | 📂 文件 | `create_folder` | 创建文件夹 | | 📂 文件 | `search_files` | 搜索文件 | | 🔗 分享 | `create_share` | 创建分享 | | 🔗 分享 | `cancel_share` | 取消分享 | | 🔗 分享 | `list_shares` | 列出分享 | | 💾 空间 | `get_space_info` | 空间信息 | | 💾 空间 | `get_recycle_bin` | 回收站 | | 💾 空间 | `restore_file` | 恢复文件 | | 💾 空间 | `clear_recycle_bin` | 清空回收站 | --- ## 📝 注意事项 1. **Cookies 有效期**：腾讯微云的 Cookies 通常在 24 小时后过期，届时需重新登录 2. **扫码登录**：需要终端支持 Unicode 字符显示（大部分现代终端均支持） 3. **频率限制**：请避免短时间内大量请求，以免触发风控 4. **大文件上传**：支持分片上传，默认分片大小为 4MB --- ## 📄 许可证 [MIT License](./LICENSE) © 2026 enoyao FILE:requirements.txt requests>=2.31.0 qrcode>=7.4 tabulate>=0.9.0 # Optional: Pillow (for QR code image rendering) # pip install Pillow>=10.0.0 FILE:upload_readme.py #!/usr/bin/env python3 """Upload README.md to Weiyun - interactive login + upload script.""" import sys import os # Add project root to path sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from weiyun_skills.login import qrcode_login, cookies_login, load_cookies from weiyun_skills.client import WeiyunClient def main(): readme_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "README.md") cookies_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cookies.json") if not os.path.isfile(readme_path): print(f"[ERROR] README.md not found at: {readme_path}") sys.exit(1) print("=" * 60) print(" Weiyun Skills - Upload README.md") print("=" * 60) print(f" File: {readme_path}") print(f" Size: {os.path.getsize(readme_path)} bytes") print() # Step 1: Check if already logged in saved = load_cookies(cookies_path) if saved and saved.get("cookies_str"): print("[*] Found saved cookies, verifying...") client = WeiyunClient(cookies_path=cookies_path) space = client.get_space_info() if space["success"]: print(f"[✓] Already logged in as: {saved.get('uin', 'unknown')}") else: print("[*] Saved cookies expired, need to re-login.") saved = {} # Step 2: Login if needed if not saved or not saved.get("cookies_str"): print() print("Choose login method:") print(" 1) QR code scan (recommended)") print(" 2) Paste cookies from browser") print() choice = input("Enter choice [1/2]: ").strip() if choice == "2": print() print("Steps to get cookies:") print(" 1. Open https://www.weiyun.com in browser and login") print(" 2. Press F12 -> Network tab -> click any request") print(" 3. Copy the 'Cookie:' header value") print() cookies_str = input("Paste cookies here: ").strip() result = cookies_login(cookies_str, save_path=cookies_path) else: result = qrcode_login(save_path=cookies_path) if not result["success"]: print(f"\n[ERROR] Login failed: {result['message']}") sys.exit(1) print(f"\n[✓] Login successful! Cookies saved to: {cookies_path}") # Step 3: Upload README.md print() print("-" * 60) print("[*] Uploading README.md to Weiyun root directory...") print("-" * 60) client = WeiyunClient(cookies_path=cookies_path) result = client.upload_file(readme_path, "/README.md", overwrite=True) if result["success"]: data = result["data"] print() print("=" * 60) print(" ✅ Upload successful!") print(f" 📄 File: {data.get('name', 'README.md')}") print(f" 📦 Size: {data.get('size', 0)} bytes") print(f" 📂 Path: {data.get('remote_path', '/README.md')}") print(f" 🔑 MD5: {data.get('md5', 'N/A')}") print(f" 🕐 Time: {data.get('uploaded_at', 'N/A')}") print("=" * 60) else: print(f"\n[ERROR] Upload failed: {result['message']}") print(f" Error code: {result.get('error_code', 'UNKNOWN')}") sys.exit(1) if __name__ == "__main__": main() FILE:weiyun_skills/client.py """Weiyun API client - core module for interacting with Weiyun services.""" import os import json import time import hashlib import requests import random from typing import Optional from weiyun_skills.login import load_cookies, DEFAULT_COOKIES_PATH from weiyun_skills.utils import ( format_size, get_file_md5, parse_cookies_str, get_timestamp, ensure_dir, build_response, ) # Weiyun API base URLs API_BASE = "https://www.weiyun.com" # Command name to protocol mapping (URL path component) CMD_PROTOCOLS = { "DiskUserInfoGet": "weiyunQdiskClient", "DiskUserConfigGet": "weiyunQdiskClient", "DiskDirList": "weiyunQdisk", "DiskDirBatchList": "weiyunQdisk", "DiskFileBatchDownload": "weiyunQdiskClient", "DiskFilePackageDownload": "weiyunQdisk", "DiskFileDocDownloadAbs": "weiyunQdiskClient", "DiskDirFileBatchDeleteEx": "weiyunQdiskClient", "DiskFileRename": "weiyunQdiskClient", "DiskDirCreate": "weiyunQdiskClient", "DiskDirAttrModify": "weiyunQdiskClient", "DiskDirFileBatchMove": "weiyunQdiskClient", "DiskRecycleList": "weiyunQdiskClient", "DiskRecycleDirFileBatchRestore": "weiyunQdiskClient", "DiskRecycleDirFileClear": "weiyunQdiskClient", "WeiyunShareList": "weiyunShare", "WeiyunShareDelete": "weiyunShare", "WeiyunShareAddV2": "weiyunShare", "WeiyunSharePwdCreate": "weiyunShare", "WeiyunSharePwdDelete": "weiyunShare", "WeiyunShareRenewal": "weiyunShare", "WeiyunShareTraceInfo": "weiyunShare", "FileSearchbyKeyWord": "weiyunFileSearch", "FileSearchTipsList": "weiyunFileSearch", } # Command name to numeric ID mapping (extracted from Weiyun JS SDK) CMD_IDS = { "DiskUserInfoGet": 2201, "DiskDirList": 2208, "DiskDirBatchList": 2209, "DiskUserConfigGet": 2225, "DiskFileBatchDownload": 2402, "DiskFilePackageDownload": 2403, "DiskFileDocDownloadAbs": 2414, "DiskDirFileBatchDeleteEx": 2509, "DiskFileRename": 2605, "DiskDirCreate": 2614, "DiskDirAttrModify": 2615, "DiskDirFileBatchMove": 2618, "DiskRecycleList": 2702, "DiskRecycleDirFileBatchRestore": 2708, "DiskRecycleDirFileClear": 2710, "WeiyunShareList": 12008, "WeiyunShareDelete": 12007, "WeiyunShareAddV2": 12100, "WeiyunSharePwdCreate": 12012, "WeiyunSharePwdDelete": 12014, "WeiyunShareRenewal": 12035, "WeiyunShareTraceInfo": 12033, "FileSearchbyKeyWord": 247251, "FileSearchTipsList": 247250, } class WeiyunClient: """Tencent Weiyun API client. Supports file management, sharing, and space operations. Authentication via saved cookies file or direct cookies string. Usage: # Auto-load cookies from cookies.json client = WeiyunClient() # Or pass cookies string directly client = WeiyunClient(cookies_str="uin=xxx; skey=xxx; ...") """ def __init__(self, cookies_str: str = None, cookies_path: str = DEFAULT_COOKIES_PATH): """Initialize Weiyun client. Args: cookies_str: Optional cookies string. If not provided, will try to load from cookies_path. cookies_path: Path to cookies JSON file. """ self.session = requests.Session() self.session.headers.update({ "User-Agent": ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/134.0.0.0 Safari/537.36" ), "Referer": "https://www.weiyun.com/disk", "Origin": "https://www.weiyun.com", }) # Store p_skey separately to avoid cookie domain conflicts self._p_skey = "" self._uin = "" self._warmed = False if cookies_str: cookies_dict = parse_cookies_str(cookies_str) self._p_skey = cookies_dict.get("p_skey", "") self._uin = cookies_dict.get("uin", "").lstrip("o").lstrip("0") self.session.cookies.update(cookies_dict) else: cookies_data = load_cookies(cookies_path) if cookies_data.get("cookies_dict"): self._p_skey = cookies_data["cookies_dict"].get("p_skey", "") self._uin = cookies_data["cookies_dict"].get( "uin", "" ).lstrip("o").lstrip("0") self.session.cookies.update(cookies_data["cookies_dict"]) elif cookies_data.get("cookies_str"): cookies_dict = parse_cookies_str(cookies_data["cookies_str"]) self._p_skey = cookies_dict.get("p_skey", "") self._uin = cookies_dict.get( "uin", "" ).lstrip("o").lstrip("0") self.session.cookies.update(cookies_dict) def _warm_session(self) -> None: """Warm up session by visiting /disk page to get wyctoken cookie. This is required before making API calls - the server sets a wyctoken cookie on the HTML page response that is needed as the g_tk CSRF token for API requests. """ if self._warmed: return try: self.session.get(f"{API_BASE}/disk", timeout=10) self._warmed = True except requests.RequestException: pass # Will proceed without warming, API may fail def _get_gtk(self) -> int: """Get g_tk token from wyctoken cookie. Must call _warm_session() first to obtain wyctoken. Returns: g_tk integer value. """ for c in self.session.cookies: if c.name == "wyctoken": try: return int(c.value) except ValueError: pass return 5381 def _get_uin(self) -> str: """Get numeric UIN from cookies. Returns: UIN string without 'o' prefix and leading zeros. """ return self._uin or "0" def _get_token_info(self) -> dict: """Build token_info for API authentication. Returns: Token info dict matching the Weiyun JS SDK format. """ return { "token_type": 0, "login_key_type": 27, "login_key_value": self._p_skey, } def _api_request(self, cmd: str, body: dict = None) -> dict: """Make an API request to Weiyun. Follows the exact format from Weiyun's JS SDK: - Session must be pre-warmed to obtain wyctoken cookie - req_header and req_body are JSON-stringified in the POST data - req_header.cmd uses numeric command ID - req_body contains ReqMsg_body with token_info and command data - g_tk (wyctoken) is passed as a URL query parameter Args: cmd: API command name (e.g., 'DiskDirList'). body: Request body data for the specific command. Returns: API response as dict. """ # Ensure session is warmed (gets wyctoken cookie) self._warm_session() g_tk = self._get_gtk() uin = self._get_uin() cmd_id = CMD_IDS.get(cmd, 0) protocol = CMD_PROTOCOLS.get(cmd, "weiyunQdiskClient") url = f"{API_BASE}/webapp/json/{protocol}/{cmd}" params = { "refer": "Chrome_Mac", "g_tk": str(g_tk), "r": str(random.random()), } # Build req_header matching JS SDK format req_header = { "seq": int(time.time()) + random.randint(0, 9999), "type": 1, "cmd": cmd_id, "appid": 30013, "version": 3, "major_version": 3, "minor_version": 3, "fix_version": 3, "wx_openid": "", "qq_openid": "", "user_flag": 0, "env_id": "", "uin": uin, "uid": uin, } # Build req_body matching JS SDK format req_body = { "ReqMsg_body": { "ext_req_head": { "token_info": self._get_token_info(), "language_info": { "language_type": 2052, }, }, f".weiyun.{cmd}MsgReq_body": body or {}, }, } # POST data: req_header and req_body are JSON strings post_data = { "req_header": json.dumps(req_header), "req_body": json.dumps(req_body), } try: resp = self.session.post(url, params=params, json=post_data, timeout=30) resp.raise_for_status() data = resp.json() # Response format: # - WAF block: {"ret": 403, "msg": "..."} # - Parse error: {"ret": 500, "msg": "..."} # - Success: {"data": {"rsp_header": {...}, "rsp_body": {...}}} if "ret" in data and data["ret"] != 0: return build_response( False, message=data.get("msg", f"API error (ret={data['ret']})"), error_code="API_ERROR", ) resp_data = data.get("data", data) rsp_header = resp_data.get("rsp_header", {}) rsp_body = resp_data.get("rsp_body", {}) rsp_msg_body = rsp_body.get("RspMsg_body", rsp_body) if rsp_header.get("retcode", 0) == 0: return build_response(True, data=rsp_msg_body) else: retcode = rsp_header.get("retcode", -1) msg = rsp_header.get("retmsg", f"API error (retcode={retcode})") return build_response(False, message=msg, error_code="API_ERROR") except requests.RequestException as e: return build_response(False, message=str(e), error_code="NETWORK_ERROR") except (json.JSONDecodeError, KeyError) as e: return build_response(False, message=f"Invalid response: {e}", error_code="API_ERROR") # ==================== File Management ==================== def _get_root_dir_key(self) -> str: """Get the root directory key from DiskUserInfoGet. Returns: Root directory key string. """ if hasattr(self, "_root_dir_key") and self._root_dir_key: return self._root_dir_key result = self._api_request("DiskUserInfoGet", {}) if result["success"]: self._root_dir_key = result["data"].get("root_dir_key", "") self._main_dir_key = result["data"].get("main_dir_key", "") return self._root_dir_key return "" def _get_file_sha1(self, path: str) -> str: """Calculate SHA1 hash of a file. Args: path: Path to local file. Returns: SHA1 hex digest string. """ sha1 = hashlib.sha1() with open(path, "rb") as f: while True: chunk = f.read(65536) if not chunk: break sha1.update(chunk) return sha1.hexdigest() def list_files(self, remote_path: str = "/", sort_by: str = "name", sort_order: str = "asc", page: int = 1, page_size: int = 100) -> dict: """List files and folders in a directory. Args: remote_path: Remote directory path (currently uses root dir). sort_by: Sort field - 'name', 'size', or 'time'. sort_order: Sort order - 'asc' or 'desc'. page: Page number. page_size: Items per page. Returns: Response dict with file list. """ sort_map = {"name": 0, "size": 1, "time": 2} start = (page - 1) * page_size # Use root dir key for "/" path root_key = self._get_root_dir_key() body = { "pdir_key": root_key if remote_path == "/" else "", "dir_key": root_key if remote_path == "/" else remote_path, "get_type": 0, "start": start, "count": page_size, "sort_field": sort_map.get(sort_by, 0), "reverse_order": sort_order == "desc", "get_abstract_url": True, } result = self._api_request("DiskDirList", body) if not result["success"]: return result # Format file list raw_files = result["data"].get("file_list", []) raw_dirs = result["data"].get("dir_list", []) files = [] for d in raw_dirs: files.append({ "file_id": d.get("dir_key", ""), "name": d.get("dir_name", ""), "type": "folder", "size": 0, "size_str": "-", "path": d.get("dir_path", ""), "updated_at": d.get("dir_mtime", ""), }) for f in raw_files: size = f.get("file_size", 0) files.append({ "file_id": f.get("file_id", ""), "name": f.get("filename", "") or f.get("file_name", ""), "type": "file", "size": size, "size_str": format_size(size), "path": f.get("file_path", ""), "updated_at": f.get("file_mtime", ""), }) total_dirs = result["data"].get("total_dir_count", len(raw_dirs)) total_files = result["data"].get("total_file_count", len(raw_files)) return build_response(True, data={ "files": files, "total": total_dirs + total_files, "pdir_key": result["data"].get("pdir_key", ""), }) def upload_file(self, local_path: str, remote_path: str, overwrite: bool = False) -> dict: """Upload a local file to Weiyun. Uses the real Weiyun upload API via FormData: 1. PreUpload (cmd=247120) - send file metadata + blob 2. UploadPiece (cmd=247121) - continue upload if needed Args: local_path: Path to local file. remote_path: Target path on Weiyun (e.g., '/README.md'). overwrite: Whether to overwrite existing file. Returns: Response dict with upload result. """ if not os.path.isfile(local_path): return build_response(False, message=f"Local file not found: {local_path}", error_code="FILE_NOT_FOUND") file_name = os.path.basename(remote_path) or os.path.basename(local_path) # Get root dir key for upload target root_key = self._get_root_dir_key() main_dir_key = getattr(self, "_main_dir_key", root_key) result = self._do_upload( local_path=local_path, ppdir_key=root_key, pdir_key=main_dir_key, file_name=file_name, overwrite=overwrite, ) if result["success"]: result["data"]["remote_path"] = remote_path return result def _do_upload(self, local_path: str, ppdir_key: str, pdir_key: str, file_name: str, overwrite: bool = False) -> dict: """Core upload logic using FormData via upload.weiyun.com. Matches the Weiyun JS SDK upload flow for QQ accounts: 1. PreUpload (cmd=247120) via FormData - sends file metadata + blob 2. If not instant upload, continue with UploadPiece (cmd=247121) Args: local_path: Path to local file. ppdir_key: Grandparent directory key. pdir_key: Target parent directory key. file_name: File name on Weiyun. overwrite: Whether to overwrite existing file. Returns: Response dict with upload result. """ if not os.path.isfile(local_path): return build_response(False, message=f"Local file not found: {local_path}", error_code="FILE_NOT_FOUND") self._warm_session() file_size = os.path.getsize(local_path) file_md5 = get_file_md5(local_path) file_sha = self._get_file_sha1(local_path) # Build request header (non-preUpload style per JS SDK) req_header = { "cmd": 247120, "appid": 30013, "version": 0, "major_version": 3, "minor_version": 0, "fix_version": 0, "user_flag": 0, } # Build request body (key without leading dot per JS SDK) req_body = { "ReqMsg_body": { "ext_req_head": { "token_info": self._get_token_info(), "language_info": {"language_type": 2052}, }, "weiyun.PreUploadMsgReq_body": { "req": { "common_upload_req": { "ppdir_key": ppdir_key, "pdir_key": pdir_key, "file_size": file_size, "filename": file_name, "file_exist_option": 6 if overwrite else 0, "use_mutil_channel": True, "file_sha": file_sha, "file_md5": file_md5, }, "upload_scr": 0, }, }, }, } # PreUpload via FormData (same as JS SDK uploadRequest) upload_url = "https://upload.weiyun.com/ftnup_v2/weiyun?cmd=247120" json_payload = json.dumps({ "req_header": req_header, "req_body": req_body, }) try: with open(local_path, "rb") as f: files = { "json": (None, json_payload, "application/json"), "upload": (file_name, f, "application/octet-stream"), } pre_resp = self.session.post( upload_url, files=files, timeout=300, ) pre_data = pre_resp.json() except requests.RequestException as e: return build_response(False, message=f"PreUpload failed: {e}", error_code="NETWORK_ERROR") except json.JSONDecodeError as e: return build_response(False, message=f"PreUpload invalid response: {e}", error_code="API_ERROR") # Check response header rsp_header = pre_data.get("rsp_header", {}) retcode = rsp_header.get("retcode", -1) if retcode != 0: return build_response( False, message=rsp_header.get("retmsg", f"PreUpload error (retcode={retcode})"), error_code="API_ERROR", ) rsp_body = pre_data.get("rsp_body", {}) rsp_msg = rsp_body.get("RspMsg_body", rsp_body) pre_result = rsp_msg.get("weiyun.PreUploadMsgRsp_body", rsp_msg.get(".weiyun.PreUploadMsgRsp_body", rsp_msg)) file_exist = pre_result.get("file_exist", False) common_rsp = pre_result.get("common_upload_rsp", pre_result.get("rsp", {})) if isinstance(common_rsp, dict) and "common_upload_rsp" in common_rsp: common_rsp = common_rsp["common_upload_rsp"] file_id = common_rsp.get("file_id", "") upload_filename = common_rsp.get("filename", file_name) # Instant upload (sec upload) — server already has the file if file_exist: return build_response(True, data={ "file_id": file_id, "name": upload_filename, "size": file_size, "md5": file_md5, "uploaded_at": get_timestamp(), "instant_upload": True, }) # Step 2: Upload file data if needed (flow_state check) flow_state = pre_result.get("flow_state", 0) channel = common_rsp.get("channel", {}) if flow_state == 1 or not channel: # Upload completed in preUpload step (small file or sec upload) return build_response(True, data={ "file_id": file_id, "name": upload_filename, "size": file_size, "md5": file_md5, "uploaded_at": get_timestamp(), "instant_upload": False, }) # Continue with UploadPiece for larger files piece_header = { "cmd": 247121, "appid": 30013, "version": 0, "major_version": 3, "minor_version": 0, "fix_version": 0, "user_flag": 0, } piece_body = { "ReqMsg_body": { "weiyun.UploadPieceMsgReq_body": { "req": channel, }, }, } piece_json = json.dumps({ "req_header": piece_header, "req_body": piece_body, }) piece_url = "https://upload.weiyun.com/ftnup_v2/weiyun?cmd=247121" try: with open(local_path, "rb") as f: files = { "json": (None, piece_json, "application/json"), "upload": (file_name, f, "application/octet-stream"), } piece_resp = self.session.post( piece_url, files=files, timeout=300, ) if piece_resp.status_code != 200: # Fallback to backup URL with open(local_path, "rb") as f: files = { "json": (None, piece_json, "application/json"), "upload": (file_name, f, "application/octet-stream"), } piece_resp = self.session.post( f"{API_BASE}/ftnup_v2/weiyun?cmd=247121", files=files, timeout=300, ) try: piece_data = piece_resp.json() p_rsp = piece_data.get("rsp_header", {}) p_retcode = p_rsp.get("retcode", 0) if p_retcode != 0: return build_response( False, message=f"Upload failed: {p_rsp.get('retmsg', f'retcode={p_retcode}')}", error_code="API_ERROR", ) except (json.JSONDecodeError, ValueError): if piece_resp.status_code != 200: return build_response( False, message=f"Upload failed: HTTP {piece_resp.status_code}", error_code="NETWORK_ERROR", ) except requests.RequestException as e: return build_response(False, message=f"Upload error: {e}", error_code="NETWORK_ERROR") return build_response(True, data={ "file_id": file_id, "name": upload_filename, "size": file_size, "md5": file_md5, "uploaded_at": get_timestamp(), "instant_upload": False, }) def _find_or_create_folder(self, parent_dir_key: str, folder_name: str) -> dict: """Find an existing folder by name under a parent dir, or create it. Args: parent_dir_key: Parent directory key. folder_name: Folder name to find or create. Returns: Response dict with folder dir_key in data. """ # List the parent directory to look for the folder list_result = self.list_files(parent_dir_key) if not list_result["success"]: return list_result for item in list_result["data"]["files"]: if item["type"] == "folder" and item["name"] == folder_name: return build_response(True, data={ "dir_key": item["file_id"], "dir_name": folder_name, "created": False, }) # Folder not found, create it root_key = self._get_root_dir_key() body = { "ppdir_key": root_key, "pdir_key": parent_dir_key, "dir_name": folder_name, } result = self._api_request("DiskDirCreate", body) if not result["success"]: return result new_dir_key = result["data"].get("dir_key", "") return build_response(True, data={ "dir_key": new_dir_key, "dir_name": folder_name, "created": True, }) def _upload_file_to_dir(self, local_path: str, pdir_key: str, ppdir_key: str, file_name: str, overwrite: bool = False) -> dict: """Upload a single file to a specific directory by dir_key. Uses the same PreUpload + UploadPiece flow as upload_file, but targets a specific parent directory. Args: local_path: Path to local file. pdir_key: Target parent directory key. ppdir_key: Grandparent directory key. file_name: File name on Weiyun. overwrite: Whether to overwrite existing file. Returns: Response dict with upload result. """ if not os.path.isfile(local_path): return build_response(False, message=f"Local file not found: {local_path}", error_code="FILE_NOT_FOUND") self._warm_session() file_size = os.path.getsize(local_path) file_md5 = get_file_md5(local_path) file_sha = self._get_file_sha1(local_path) g_tk = self._get_gtk() uin = self._get_uin() # Step 1: PreUpload pre_upload_header = { "cmd": 247120, "appid": 30013, "version": 3, "major_version": 3, "minor_version": 0, "fix_version": 0, "type": 1, "user_flag": 0, "env_id": "", "login_keytype": 27, "uin": uin, "uid": uin, } pre_upload_body = { "ReqMsg_body": { "ext_req_head": { "token_info": self._get_token_info(), "language_info": {"language_type": 2052}, }, ".weiyun.PreUploadMsgReq_body": { "req": { "common_upload_req": { "ppdir_key": ppdir_key, "pdir_key": pdir_key, "file_size": file_size, "filename": file_name, "file_exist_option": 6 if overwrite else 0, "use_mutil_channel": True, "file_sha": file_sha, "file_md5": file_md5, }, "upload_scr": 0, }, }, }, } pre_upload_data = { "req_header": pre_upload_header, "req_body": pre_upload_body, } try: pre_resp = self.session.post( f"{API_BASE}/api/v3/ftn_pre_upload", params={"g_tk": str(g_tk), "r": str(random.random())}, json=pre_upload_data, timeout=30, ) pre_resp.raise_for_status() pre_data = pre_resp.json() except requests.RequestException as e: return build_response(False, message=f"PreUpload failed: {e}", error_code="NETWORK_ERROR") except json.JSONDecodeError as e: return build_response(False, message=f"PreUpload invalid response: {e}", error_code="API_ERROR") err_code = pre_data.get("retcode", pre_data.get("ret")) if err_code is not None and err_code != 0: return build_response( False, message=pre_data.get("msg", f"PreUpload error (code={err_code})"), error_code="API_ERROR", ) resp_data = pre_data.get("data", {}) rsp_body = resp_data.get("rsp_body", {}) rsp_msg = rsp_body.get("RspMsg_body", {}) pre_result = rsp_msg.get( ".weiyun.PreUploadMsgRsp_body", rsp_msg, ) file_exist = pre_result.get("file_exist", False) common_rsp = pre_result.get("common_upload_rsp", pre_result.get("rsp", {})) if isinstance(common_rsp, dict) and "common_upload_rsp" in common_rsp: common_rsp = common_rsp["common_upload_rsp"] file_id = common_rsp.get("file_id", "") upload_filename = common_rsp.get("filename", file_name) # Instant upload (sec upload) — file already exists on server if file_exist: return build_response(True, data={ "file_id": file_id, "name": upload_filename, "size": file_size, "md5": file_md5, "uploaded_at": get_timestamp(), "instant_upload": True, }) # Step 2: Upload file data channel = common_rsp.get("channel", {}) upload_url = ( f"https://upload.weiyun.com/ftnup_v2/weiyun?cmd=247121" ) upload_header = { "cmd": 247121, "appid": 30013, "version": 0, "major_version": 3, "minor_version": 0, "fix_version": 0, "user_flag": 0, } upload_body = { "ReqMsg_body": { ".weiyun.UploadPieceMsgReq_body": { "req": channel if channel else common_rsp, }, }, } upload_json_str = json.dumps({ "req_header": upload_header, "req_body": upload_body, }) try: with open(local_path, "rb") as f: files = { "json": (None, upload_json_str, "application/json"), "upload": (file_name, f, "application/octet-stream"), } upload_resp = self.session.post( upload_url, files=files, timeout=300, ) if upload_resp.status_code != 200: with open(local_path, "rb") as f: files = { "json": (None, upload_json_str, "application/json"), "upload": (file_name, f, "application/octet-stream"), } upload_resp = self.session.post( f"{API_BASE}/ftnup_v2/weiyun?cmd=247121", files=files, timeout=300, ) if upload_resp.status_code != 200: return build_response( False, message=f"Upload failed: HTTP {upload_resp.status_code}", error_code="NETWORK_ERROR", ) # Check upload response retcode try: up_data = upload_resp.json() up_rsp_header = up_data.get("rsp_header", {}) up_retcode = up_rsp_header.get("retcode", 0) if up_retcode != 0: return build_response( False, message=f"Upload failed: {up_rsp_header.get('retmsg', f'retcode={up_retcode}')}", error_code="API_ERROR", ) except (json.JSONDecodeError, ValueError): pass # Non-JSON response with HTTP 200 is ok except requests.RequestException as e: return build_response(False, message=f"Upload error: {e}", error_code="NETWORK_ERROR") return build_response(True, data={ "file_id": file_id, "name": upload_filename, "size": file_size, "md5": file_md5, "uploaded_at": get_timestamp(), "instant_upload": False, }) def upload_folder(self, local_path: str, remote_path: str = "/", overwrite: bool = False) -> dict: """Upload a local folder to Weiyun, preserving directory structure. Recursively traverses the local folder and uploads all files, creating corresponding folders on Weiyun as needed. Args: local_path: Path to local folder. remote_path: Target path on Weiyun ('/' for root, or a folder name under root). overwrite: Whether to overwrite existing files. Returns: Response dict with upload summary. """ if not os.path.isdir(local_path): return build_response( False, message=f"Local folder not found: {local_path}", error_code="FOLDER_NOT_FOUND") root_key = self._get_root_dir_key() main_dir_key = getattr(self, "_main_dir_key", root_key) folder_name = os.path.basename(os.path.normpath(local_path)) # Determine the parent dir key where we create the folder if remote_path == "/" or remote_path == main_dir_key: parent_dir_key = main_dir_key ppdir_key = root_key else: # remote_path is a folder name — find it list_result = self.list_files(main_dir_key) if not list_result["success"]: return list_result target = None for item in list_result["data"]["files"]: if item["type"] == "folder" and item["name"] == remote_path: target = item break if target: parent_dir_key = target["file_id"] ppdir_key = main_dir_key else: parent_dir_key = main_dir_key ppdir_key = root_key # Create the top-level folder on Weiyun folder_result = self._find_or_create_folder(parent_dir_key, folder_name) if not folder_result["success"]: return folder_result target_dir_key = folder_result["data"]["dir_key"] start_time = time.time() # Recursively upload result = self._upload_folder_recursive( local_path=local_path, pdir_key=target_dir_key, ppdir_key=parent_dir_key, overwrite=overwrite, ) elapsed = round(time.time() - start_time, 2) if result["success"]: result["data"]["folder_name"] = folder_name result["data"]["elapsed"] = elapsed return result def _upload_folder_recursive(self, local_path: str, pdir_key: str, ppdir_key: str, overwrite: bool = False) -> dict: """Recursively upload folder contents. Args: local_path: Local folder path. pdir_key: Target directory key on Weiyun. ppdir_key: Parent of target directory key. overwrite: Whether to overwrite existing files. Returns: Response dict with upload summary. """ uploaded_files = [] failed_files = [] total_size = 0 try: entries = sorted(os.listdir(local_path)) except OSError as e: return build_response(False, message=f"Cannot read directory: {e}", error_code="IO_ERROR") for entry in entries: full_path = os.path.join(local_path, entry) # Skip hidden files and common cache directories if entry.startswith(".") or entry == "__pycache__": continue if os.path.isdir(full_path): # Create subfolder on Weiyun and recurse sub_result = self._find_or_create_folder(pdir_key, entry) if not sub_result["success"]: failed_files.append({ "name": entry + "/", "error": sub_result["message"], }) continue sub_dir_key = sub_result["data"]["dir_key"] recurse_result = self._upload_folder_recursive( local_path=full_path, pdir_key=sub_dir_key, ppdir_key=pdir_key, overwrite=overwrite, ) if recurse_result["success"]: uploaded_files.extend( recurse_result["data"].get("uploaded_files", [])) failed_files.extend( recurse_result["data"].get("failed_files", [])) total_size += recurse_result["data"].get("total_size", 0) else: failed_files.append({ "name": entry + "/", "error": recurse_result["message"], }) elif os.path.isfile(full_path): result = self._do_upload( local_path=full_path, ppdir_key=ppdir_key, pdir_key=pdir_key, file_name=entry, overwrite=overwrite, ) if result["success"]: file_size = result["data"].get("size", 0) uploaded_files.append({ "name": result["data"].get("name", entry), "size": file_size, "size_str": format_size(file_size), "instant_upload": result["data"].get( "instant_upload", False), }) total_size += file_size else: failed_files.append({ "name": entry, "error": result["message"], }) return build_response(True, data={ "uploaded_files": uploaded_files, "failed_files": failed_files, "uploaded_count": len(uploaded_files), "failed_count": len(failed_files), "total_size": total_size, "total_size_str": format_size(total_size), }) def _download_single_file(self, file_id: str, pdir_key: str, local_path: str) -> dict: """Download a single file using DiskFileBatchDownload API. Args: file_id: File ID on Weiyun. pdir_key: Parent directory key. local_path: Local destination path. Returns: Response dict with download result. """ body = { "file_list": [{"file_id": file_id, "pdir_key": pdir_key}], "download_type": 0, } result = self._api_request("DiskFileBatchDownload", body) if not result["success"]: return result file_list = result["data"].get("file_list", []) if not file_list: return build_response(False, message="No download info returned", error_code="API_ERROR") info = file_list[0] download_url = info.get("https_download_url", "") or info.get( "download_url", "") if not download_url: return build_response(False, message="No download URL returned", error_code="API_ERROR") # Set FTN cookie for download authentication cookie_name = info.get("cookie_name", "FTN5K") cookie_value = info.get("cookie_value", "") if cookie_value: self.session.cookies.set(cookie_name, cookie_value, domain=".weiyun.com", path="/") ensure_dir(local_path) start_time = time.time() try: resp = self.session.get(download_url, stream=True, timeout=300) resp.raise_for_status() with open(local_path, "wb") as f: for chunk in resp.iter_content(chunk_size=65536): if chunk: f.write(chunk) except requests.RequestException as e: return build_response(False, message=f"Download error: {e}", error_code="NETWORK_ERROR") elapsed = round(time.time() - start_time, 2) file_size = os.path.getsize(local_path) file_md5 = get_file_md5(local_path) return build_response(True, data={ "local_path": local_path, "size": file_size, "md5": file_md5, "elapsed": elapsed, }) def _download_package(self, pdir_key: str, dir_list: list, file_list: list, zip_filename: str, local_path: str) -> dict: """Download files/folders as a zip package using DiskFilePackageDownload. Args: pdir_key: Parent directory key. dir_list: List of dicts with dir_key and dir_name. file_list: List of dicts with file_id and pdir_key. zip_filename: Name for the zip file (URL-encoded). local_path: Local destination path for the zip file. Returns: Response dict with download result. """ import urllib.parse body = { "pdir_list": [{ "pdir_key": pdir_key, "dir_list": dir_list, "file_list": file_list, }], "zip_filename": urllib.parse.quote(zip_filename), } result = self._api_request("DiskFilePackageDownload", body) if not result["success"]: return result download_url = result["data"].get("https_download_url", "") or \ result["data"].get("download_url", "") if not download_url: return build_response(False, message="No download URL returned", error_code="API_ERROR") # Set FTN cookie for download authentication cookie_name = result["data"].get("cookie_name", "FTN5K") cookie_value = result["data"].get("cookie_value", "") if cookie_value: self.session.cookies.set(cookie_name, cookie_value, domain=".weiyun.com", path="/") ensure_dir(local_path) start_time = time.time() try: resp = self.session.get(download_url, stream=True, timeout=600) resp.raise_for_status() with open(local_path, "wb") as f: for chunk in resp.iter_content(chunk_size=65536): if chunk: f.write(chunk) except requests.RequestException as e: return build_response(False, message=f"Download error: {e}", error_code="NETWORK_ERROR") elapsed = round(time.time() - start_time, 2) file_size = os.path.getsize(local_path) file_md5 = get_file_md5(local_path) return build_response(True, data={ "local_path": local_path, "size": file_size, "size_str": format_size(file_size), "md5": file_md5, "elapsed": elapsed, }) def download_file(self, remote_path: str, local_path: str, overwrite: bool = False) -> dict: """Download a file from Weiyun to local. Supports both file path names and file_id based lookups. The remote_path can be a file name found via list_files. Args: remote_path: Remote file path or name on Weiyun. local_path: Local destination path. overwrite: Whether to overwrite local file. Returns: Response dict with download result. """ if os.path.exists(local_path) and not overwrite: return build_response( False, message=f"Local file already exists: {local_path}", error_code="DUPLICATE_NAME" ) # Need to find the file in the directory listing # Use root dir to search for the file root_key = self._get_root_dir_key() main_dir_key = getattr(self, "_main_dir_key", root_key) # List files in the main dir to find the target file list_result = self.list_files(main_dir_key) if not list_result["success"]: return list_result target_file = None for f in list_result["data"]["files"]: if f["type"] == "file" and f["name"] == remote_path: target_file = f break if not target_file: return build_response( False, message=f"File not found: {remote_path}", error_code="FILE_NOT_FOUND" ) return self._download_single_file( file_id=target_file["file_id"], pdir_key=main_dir_key, local_path=local_path, ) def download_folder(self, folder_name: str, local_path: str, overwrite: bool = False, as_zip: bool = False) -> dict: """Download a folder from Weiyun to local. Supports two modes: 1. Recursive download: downloads each file individually, preserving the folder structure locally. 2. Zip download: downloads the entire folder as a zip file. Args: folder_name: Folder name on Weiyun (as shown in list_files). local_path: Local destination directory (or zip file path). overwrite: Whether to overwrite existing local files. as_zip: If True, download as a single zip file. Returns: Response dict with download result. """ # Find the folder in root listing root_key = self._get_root_dir_key() main_dir_key = getattr(self, "_main_dir_key", root_key) list_result = self.list_files(main_dir_key) if not list_result["success"]: return list_result target_folder = None for f in list_result["data"]["files"]: if f["type"] == "folder" and f["name"] == folder_name: target_folder = f break if not target_folder: return build_response( False, message=f"Folder not found: {folder_name}", error_code="FOLDER_NOT_FOUND" ) folder_dir_key = target_folder["file_id"] if as_zip: # Download as zip package zip_path = local_path if os.path.isdir(local_path): zip_path = os.path.join(local_path, f"{folder_name}.zip") if os.path.exists(zip_path) and not overwrite: return build_response( False, message=f"Local file already exists: {zip_path}", error_code="DUPLICATE_NAME" ) return self._download_package( pdir_key=main_dir_key, dir_list=[{ "dir_key": folder_dir_key, "dir_name": folder_name, }], file_list=[], zip_filename=folder_name, local_path=zip_path, ) # Recursive download mode return self._download_folder_recursive( pdir_key=main_dir_key, dir_key=folder_dir_key, dir_name=folder_name, local_base=local_path, overwrite=overwrite, ) def _download_folder_recursive(self, pdir_key: str, dir_key: str, dir_name: str, local_base: str, overwrite: bool = False) -> dict: """Recursively download folder contents. Args: pdir_key: Parent directory key. dir_key: Directory key to download. dir_name: Directory name (for local path). local_base: Local base directory path. overwrite: Whether to overwrite existing files. Returns: Response dict with download summary. """ local_dir = os.path.join(local_base, dir_name) os.makedirs(local_dir, exist_ok=True) # List contents of this folder list_result = self.list_files(dir_key) if not list_result["success"]: return list_result downloaded_files = [] failed_files = [] total_size = 0 start_time = time.time() for item in list_result["data"]["files"]: if item["type"] == "folder": # Recurse into subfolder sub_result = self._download_folder_recursive( pdir_key=dir_key, dir_key=item["file_id"], dir_name=item["name"], local_base=local_dir, overwrite=overwrite, ) if sub_result["success"]: downloaded_files.extend( sub_result["data"].get("downloaded_files", [])) failed_files.extend( sub_result["data"].get("failed_files", [])) total_size += sub_result["data"].get("total_size", 0) else: failed_files.append({ "name": item["name"], "error": sub_result["message"], }) else: # Download file if not item["name"]: # Skip files with empty names failed_files.append({ "name": "(unnamed)", "error": "File has no name, skipped", }) continue file_local_path = os.path.join(local_dir, item["name"]) if os.path.exists(file_local_path) and not overwrite: failed_files.append({ "name": item["name"], "error": "File already exists", }) continue result = self._download_single_file( file_id=item["file_id"], pdir_key=dir_key, local_path=file_local_path, ) if result["success"]: downloaded_files.append({ "name": item["name"], "local_path": file_local_path, "size": result["data"]["size"], "size_str": format_size(result["data"]["size"]), }) total_size += result["data"]["size"] else: failed_files.append({ "name": item["name"], "error": result["message"], }) elapsed = round(time.time() - start_time, 2) return build_response(True, data={ "folder_name": dir_name, "local_path": local_dir, "downloaded_files": downloaded_files, "failed_files": failed_files, "downloaded_count": len(downloaded_files), "failed_count": len(failed_files), "total_size": total_size, "total_size_str": format_size(total_size), "elapsed": elapsed, }) def delete_file(self, remote_path: str, permanent: bool = False) -> dict: """Delete a file or folder on Weiyun. Args: remote_path: Path to delete. permanent: If True, permanently delete (skip recycle bin). Returns: Response dict with deletion result. """ body = { "file_path": remote_path, "permanent": permanent, } result = self._api_request("DiskFileDelete", body) if not result["success"]: return result return build_response(True, data={ "deleted_path": remote_path, "is_permanent": permanent, "deleted_at": get_timestamp(), }) def move_file(self, source_path: str, target_path: str) -> dict: """Move a file or folder to another directory. Args: source_path: Source file/folder path. target_path: Target directory path. Returns: Response dict with move result. """ body = { "src_path": source_path, "dst_path": target_path, } result = self._api_request("DiskFileMove", body) if not result["success"]: return result file_name = os.path.basename(source_path) new_path = os.path.join(target_path, file_name) return build_response(True, data={ "source_path": source_path, "target_path": new_path, }) def copy_file(self, source_path: str, target_path: str) -> dict: """Copy a file or folder to another directory. Args: source_path: Source file/folder path. target_path: Target directory path. Returns: Response dict with copy result. """ body = { "src_path": source_path, "dst_path": target_path, } result = self._api_request("DiskFileCopy", body) if not result["success"]: return result file_name = os.path.basename(source_path) new_path = os.path.join(target_path, file_name) return build_response(True, data={ "source_path": source_path, "target_path": new_path, "new_file_id": result["data"].get("file_id", ""), }) def rename_file(self, remote_path: str, new_name: str) -> dict: """Rename a file or folder. Args: remote_path: Current file path. new_name: New name (without path). Returns: Response dict with rename result. """ body = { "file_path": remote_path, "new_name": new_name, } result = self._api_request("DiskFileRename", body) if not result["success"]: return result parent_dir = os.path.dirname(remote_path) new_path = os.path.join(parent_dir, new_name) return build_response(True, data={ "old_path": remote_path, "new_path": new_path, }) def create_folder(self, remote_path: str) -> dict: """Create a folder on Weiyun. Args: remote_path: Folder path to create. Returns: Response dict with folder creation result. """ body = {"dir_path": remote_path} result = self._api_request("DiskDirCreate", body) if not result["success"]: return result return build_response(True, data={ "folder_id": result["data"].get("dir_key", ""), "path": remote_path, "created_at": get_timestamp(), }) def search_files(self, keyword: str, file_type: str = "all", page: int = 1, page_size: int = 50) -> dict: """Search files by keyword. Args: keyword: Search keyword. file_type: Type filter - 'all', 'document', 'image', 'video', 'audio'. page: Page number. page_size: Items per page. Returns: Response dict with search results. """ type_map = {"all": 0, "document": 1, "image": 2, "video": 3, "audio": 4} body = { "keyword": keyword, "search_type": type_map.get(file_type, 0), "start": (page - 1) * page_size, "count": page_size, } result = self._api_request("DiskFileSearch", body) if not result["success"]: return result raw_files = result["data"].get("file_list", []) results = [] for f in raw_files: size = f.get("file_size", 0) results.append({ "file_id": f.get("file_id", ""), "name": f.get("file_name", ""), "type": "file", "size_str": format_size(size), "path": f.get("file_path", ""), }) return build_response(True, data={ "results": results, "total": result["data"].get("total_count", len(results)), }) # ==================== Share Management ==================== def create_share(self, remote_path: str, expire_days: int = 0, password: str = None) -> dict: """Create a share link for a file or folder. Args: remote_path: Path to share. expire_days: Expiry in days (0 = permanent). password: Optional 4-char password. Returns: Response dict with share link info. """ body = { "file_path": remote_path, "expire_days": expire_days, } if password: body["password"] = password result = self._api_request("DiskShareCreate", body) if not result["success"]: return result return build_response(True, data={ "share_id": result["data"].get("share_id", ""), "share_url": result["data"].get("share_url", ""), "password": password or "", "expire_at": result["data"].get("expire_time", ""), "created_at": get_timestamp(), }) def cancel_share(self, share_id: str) -> dict: """Cancel an existing share. Args: share_id: Share ID to cancel. Returns: Response dict with cancellation result. """ body = {"share_id": share_id} result = self._api_request("DiskShareCancel", body) if not result["success"]: return result return build_response(True, data={ "share_id": share_id, "cancelled_at": get_timestamp(), }) def list_shares(self, status: str = "all", page: int = 1, page_size: int = 20) -> dict: """List all share links. Args: status: Filter by status - 'all', 'active', 'expired'. page: Page number. page_size: Items per page. Returns: Response dict with share list. """ status_map = {"all": 0, "active": 1, "expired": 2} body = { "status": status_map.get(status, 0), "start": (page - 1) * page_size, "count": page_size, } result = self._api_request("DiskShareList", body) if not result["success"]: return result raw_shares = result["data"].get("share_list", []) shares = [] for s in raw_shares: shares.append({ "share_id": s.get("share_id", ""), "share_url": s.get("share_url", ""), "file_name": s.get("file_name", ""), "status": s.get("status", ""), "view_count": s.get("view_count", 0), "download_count": s.get("download_count", 0), "created_at": s.get("create_time", ""), "expire_at": s.get("expire_time", ""), }) return build_response(True, data={ "shares": shares, "total": result["data"].get("total_count", len(shares)), }) # ==================== Space Management ==================== def get_space_info(self) -> dict: """Get storage space usage information. Returns: Response dict with space info. """ result = self._api_request("DiskSpaceQuery", {}) if not result["success"]: return result total = result["data"].get("total_space", 0) used = result["data"].get("used_space", 0) free = max(0, total - used) percent = round((used / total * 100), 1) if total > 0 else 0 return build_response(True, data={ "total_space": total, "total_space_str": format_size(total), "used_space": used, "used_space_str": format_size(used), "free_space": free, "free_space_str": format_size(free), "usage_percent": percent, "file_count": result["data"].get("file_count", 0), "folder_count": result["data"].get("dir_count", 0), }) def get_recycle_bin(self, page: int = 1, page_size: int = 50) -> dict: """Get files in recycle bin. Args: page: Page number. page_size: Items per page. Returns: Response dict with recycle bin contents. """ body = { "start": (page - 1) * page_size, "count": page_size, } result = self._api_request("DiskRecycleList", body) if not result["success"]: return result raw_files = result["data"].get("file_list", []) files = [] total_size = 0 for f in raw_files: size = f.get("file_size", 0) total_size += size files.append({ "file_id": f.get("file_id", ""), "name": f.get("file_name", ""), "size_str": format_size(size), "original_path": f.get("original_path", ""), "deleted_at": f.get("delete_time", ""), }) return build_response(True, data={ "files": files, "total": result["data"].get("total_count", len(files)), "total_size_str": format_size(total_size), }) def restore_file(self, file_id: str) -> dict: """Restore a file from recycle bin. Args: file_id: File ID in recycle bin. Returns: Response dict with restoration result. """ body = {"file_id": file_id} result = self._api_request("DiskRecycleRestore", body) if not result["success"]: return result return build_response(True, data={ "file_id": file_id, "restored_path": result["data"].get("restore_path", ""), "restored_at": get_timestamp(), }) def clear_recycle_bin(self, confirm: bool = False) -> dict: """Clear all files in recycle bin. This action is irreversible! Args: confirm: Must be True to execute. Returns: Response dict with clear result. """ if not confirm: return build_response( False, message="Must set confirm=True to clear recycle bin", error_code="INVALID_PARAM" ) result = self._api_request("DiskRecycleClear", {}) if not result["success"]: return result return build_response(True, data={ "deleted_count": result["data"].get("delete_count", 0), "freed_space_str": format_size( result["data"].get("freed_space", 0) ), "cleared_at": get_timestamp(), }) FILE:weiyun_skills/login.py """Weiyun login module - supports QR code login and cookies login.""" import io import json import time import argparse import requests try: import qrcode except ImportError: qrcode = None from weiyun_skills.utils import ( parse_cookies_str, cookies_dict_to_str, get_timestamp, build_response, ) # Weiyun QR code login endpoints XLOGIN_URL = "https://xui.ptlogin2.qq.com/cgi-bin/xlogin" PTQRSHOW_URL = "https://ssl.ptlogin2.qq.com/ptqrshow" PTQRLOGIN_URL = "https://ssl.ptlogin2.qq.com/ptqrlogin" CHECK_URL = "https://www.weiyun.com/disk" DEFAULT_COOKIES_PATH = "cookies.json" def _save_cookies(cookies_data: dict, save_path: str) -> None: """Save cookies data to a JSON file. Args: cookies_data: Cookies data to save. save_path: File path to save cookies. """ cookies_data["update_time"] = get_timestamp() with open(save_path, "w", encoding="utf-8") as f: json.dump(cookies_data, f, ensure_ascii=False, indent=2) def load_cookies(save_path: str = DEFAULT_COOKIES_PATH) -> dict: """Load cookies from a JSON file. Args: save_path: Path to cookies file. Returns: Cookies data dict, or empty dict if file not found. """ try: with open(save_path, "r", encoding="utf-8") as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError): return {} def _display_qr_terminal(img_bytes: bytes) -> None: """Display QR code image in terminal using half-block characters. Also saves the QR code as PNG files (original + enlarged) for easy scanning. Args: img_bytes: QR code image bytes. """ import os # Save QR code as image file for easy access qr_path = os.path.join(os.getcwd(), "weiyun_qrcode.png") try: with open(qr_path, "wb") as f: f.write(img_bytes) print(f"[*] QR code saved to: {qr_path}") except Exception as e: print(f"[WARN] Failed to save QR code image: {e}") # Try to display in terminal using pixel-accurate half-block rendering try: from PIL import Image img = Image.open(io.BytesIO(img_bytes)).convert("L") w, h = img.size # Save an enlarged version for easier phone scanning if w < 300: scale_factor = 300 // w + 1 enlarged = img.resize( (w * scale_factor, h * scale_factor), Image.NEAREST ) enlarged_path = qr_path.replace(".png", "_large.png") enlarged.save(enlarged_path) print(f"[*] Enlarged QR saved to: {enlarged_path}") # Render using half-block characters (▀ ▄ █ and space) # Each character represents 2 vertical pixels for accurate display print() for y in range(0, h - 1, 2): line = " " # indent for x in range(w): top = img.getpixel((x, y)) < 128 bottom = img.getpixel((x, y + 1)) < 128 if y + 1 < h else False if top and bottom: line += "█" elif top and not bottom: line += "▀" elif not top and bottom: line += "▄" else: line += " " print(line) if h % 2 == 1: line = " " for x in range(w): line += "▀" if img.getpixel((x, h - 1)) < 128 else " " print(line) print() except ImportError: print("[WARN] Pillow not installed. Please open the PNG file to scan.") print(" Install: pip install Pillow") def qrcode_login(save_path: str = DEFAULT_COOKIES_PATH) -> dict: """Login to Weiyun via QR code scanning. Displays a QR code in terminal for the user to scan with WeChat/QQ. After successful login, cookies are saved to the specified file. Args: save_path: Path to save cookies file. Returns: Response dict with login result. """ session = requests.Session() session.headers.update({ "User-Agent": ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36" ), "Referer": "https://www.weiyun.com/", }) # Step 1: Visit xlogin page to get initial cookies print("[*] Initializing QR code login...") try: xlogin_params = { "appid": "527020901", "daid": "372", "style": "33", "login_text": "登录", "hide_title_bar": "1", "hide_border": "1", "target": "self", "s_url": "https://www.weiyun.com/web/callback/common_qq_login_ok.html", "pt_3rd_aid": "0", "pt_feedback_link": "https://support.qq.com/", } session.get(XLOGIN_URL, params=xlogin_params, timeout=10) except requests.RequestException as e: return build_response(False, message=f"Failed to init login: {e}", error_code="NETWORK_ERROR") # Step 2: Get QR code image print("[*] Fetching QR code...") try: qr_params = { "appid": "527020901", "e": "2", "l": "M", "s": "3", "d": "72", "v": "4", "t": str(time.time()), "daid": "372", "pt_3rd_aid": "0", } qr_resp = session.get(PTQRSHOW_URL, params=qr_params, timeout=10) if qr_resp.status_code != 200: return build_response(False, message="Failed to get QR code", error_code="NETWORK_ERROR") # Get qrsig cookie for polling qrsig = session.cookies.get("qrsig", "") if not qrsig: return build_response(False, message="Failed to get qrsig cookie", error_code="AUTH_FAILED") except requests.RequestException as e: return build_response(False, message=f"Failed to fetch QR code: {e}", error_code="NETWORK_ERROR") # Step 3: Display QR code in terminal print("\n" + "=" * 50) print(" Scan the QR code below with WeChat/QQ") print("=" * 50 + "\n") _display_qr_terminal(qr_resp.content) print("\n" + "=" * 50) print(" Waiting for scan...") print("=" * 50 + "\n") # Step 4: Poll for login result def _hash33(t: str) -> int: """Hash function for ptqrtoken.""" e = 0 for c in t: e += (e << 5) + ord(c) return e & 0x7FFFFFFF ptqrtoken = _hash33(qrsig) max_attempts = 60 # Wait up to ~120 seconds for attempt in range(max_attempts): try: poll_params = { "u1": "https://www.weiyun.com/web/callback/common_qq_login_ok.html", "ptqrtoken": str(ptqrtoken), "ptredirect": "0", "h": "1", "t": "1", "g": "1", "from_ui": "1", "ptlang": "2052", "action": f"0-0-{int(time.time() * 1000)}", "js_ver": "24012815", "js_type": "1", "login_sig": session.cookies.get("pt_login_sig", ""), "pt_uistyle": "40", "aid": "527020901", "daid": "372", "pt_3rd_aid": "0", "o1vId": "", } poll_resp = session.get(PTQRLOGIN_URL, params=poll_params, timeout=10) resp_text = poll_resp.text if "登录成功" in resp_text or "登陆成功" in resp_text: print("[✓] Login successful!") # Extract redirect URL and follow it to get final cookies import re url_match = re.search(r"'(https?://[^']+)'", resp_text) if url_match: redirect_url = url_match.group(1) session.get(redirect_url, timeout=10, allow_redirects=True) # Collect all cookies all_cookies = {} for cookie in session.cookies: all_cookies[cookie.name] = cookie.value cookies_str = cookies_dict_to_str(all_cookies) cookies_data = { "cookies_str": cookies_str, "cookies_dict": all_cookies, "uin": all_cookies.get("uin", ""), } _save_cookies(cookies_data, save_path) return build_response(True, data={ "uin": all_cookies.get("uin", ""), "nickname": "", "save_path": save_path, }) elif "二维码未失效" in resp_text: pass # QR code still valid, keep waiting elif "二维码认证中" in resp_text or "已扫描" in resp_text: print("[*] QR code scanned, waiting for confirmation...") elif "二维码已失效" in resp_text: print("[✗] QR code expired!") return build_response(False, message="QR code expired", error_code="QR_EXPIRED") elif "本次登录已被拒绝" in resp_text: print("[✗] Login cancelled by user!") return build_response(False, message="Login cancelled", error_code="QR_CANCELLED") except requests.RequestException: pass time.sleep(2) return build_response(False, message="Login timeout", error_code="QR_EXPIRED") def cookies_login(cookies_str: str, save_path: str = DEFAULT_COOKIES_PATH) -> dict: """Login to Weiyun using cookies copied from browser. Args: cookies_str: Cookie string from browser DevTools. save_path: Path to save cookies file. Returns: Response dict with login result. """ if not cookies_str or not cookies_str.strip(): return build_response(False, message="Cookies string is empty", error_code="INVALID_PARAM") cookies_dict = parse_cookies_str(cookies_str) # Verify cookies by making a test request print("[*] Verifying cookies...") session = requests.Session() session.headers.update({ "User-Agent": ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36" ), "Referer": "https://www.weiyun.com/", }) session.cookies.update(cookies_dict) try: resp = session.get("https://www.weiyun.com/disk", timeout=10, allow_redirects=False) # If we get redirected to login page, cookies are invalid if resp.status_code in (301, 302): location = resp.headers.get("Location", "") if "login" in location.lower() or "xui.ptlogin2" in location: return build_response(False, message="Cookies are invalid or expired", error_code="AUTH_EXPIRED") except requests.RequestException as e: return build_response(False, message=f"Verification failed: {e}", error_code="NETWORK_ERROR") print("[✓] Cookies verified successfully!") # Save cookies cookies_data = { "cookies_str": cookies_str.strip(), "cookies_dict": cookies_dict, "uin": cookies_dict.get("uin", ""), } _save_cookies(cookies_data, save_path) return build_response(True, data={ "uin": cookies_dict.get("uin", ""), "nickname": "", "save_path": save_path, }) def main(): """CLI entry point for login.""" parser = argparse.ArgumentParser(description="Weiyun Login Tool") parser.add_argument( "--method", choices=["qrcode", "cookies"], default="qrcode", help="Login method: qrcode (scan QR code) or cookies (paste cookies)" ) parser.add_argument( "--cookies", type=str, default="", help="Cookie string from browser (required for cookies method)" ) parser.add_argument( "--save-path", type=str, default=DEFAULT_COOKIES_PATH, help=f"Path to save cookies file (default: {DEFAULT_COOKIES_PATH})" ) args = parser.parse_args() if args.method == "qrcode": result = qrcode_login(save_path=args.save_path) elif args.method == "cookies": if not args.cookies: print("[ERROR] --cookies argument is required for cookies method") print("Usage: python login.py --method cookies --cookies \"your_cookies\"") return result = cookies_login(args.cookies, save_path=args.save_path) else: print(f"[ERROR] Unknown method: {args.method}") return if result["success"]: print(f"\n[✓] Login successful! Cookies saved to: {result['data']['save_path']}") else: print(f"\n[✗] Login failed: {result['message']}") if __name__ == "__main__": main() FILE:weiyun_skills/main.py """Weiyun Skills CLI - command line interface for Weiyun management.""" import sys import json import argparse from weiyun_skills.client import WeiyunClient from weiyun_skills.utils import format_size try: from tabulate import tabulate except ImportError: tabulate = None def _print_json(data: dict) -> None: """Pretty print JSON data.""" print(json.dumps(data, ensure_ascii=False, indent=2)) def _print_table(headers: list, rows: list) -> None: """Print data as a formatted table.""" if tabulate: print(tabulate(rows, headers=headers, tablefmt="grid")) else: # Fallback: simple table output print("\t".join(headers)) print("-" * (len(headers) * 20)) for row in rows: print("\t".join(str(c) for c in row)) def cmd_list(client: WeiyunClient, args) -> None: """Handle 'list' command.""" result = client.list_files( remote_path=args.path, sort_by=getattr(args, "sort", "name"), sort_order=getattr(args, "order", "asc"), ) if not result["success"]: print(f"[ERROR] {result['message']}") return files = result["data"]["files"] if not files: print(f"(empty directory: {args.path})") return headers = ["Type", "Name", "Size", "Updated"] rows = [] for f in files: icon = "📁" if f["type"] == "folder" else "📄" rows.append([icon, f["name"], f["size_str"], f["updated_at"]]) _print_table(headers, rows) print(f"\nTotal: {result['data']['total']} items") def cmd_upload(client: WeiyunClient, args) -> None: """Handle 'upload' command.""" print(f"[*] Uploading {args.local} -> {args.remote}") result = client.upload_file( args.local, args.remote, overwrite=getattr(args, "overwrite", False) ) if result["success"]: d = result["data"] print(f"[✓] Uploaded: {d['name']} ({format_size(d['size'])})") print(f" Path: {d['remote_path']}") print(f" MD5: {d['md5']}") else: print(f"[✗] Upload failed: {result['message']}") def cmd_upload_folder(client: WeiyunClient, args) -> None: """Handle 'upload-folder' command.""" remote = getattr(args, "remote", "/") print(f"[*] Uploading folder '{args.local}' -> Weiyun:{remote}") result = client.upload_folder( args.local, remote, overwrite=getattr(args, "overwrite", False) ) if result["success"]: d = result["data"] print(f"[✓] Uploaded folder: {d['folder_name']}") print(f" Files: {d['uploaded_count']} uploaded, " f"{d['failed_count']} failed") print(f" Size: {d['total_size_str']}") print(f" Time: {d['elapsed']}s") if d["uploaded_files"]: print("\n Uploaded files:") for f in d["uploaded_files"]: instant = " ⚡" if f.get("instant_upload") else "" print(f" 📄 {f['name']} ({f['size_str']}){instant}") if d["failed_files"]: print("\n Failed files:") for f in d["failed_files"]: print(f" ❌ {f['name']}: {f['error']}") else: print(f"[✗] Upload failed: {result['message']}") def cmd_download(client: WeiyunClient, args) -> None: """Handle 'download' command.""" print(f"[*] Downloading {args.remote} -> {args.local}") result = client.download_file( args.remote, args.local, overwrite=getattr(args, "overwrite", False) ) if result["success"]: d = result["data"] print(f"[✓] Downloaded: {d['local_path']} ({format_size(d['size'])})") print(f" MD5: {d['md5']}") print(f" Time: {d['elapsed']}s") else: print(f"[✗] Download failed: {result['message']}") def cmd_download_folder(client: WeiyunClient, args) -> None: """Handle 'download-folder' command.""" as_zip = getattr(args, "zip", False) mode = "zip" if as_zip else "recursive" print(f"[*] Downloading folder '{args.folder}' -> {args.local} (mode: {mode})") result = client.download_folder( args.folder, args.local, overwrite=getattr(args, "overwrite", False), as_zip=as_zip, ) if result["success"]: d = result["data"] if as_zip: print(f"[✓] Downloaded zip: {d['local_path']}") print(f" Size: {d.get('size_str', format_size(d.get('size', 0)))}") print(f" MD5: {d['md5']}") print(f" Time: {d['elapsed']}s") else: print(f"[✓] Downloaded folder: {d['local_path']}") print(f" Files: {d['downloaded_count']} downloaded, " f"{d['failed_count']} failed") print(f" Size: {d['total_size_str']}") print(f" Time: {d['elapsed']}s") if d["downloaded_files"]: print("\n Downloaded files:") for f in d["downloaded_files"]: print(f" 📄 {f['name']} ({f['size_str']})") if d["failed_files"]: print("\n Failed files:") for f in d["failed_files"]: print(f" ❌ {f['name']}: {f['error']}") else: print(f"[✗] Download failed: {result['message']}") def cmd_delete(client: WeiyunClient, args) -> None: """Handle 'delete' command.""" permanent = getattr(args, "permanent", False) action = "Permanently deleting" if permanent else "Deleting" print(f"[*] {action} {args.path}") result = client.delete_file(args.path, permanent=permanent) if result["success"]: d = result["data"] msg = "permanently deleted" if d["is_permanent"] else "moved to recycle bin" print(f"[✓] {d['deleted_path']} {msg}") else: print(f"[✗] Delete failed: {result['message']}") def cmd_move(client: WeiyunClient, args) -> None: """Handle 'move' command.""" print(f"[*] Moving {args.source} -> {args.target}") result = client.move_file(args.source, args.target) if result["success"]: d = result["data"] print(f"[✓] Moved to: {d['target_path']}") else: print(f"[✗] Move failed: {result['message']}") def cmd_copy(client: WeiyunClient, args) -> None: """Handle 'copy' command.""" print(f"[*] Copying {args.source} -> {args.target}") result = client.copy_file(args.source, args.target) if result["success"]: d = result["data"] print(f"[✓] Copied to: {d['target_path']}") else: print(f"[✗] Copy failed: {result['message']}") def cmd_rename(client: WeiyunClient, args) -> None: """Handle 'rename' command.""" print(f"[*] Renaming {args.path} -> {args.name}") result = client.rename_file(args.path, args.name) if result["success"]: d = result["data"] print(f"[✓] Renamed: {d['old_path']} -> {d['new_path']}") else: print(f"[✗] Rename failed: {result['message']}") def cmd_mkdir(client: WeiyunClient, args) -> None: """Handle 'mkdir' command.""" print(f"[*] Creating folder: {args.path}") result = client.create_folder(args.path) if result["success"]: print(f"[✓] Created: {result['data']['path']}") else: print(f"[✗] Create failed: {result['message']}") def cmd_search(client: WeiyunClient, args) -> None: """Handle 'search' command.""" result = client.search_files( keyword=args.keyword, file_type=getattr(args, "type", "all"), ) if not result["success"]: print(f"[ERROR] {result['message']}") return results = result["data"]["results"] if not results: print(f"No files found matching '{args.keyword}'") return headers = ["Name", "Size", "Path"] rows = [[r["name"], r["size_str"], r["path"]] for r in results] _print_table(headers, rows) print(f"\nFound: {result['data']['total']} items") def cmd_share(client: WeiyunClient, args) -> None: """Handle 'share' command.""" print(f"[*] Creating share for: {args.path}") result = client.create_share( args.path, expire_days=getattr(args, "expire", 0), password=getattr(args, "password", None), ) if result["success"]: d = result["data"] print(f"[✓] Share created!") print(f" URL: {d['share_url']}") if d.get("password"): print(f" Password: {d['password']}") if d.get("expire_at"): print(f" Expires: {d['expire_at']}") else: print(f"[✗] Share failed: {result['message']}") def cmd_unshare(client: WeiyunClient, args) -> None: """Handle 'unshare' command.""" result = client.cancel_share(args.share_id) if result["success"]: print(f"[✓] Share {args.share_id} cancelled") else: print(f"[✗] Cancel failed: {result['message']}") def cmd_shares(client: WeiyunClient, args) -> None: """Handle 'shares' command.""" result = client.list_shares( status=getattr(args, "status", "all") ) if not result["success"]: print(f"[ERROR] {result['message']}") return shares = result["data"]["shares"] if not shares: print("No shares found") return headers = ["ID", "File", "URL", "Status", "Views", "Downloads", "Expires"] rows = [] for s in shares: rows.append([ s["share_id"], s["file_name"], s["share_url"], s["status"], s["view_count"], s["download_count"], s["expire_at"], ]) _print_table(headers, rows) print(f"\nTotal: {result['data']['total']} shares") def cmd_space(client: WeiyunClient, args) -> None: """Handle 'space' command.""" result = client.get_space_info() if not result["success"]: print(f"[ERROR] {result['message']}") return d = result["data"] print("=" * 40) print(" Weiyun Space Usage") print("=" * 40) print(f" Total: {d['total_space_str']}") print(f" Used: {d['used_space_str']} ({d['usage_percent']}%)") print(f" Free: {d['free_space_str']}") print(f" Files: {d['file_count']}") print(f" Folders: {d['folder_count']}") print("=" * 40) # Simple progress bar bar_width = 30 filled = int(bar_width * d["usage_percent"] / 100) bar = "█" * filled + "░" * (bar_width - filled) print(f" [{bar}] {d['usage_percent']}%") def cmd_recycle(client: WeiyunClient, args) -> None: """Handle 'recycle' command.""" result = client.get_recycle_bin() if not result["success"]: print(f"[ERROR] {result['message']}") return files = result["data"]["files"] if not files: print("Recycle bin is empty") return headers = ["ID", "Name", "Size", "Original Path", "Deleted At"] rows = [] for f in files: rows.append([ f["file_id"], f["name"], f["size_str"], f["original_path"], f["deleted_at"], ]) _print_table(headers, rows) print(f"\nTotal: {result['data']['total']} items " f"({result['data']['total_size_str']})") def cmd_restore(client: WeiyunClient, args) -> None: """Handle 'restore' command.""" result = client.restore_file(args.file_id) if result["success"]: d = result["data"] print(f"[✓] Restored to: {d['restored_path']}") else: print(f"[✗] Restore failed: {result['message']}") def cmd_clear_recycle(client: WeiyunClient, args) -> None: """Handle 'clear-recycle' command.""" if not getattr(args, "confirm", False): print("[!] This will permanently delete all files in recycle bin!") print(" Add --confirm flag to proceed.") return result = client.clear_recycle_bin(confirm=True) if result["success"]: d = result["data"] print(f"[✓] Recycle bin cleared!") print(f" Deleted: {d['deleted_count']} files") print(f" Freed: {d['freed_space_str']}") else: print(f"[✗] Clear failed: {result['message']}") def main(): """Main CLI entry point.""" parser = argparse.ArgumentParser( description="Weiyun Skills - Tencent Weiyun Management CLI", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python -m weiyun_skills.main list / python -m weiyun_skills.main upload ./file.pdf /docs/ python -m weiyun_skills.main download /docs/file.pdf ./ python -m weiyun_skills.main share /docs/file.pdf --expire 7 python -m weiyun_skills.main space """ ) parser.add_argument( "--cookies", type=str, default=None, help="Cookies string (overrides cookies.json)" ) subparsers = parser.add_subparsers(dest="command", help="Available commands") # list p_list = subparsers.add_parser("list", help="List files in a directory") p_list.add_argument("path", nargs="?", default="/", help="Directory path") p_list.add_argument("--sort", choices=["name", "size", "time"], default="name") p_list.add_argument("--order", choices=["asc", "desc"], default="asc") # upload p_upload = subparsers.add_parser("upload", help="Upload a file") p_upload.add_argument("local", help="Local file path") p_upload.add_argument("remote", help="Remote target path") p_upload.add_argument("--overwrite", action="store_true") # upload-folder p_ul_folder = subparsers.add_parser("upload-folder", help="Upload a folder") p_ul_folder.add_argument("local", help="Local folder path") p_ul_folder.add_argument("remote", nargs="?", default="/", help="Remote target path (default: root)") p_ul_folder.add_argument("--overwrite", action="store_true", help="Overwrite existing files on Weiyun") # download p_download = subparsers.add_parser("download", help="Download a file") p_download.add_argument("remote", help="Remote file path") p_download.add_argument("local", help="Local save path") p_download.add_argument("--overwrite", action="store_true") # download-folder p_dl_folder = subparsers.add_parser("download-folder", help="Download a folder") p_dl_folder.add_argument("folder", help="Folder name on Weiyun") p_dl_folder.add_argument("local", help="Local save directory") p_dl_folder.add_argument("--overwrite", action="store_true", help="Overwrite existing local files") p_dl_folder.add_argument("--zip", action="store_true", help="Download as zip file instead of " "individual files") # delete p_delete = subparsers.add_parser("delete", help="Delete a file") p_delete.add_argument("path", help="File path to delete") p_delete.add_argument("--permanent", action="store_true", help="Permanently delete (skip recycle bin)") # move p_move = subparsers.add_parser("move", help="Move a file") p_move.add_argument("source", help="Source path") p_move.add_argument("target", help="Target directory path") # copy p_copy = subparsers.add_parser("copy", help="Copy a file") p_copy.add_argument("source", help="Source path") p_copy.add_argument("target", help="Target directory path") # rename p_rename = subparsers.add_parser("rename", help="Rename a file") p_rename.add_argument("path", help="File path") p_rename.add_argument("name", help="New name") # mkdir p_mkdir = subparsers.add_parser("mkdir", help="Create a folder") p_mkdir.add_argument("path", help="Folder path") # search p_search = subparsers.add_parser("search", help="Search files") p_search.add_argument("keyword", help="Search keyword") p_search.add_argument("--type", choices=["all", "document", "image", "video", "audio"], default="all") # share p_share = subparsers.add_parser("share", help="Create a share link") p_share.add_argument("path", help="File path to share") p_share.add_argument("--expire", type=int, default=0, help="Expire in days (0=permanent)") p_share.add_argument("--password", type=str, default=None, help="Share password (4 chars)") # unshare p_unshare = subparsers.add_parser("unshare", help="Cancel a share") p_unshare.add_argument("share_id", help="Share ID to cancel") # shares p_shares = subparsers.add_parser("shares", help="List all shares") p_shares.add_argument("--status", choices=["all", "active", "expired"], default="all") # space subparsers.add_parser("space", help="Show space usage") # recycle subparsers.add_parser("recycle", help="Show recycle bin") # restore p_restore = subparsers.add_parser("restore", help="Restore from recycle bin") p_restore.add_argument("file_id", help="File ID to restore") # clear-recycle p_clear = subparsers.add_parser("clear-recycle", help="Clear recycle bin") p_clear.add_argument("--confirm", action="store_true", help="Confirm clear operation") args = parser.parse_args() if not args.command: parser.print_help() sys.exit(1) # Initialize client client = WeiyunClient(cookies_str=args.cookies) # Dispatch command commands = { "list": cmd_list, "upload": cmd_upload, "upload-folder": cmd_upload_folder, "download": cmd_download, "download-folder": cmd_download_folder, "delete": cmd_delete, "move": cmd_move, "copy": cmd_copy, "rename": cmd_rename, "mkdir": cmd_mkdir, "search": cmd_search, "share": cmd_share, "unshare": cmd_unshare, "shares": cmd_shares, "space": cmd_space, "recycle": cmd_recycle, "restore": cmd_restore, "clear-recycle": cmd_clear_recycle, } handler = commands.get(args.command) if handler: handler(client, args) else: parser.print_help() if __name__ == "__main__": main() FILE:weiyun_skills/utils.py """Utility functions for Weiyun Skills.""" import os import hashlib from datetime import datetime def format_size(size_bytes: int) -> str: """Convert bytes to human-readable size string. Args: size_bytes: File size in bytes. Returns: Human-readable size string like '2.50 MB'. """ if size_bytes == 0: return "0 B" units = ["B", "KB", "MB", "GB", "TB"] unit_index = 0 size = float(size_bytes) while size >= 1024 and unit_index < len(units) - 1: size /= 1024 unit_index += 1 return f"{size:.2f} {units[unit_index]}" def get_file_md5(file_path: str) -> str: """Calculate MD5 hash of a file. Args: file_path: Path to the file. Returns: MD5 hex digest string. """ md5 = hashlib.md5() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): md5.update(chunk) return md5.hexdigest() def parse_cookies_str(cookies_str: str) -> dict: """Parse cookies string into a dictionary. Args: cookies_str: Cookie string from browser, e.g. 'key1=val1; key2=val2'. Returns: Dictionary of cookie key-value pairs. """ cookies = {} if not cookies_str: return cookies for item in cookies_str.split(";"): item = item.strip() if "=" in item: key, value = item.split("=", 1) cookies[key.strip()] = value.strip() return cookies def cookies_dict_to_str(cookies_dict: dict) -> str: """Convert cookies dictionary back to string. Args: cookies_dict: Dictionary of cookies. Returns: Cookie string like 'key1=val1; key2=val2'. """ return "; ".join(f"{k}={v}" for k, v in cookies_dict.items()) def get_timestamp() -> str: """Get current timestamp string. Returns: Timestamp string in 'YYYY-MM-DD HH:MM:SS' format. """ return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def ensure_dir(path: str) -> None: """Ensure the directory for a file path exists. Args: path: File path whose parent directory should exist. """ dir_path = os.path.dirname(path) if dir_path: os.makedirs(dir_path, exist_ok=True) def build_response(success: bool, data=None, message: str = "ok", error_code: str = None) -> dict: """Build a standardized response dict. Args: success: Whether the operation succeeded. data: Response data payload. message: Response message. error_code: Error code string if failed. Returns: Standardized response dictionary. """ resp = {"success": success, "data": data, "message": message} if error_code: resp["error_code"] = error_code return resp FILE:weiyun_skills/__init__.py """Weiyun Skills - Tencent Weiyun cloud storage management toolkit.""" from weiyun_skills.client import WeiyunClient __version__ = "1.0.0" __all__ = ["WeiyunClient"]

super-weiyun-skill

Comments (0)