feat: DocSystem (MxsDoc) 文件操作 skill

从公司 DocSystem / MxsDoc V2.02.71 逆向出 HTTP 接口,封装为零依赖 Python CLI,
让 AI 可以完全通过命令行操作文档系统的文件系统。

功能:
- 浏览:ls / tree / repos / auth
- 读写:mkdir(-p) / touch / upload / download / cat
- 变更:rename / rm / mv / cp
- 检索:search(文件名 + 正文)

要点:
- 不指定仓库时默认操作「研发管理配置库」(id=2);首个位置参数以 / 开头时自动判定为路径
- 凭证不入库,从 ~/.docsys/config.json 或环境变量读取
- 会话 cookie 缓存于 ~/.docsys/session.json,过期自动重登
This commit is contained in:
Cheney
2026-10-20 15:50:57 +08:00
commit 8ae1694b6d
4 changed files with 1431 additions and 0 deletions
+15
View File
@@ -0,0 +1,15 @@
# 本地工作区数据(会话记忆等,不入库)
.workbuddy/
__pycache__/
*.py[cod]
# 可能含个人凭证 / 会话的本地文件
config.json
session.json
*.local.json
.idea/
.vscode/
.DS_Store
Thumbs.db
+260
View File
@@ -0,0 +1,260 @@
# DocSystem(MxsDoc)HTTP 接口文档
目标系统:MxsDoc / DocSystem V2.02.71
地址:`http://172.1.4.31:8100/DocSystem`
前端入口:`/DocSystem/web/index.html` → `projects.html` → `project.html?vid=<仓库ID>`
本文档由前端 JS 逆向 + 逐接口实测得出,用于支撑 `scripts/docsys.py`。
---
## 通用约定
- 全部接口为 `POST`,表单编码 `application/x-www-form-urlencoded; charset=UTF-8`
(仅 `uploadDoc.do` 为 `multipart/form-data`)。
- 必须带请求头 `X-Requested-With: XMLHttpRequest`。
- 会话靠 Cookie 维持:`JSESSIONID`、`dstoken`(= base64(密码))、`dsuser`(= 用户名)。
`rememberMe=1` 时服务端下发 `dstoken`/`dsuser` 用于自动登录。
- 响应统一 JSON:`{"status":"ok"|"fail","data":...,"msgInfo":"...","msgData":...}`。
少数接口(如 `getDocContent.do`)返回纯文本。
- 节点类型:`type=1` 文件,`type=2` 目录。
- 时间字段为毫秒时间戳。
## 鉴权
### POST `User/login.do`
| 参数 | 说明 |
|---|---|
| `userName` | 邮箱或手机号 |
| `pwd` | 密码经 base64 编码后的字符串(标准 base64,非 URL-safe) |
| `rememberMe` | `1` 记住登录,`0` 不记住 |
成功返回 `data` 为用户对象(`id`/`name`/`realName`/`role`/…)。
### POST `User/logout.do`
无参数。
### POST `User/getLoginUser.do`
无参数,返回当前登录用户。可用于探测会话是否有效。
---
## 仓库
### POST `Repos/getReposList.do`
无参数。返回 `data` 为仓库数组:
`{id, name, info, type, verCtrl, path, owner, ...}`
实测可见仓库:
| id | 名称 | 说明 |
|---|---|---|
| 1 | 意见与反馈 | 当前账号可新增/编辑,不可删除/下载 |
| **2** | **研发管理配置库** | **CLI 默认仓库**(`DEFAULT_REPO_NAME`),基本只读,部分目录可下载 |
| 5 | 常用工具 | 只读 |
| 6 | 临时仓库 | 无访问权限 |
> **约定**:`scripts/docsys.py` 中若省略仓库参数,默认操作 id=2「研发管理配置库」。
> 判定规则:第一个位置参数以 `/` 或 `./` 开头则视为路径(仓库取默认),否则视为仓库 id/名称。
### POST `Repos/getDocSysConfig.do`
无参数,返回系统配置(如 `defaultReposStorePath`)。
---
## 目录列举
### POST `Repos/getSubDocList.do`
| 参数 | 说明 |
|---|---|
| `vid` | 仓库 id |
| `docId` | 目录 docId(**实测服务端不使用**,仅随请求透传) |
| `path` | 目录路径,仓库根为 `""`,其余以 `/` 结尾 |
| `name` | 目录名。服务端取 `path + name` 作为目标目录 |
| `sort` | 排序方式,`0` 名称 |
| `needLockState` | `1` |
| `shareId` | 可传空 |
**关键结论**:目标目录 = `path + name`。传 `path="a/"`、`name=""` 与 `path=""`、`name="a"`
等价;但传 `path="a/"` 且 `name="a"` 会变成 `a/a` 从而返回空。
返回 `data` 为节点数组,节点字段:
`docId, pid, path, name, type, level, size, checkSum, createTime, latestEditTime,
latestEditorName, creatorName, localRootPath, reposPath, ...`
其中 `path` 是**节点所在目录**的路径,节点完整路径 = `path + name`。
---
## 权限
### POST `Repos/getUserDocAuth.do`
| 参数 | 说明 |
|---|---|
| `docId` | 目录 docId,仓库根为 `0` |
| `reposId` | 仓库 id |
| `path` | 目录自身路径 |
| `name` | 目录名 |
返回 `data`:`{access, addEn, editEn, deleteEn, downloadEn, isAdmin, uploadSize, ...}`
实测(账号 `qian.huang@sunyard.com`,仓库 1):
```
access=1 addEn=1 editEn=1 deleteEn=0 downloadEn=0 isAdmin=0
```
注意:**rename / move 在服务端按"删除 + 新增"鉴权**,因此即使 `addEn=1`,
`deleteEn=0` 时重命名和移动同样会被拒绝,报错为「您没有该目录的删除权限」。
### 权限是逐目录授予的,不是仓库级
同一个仓库里不同目录的权限可以完全不同。实测:
| 仓库 / 目录 | access | addEn | editEn | deleteEn | downloadEn |
|---|---|---|---|---|---|
| 2 研发管理配置库 `/`(根) | 1 | 0 | 0 | 0 | 0 |
| 2 `/Sunyard/Docs/开发流程` | 1 | 0 | 0 | 0 | **1** |
| 2 `/Sunyard/Dev/产品/系统研发/工具` | 1 | **1** | **1** | **1** | **1** |
| 1 意见与反馈 `/`(根) | 1 | 1 | 1 | 0 | 0 |
所以**不能凭仓库名推断权限**,写操作前务必先查 `getUserDocAuth`。
---
## 写操作
### POST `Doc/addDoc.do`(新建文件 / 目录)
| 参数 | 说明 |
|---|---|
| `reposId` | 仓库 id |
| `type` | `1` 文件,`2` 目录 |
| `pid` | 父目录 docId,仓库根为 `0` |
| `path` | **父目录完整路径 + `/`**,仓库根为 `""` |
| `name` | 新节点名 |
| `content` | 文本文件内容;建目录传空 |
返回 `data` 为新节点,`msgData` 为 `isNewNode`。
> ⚠️ **实测:`content` 被服务端忽略。** 即使传了内容,建出来也是 0 字节空文件
> (响应 `size=0`,随后 `getDocContent` 读不到任何内容)。
> 要创建带内容的文件必须改用 `Doc/uploadDoc.do`(CLI 的 `touch` 已自动走这条路)。
### POST `Doc/renameDoc.do`
| 参数 | 说明 |
|---|---|
| `reposId` | 仓库 id |
| `docId` | 节点 docId |
| `pid` | 父目录 docId |
| `type` | 节点类型 |
| `level` | 节点层级 |
| `path` | 节点**自身所在目录**路径 |
| `name` | 原名 |
| `dstName` | 新名 |
### POST `Doc/deleteDoc.do`
| 参数 | 说明 |
|---|---|
| `reposId` | 仓库 id |
| `docId` | 节点 docId |
| `pid` | 父目录 docId |
| `path` | 节点自身所在目录路径 |
| `name` | 节点名 |
删除目录时为递归删除。
### POST `Doc/moveDoc.do`
| 参数 | 说明 |
|---|---|
| `reposId` | 仓库 id |
| `docId` | 节点 docId |
| `type` | 节点类型 |
| `srcLevel` | 源层级 |
| `dstLevel` | 目标层级(父目录 level + 1;移到仓库根为 `0`) |
| `srcPid` | 源父目录 docId |
| `dstPid` | 目标父目录 docId,仓库根为 `0` |
| `srcPath` | 源目录路径 |
| `srcName` | 源名 |
| `dstPath` | **目标父目录完整路径 + `/`**,仓库根为 `""` |
| `dstName` | 目标名(可与源名不同,即改名 + 移动) |
### POST `Doc/copyDoc.do`
参数同 `moveDoc`。
---
## 上传 / 下载
### POST `Doc/uploadDoc.do`(multipart/form-data)
| 字段 | 说明 |
|---|---|
| `reposId` | 仓库 id |
| `path` | **目标目录完整路径 + `/`** |
| `name` | 远端文件名 |
| `size` | 文件字节数 |
| `checkSum` | 文件 MD5 |
| `uploadFile` | 文件内容(文件域) |
| `commitMsg` | 提交说明,可空 |
| `dirPath` | 文件夹上传标记,可空 |
大文件走分片:`chunkNum` / `cutSize` / `chunkIndex` / `chunkSize` / `chunkHash` /
`combineDisabled=1`,分片全部传完再调 `Doc/combineChunks.do`。
### POST `Doc/downloadDocPrepare.do`
| 参数 | 说明 |
|---|---|
| `reposId`, `docId`, `pid`, `path`, `name` | 目标节点 |
| `downloadType` | `1` 单文件 |
| `downloadList` | 可空 |
返回 `data`:`{path, name, targetPath, targetName}`,`msgData` 为 `deleteFlag`。
若 `msgData == 5` 表示目录正在打包,用 `data.id` 轮询
`Doc/queryDownloadPrepareTask.do`(参数 `taskId`),每 2 秒一次,直到 `msgData == 6`。
### GET `Doc/downloadDoc.do?...`
查询串:`vid, path, name, targetPath, targetName, deleteFlag, encryptEn=1`(值需 URL 编码)。
> 实测:直接调 `Doc/getDocFileLink.do` 拿到的链接指向 `tmp/History/rdata/<commitId>/...`
> 历史目录,该文件通常不存在,会返回 500。**必须走 `downloadDocPrepare` 流程。**
---
## 读取 / 搜索
### POST `Doc/searchDoc.do`
| 参数 | 说明 |
|---|---|
| `searchWord` | 关键词 |
| `sort` | `0` |
| `reposId` | 仓库 id |
| `path` | 限定目录范围,全库传 `""` |
返回 `data` 为命中项,`hitType`:`1` 文件名命中,`3` 正文命中;
`content` 为正文的 base64 编码;`dataEx` 为命中总数。
### POST `Doc/getDocContent.do`
| 参数 | 说明 |
|---|---|
| `reposId`, `docId`, `pid`, `path`, `name` | 目标文件 |
返回**纯文本**,响应以 `ok` 开头,后面才是文件正文,解析时须去掉前两个字符。
### POST `Doc/getDoc.do`
| 参数 | 说明 |
|---|---|
| `reposId`, `docId`, `pid`, `path`, `name` | 目标节点 |
| `docType` | `0` 不读取正文和备注 |
返回节点元信息。
---
## 已探明但未使用的接口
- `Doc/lockDoc.do` / `Doc/unlockDoc.do` — 文件锁定
- `Doc/executeDoc.do`、`Doc/refreshDoc.do`
- `Doc/checkChunkUploaded.do`、`Doc/combineChunks.do`、`Doc/getMaxThreadCount.do` — 分片上传
- `Doc/getDocFileLink.do`、`Doc/getZipDocFileLink.do` — 预览/下载链接(下载场景不可靠,见上)
- `Bussiness/getDocOfficeLink.do`、`Bussiness/getDocCadLink.do` — 在线预览
- `Bussiness/addDocShare.do`、`Doc/getDocShare.do` — 分享
- `Doc/querySystemLog.do` — 操作日志
- `Repos/getReposInitMenu.do` — 目录树初始化菜单
+159
View File
@@ -0,0 +1,159 @@
---
name: docsys
description: 操作公司 MxsDoc / DocSystem 网络文档系统(http://172.1.4.31:8100/DocSystem)的文件系统——仓库浏览、目录创建/删除/重命名、文件上传/删除/重命名/移动/复制、以及按文件名或正文搜索。当用户提到"文档系统""DocSystem""MxsDoc""文档库""上传文件到文档系统""在文档系统里建目录/搜文件"时使用。
agent_created: true
---
# DocSystem 文档系统操作 Skill
通过纯 HTTP 调用 MxsDoc(DocSystem)后端接口,让 AI 直接操作公司文档系统的文件系统:
浏览、建目录、删除、重命名、移动、复制、上传、下载、搜索。
底层是一个零依赖的 Python CLI:只用标准库,不需要 pip 安装任何东西。
## 快速开始
```bash
PY="C:/Users/occul/.workbuddy/binaries/python/versions/3.13.12/python.exe"
SK="C:/Users/occul/WorkBuddy/docsys-skill/docsys"
$PY "$SK/scripts/docsys.py" login # 登录并缓存会话(首次执行即可)
$PY "$SK/scripts/docsys.py" repos # 列出可访问仓库
$PY "$SK/scripts/docsys.py" ls "意见与反馈" / # 列出仓库根目录
```
会话 cookie 缓存在 `~/.docsys/session.json`,过期时会自动重新登录,不需要每次显式 login。
## ⚠️ 默认仓库约定
**未指定仓库时一律使用 `研发管理配置库`(id=2)。** 用户说"上传个文件""搜一下 XX"而没提别的库时,不要反问,直接对这个库操作。
```bash
$PY "$SK/scripts/docsys.py" ls # = ls 研发管理配置库 /
$PY "$SK/scripts/docsys.py" ls /Sunyard # 路径以 / 开头,仓库自动取默认
$PY "$SK/scripts/docsys.py" ls 1 / # 显式切到其他库才带仓库参数
$PY "$SK/scripts/docsys.py" search 详细设计 # 默认搜默认库
$PY "$SK/scripts/docsys.py" search 详细设计 --all-repos # 例外:搜全部库
```
`repos` 命令会给默认库打 `← 默认` 标记。默认库可用环境变量 `DOCSYS_DEFAULT_REPO` 覆盖。
**判定规则**:第一个位置参数若以 `/` 或 `./` 开头 → 视为路径,仓库取默认;否则视为仓库名/id。所以 `ls /Sunyard` 和 `ls 研发管理配置库 /Sunyard` 等价。
内部实现是"位置参数整体左移一位",所以 `touch /a/b.txt "内容"` 这类后面还带可选参数的命令也不会错位。
## 命令一览
`<repo>` 全部可省略(省略即默认库):
| 命令 | 说明 |
|---|---|
| `login` / `logout` / `whoami` | 会话管理 |
| `repos` | 列出可访问仓库(id + 名称,标注默认) |
| `ls [repo] [path]` | 列出目录内容 |
| `tree [repo] [path] [--depth N]` | 递归列目录树 |
| `mkdir [repo] <path> [-p]` | 创建目录;默认要求父目录已存在,`-p` 递归创建中间目录 |
| `touch [repo] <path> [content]` | 创建文本文件;`content=@file --file x.txt` 从本地读内容 |
| `rename [repo] <path> <new_name>` | 重命名(不能跨目录) |
| `rm [repo] <path>` | 删除文件或目录(目录为递归删除) |
| `mv [repo] <src> <dst>` | 移动/改名。`dst` 是已存在目录则移入其中,否则视为目标完整路径 |
| `cp [repo] <src> <dst>` | 复制,语义同 `mv` |
| `upload [repo] <remote_dir> <local_file> [--name X]` | 上传本地文件 |
| `download [repo] <path> <local_file>` | 下载;`local_file` 为 `-` 时输出到 stdout |
| `cat [repo] <path>` | 读取文本文件内容 |
| `search <keyword> [--repo R] [--all-repos] [--path P]` | 按文件名或正文搜索;默认搜默认库,`--all-repos` 搜全部 |
| `auth [repo] [path]` | 查看当前账号在该目录的权限 |
全局参数:`--json`(结构化输出,推荐 AI 解析用)、`--base-url`、`--user`、`--password`。
## 用法约定(给 AI 的操作说明)
**0. 仓库默认就是「研发管理配置库」(id=2)。** 用户没指名别的仓库时不要反问、不要先跑 `repos` 确认,直接操作。切库时才带 repo 参数。
**1. 仓库也可以用 id 或名称指定**,名称支持子串匹配:
```
$PY ... ls 1 / # 按 id
$PY ... ls 常用工具 / # 按名称
$PY ... ls 研发 / # 子串,唯一匹配
```
**2. 路径一律用 `/a/b/c` 形式**,前导斜杠可省略,仓库根写 `/` 或留空。
单独给路径时(`ls /Sunyard`)脚本会自动识别为"默认库的该路径",不会误判成仓库名。
**3. 要解析用户输入("研发库的 2024 归档目录")时,先用 `repos` + `search` / `tree` 定位,不要猜。**
**4. 破坏性操作(`rm`、覆盖式 `mv`/`cp`)执行前先向用户确认**,并优先用 `ls` 或 `auth` 确认目标存在且有权限。
**5. 推荐加 `--json`**,输出结构稳定:
```json
[{"name":"t1.txt","fullPath":"_SELFTEST/t1.txt","type":"file","size":35,"mtime":"2026-10-20 14:33","docId":202070828624}]
```
## 权限边界(当前账号实测)
账号 `qian.huang@sunyard.com` 实测:
**权限是逐目录授予的,不是仓库级。** 同一个仓库里不同目录可以完全不同:
| 目录 | 访问 | 新增 | 编辑 | 删除 | 下载 |
|---|---|---|---|---|---|
| 研发管理配置库 `/`(根) | ✅ | ❌ | ❌ | ❌ | ❌ |
| 研发管理配置库 `/Sunyard/Docs/开发流程` | ✅ | ❌ | ❌ | ❌ | ✅ |
| **研发管理配置库 `/Sunyard/Dev/产品/系统研发/工具`** | ✅ | ✅ | ✅ | ✅ | ✅ |
| 意见与反馈 `/`(根) | ✅ | ✅ | ✅ | ❌ | ❌ |
所以**不要凭仓库名推断权限**,任何写操作前先跑 `auth <path>` 确认。
已验证可完整 CRUD 的目录:`/Sunyard/Dev/产品/系统研发/工具/测试`(增删改 + 下载全开)。
服务端权限不足时返回中文错误,CLI 原样透出:
```
[docsys] 您没有该目录的删除权限,请联系管理员
```
遇到这类报错不要反复重试、也不要自作主张换到别的目录/仓库,直接告诉用户需要管理员授权。
注意:重命名和移动在服务端按"删除 + 新增"鉴权,所以即使有新增权限也可能被拒。
## 接口契约(排障用)
全部为 `POST {BASE}/...`,`{BASE} = http://172.1.4.31:8100/DocSystem`,表单编码,
需要带 `X-Requested-With: XMLHttpRequest`,会话靠 cookie(`JSESSIONID` / `dstoken` / `dsuser`)。
登录密码先做 base64 编码再提交。响应统一为 `{status:"ok"|"fail", data, msgInfo}`。
| 接口 | 参数 |
|---|---|
| `User/login.do` | `userName`, `pwd`(base64), `rememberMe` |
| `Repos/getReposList.do` | 无 |
| `Repos/getSubDocList.do` | `vid`, `docId`, `path`, `name`, `sort`, `needLockState` |
| `Repos/getUserDocAuth.do` | `docId`, `reposId`, `path`, `name` |
| `Doc/addDoc.do` | `reposId`, `type`(1文件/2目录), `pid`, `path`, `name`, `content`(**实测 content 被忽略,建出来恒为 0 字节空文件**) |
| `Doc/renameDoc.do` | `reposId`, `docId`, `pid`, `type`, `level`, `path`, `name`, `dstName` |
| `Doc/deleteDoc.do` | `reposId`, `docId`, `pid`, `path`, `name` |
| `Doc/moveDoc.do` | `reposId`, `docId`, `type`, `srcLevel`, `dstLevel`, `srcPid`, `dstPid`, `srcPath`, `srcName`, `dstPath`, `dstName` |
| `Doc/copyDoc.do` | 同 `moveDoc` |
| `Doc/uploadDoc.do` | multipart:`reposId`, `path`, `name`, `size`, `checkSum`(md5), `uploadFile`, `commitMsg`, `dirPath` |
| `Doc/downloadDocPrepare.do` | `reposId`, `docId`, `pid`, `path`, `name`, `downloadType` |
| `Doc/downloadDoc.do` | GET:`vid`, `path`, `name`, `targetPath`, `targetName`, `deleteFlag`, `encryptEn=1` |
| `Doc/searchDoc.do` | `searchWord`, `sort`, `reposId`, `path` |
| `Doc/getDocContent.do` | `reposId`, `docId`, `pid`, `path`, `name`(返回纯文本,响应开头有 `ok` 前缀) |
### 路径语义(最容易踩的坑)
- 节点字段 `path` 是**它所在目录**的路径:根目录下为 `""`,其余以 `/` 结尾,如 `_SELFTEST/sub/`。
节点完整路径 = `path + name`。
- **列举**:`getSubDocList` 只看 `path + name` 拼接结果,`docId` 不参与定位。
要列 `_SELFTEST` 的子项,必须传 `path=_SELFTEST/`(或 `path=_SELFTEST/` 且 `name=` 空)。
- **写入**(addDoc / uploadDoc / moveDoc 的 `dstPath`):要传**父目录完整路径 + `/`**,
例如往 `_SELFTEST` 下建文件,`path` 应为 `_SELFTEST/`,而不是节点自身的 `""`。
- **删除 / 重命名**:传节点自身的 `path` 与 `name`。
### 其他服务端怪癖
- **`addDoc.do` 忽略 `content`**:新建文件永远得到 0 字节空文件,响应里 `size=0`,
`getDocContent` 也读不到内容。要写内容必须走 `uploadDoc.do`(CLI 的 `touch` 已自动改为走上传通道)。
- **`getDocContent.do` 返回纯文本**且开头带 `ok` 两个字符前缀,解析要剥掉;空文件时只返回 `ok`。
- **`getDocFileLink.do` 不可用**:链接指向 `tmp/History/rdata/<commitId>/...` 历史目录,文件不存在会 500。
下载必须走 `downloadDocPrepare.do` → `downloadDoc.do`。
## 文件
- `scripts/docsys.py` — CLI 实现(Python 3 标准库,零依赖)
- `README.md` — 逆向出的完整接口文档
+997
View File
@@ -0,0 +1,997 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
docsys.py - MxsDoc / DocSystem Web 文件系统的命令行客户端。
让 AI 以纯 HTTP 方式操作公司文档系统:建目录 / 删除 / 重命名 / 移动 /
上传 / 下载 / 搜索 / 浏览。不依赖任何第三方库,仅用 Python 标准库。
接口契约(由 MxsDoc V2.02.71 前端 JS 逆向 + 实测确认):
POST {BASE}/User/login.do userName, pwd(base64), rememberMe
POST {BASE}/User/getLoginUser.do
POST {BASE}/Repos/getReposList.do
POST {BASE}/Repos/getSubDocList.do vid, docId, path, name, sort, needLockState
POST {BASE}/Repos/getUserDocAuth.do docId, reposId, path, name
POST {BASE}/Doc/addDoc.do reposId, type(1文件/2目录), pid, path, name, content
POST {BASE}/Doc/renameDoc.do reposId, docId, pid, type, level, path, name, dstName
POST {BASE}/Doc/deleteDoc.do reposId, docId, pid, path, name
POST {BASE}/Doc/moveDoc.do reposId, docId, type, srcLevel, dstLevel,
srcPid, dstPid, srcPath, srcName, dstPath, dstName
POST {BASE}/Doc/copyDoc.do 同 moveDoc
POST {BASE}/Doc/uploadDoc.do multipart: reposId, path, name, size,
checkSum, uploadFile, commitMsg, dirPath
POST {BASE}/Doc/downloadDocPrepare.do reposId, docId, pid, path, name, downloadType
GET {BASE}/Doc/downloadDoc.do?... 实际取流
POST {BASE}/Doc/searchDoc.do searchWord, sort, reposId, path
POST {BASE}/Doc/getDocContent.do reposId, docId, pid, path, name
路径语义(关键,实测结论):
node.path 是「自身所在目录」,根目录为 "",其余以 "/" 结尾,例如 "_AI_PROBE/"。
节点的完整路径 = node.path + node.name。
getSubDocList 的 path + name 拼接后即为要列举的目录。
"""
import argparse
import base64
import hashlib
import http.cookiejar
import json
import mimetypes
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from urllib.request import HTTPCookieProcessor, Request, build_opener
DEFAULT_BASE_URL = "http://172.1.4.31:8100/DocSystem"
# 凭证不入库:从 ~/.docsys/config.json 或环境变量 DOCSYS_USER / DOCSYS_PASSWORD 读取。
# 示例(config.json):
# {"baseUrl": "http://172.1.4.31:8100/DocSystem",
# "user": "you@company.com", "password": "xxx"}
DEFAULT_USER = ""
DEFAULT_PASSWORD = ""
# 约定:不指定仓库时的默认库(研发管理配置库,vid=2)
# 可用 DOCSYS_DEFAULT_REPO 环境变量覆盖。
DEFAULT_REPO_NAME = os.environ.get("DOCSYS_DEFAULT_REPO") or "研发管理配置库"
STATE_DIR = os.environ.get("DOCSYS_HOME") or os.path.join(os.path.expanduser("~"), ".docsys")
SESSION_FILE = os.path.join(STATE_DIR, "session.json")
CONFIG_FILE = os.path.join(STATE_DIR, "config.json")
TYPE_FILE = 1
TYPE_DIR = 2
SORT_NAME = 0
SORT_TIME = 1
class DocSysError(Exception):
"""业务错误(服务端返回 status=fail 或权限不足)。"""
# --------------------------------------------------------------------------
# 配置
# --------------------------------------------------------------------------
def load_config():
cfg = {
"baseUrl": os.environ.get("DOCSYS_BASE_URL") or DEFAULT_BASE_URL,
"user": os.environ.get("DOCSYS_USER") or DEFAULT_USER,
"password": os.environ.get("DOCSYS_PASSWORD") or DEFAULT_PASSWORD,
}
if os.path.exists(CONFIG_FILE):
try:
with open(CONFIG_FILE, "r", encoding="utf-8") as fh:
cfg.update(json.load(fh))
except Exception:
pass
return cfg
def save_session(jar, base_url):
cookies = []
for c in jar:
cookies.append({
"name": c.name,
"value": c.value,
"domain": c.domain,
"path": c.path,
"expires": None if c.expires is None else int(c.expires),
})
os.makedirs(STATE_DIR, exist_ok=True)
with open(SESSION_FILE, "w", encoding="utf-8") as fh:
json.dump({"baseUrl": base_url, "cookies": cookies}, fh)
def load_session(jar, base_url):
"""把缓存的 cookie 灌进 jar;base_url 变了就丢弃旧缓存。"""
if not os.path.exists(SESSION_FILE):
return False
try:
with open(SESSION_FILE, "r", encoding="utf-8") as fh:
data = json.load(fh)
except Exception:
return False
if data.get("baseUrl") != base_url or not data.get("cookies"):
return False
for c in data["cookies"]:
jar.set_cookie(http.cookiejar.Cookie(
version=0, name=c["name"], value=c["value"],
port=None, port_specified=False,
domain=c["domain"], domain_specified=True, domain_initial_dot=False,
path=c["path"], path_specified=True,
secure=False, expires=c["expires"], discard=False,
comment=None, comment_url=None, rest={}, rfc2109=False,
))
return True
# --------------------------------------------------------------------------
# 客户端
# --------------------------------------------------------------------------
class DocSysClient(object):
def __init__(self, base_url=None, user=None, password=None, timeout=120):
cfg = load_config()
self.base = (base_url or cfg["baseUrl"]).rstrip("/")
self.user = user or cfg["user"]
self.password = password or cfg["password"]
self.timeout = timeout
self.jar = http.cookiejar.CookieJar()
self.opener = build_opener(HTTPCookieProcessor(self.jar))
self._logged_in = False
if load_session(self.jar, self.base):
self._logged_in = True
# -- HTTP 基础 ---------------------------------------------------------
def _url(self, endpoint):
return "{}/{}".format(self.base, endpoint.lstrip("/"))
def _headers(self, extra=None):
h = {
"X-Requested-With": "XMLHttpRequest",
"Accept": "application/json, text/javascript, */*; q=0.01",
"Accept-Language": "zh-CN,zh;q=0.9",
"User-Agent": "Mozilla/5.0 (compatible; docsys-skill/1.0)",
"Referer": self.base + "/web/index.html",
}
if extra:
h.update(extra)
return h
def _post(self, endpoint, fields=None, raw_body=None, content_type=None, retry=True):
url = self._url(endpoint)
if raw_body is None:
body = urllib.parse.urlencode(fields or {}, encoding="utf-8").encode("utf-8")
content_type = content_type or "application/x-www-form-urlencoded; charset=UTF-8"
else:
body = raw_body
req = Request(url, data=body, headers=self._headers({
"Content-Type": content_type,
"Origin": "{}/..{}".format(self.base, ""),
}), method="POST")
try:
with self.opener.open(req, timeout=self.timeout) as resp:
return resp.read().decode("utf-8", "replace")
except urllib.error.HTTPError as exc:
body_text = exc.read().decode("utf-8", "replace")
if retry and self._looks_like_session_lost(body_text):
self._logged_in = False
self.login()
return self._post(endpoint, fields, raw_body, content_type, retry=False)
raise DocSysError("HTTP {}: {}".format(exc.code, _short(body_text)))
except urllib.error.URLError as exc:
raise DocSysError("无法连接 {}: {}".format(self.base, exc.reason))
@staticmethod
def _looks_like_session_lost(text):
low = text.lower()
return ("<title>http status 403" in low or "<title>http status 401" in low
or "登录" in text and "html" in low)
def _get(self, url):
req = Request(url, headers=self._headers())
with self.opener.open(req, timeout=self.timeout) as resp:
return resp.read(), dict(resp.headers)
def _json(self, endpoint, fields=None, raw_body=None, content_type=None):
text = self._post(endpoint, fields, raw_body, content_type)
try:
data = json.loads(text)
except ValueError:
raise DocSysError("返回非 JSON 内容 ({}): {}".format(endpoint, _short(text)))
if isinstance(data, dict) and data.get("status") == "fail":
raise DocSysError(data.get("msgInfo") or "操作失败")
return data
# -- 会话 --------------------------------------------------------------
def login(self, force=False):
if self._logged_in and not force:
return self.whoami()
if not self.user or not self.password:
raise DocSysError(
"未配置 DocSystem 凭证。请在 {} 中写入 user/password,"
"或设置环境变量 DOCSYS_USER / DOCSYS_PASSWORD。".format(CONFIG_FILE))
self.jar.clear()
payload = {
"userName": self.user,
"pwd": base64.b64encode(self.password.encode("utf-8")).decode("ascii"),
"rememberMe": "1",
}
data = self._json("User/login.do", payload)
save_session(self.jar, self.base)
self._logged_in = True
return data.get("data") or {}
def whoami(self):
if not self._logged_in:
self.login()
data = self._json("User/getLoginUser.do")
return data.get("data") or {}
def logout(self):
try:
self._json("User/logout.do")
finally:
if os.path.exists(SESSION_FILE):
os.remove(SESSION_FILE)
self._logged_in = False
# -- 仓库 --------------------------------------------------------------
def repos(self):
self.whoami()
data = self._json("Repos/getReposList.do").get("data") or []
return [{
"id": r.get("id"), "name": r.get("name"), "info": r.get("info"),
"access": bool(r.get("isRemote")),
} for r in data]
def resolve_repo(self, key):
"""key 可以是仓库 id(数字) / 精确名称 / 名称子串。"""
repos = self.repos()
key_s = str(key).strip()
if key_s.isdigit():
for r in repos:
if r["id"] == int(key_s):
return r
raise DocSysError("仓库 id={} 不存在。可用: {}".format(
key_s, ", ".join("{}={}".format(r["id"], r["name"]) for r in repos)))
for r in repos:
if r["name"] == key_s:
return r
hits = [r for r in repos if key_s and key_s in (r["name"] or "")]
if len(hits) == 1:
return hits[0]
if not hits:
raise DocSysError("未找到仓库 '{}'。可用: {}".format(
key_s, ", ".join("{}={}".format(r["id"], r["name"]) for r in repos)))
raise DocSysError("仓库名 '{}' 匹配到多个: {}".format(
key_s, ", ".join("{}={}".format(r["id"], r["name"]) for r in hits)))
# -- 路径解析 ----------------------------------------------------------
@staticmethod
def split_path(path):
return [p for p in str(path or "").replace("\\", "/").split("/") if p and p != "."]
@staticmethod
def join_path(dir_path, name):
return "{}{}".format(dir_path or "", name or "")
@staticmethod
def parent_dir(full_path):
parts = DocSysClient.split_path(full_path)
if not parts:
raise DocSysError("仓库根目录不能作为操作对象")
return "/".join(parts[:-1])
@staticmethod
def dir_path_of(node):
"""写操作用的目录路径:目录完整路径 + '/',仓库根为 ''。
注意:节点响应里的 node.path 是「自身所在目录」,
写接口要的是「父目录的完整路径」,两者差一个目录名。
"""
if not node.get("name"):
return ""
return DocSysClient.join_path(node["path"], node["name"]) + "/"
@staticmethod
def basename(full_path):
parts = DocSysClient.split_path(full_path)
if not parts:
raise DocSysError("仓库根目录没有名字")
return parts[-1]
def ls_raw(self, vid, dir_path="", doc_id=0, sort=SORT_NAME):
"""列举目录。服务端语义:path 与 name 拼接 = 目标目录,仓库根为 ''。
实测:docId 不参与定位,服务端只看 path+name,所以必须传完整目录路径。
"""
self.whoami()
data = self._json("Repos/getSubDocList.do", {
"vid": vid, "docId": doc_id, "path": dir_path or "", "name": "",
"sort": sort, "needLockState": 1, "shareId": "",
})
return data.get("data") or []
def ls(self, vid, path=""):
node = self.resolve(vid, path)
return self.ls_raw(vid, self.dir_path_of(node), node["docId"])
def resolve(self, vid, path):
"""把 '/a/b/c' 这样的路径解析成节点 dict;根目录返回伪节点。"""
cur = {"docId": 0, "pid": 0, "path": "", "name": "", "type": TYPE_DIR, "level": -1}
for part in self.split_path(path):
children = self.ls_raw(vid, self.dir_path_of(cur), cur["docId"])
match = None
for c in children:
if c.get("name") == part:
match = c
break
if match is None:
shown = ", ".join(c.get("name", "") for c in children[:20]) or "(空)"
raise DocSysError("路径不存在: /{}/ (可用项: {})".format(
"/".join(self.split_path(path)), shown))
cur = match
return cur
def exists(self, vid, path):
try:
self.resolve(vid, path)
return True
except DocSysError:
return False
# -- 读 ----------------------------------------------------------------
def auth(self, vid, path=""):
self.whoami()
node = self.resolve(vid, path)
data = self._json("Repos/getUserDocAuth.do", {
"docId": node["docId"], "reposId": vid,
"path": node["path"], "name": node["name"], "shareId": "",
}).get("data") or {}
return {
"access": data.get("access") == 1,
"add": data.get("addEn") == 1,
"edit": data.get("editEn") == 1,
"delete": data.get("deleteEn") == 1,
"download": data.get("downloadEn") == 1,
"isAdmin": data.get("isAdmin") == 1,
"uploadSizeLimit": data.get("uploadSize"),
}
def tree(self, vid, path="", depth=2):
root = self.resolve(vid, path)
base_level = root.get("level", -1)
out = []
def walk(node, cur_depth):
if cur_depth > depth:
return
for c in self.ls_raw(vid, self.dir_path_of(node), node["docId"]):
c = dict(c)
c["fullPath"] = self.join_path(c["path"], c["name"])
out.append(c)
if c.get("type") == TYPE_DIR:
walk(c, cur_depth + 1)
walk(root, 0)
_ = base_level
return out
def search(self, vid, keyword, path=""):
self.whoami()
scope = ""
if path:
scope = self.dir_path_of(self.resolve(vid, path))
data = self._json("Doc/searchDoc.do", {
"searchWord": keyword, "sort": SORT_NAME,
"reposId": vid, "path": scope, "shareId": "",
})
results = []
for r in data.get("data") or []:
item = {
"docId": r.get("docId"), "name": r.get("name"),
"fullPath": self.join_path(r.get("path"), r.get("name")),
"type": r.get("type"), "size": r.get("size"),
"reposName": r.get("reposName"),
"latestEditTime": r.get("latestEditTime"),
"hitType": r.get("hitType"), # 1=文件名命中 3=正文命中
}
content = r.get("content") or ""
if content:
try:
item["snippet"] = base64.b64decode(content).decode("utf-8", "replace")[:400]
except Exception:
item["snippet"] = content[:400]
results.append(item)
return results
def cat(self, vid, path):
node = self.resolve(vid, path)
if node.get("type") == TYPE_DIR:
raise DocSysError("/{} 是目录,请用 ls 或 download".format(
"/".join(self.split_path(path))))
text = self._post("Doc/getDocContent.do", {
"reposId": vid, "docId": node["docId"], "pid": node["pid"],
"path": node["path"], "name": node["name"], "shareId": "",
})
if text.startswith("ok"):
text = text[2:]
return text
# -- 写 ----------------------------------------------------------------
def mkdir(self, vid, path, parents=False):
"""创建目录。parents=True 时逐级创建缺失的中间目录(类似 mkdir -p)。"""
parts = self.split_path(path)
if not parts:
raise DocSysError("不能在仓库根创建目录")
if parents:
created = []
for i in range(1, len(parts) + 1):
sub = "/".join(parts[:i])
if self.exists(vid, sub):
continue
node = self._mkdir_one(vid, sub)
created.append(node)
if not created:
raise DocSysError("已存在: /{}".format("/".join(parts)))
return created[-1]
return self._mkdir_one(vid, path)
def _mkdir_one(self, vid, path):
if self.exists(vid, path):
raise DocSysError("已存在: /{}".format("/".join(self.split_path(path))))
parent = self.parent_dir(path)
name = self.basename(path)
pnode = self.resolve(vid, parent)
data = self._json("Doc/addDoc.do", {
"reposId": vid, "type": TYPE_DIR, "pid": pnode["docId"],
"path": self.dir_path_of(pnode), "name": name,
"content": "", "shareId": "",
})
node = data.get("data") or {}
node["fullPath"] = self.join_path(self.dir_path_of(pnode), name)
return node
def mkfile(self, vid, path, content=""):
"""新建文件。
实测:addDoc.do 会**忽略** content 参数,建出来的永远是 0 字节空文件
(响应 size=0,getDocContent 也读不到内容)。所以要带内容时改为
走 uploadDoc.do 上传字节,这样内容才真的落盘。
"""
if self.exists(vid, path):
raise DocSysError("已存在: /{}".format("/".join(self.split_path(path))))
if content:
return self.upload_bytes(vid, self.parent_dir(path),
self.basename(path), content.encode("utf-8"))
parent = self.parent_dir(path)
name = self.basename(path)
pnode = self.resolve(vid, parent)
data = self._json("Doc/addDoc.do", {
"reposId": vid, "type": TYPE_FILE, "pid": pnode["docId"],
"path": self.dir_path_of(pnode), "name": name,
"content": "", "shareId": "",
})
node = data.get("data") or {}
node["fullPath"] = self.join_path(self.dir_path_of(pnode), name)
return node
def rename(self, vid, path, new_name):
node = self.resolve(vid, path)
if not new_name or "/" in new_name:
raise DocSysError("非法的新名字: {!r}".format(new_name))
data = self._json("Doc/renameDoc.do", {
"reposId": vid, "docId": node["docId"], "pid": node["pid"],
"type": node["type"], "level": node["level"],
"path": node["path"], "name": node["name"], "dstName": new_name,
"shareId": "",
})
res = data.get("data") or {}
res["fullPath"] = self.join_path(node["path"], new_name)
return res
def delete(self, vid, path):
node = self.resolve(vid, path)
self._json("Doc/deleteDoc.do", {
"reposId": vid, "docId": node["docId"], "pid": node["pid"],
"path": node["path"], "name": node["name"], "shareId": "",
})
return {"deleted": self.join_path(node["path"], node["name"]),
"type": "dir" if node["type"] == TYPE_DIR else "file"}
def move(self, vid, src, dst):
"""dst 为已存在目录 → 移入其中;否则视为目标完整路径(改名+移动)。"""
node = self.resolve(vid, src)
if self.exists(vid, dst) and self.resolve(vid, dst)["type"] == TYPE_DIR:
dst_node = self.resolve(vid, dst)
dpath = self.join_path(dst_node["path"], dst_node["name"])
dname = node["name"]
else:
dpath = self.parent_dir(dst)
dname = self.basename(dst)
dnode = self.resolve(vid, dpath)
data = self._json("Doc/moveDoc.do", {
"reposId": vid, "docId": node["docId"], "type": node["type"],
"srcLevel": node["level"], "dstLevel": dnode["level"] + 1,
"srcPid": node["pid"], "dstPid": dnode["docId"],
"srcPath": node["path"], "srcName": node["name"],
"dstPath": self.dir_path_of(dnode), "dstName": dname, "shareId": "",
})
res = data.get("data") or {}
res["fullPath"] = self.join_path(self.dir_path_of(dnode), dname)
return res
def copy(self, vid, src, dst):
node = self.resolve(vid, src)
if self.exists(vid, dst) and self.resolve(vid, dst)["type"] == TYPE_DIR:
dst_node = self.resolve(vid, dst)
dpath = self.join_path(dst_node["path"], dst_node["name"])
dname = node["name"]
else:
dpath = self.parent_dir(dst)
dname = self.basename(dst)
dnode = self.resolve(vid, dpath)
data = self._json("Doc/copyDoc.do", {
"reposId": vid, "docId": node["docId"], "type": node["type"],
"srcLevel": node["level"], "dstPid": dnode["docId"],
"srcPath": node["path"], "srcName": node["name"],
"dstLevel": dnode["level"] + 1,
"dstPath": self.dir_path_of(dnode), "dstName": dname, "shareId": "",
})
res = data.get("data") or {}
res["fullPath"] = self.join_path(self.dir_path_of(dnode), dname)
return res
# -- 上传 / 下载 -------------------------------------------------------
@staticmethod
def _multipart(fields, files):
boundary = "----DocSysBoundary" + uuid.uuid4().hex
parts = []
for k, v in fields.items():
if v is None:
continue
parts.append((
'--{b}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'
).format(b=boundary, k=k, v=v).encode("utf-8"))
for k, (fname, data, ctype) in files.items():
quoted = urllib.parse.quote(fname)
head = (
'--{b}\r\nContent-Disposition: form-data; name="{k}"; filename="{f}"; '
"filename*=UTF-8''{q}\r\n"
"Content-Type: {c}\r\n\r\n"
).format(b=boundary, k=k, f=fname, q=quoted, c=ctype or "application/octet-stream")
parts.append(head.encode("utf-8") + data + b"\r\n")
parts.append("--{}--\r\n".format(boundary).encode("utf-8"))
return b"".join(parts), "multipart/form-data; boundary={}".format(boundary)
def upload(self, vid, remote_dir, local_file, remote_name=None):
if not os.path.isfile(local_file):
raise DocSysError("本地文件不存在: {}".format(local_file))
with open(local_file, "rb") as fh:
payload = fh.read()
name = remote_name or os.path.basename(local_file)
return self.upload_bytes(vid, remote_dir, name, payload)
def upload_bytes(self, vid, remote_dir, name, payload):
node = self.resolve(vid, remote_dir)
ctype = mimetypes.guess_type(name)[0] or "application/octet-stream"
body, ct = self._multipart({
"reposId": vid,
"path": self.dir_path_of(node),
"name": name,
"size": len(payload),
"checkSum": hashlib.md5(payload).hexdigest(),
"commitMsg": "",
"dirPath": "",
}, {"uploadFile": (name, payload, ctype)})
data = self._json("Doc/uploadDoc.do", None, raw_body=body, content_type=ct)
res = data.get("data") or {}
res["fullPath"] = self.join_path(self.dir_path_of(node), name)
return res
def download(self, vid, path, local_file):
node = self.resolve(vid, path)
data = self._json("Doc/downloadDocPrepare.do", {
"reposId": vid, "docId": node["docId"], "pid": node["pid"],
"path": node["path"], "name": node["name"],
"downloadType": 1, "downloadList": "", "shareId": "",
})
delete_flag = data.get("msgData")
info = data.get("data") or {}
# msgData == 5 表示目录正在打包,轮询任务
if delete_flag == 5 or info.get("info"):
task_id = info.get("id")
for _ in range(60):
time.sleep(2)
task = self._json("Doc/queryDownloadPrepareTask.do", {"taskId": task_id})
if task.get("msgData") == 6:
info = task.get("data") or {}
delete_flag = task.get("msgData2") or delete_flag
break
else:
raise DocSysError("打包超时")
qs = urllib.parse.urlencode({
"vid": vid,
"path": info.get("path", ""),
"name": info.get("name", ""),
"targetPath": info.get("targetPath", ""),
"targetName": info.get("targetName", ""),
"deleteFlag": delete_flag,
"encryptEn": 1,
}, safe="")
blob, headers = self._get(self._url("Doc/downloadDoc.do?") + qs)
ctype = headers.get("Content-Type", "")
if "html" in ctype.lower() and b"<html" in blob[:400].lower():
raise DocSysError("下载返回了 HTML 错误页,通常是权限不足或文件已不存在")
if local_file in (None, "-"):
return blob, ctype
target = local_file
if os.path.isdir(target):
target = os.path.join(target, info.get("targetName") or node["name"])
parent = os.path.dirname(os.path.abspath(target))
if parent and not os.path.isdir(parent):
os.makedirs(parent, exist_ok=True)
with open(target, "wb") as fh:
fh.write(blob)
return {"saved": target, "bytes": len(blob)}, ctype
def _short(text, limit=300):
text = (text or "").replace("\n", " ").strip()
return text[:limit] + ("..." if len(text) > limit else "")
def human_size(n):
try:
n = float(n or 0)
except (TypeError, ValueError):
return "?"
for unit in ("B", "KB", "MB", "GB", "TB"):
if n < 1024:
return "{:.0f}{}".format(n, unit) if unit == "B" else "{:.1f}{}".format(n, unit)
n /= 1024.0
return "{:.1f}PB".format(n)
def fmt_time(ms):
if not ms:
return ""
try:
return time.strftime("%Y-%m-%d %H:%M", time.localtime(int(ms) / 1000.0))
except Exception:
return ""
# --------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------
def emit(args, data, text_fn):
if getattr(args, "json", False):
print(json.dumps(data, ensure_ascii=False, indent=2))
else:
text = text_fn(data)
if text:
print(text)
def cmd_login(c, args):
user = c.login(force=True)
emit(args, user, lambda u: "已登录: {} ({}) id={}".format(
u.get("name"), u.get("realName", ""), u.get("id")))
def cmd_whoami(c, args):
user = c.whoami()
emit(args, user, lambda u: "{} | {} | id={} | role={}".format(
u.get("name"), u.get("realName", ""), u.get("id"), u.get("role")))
def cmd_logout(c, args):
c.logout()
emit(args, {"ok": True}, lambda _: "已退出登录并清除本地会话缓存")
def cmd_repos(c, args):
repos = c.repos()
emit(args, repos, lambda rs: "\n".join(
"{:<4} {}{}".format(r["id"], r["name"],
" ← 默认" if r["name"] == DEFAULT_REPO_NAME else "")
for r in rs) or "(无可见仓库)")
def looks_like_path(value):
return bool(value) and (str(value).startswith("/") or str(value).startswith("./"))
def normalize_optional_repo(args):
"""仓库可省略:若首个位置参数像路径,说明用户压根没给仓库,整体左移一位。
必须"整体左移"而不是只交换 repo/path——像 `touch <path> <content>` 这种
后面还有可选位置参数的命令,argparse 会把 path 塞进 repo 槽、content 塞进
path 槽,只交换两个会丢内容。
"""
dests = getattr(args, "positional_dests", None)
if not dests:
return args
if not looks_like_path(getattr(args, dests[0], "")):
return args
vals = [getattr(args, d) for d in dests]
setattr(args, dests[0], DEFAULT_REPO_NAME)
for i in range(1, len(dests)):
setattr(args, dests[i], vals[i - 1])
return args
def pick_repo(c, args):
"""解析命令里的仓库参数;省略时用 DEFAULT_REPO_NAME。"""
if not getattr(args, "repo", None):
args.repo = DEFAULT_REPO_NAME
return c.resolve_repo(args.repo)
def cmd_ls(c, args):
repo = pick_repo(c, args)
nodes = c.ls(repo["id"], args.path)
out = [{"name": n["name"], "fullPath": c.join_path(n["path"], n["name"]),
"type": "dir" if n["type"] == TYPE_DIR else "file",
"size": n.get("size"), "mtime": fmt_time(n.get("latestEditTime")),
"docId": n["docId"]} for n in nodes]
def render(items):
if not items:
return "(空目录)"
lines = []
for it in items:
tag = "[DIR] " if it["type"] == "dir" else " "
lines.append("{}{:<50} {:>10} {}".format(
tag, it["fullPath"], human_size(it["size"]) if it["type"] == "file" else "",
it["mtime"]))
return "\n".join(lines)
emit(args, out, render)
def cmd_tree(c, args):
repo = pick_repo(c, args)
nodes = c.tree(repo["id"], args.path, args.depth)
out = [{"fullPath": n.get("fullPath"), "type": "dir" if n["type"] == TYPE_DIR else "file",
"size": n.get("size"), "docId": n["docId"]} for n in nodes]
emit(args, out, lambda items: "\n".join(
"{}{}".format(" " * max(0, i["fullPath"].count("/")), i["fullPath"])
for i in items) or "(空)")
def cmd_mkdir(c, args):
repo = pick_repo(c, args)
node = c.mkdir(repo["id"], args.path, parents=args.parents)
emit(args, node, lambda n: "已创建目录 /{} (docId={})".format(
n.get("fullPath", "").strip("/"), n.get("docId")))
def cmd_touch(c, args):
repo = pick_repo(c, args)
content = args.content or ""
if content == "@file" and args.file:
with open(args.file, "r", encoding="utf-8") as fh:
content = fh.read()
node = c.mkfile(repo["id"], args.path, content)
emit(args, node, lambda n: "已创建文件 /{} (docId={})".format(
n.get("fullPath", "").strip("/"), n.get("docId")))
def cmd_rename(c, args):
repo = pick_repo(c, args)
node = c.rename(repo["id"], args.path, args.new_name)
emit(args, node, lambda n: "已重命名为 /{} (docId={})".format(
n.get("fullPath", "").strip("/"), n.get("docId")))
def cmd_delete(c, args):
repo = pick_repo(c, args)
res = c.delete(repo["id"], args.path)
emit(args, res, lambda r: "已删除 {} /{}".format(r["type"], r["deleted"].strip("/")))
def cmd_move(c, args):
repo = pick_repo(c, args)
res = c.move(repo["id"], args.src, args.dst)
emit(args, res, lambda r: "已移动到 /{}".format(r.get("fullPath", "").strip("/")))
def cmd_copy(c, args):
repo = pick_repo(c, args)
res = c.copy(repo["id"], args.src, args.dst)
emit(args, res, lambda r: "已复制到 /{}".format(r.get("fullPath", "").strip("/")))
def cmd_upload(c, args):
repo = pick_repo(c, args)
node = c.upload(repo["id"], args.remote_dir, args.local_file, args.name)
emit(args, node, lambda n: "已上传 /{} ({} , docId={})".format(
n.get("fullPath", "").strip("/"), human_size(n.get("size")), n.get("docId")))
def cmd_download(c, args):
repo = pick_repo(c, args)
res, _ = c.download(repo["id"], args.path, args.local_file)
if isinstance(res, bytes):
sys.stdout.buffer.write(res)
else:
emit(args, res, lambda r: "已下载到 {} ({} )".format(r["saved"], human_size(r["bytes"])))
def cmd_cat(c, args):
repo = pick_repo(c, args)
text = c.cat(repo["id"], args.path)
if args.json:
print(json.dumps({"path": args.path, "content": text}, ensure_ascii=False, indent=2))
else:
sys.stdout.write(text)
def cmd_search(c, args):
if args.all_repos:
repos = c.repos()
else:
repos = [c.resolve_repo(args.repo or DEFAULT_REPO_NAME)]
all_hits = []
for r in repos:
try:
for hit in c.search(r["id"], args.keyword, args.path or ""):
hit["repoId"] = r["id"]
all_hits.append(hit)
except DocSysError as exc:
if not args.all_repos:
raise
all_hits.append({"error": "{}: {}".format(r["name"], exc)})
emit(args, all_hits, lambda hs: "\n".join(
("[{}] {}{}".format(h.get("repoName", h.get("repoId")), h.get("fullPath"),
" ← 正文命中" if h.get("hitType") == 3 else ""))
if "fullPath" in h else "[!] " + h.get("error", "")
for h in hs) or "(无结果)")
def cmd_auth(c, args):
repo = pick_repo(c, args)
a = c.auth(repo["id"], args.path or "")
emit(args, a, lambda x: "仓库 {}\n 访问={} 新增={} 编辑={} 删除={} 下载={} 管理员={}".format(
repo["name"], x["access"], x["add"], x["edit"], x["delete"], x["download"], x["isAdmin"]))
def build_parser():
p = argparse.ArgumentParser(
prog="docsys.py",
description="MxsDoc / DocSystem 文档系统命令行客户端(供 AI 自动化操作)")
p.add_argument("--json", action="store_true", help="以 JSON 输出")
p.add_argument("--base-url", help="覆盖服务地址,默认 " + DEFAULT_BASE_URL)
p.add_argument("--user", help="覆盖登录账号")
p.add_argument("--password", help="覆盖登录密码")
sub = p.add_subparsers(dest="cmd", required=True)
sub.add_parser("login", help="登录并缓存会话").set_defaults(func=cmd_login)
sub.add_parser("whoami", help="查看当前登录用户").set_defaults(func=cmd_whoami)
sub.add_parser("logout", help="退出登录").set_defaults(func=cmd_logout)
sub.add_parser("repos", help="列出可访问的仓库").set_defaults(func=cmd_repos)
sp = sub.add_parser("ls", help="列出目录内容(省略仓库则用默认库)")
sp.add_argument("repo", nargs="?", default="")
sp.add_argument("path", nargs="?", default="")
sp.set_defaults(func=cmd_ls)
sp.set_defaults(positional_dests=['repo', 'path'])
sp = sub.add_parser("tree", help="递归列出目录树")
sp.add_argument("repo", nargs="?", default="")
sp.add_argument("path", nargs="?", default="")
sp.add_argument("--depth", type=int, default=2)
sp.set_defaults(func=cmd_tree)
sp.set_defaults(positional_dests=['repo', 'path'])
sp = sub.add_parser("mkdir", help="创建目录")
sp.add_argument("repo", nargs="?", default="")
sp.add_argument("path", help="如 /a/b/c")
sp.add_argument("-p", "--parents", action="store_true",
help="递归创建缺失的中间目录(默认要求父目录已存在)")
sp.set_defaults(func=cmd_mkdir)
sp.set_defaults(positional_dests=['repo', 'path'])
sp = sub.add_parser("touch", help="创建文本文件(content 直接给文本,@file 从本地文件读)")
sp.add_argument("repo", nargs="?", default="")
sp.add_argument("path")
sp.add_argument("content", nargs="?", default="")
sp.add_argument("--file", help="配合 content=@file 从本地文件读取内容")
sp.set_defaults(func=cmd_touch)
sp.set_defaults(positional_dests=['repo', 'path', 'content'])
sp = sub.add_parser("rename", help="重命名(不支持跨目录)")
sp.add_argument("repo", nargs="?", default="")
sp.add_argument("path")
sp.add_argument("new_name")
sp.set_defaults(func=cmd_rename)
sp.set_defaults(positional_dests=['repo', 'path', 'new_name'])
sp = sub.add_parser("rm", help="删除文件或目录(目录为递归删除)")
sp.add_argument("repo", nargs="?", default="")
sp.add_argument("path")
sp.set_defaults(func=cmd_delete)
sp.set_defaults(positional_dests=['repo', 'path'])
sp = sub.add_parser("mv", help="移动 / 改名")
sp.add_argument("repo", nargs="?", default="")
sp.add_argument("src")
sp.add_argument("dst", help="目标目录(已存在)或目标完整路径")
sp.set_defaults(func=cmd_move)
sp.set_defaults(positional_dests=['repo', 'src', 'dst'])
sp = sub.add_parser("cp", help="复制")
sp.add_argument("repo", nargs="?", default="")
sp.add_argument("src")
sp.add_argument("dst")
sp.set_defaults(func=cmd_copy)
sp.set_defaults(positional_dests=['repo', 'src', 'dst'])
sp = sub.add_parser("upload", help="上传本地文件")
sp.add_argument("repo", nargs="?", default="")
sp.add_argument("remote_dir", help="远端目标目录,如 / 或 /a/b")
sp.add_argument("local_file")
sp.add_argument("--name", help="覆盖远端文件名")
sp.set_defaults(func=cmd_upload)
sp.set_defaults(positional_dests=['repo', 'remote_dir', 'local_file'])
sp = sub.add_parser("download", help="下载文件,local_file 为 - 时输出到 stdout")
sp.add_argument("repo", nargs="?", default="")
sp.add_argument("path")
sp.add_argument("local_file")
sp.set_defaults(func=cmd_download)
sp.set_defaults(positional_dests=['repo', 'path', 'local_file'])
sp = sub.add_parser("cat", help="读取文本文件内容")
sp.add_argument("repo", nargs="?", default="")
sp.add_argument("path")
sp.set_defaults(func=cmd_cat)
sp.set_defaults(positional_dests=['repo', 'path'])
sp = sub.add_parser("search", help="按文件名或正文搜索(默认搜默认库)")
sp.add_argument("keyword")
sp.add_argument("--repo", help="限定仓库;省略则搜默认库")
sp.add_argument("--all-repos", action="store_true", help="搜索全部可见仓库")
sp.add_argument("--path", help="限定目录范围")
sp.set_defaults(func=cmd_search)
sp = sub.add_parser("auth", help="查看某目录的操作权限")
sp.add_argument("repo", nargs="?", default="")
sp.add_argument("path", nargs="?", default="")
sp.set_defaults(func=cmd_auth)
sp.set_defaults(positional_dests=['repo', 'path'])
return p
def main(argv=None):
args = build_parser().parse_args(argv)
normalize_optional_repo(args)
client = DocSysClient(args.base_url, args.user, args.password)
try:
args.func(client, args)
return 0
except DocSysError as exc:
sys.stderr.write("[docsys] {}\n".format(exc))
return 2
except KeyboardInterrupt:
return 130
if __name__ == "__main__":
sys.exit(main())