Cheney
8ae1694b6d
从公司 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，过期自动重登
997 lines
39 KiB
Python
997 lines
39 KiB
Python
#!/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()) |