"""vjmap Python SDK（单文件版）。
对接 vjmap 后端 REST 接口（web服务 + 图形服务）。设计原则与现有前端
`vjmap` 包的 ``Service`` 语义保持一致，方法名做成 Python 风格
（snake_case），参数尽量直接对齐其接口定义。
依赖
----
- Python >= 3.8
- ``requests``
使用
----
.. code-block:: python
from vjmap import Vjmap
svc = Vjmap("http://127.0.0.1:27660/api/v1", "<jwt-token>")
svc.open_map(mapid="sys_zp", version="v1", mapopenway="GeomRender")
meta = svc.metadata()
print(len(meta.get("layers", [])), "layers")
异常
----
所有 HTTP 非 2xx 都会抛 :class:`VjmapError`，包含 ``status``、``url`` 与
已解析的 ``body``，便于日志与排障。
"""
from __future__ import annotations
import base64
import hashlib
import io
import json
import os
import time
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple, Union
from urllib.parse import quote, urlencode, urlsplit, urlunsplit
import requests
__all__ = [
"Vjmap",
"VjmapError",
"MapOpenWay",
"compute_secret_key",
"file_md5",
"md5_hex",
]
# ---------------------------------------------------------------------------
# 常量 / 小工具
# ---------------------------------------------------------------------------
class MapOpenWay:
"""与官方 JS SDK 的 ``MapOpenWay`` 枚举对齐（仅客户端枚举，后端不读该字段，
SDK 换算成请求体布尔 ``geom = mapopenway != "Memory"``）。
- ``Memory`` : 内存展开图档，不生成几何数据，适合只读元数据（``geom:false``）。
- ``GeomRender`` : 几何渲染模式，首次打开异步生成几何库（``geom:true``）。
"""
Memory = "Memory"
GeomRender = "GeomRender"
class VjmapError(Exception):
"""SDK 统一异常。
Attributes:
status: HTTP 状态码（网络错误时为 ``None``）。
url: 实际请求的 URL。
body: 响应体（JSON dict / str / bytes）。
"""
def __init__(
self,
message: str,
*,
status: Optional[int] = None,
url: Optional[str] = None,
body: Any = None,
) -> None:
super().__init__(message)
self.status = status
self.url = url
self.body = body
def __repr__(self) -> str: # pragma: no cover - debug helper
return f"VjmapError(status={self.status}, url={self.url!r}, body={self.body!r})"
def md5_hex(data: Union[bytes, str]) -> str:
"""计算字节/字符串的 MD5 十六进制串（小写，32 字符）。"""
if isinstance(data, str):
data = data.encode("utf-8")
return hashlib.md5(data).hexdigest()
def compute_secret_key(pwd: str) -> str:
"""把明文密码转成 vjmap 的 secretKey（16 位）。
算法（与 ``Service.pwdToSecretKey`` 一致）：
- 连续 4 次 md5 后取前 16 字符；
- 空字符串 → 返回空；
- 长度 == 18 且以 ``ak`` 开头 → 视为 accessKey，原样返回。
"""
if not pwd:
return ""
if len(pwd) == 18 and pwd.startswith("ak"):
return pwd
s = pwd
for _ in range(4):
s = md5_hex(s)
return s[:16]
def file_md5(src: Union[str, bytes, io.IOBase]) -> str:
"""流式计算文件 MD5。``src`` 可以是路径、bytes 或一个已打开的二进制流。"""
h = hashlib.md5()
if isinstance(src, (bytes, bytearray)):
h.update(src)
return h.hexdigest()
if isinstance(src, str):
with open(src, "rb") as f:
for chunk in iter(lambda: f.read(1024 * 1024), b""):
h.update(chunk)
return h.hexdigest()
if hasattr(src, "read"):
while True:
chunk = src.read(1024 * 1024)
if not chunk:
break
h.update(chunk)
return h.hexdigest()
raise TypeError(f"Unsupported src type for file_md5: {type(src)!r}")
def _normalize_base(base: str) -> str:
"""保证 base_url 以 `/` 结尾；``http://h/api/v1`` → ``http://h/api/v1/``。"""
if not base:
raise ValueError("base_url is required")
if not base.endswith("/"):
base = base + "/"
return base
def _encode_query(params: Mapping[str, Any]) -> str:
"""与 JS 的 ``encodeURIComponent`` 行为对齐的 query 编码。"""
parts: List[str] = []
for k, v in params.items():
if v is None:
continue
if isinstance(v, bool):
v = "true" if v else "false"
parts.append(f"{quote(str(k), safe='')}={quote(str(v), safe='')}")
return "&".join(parts)
# ---------------------------------------------------------------------------
# Vjmap 主类
# ---------------------------------------------------------------------------
class Vjmap:
"""vjmap 服务客户端。
Parameters
----------
base_url:
后端前缀，如 ``http://127.0.0.1:27660/api/v1``。会自动补 ``/``。
token:
JWT Token。所有请求通过 ``?token=`` 或 header ``token`` 传入；
SDK 默认使用 query string 形式，便于调试。
timeout:
单次请求超时（秒）。默认 60。
session:
可选自定义 :class:`requests.Session`。未传时内部建一个。
"""
def __init__(
self,
base_url: str,
token: str = "",
*,
timeout: float = 60.0,
session: Optional[requests.Session] = None,
) -> None:
self.base_url = _normalize_base(base_url)
self.token = token or ""
self.timeout = timeout
self._session = session or requests.Session()
self._cur_map: Dict[str, str] = {} # {mapid, version, layer, fileid}
# 已缓存的图密钥：发请求时按形态拆头——18 位 ``ak`` 开头 → ``accessKey`` 头，其余 → ``secretKey`` 头
self._secret_keys: List[str] = []
# 始终走 ``accessKey`` 头的密钥（如服务端 config ``map_super_key``，后端只在 accessKey 列表里认它）
self._access_keys: List[str] = []
self._workspace: str = "" # 命名工作区名；空表示默认工作区
# -- 上下文辅助 ---------------------------------------------------------
@property
def workspace(self) -> str:
return self._workspace
def switch_workspace(self, name: str = "") -> None:
"""切换命名工作区；空字符串切回默认工作区。"""
self._workspace = name or ""
def current_map_param(self) -> Dict[str, str]:
"""返回当前打开的 mapid/version/layer/fileid。"""
return dict(self._cur_map)
def set_current_map(self, mapid: str, version: str = "v1",
layer: str = "", fileid: str = "") -> None:
self._cur_map = {
"mapid": mapid,
"version": version or "v1",
"layer": layer or "",
"fileid": fileid or mapid,
}
# -- URL 构造 -----------------------------------------------------------
def _map_prefix(self) -> str:
"""默认工作区 → ``map/``；命名工作区 → ``workspace/{name}/``。"""
if self._workspace:
return f"workspace/{quote(self._workspace, safe='')}/"
return "map/"
def service_url(self, path: str, params: Optional[Mapping[str, Any]] = None) -> str:
"""构造 ``/api/v1/service/<path>`` URL，附带 token。"""
params = dict(params or {})
params.setdefault("token", self.token)
url = f"{self.base_url}service/{path.lstrip('/')}"
qs = _encode_query(params)
return f"{url}?{qs}" if qs else url
def map_url(self, path: str, params: Optional[Mapping[str, Any]] = None) -> str:
"""构造 ``/api/v1/map/<path>`` 或 ``workspace/{ws}/<path>`` URL。"""
params = dict(params or {})
params.setdefault("token", self.token)
url = f"{self.base_url}{self._map_prefix()}{path.lstrip('/')}"
qs = _encode_query(params)
return f"{url}?{qs}" if qs else url
def base_gateway(self) -> str:
"""``http://h[/prefix]/api/v1/`` → ``http://h[/prefix]/``。仅去 api/vN 尾段，保留部署前缀（公有云 ``/server``）。"""
import re
parts = urlsplit(self.base_url)
path = re.sub(r"/api/v\d+/?$", "/", parts.path)
if not path.startswith("/"):
path = "/" + path
return urlunsplit((parts.scheme, parts.netloc, path, "", ""))
def gateway_url(self, path: str, params: Optional[Mapping[str, Any]] = None) -> str:
"""网关根级 URL（``/heart``、``/version`` 等不在 /api/v1 下）。"""
params = dict(params or {})
params.setdefault("token", self.token)
url = f"{self.base_gateway()}{path.lstrip('/')}"
qs = _encode_query({k: v for k, v in params.items() if v is not None})
return f"{url}?{qs}" if qs else url
def cmd_url(
self,
cmd: str,
mapid: str = "",
version: str = "",
params: Optional[Mapping[str, Any]] = None,
) -> str:
"""构造 ``/api/v1/map/cmd/{cmd}/{mapid}/{version}`` URL。"""
mapid = mapid or self._cur_map.get("mapid") or "_"
version = version or self._cur_map.get("version") or "_"
path = f"cmd/{cmd}/{quote(mapid, safe='_*')}/{quote(version, safe='_*')}"
return self.map_url(path, params)
# -- HTTP 核心 ----------------------------------------------------------
def _extract_body(self, resp: requests.Response) -> Any:
ctype = (resp.headers.get("Content-Type") or "").lower()
if "application/json" in ctype:
try:
return resp.json()
except ValueError:
return resp.text
if ctype.startswith("text/"):
return resp.text
# 图形引擎部分响应 content-type 不规范（octet-stream 里实为 JSON），按首字符嗅探
head = resp.content[:64].lstrip()
if head.startswith(b"{") or head.startswith(b"["):
try:
return resp.json()
except ValueError:
pass
return resp.content
def _raise_if_error(self, resp: requests.Response, body: Any) -> None:
if 200 <= resp.status_code < 300:
return
raise VjmapError(
f"HTTP {resp.status_code} for {resp.request.method} {resp.url}",
status=resp.status_code,
url=resp.url,
body=body,
)
def _unwrap_gateway(self, body: Any) -> Any:
"""web服务标准响应 ``{code, msg, data}`` 自动 unwrap 为 ``data``。
服务服务 JSON 没有 ``code`` 字段，保持原样。
"""
if not isinstance(body, dict):
return body
if "code" in body and "data" in body and "msg" in body:
code = body.get("code")
if code == 0:
return body.get("data")
raise VjmapError(
f"gateway code={code}: {body.get('msg')!r}",
status=200,
body=body,
)
return body
def _key_headers(self) -> Dict[str, str]:
"""由缓存密钥组装 ``secretKey`` / ``accessKey`` 请求头（多个用 ``_;_`` 连接）。
后端 HavePermissionInMap：accessKey 列表含 ``map_super_key`` → admin；secretKey
列表含该图 secretKey → admin；accessKey 列表含该图 accessKey → 仅浏览。
头部同名键会**覆盖** body 里的值。
"""
def is_ak(k: str) -> bool:
return isinstance(k, str) and len(k) == 18 and k.startswith("ak")
sk = [k for k in self._secret_keys if k and not is_ak(k)]
ak = [k for k in self._secret_keys if is_ak(k)] + [k for k in self._access_keys if k]
h: Dict[str, str] = {}
if sk:
h["secretKey"] = "_;_".join(sk)
if ak:
h["accessKey"] = "_;_".join(ak)
return h
def _request(
self,
method: str,
url: str,
*,
json_body: Any = None,
form: Optional[Mapping[str, Any]] = None,
files: Any = None,
headers: Optional[Mapping[str, str]] = None,
timeout: Optional[float] = None,
stream: bool = False,
) -> Any:
hdrs: Dict[str, str] = {"token": self.token} if self.token else {}
hdrs.update(self._key_headers())
if headers:
hdrs.update(headers)
try:
resp = self._session.request(
method.upper(),
url,
json=json_body,
data=form,
files=files,
headers=hdrs,
timeout=timeout or self.timeout,
stream=stream,
)
except requests.RequestException as exc:
raise VjmapError(str(exc), url=url) from exc
if stream:
return resp # 二进制场景由调用方处理
body = self._extract_body(resp)
self._raise_if_error(resp, body)
return self._unwrap_gateway(body)
def heart(self) -> Dict[str, Any]:
"""``GET /heart``。返回 ``{"status":"ok","timestamp":...}``。"""
return self._request("GET", self.gateway_url("heart", {}))
def version(self) -> Dict[str, Any]:
"""``GET /version``。返回 ``{"arch","os","status","time","version"}``。"""
return self._request("GET", self.gateway_url("version", {}))
def run_status(self, detail: bool = True) -> Dict[str, Any]:
"""``GET /api/v1/map/cmd/runstatus/_/v1``。
Parameters
----------
detail:
是否返回详细字段（process / slice / store 等）。
"""
return self._request(
"GET",
self.cmd_url("runstatus", "_", "v1", {"detail": detail}),
)
def view_logs(self, record: int = 100) -> Any:
"""``GET /api/v1/map/cmd/viewlogs/_/v1?record=N``。"""
return self._request(
"GET", self.cmd_url("viewlogs", "_", "v1", {"record": int(record)})
)
# ========================================================================
# 分类 1：地图管理（Map Management）
# ========================================================================
def check_map_file(self, md5: str) -> Dict[str, Any]:
"""``GET /api/v1/map/mapfile?md5=...``。
秒传检测：客户端先算文件 md5，若后端返回 ``{"result": true, "fileid", "mapid", ...}``
则跳过上传；不存在时返回 ``{"result": false}``。
"""
return self._request("GET", self.map_url("mapfile", {"md5": md5}))
def upload_map(
self,
path_or_bytes: Union[str, bytes, Tuple[str, bytes]],
*,
filename: Optional[str] = None,
md5: Optional[str] = None,
extra_fields: Optional[Mapping[str, Any]] = None,
) -> Dict[str, Any]:
"""``POST /api/v1/map/uploads``（multipart/form-data）。
Parameters
----------
path_or_bytes:
文件路径、字节串，或 ``(filename, bytes)`` 元组。
filename:
显示文件名；``path_or_bytes`` 是路径时自动取 basename。
md5:
预计算 MD5；不给则自动算（后端不读该表单字段，按文件内容自算）。
extra_fields:
额外随 form 一起 POST 的字段（后端只取文件部分，表单字段均不读；
加密请在首次 :meth:`open_map` 时传 ``secretKey``）。
返回样例（真实回填见 C.1 文档）::
{"fileid": "caf27de8a225", "mapid": "caf27de8a225", "uploadname": "my.dwg"}
同内容文件已被某图使用时还带 ``version`` / ``geom``；图片另带 ``maptype``；
失败时为 ``{"uploadname", "error"}``。
"""
if isinstance(path_or_bytes, tuple) and len(path_or_bytes) == 2:
name, content = path_or_bytes
elif isinstance(path_or_bytes, (bytes, bytearray)):
if not filename:
raise ValueError("upload_map: bytes 模式必须给 filename")
name, content = filename, bytes(path_or_bytes)
else:
name = filename or os.path.basename(str(path_or_bytes))
with open(str(path_or_bytes), "rb") as f:
content = f.read()
if md5 is None:
md5 = file_md5(content)
data = {"md5": md5}
if extra_fields:
data.update({k: str(v) for k, v in extra_fields.items()})
files = {"file": (name, content, "application/octet-stream")}
return self._request(
"POST", self.map_url("uploads", {}), form=data, files=files
)
def open_map(
self,
mapid: str,
version: str = "v1",
*,
mapopenway: str = MapOpenWay.GeomRender,
fileid: Optional[str] = None,
style: Optional[Mapping[str, Any]] = None,
secretKey: Optional[str] = None,
accessKey: Optional[str] = None,
mapbounds: Optional[Any] = None,
mapLayerIndex: Optional[Any] = None,
layers: Optional[Any] = None,
useMethod: str = "POST",
**extra: Any,
) -> Dict[str, Any]:
"""``GET/POST /api/v1/map/openmap/{mapid}?version=...``。
打开已上传的图或首次创建。成功返回 ``{"mapid","version","layer","layers",...}``
并隐式记录 ``self._cur_map``（``layer`` 取响应的 ``layer`` 字段）。
后端不读 ``mapopenway``，只认布尔 ``geom``（缺省 false = 内存模式）；SDK 按官方
按官方 JS SDK 的做法换算 ``geom = mapopenway != "Memory"``：
- ``Memory``：``geom:false``，内存打开（从未以 GeomRender 打开过的图，cmd 默认走内存进程，condition 查询不可用）
- ``GeomRender``：``geom:true``，几何渲染。首次打开是异步的——响应
``{"status":"working"}``，需 :meth:`wait_map_open_finish` 轮询到 finish 后再打开一次
显式给 ``secretKey`` / ``accessKey`` 时，本次请求对应的头只带该值（头会覆盖 body 同名键）。
"""
params = {"version": version}
body: Dict[str, Any] = {
"mapid": mapid,
"version": version,
"geom": mapopenway != MapOpenWay.Memory,
}
if fileid:
body["fileid"] = fileid
if style is not None:
body["style"] = style
headers: Dict[str, str] = {}
if secretKey:
body["secretKey"] = secretKey
headers["secretKey"] = secretKey
if accessKey:
body["accessKey"] = accessKey
headers["accessKey"] = accessKey
if mapbounds is not None:
body["mapbounds"] = mapbounds
if mapLayerIndex is not None:
body["mapLayerIndex"] = mapLayerIndex
if layers is not None:
body["layers"] = layers
for k, v in extra.items():
if v is not None:
body[k] = v
url = self.map_url(f"openmap/{quote(mapid, safe='_*')}", params)
res = self._request(useMethod, url, json_body=body, headers=headers or None)
if isinstance(res, dict) and res.get("mapid"):
self.set_current_map(
mapid=res.get("mapid", mapid),
version=res.get("version", version),
layer=res.get("layer") or "",
fileid=res.get("fileid", fileid or mapid),
)
return res
def update_map(
self,
mapid: str,
version: str = "v1",
*,
body: Optional[Mapping[str, Any]] = None,
) -> Dict[str, Any]:
"""``POST /api/v1/map/updatemap/{mapid}``。
body 给**新** ``fileid`` 时为已有 mapid 新建版本 vN+1（可带 ``deleteOldVersion: True``
删除旧版本）；``fileid`` 为空或与当前版本相同则按原样重开。打开方式同
:meth:`open_map` 由 ``geom`` 决定——body 未给 ``geom`` 时按 ``mapopenway`` 换算
（缺省 GeomRender → ``geom: True``）。
"""
b = dict(body or {})
mapopenway = b.pop("mapopenway", None)
b.setdefault("geom", mapopenway != MapOpenWay.Memory)
url = self.map_url(
f"updatemap/{quote(mapid, safe='_*')}", {"version": version}
)
return self._request("POST", url, json_body=b)
def wait_map_open_finish(
self,
mapid: str,
version: str = "_",
*,
interval: float = 2.0,
max_tries: int = 150,
) -> Dict[str, Any]:
"""轮询 ``cmd/listmaps/{mapid}/{version}`` 直到该版本 ``status`` 为 ``finish``（或 error/runerror）。
对应官方 JS SDK 的 ``waitMapOpenFinish``；GeomRender 首次打开返回 ``status:"working"`` 后使用。
返回 ``{"status":"finish","version":..}`` 或 ``{"status":"error","error":..}``。
"""
for _ in range(max_tries):
res = self._request("GET", self.cmd_url("listmaps", mapid, version or "_"))
vers = (res or {}).get(mapid) or [] if isinstance(res, dict) else []
item = next(
(v for v in vers if version in ("", "_") or v.get("version") == version),
vers[0] if vers else None,
)
if item:
st = item.get("status")
if st == "finish":
return {"status": "finish", "version": item.get("version")}
if st in ("error", "runerror"):
return {"status": "error", "version": item.get("version"),
"error": item.get("errmsg") or st}
time.sleep(interval)
return {"status": "error", "error": "try max times"}
def close_map(self, mapid: str, version: str = "v1") -> Dict[str, Any]:
"""``GET /api/v1/map/cmd/closemap/{mapid}/{version}``。
仅影响当前内存/渲染进程；不会删除数据库文件。
实现备注：图形服务 端对"无 body POST"会回 500 ``param error``，
SDK 统一用 GET 变体规避该陷阱。
"""
return self._request("GET", self.cmd_url("closemap", mapid, version))
def delete_map(
self, mapid: str, version: str = "v1", *, delete_dwg_file: bool = False
) -> Dict[str, Any]:
"""``GET /api/v1/map/cmd/deletemap/{mapid}/{version}``。
``delete_dwg_file=True`` 时同时删除上传的源文件；默认 ``False``
时是否删源文件取决于服务端配置 ``noAutoDeleteDwgFile``。
旧版的 ``keep_source`` 参数后端从未消费过，已移除。
"""
params = {"deleteDwgFile": True} if delete_dwg_file else {}
return self._request(
"GET",
self.cmd_url("deletemap", mapid, version, params),
)
def rename_map(self, old_mapid: str, new_mapid: str) -> Dict[str, Any]:
"""``GET /api/v1/map/cmd/renamemap/{old}/_?newMapId=...``。
注意后端读的是驼峰 ``newMapId``，且要求当前用户对该图有 admin 权限。
"""
return self._request(
"GET",
self.cmd_url("renamemap", old_mapid, "_", {"newMapId": new_mapid}),
)
def delete_map_file(self, filenames: Union[str, Sequence[str]]) -> Dict[str, Any]:
"""``GET /api/v1/map/cmd/deletemapfile/_/v1?files=a.dwg;b.dwg``。"""
if isinstance(filenames, (list, tuple)):
files = ";".join(filenames)
else:
files = str(filenames)
return self._request(
"GET",
self.cmd_url("deletemapfile", "_", "v1", {"files": files}),
)
def list_maps(
self,
*,
cur_page: Optional[int] = None,
page_count: Optional[int] = None,
map_ids: Optional[Union[str, Sequence[str]]] = None,
) -> Dict[str, Any]:
"""``GET/POST /api/v1/map/cmd/listmaps/_/v1``。
后端实际参数为 ``curPage`` / ``pageCount`` / ``mapIds``（数组），
无关键字搜索；``mapIds`` 由 ``toStringList()`` 读取，必须放 JSON
body，因此带参数时走 POST。
不带分页参数时返回 ``{mapid: [ {version,db,filename,mtime,uploadname,status,geom}, ... ]}``；
带 ``cur_page``/``page_count`` 时返回分页信封 ``{count, curPage, pageCount,
mapIds: [本页 mapid], mapInfos: {mapid: [版本…]}, allMapIds: [全部 mapid（不过滤）]}``。
非 root 默认隐藏 ``ns_`` 前缀（配置 ``map_list_exclude_prefix``），root 隐藏 ``ns_temp_``。
"""
body: Dict[str, Any] = {}
if cur_page is not None:
body["curPage"] = int(cur_page)
if page_count is not None:
body["pageCount"] = int(page_count)
if map_ids is not None:
body["mapIds"] = (
list(map_ids) if isinstance(map_ids, (list, tuple)) else [str(map_ids)]
)
if not body:
return self._request(
"GET", self.cmd_url("listmaps", "_", "v1")
)
return self._request(
"POST", self.cmd_url("listmaps", "_", "v1"), json_body=body
)
def metadata(
self,
mapid: Optional[str] = None,
version: Optional[str] = None,
) -> Dict[str, Any]:
"""``GET /api/v1/map/cmd/metadata/{mapid}/{version}``。
返回图的元数据：``layers / bounds / linetypes / textstyles / projection / ...``。
注：``cmdMetadata`` 只解析路径 ``mapid``/``version``，不读业务参数；
旧版的 ``enable_edit_meta`` / ``fileid`` 均无消费点，已移除。
"""
return self._request(
"GET",
self.cmd_url("metadata", mapid or "", version or ""),
)
def update_metadata(
self,
mapid: str,
version: str = "v1",
*,
body: Optional[Mapping[str, Any]] = None,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/updateMetadata/{mapid}/{version}``。
body 的**顶层键**逐个合并进元数据（``meta[key] = value``，同名覆盖），形如
``{"description": "xx", "tags": "a,b"}``——不是 ``{"fields":[...]}`` 数组形式。
"""
return self._request(
"POST",
self.cmd_url("updateMetadata", mapid, version),
json_body=dict(body or {}),
)
def thumbnail_url(
self,
mapid: Optional[str] = None,
version: Optional[str] = None,
*,
width: int = 200,
height: int = 200,
dark_theme: Optional[bool] = None,
) -> str:
"""构造缩略图 URL（不发起请求）。直接在 ``<img src=...>`` 中使用。
后端实际只消费 ``width`` / ``height`` / ``darkTheme``
（``layers`` / ``backcolor`` 无消费点，已移除）。
"""
params: Dict[str, Any] = {"width": width, "height": height}
if dark_theme is not None:
params["darkTheme"] = dark_theme
return self.cmd_url(
"thumbnail", mapid or "", version or "", params,
)
def thumbnail_bytes(
self,
mapid: Optional[str] = None,
version: Optional[str] = None,
*,
width: int = 200,
height: int = 200,
) -> bytes:
"""拉取缩略图 PNG 字节流。"""
url = self.thumbnail_url(mapid, version, width=width, height=height)
resp = self._request("GET", url, stream=True)
try:
self._raise_if_error(resp, None)
return resp.content
finally:
resp.close()
def support_format(self) -> Dict[str, Any]:
"""``GET /api/v1/map/cmd/supportFormat/_/v1``。返回所有支持的上传格式。"""
return self._request("GET", self.cmd_url("supportFormat", "_", "v1"))
def const_data(self) -> Dict[str, Any]:
"""``GET /api/v1/map/cmd/constData/_/v1``。返回系统常量（色号、单位等）。"""
return self._request("GET", self.cmd_url("constData", "_", "v1"))
def get_data_bounds(
self, mapid: Optional[str] = None, version: Optional[str] = None
) -> Dict[str, Any]:
"""``GET /api/v1/map/cmd/getDataBounds/{mapid}/{version}``。返回 ``[minx,miny,maxx,maxy]``。"""
return self._request(
"GET", self.cmd_url("getDataBounds", mapid or "", version or "")
)
# ========================================================================
# 分类 2：瓦片 / 样式（Tiles & Styles）
# ========================================================================
def raster_tile_url(
self,
mapid: Optional[str] = None,
version: Optional[str] = None,
*,
layer: Optional[str] = None,
fileid: Optional[str] = None,
z: str = "{z}",
x: str = "{x}",
y: str = "{y}",
) -> str:
"""构造栅格瓦片 URL；含 ``{z}/{x}/{y}`` 占位符，可直接作为 vjmap source。"""
mapid = mapid or self._cur_map.get("mapid") or "_"
version = version or self._cur_map.get("version") or "v1"
layer = layer or self._cur_map.get("layer") or "_"
fileid = fileid or self._cur_map.get("fileid") or ""
base = f"{self.base_url}{self._map_prefix()}tile/{mapid}/{version}/{layer}/{z}/{x}/{y}"
qs = _encode_query({"tag": fileid, "token": self.token})
return f"{base}?{qs}"
def vector_tile_url(
self,
mapid: Optional[str] = None,
version: Optional[str] = None,
*,
layer: Optional[str] = None,
fileid: Optional[str] = None,
z: str = "{z}",
x: str = "{x}",
y: str = "{y}",
) -> str:
"""构造矢量瓦片 (MVT) URL。"""
mapid = mapid or self._cur_map.get("mapid") or "_"
version = version or self._cur_map.get("version") or "v1"
layer = layer or self._cur_map.get("layer") or "main"
fileid = fileid or self._cur_map.get("fileid") or ""
base = f"{self.base_url}{self._map_prefix()}tile/{mapid}/{version}/{layer}/{z}/{x}/{y}.mvt"
qs = _encode_query({"tag": fileid, "token": self.token})
return f"{base}?{qs}"
def wms_tile_url(
self,
mapid: Union[str, Sequence[str]],
version: Union[str, Sequence[str]] = "v1",
*,
layer: Union[str, Sequence[str], None] = None,
width: int = 256,
height: int = 256,
bbox: str = "{bbox-epsg-3857}",
srs: str = "EPSG:3857",
background_color: Optional[str] = None,
transparent: Optional[bool] = None,
crs: Union[str, Sequence[str], None] = None,
four_parameter: Union[str, Sequence[str], None] = None,
mvt: bool = False,
) -> str:
"""构造 WMS 单图 / 多图 URL；可直接作为 vjmap ``raster`` source。
单图：`/api/v1/map/cmd/wms/{mapid}/{version}`（``mvt=True`` 时
version 后加 ``.mvt`` 后缀）。
多图：走 ``/api/v1/service/wms``；Go 端按 ``"_,"`` 切分
``mapid/version/layers/crs/fourParameter``，五者长度必须一致，
否则返回 ``paramter error``；不支持 ``mvt``。
后端实际消费 ``layers``（样式名，复数）与 ``backgroundColor``
（``"r,g,b[,a]"`` 串，仅 ``transparent=false`` 时生效，且仅
单图 ``cmd/wms`` 有效——Go 多图代理不转发它）；旧版
参数名 ``layer`` / ``backcolor`` 没有消费点，已修正。
"""
qs_params: Dict[str, Any] = {
"srs": srs, "width": width, "height": height,
"token": self.token,
}
if isinstance(mapid, (list, tuple)):
if mvt:
raise ValueError(
"/service/wms 不支持 mvt（仅单图 cmd/wms 路径支持 .mvt 后缀）"
)
n = len(mapid)
def _rep(v: Any) -> str:
arr = list(v) if isinstance(v, (list, tuple)) else [v] * n
if len(arr) != n:
raise ValueError("wms_tile_url: 数组参数长度须与 mapid 一致")
return "_,".join("" if x is None else str(x) for x in arr)
base = f"{self.base_url}service/wms"
qs_params.update({
"mapid": _rep(mapid),
"version": _rep(version),
"layers": _rep(layer),
"crs": _rep(crs),
"fourParameter": _rep(four_parameter),
})
else:
ext = ".mvt" if mvt else ""
base = (
f"{self.base_url}{self._map_prefix()}cmd/wms/"
f"{quote(mapid, safe='_*')}/{quote(version, safe='_*')}{ext}"
)
if layer is not None:
qs_params["layers"] = layer
if crs is not None:
qs_params["crs"] = crs
if four_parameter is not None:
qs_params["fourParameter"] = four_parameter
if background_color is not None:
qs_params["backgroundColor"] = background_color
if transparent is not None:
qs_params["transparent"] = str(transparent).lower()
qs = _encode_query(qs_params)
# bbox 常包含 ``{bbox-epsg-3857}`` 等 如Leaflet 占位符，
# 故保留原文，不做 percent-encode。
return f"{base}?bbox={bbox}&{qs}"
def create_map_style(
self,
mapid: Optional[str] = None,
version: Optional[str] = None,
*,
layeron: Optional[Union[Sequence[int], str]] = None,
layeroff: Optional[Union[Sequence[int], str]] = None,
backcolor: Optional[Union[int, str]] = None,
clipbounds: Optional[str] = None,
expression: Optional[str] = None,
lineweight: Optional[Sequence[int]] = None,
is_geom_layer: bool = True,
name: Optional[str] = None,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/createMapStyle/{mapid}/{version}``。
返回 ``{"stylename": "<短哈希>"}``；该 stylename 就是后续
:meth:`raster_tile_url` / :meth:`vector_tile_url` 中的 ``layer`` 段。
``name`` 以 ``stylename`` 键发送（后端读 ``stylename``，不读 ``name``）。
"""
body: Dict[str, Any] = {"geom": is_geom_layer}
if name is not None:
body["stylename"] = name
if layeron is not None:
body["layeron"] = (
f"({','.join(str(x) for x in layeron)})"
if isinstance(layeron, (list, tuple)) else layeron
)
if layeroff is not None:
body["layeroff"] = (
f"({','.join(str(x) for x in layeroff)})"
if isinstance(layeroff, (list, tuple)) else layeroff
)
if backcolor is not None:
body["backcolor"] = backcolor
if clipbounds is not None:
body["clipbounds"] = clipbounds
if expression is not None:
body["expression"] = expression
if lineweight is not None:
body["lineweight"] = json.dumps(list(lineweight))
return self._request(
"POST",
self.cmd_url("createMapStyle", mapid or "", version or ""),
json_body=body,
)
def cmd_switch_layers(
self,
visible_layers: Sequence[Union[int, str]],
*,
expression: Optional[str] = None,
dark_mode: Optional[bool] = None,
mapid: Optional[str] = None,
version: Optional[str] = None,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/switchlayer/{mapid}/{version}``。"""
body: Dict[str, Any] = {"visibleLayer": list(visible_layers)}
if expression is not None:
body["expression"] = expression
if dark_mode is not None:
body["darkMode"] = bool(dark_mode)
return self._request(
"POST",
self.cmd_url("switchlayer", mapid or "", version or ""),
json_body=body,
)
def cmd_update_style(
self,
*,
name: str = "",
layeron: Optional[Union[Sequence[int], str]] = None,
layeroff: Optional[Union[Sequence[int], str]] = None,
backcolor: Optional[Union[int, str]] = None,
clipbounds: Optional[Any] = None,
expression: Optional[str] = None,
lineweight: Optional[Sequence[int]] = None,
mapid: Optional[str] = None,
version: Optional[str] = None,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/updateStyle/{mapid}/{version}``。"""
def _to_layer_str(v):
if v is None:
return ""
if isinstance(v, (list, tuple)):
return "(" + ",".join(str(x) for x in v) + ")"
return str(v)
body: Dict[str, Any] = {
"stylename": name,
"layeron": _to_layer_str(layeron),
"layeroff": _to_layer_str(layeroff),
"bkcolor": backcolor if backcolor is not None else "",
"expression": expression or "",
"lineweight": json.dumps(list(lineweight)) if lineweight else "",
}
if clipbounds is not None:
body["clipbounds"] = (
json.dumps(clipbounds) if not isinstance(clipbounds, str) else clipbounds
)
else:
body["clipbounds"] = ""
return self._request(
"POST",
self.cmd_url("updateStyle", mapid or "", version or ""),
json_body=body,
)
def slice_layer(
self,
*,
layer: Optional[str] = None,
zoom: Union[int, str, Sequence[int]] = 0,
ismvt: bool = False,
iscancel: bool = False,
is_all_cancel: bool = False,
batch_num: Optional[int] = None,
idle_batch_sleep_ms: Optional[int] = None,
busy_batch_sleep_ms: Optional[int] = None,
mapid: Optional[str] = None,
version: Optional[str] = None,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/slicelayer/{mapid}/{version}``。触发/取消图层预切片。"""
body: Dict[str, Any] = {
"layer": layer or self._cur_map.get("layer") or "",
"zoom": zoom,
"ismvt": bool(ismvt),
"iscancel": bool(iscancel),
"isAllCancel": bool(is_all_cancel),
}
if batch_num is not None:
body["batchnum"] = batch_num
if idle_batch_sleep_ms is not None:
body["idle_batch_sleepms"] = idle_batch_sleep_ms
if busy_batch_sleep_ms is not None:
body["busy_batch_sleepms"] = busy_batch_sleep_ms
return self._request(
"POST",
self.cmd_url("slicelayer", mapid or "", version or ""),
json_body=body,
)
def get_slice_cache_zoom(
self,
*,
layer: Optional[str] = None,
ismvt: bool = False,
mapid: Optional[str] = None,
version: Optional[str] = None,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/getSliceCacheZoom/{mapid}/{version}``。"""
body: Dict[str, Any] = {
"layer": layer or self._cur_map.get("layer") or "",
"ismvt": bool(ismvt),
}
return self._request(
"POST",
self.cmd_url("getSliceCacheZoom", mapid or "", version or ""),
json_body=body,
)
def delete_style(
self,
stylename: str,
*,
mapid: Optional[str] = None,
version: Optional[str] = None,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/deletestyle/{mapid}/{version}``。
后端按 ``styleid`` 键匹配 layername 或 style name；``*``/``a*``/``m*``/``s*`` 为通配。
"""
return self._request(
"POST",
self.cmd_url("deletestyle", mapid or "", version or ""),
json_body={"styleid": stylename},
)
# ========================================================================
# 分类 3：要素查询 / 几何（Query）
# ========================================================================
def query_features(
self,
*,
querytype: str,
mapid: Optional[str] = None,
version: Optional[str] = None,
layer: Optional[str] = None,
# 点查询
x: Optional[float] = None,
y: Optional[float] = None,
pixelsize: Optional[float] = None,
# 矩形查询
x1: Optional[float] = None,
y1: Optional[float] = None,
x2: Optional[float] = None,
y2: Optional[float] = None,
# 条件查询 / 表达式查询
expr: Optional[str] = None,
condition: Optional[str] = None,
beginpos: Optional[int] = None,
bounds: Optional[Any] = None,
# 通用
zoom: int = 1,
limit: Optional[int] = None,
fields: Optional[str] = None,
geom: Optional[bool] = None,
simplify_tolerance: Optional[float] = None,
use_cache: Optional[bool] = None,
to_map_coordinate: Optional[bool] = None,
max_geom_bytes_size: Optional[int] = None,
pixel_to_geo_length: Optional[float] = None,
extra: Optional[Mapping[str, Any]] = None,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/queryFeatures/{mapid}/{version}``。
``querytype`` 可选值:
- ``point``: 传 ``x/y/pixelsize``
- ``rect``: 传 ``x1/y1/x2/y2``
- ``expresion``: 传 ``expr``
- ``condition``: 传 ``condition`` + 可选 ``bounds``
``bounds`` 后端只按**字符串**读（传数组会被转成空串、空间过滤被静默丢弃），SDK 转换：
点列表 ``[(x,y),...]`` → ``"x,y;x,y"``（折线/多边形查询，首尾相同为闭合）；
4 数 ``[x1,y1,x2,y2]`` → ``"[x1,y1,x2,y2]"``；字符串原样。
``geom`` 不是返回几何的开关，而是命令路由：缺省按图打开方式；``geom=False`` 强制走
内存进程，只支持 point/rect/expresion，condition 会返回 ``not support query type!``。
返回 ``{result:[...], ...}``，每个 result 的坐标已展开为 ``geom``。
"""
body: Dict[str, Any] = {
"querytype": querytype,
"zoom": zoom,
"layername": layer or self._cur_map.get("layer") or "",
}
for k, v in {
"x": x, "y": y, "pixelsize": pixelsize,
"x1": x1, "y1": y1, "x2": x2, "y2": y2,
"expr": expr, "condition": condition,
"beginpos": beginpos,
"maxReturnCount": limit, "limit": limit,
"fields": fields,
"geom": geom, "simplifyTolerance": simplify_tolerance,
"useCache": use_cache, "toMapCoordinate": to_map_coordinate,
"maxGeomBytesSize": max_geom_bytes_size,
"pixelToGeoLength": pixel_to_geo_length,
}.items():
if v is not None:
body[k] = v
if bounds is not None:
if (
isinstance(bounds, (list, tuple))
and bounds and isinstance(bounds[0], (list, tuple))
):
body["bounds"] = ";".join(
f"{p[0]},{p[1]}" for p in bounds # type: ignore[index]
)
elif isinstance(bounds, (list, tuple)):
body["bounds"] = json.dumps(list(bounds), separators=(",", ":"))
else:
body["bounds"] = str(bounds)
if extra:
body.update(dict(extra))
return self._request(
"POST",
self.cmd_url("queryFeatures", mapid or "", version or ""),
json_body=body,
)
def create_entity_geo_data(
self,
*,
mapdata: Any,
map_extent: Optional[Any] = None,
render_accuracy: Optional[float] = None,
exclude_attribute: bool = True,
use_zip: bool = True,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/createEntityGeoData/_null/v1``。
``mapdata`` 是客户端生成的 DWG 结构体描述（通常是 JSON 字符串），
后端把它几何化后回显。
"""
body: Dict[str, Any] = {
"mapdata": mapdata,
"excludeAttribute": bool(exclude_attribute),
"useZip": bool(use_zip),
}
if map_extent is not None:
body["mapExtent"] = (
map_extent if isinstance(map_extent, str) else json.dumps(map_extent)
)
if render_accuracy is not None:
body["renderAccuracy"] = render_accuracy
return self._request(
"POST",
self.cmd_url("createEntityGeoData", "_null", "v1"),
json_body=body,
)
def coord_transform(
self,
*,
srs: str,
crs: str,
points: Union[Sequence[Sequence[float]], Sequence[float], Mapping[str, float]],
four_parameter: Optional[Union[str, Sequence[Union[int, float]]]] = None,
is_inverse_four_parameter: Optional[bool] = None,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/coordtransform/_/_``。
``points`` 支持：
- 单点: ``(x,y)`` 或 ``{"x":1,"y":2}``
- 多点: ``[(x1,y1),(x2,y2)]``
"""
def _norm_one(p):
if isinstance(p, Mapping):
return f"{p['x']},{p['y']}"
return f"{p[0]},{p[1]}"
if isinstance(points, Mapping):
pts = [_norm_one(points)]
elif (
isinstance(points, (list, tuple))
and points
and isinstance(points[0], (int, float))
):
pts = [_norm_one(points)]
else:
pts = [_norm_one(p) for p in points] # type: ignore[arg-type]
body: Dict[str, Any] = {
"srs": srs,
"crs": crs,
"points": ";".join(pts),
}
if four_parameter is not None:
if isinstance(four_parameter, (list, tuple)):
body["fourParameter"] = ",".join(str(x) for x in four_parameter)
else:
body["fourParameter"] = four_parameter
if is_inverse_four_parameter is True:
body["isInverseFourParamter"] = True
return self._request(
"POST",
self.cmd_url("coordtransform", "_", "_"),
json_body=body,
)
def prj_wkt_to_prj4_str(
self,
wkt: str,
*,
from_: str = "",
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/prjWktToPrj4Str/_/_``。WKT/ESRI prj → proj4 字符串。
``from_`` 可空或 ``wmsauto / xml / urn / crsurl / url / micoordsys / pci``。
"""
return self._request(
"POST",
self.cmd_url("prjWktToPrj4Str", "_", "_"),
json_body={"wkt": wkt, "from": from_},
)
# ========================================================================
# 分类 4：图形分析（Analysis）
#
# 分析类接口参数差异大、依赖真实数据，SDK 统一使用 pass-through 风格：
# 调用方只需按接口文档组好参数，SDK 负责打包 POST body 与错误处理。
# ========================================================================
def _post_cmd_null(self, cmd: str, params: Mapping[str, Any]) -> Dict[str, Any]:
"""分析类接口统一走 ``{mapid}=_null``、``{version}=v1``，POST JSON。"""
return self._request(
"POST",
self.cmd_url(cmd, "_null", "v1"),
json_body=dict(params),
)
def cmd_map_diff(self, **params: Any) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/mapDiff/_null/v1``。像素差异对比。
关键参数: ``mapid1``, ``version1``, ``layer1``, ``mapid2``, ``version2``,
``layer2``；调优 ``size``(4096) ``cellsize``(10) ``diffMinPixel``(0)
``diffMinAlpha``(25) ``diffMinColor``(6) ``noCompareNew`` ``noCompareDelete``。
返回 ``{status, modify, new, del}``（bbox 分号串）。
"""
return self._post_cmd_null("mapDiff", params)
def cmd_map_compare(self, **params: Any) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/mapCompare/_null/v1``。实体级对比。
关键参数: ``mapid1 / version1`` 和 ``mapid2 / version2``（``layer1`` 读了但未使用，
``layer2`` 不读）；``noCompareProps``（不比属性）、``comparePropMaxSize``(5000)。
返回 ``addEntities / deleteEntities / modifyEntities`` + 两图实体计数。
"""
return self._post_cmd_null("mapCompare", params)
def cmd_extract_table(self, **params: Any) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/extractTable/_null/v1``。自动识别 DWG 内表格。
常用参数: ``mapid``, ``version``, ``layer``, ``bounds``, ``condition``,
``isContains``, ``geom``, ``pageLimit``, ``findChildMapRects``。
返回每张表的单元格网格与文本（``tables``）。
"""
return self._post_cmd_null("extractTable", params)
def cmd_extract_center_lines(self, **params: Any) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/extractCenterLines/_null/v1``。闭合多段线中心线提取。
常用参数: ``mapid``, ``version``, ``layer``, ``bounds``, ``condition``,
``isContains``, ``pageLimit``，及 ``bMergeLine`` 等 ``bXxx`` 调优开关。
"""
return self._post_cmd_null("extractCenterLines", params)
def cmd_match_object(self, **params: Any) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/objectMatch/_null/v1``。模板匹配（找相似子图）。
关键参数: ``mapid / version``（底图）、``objectBounds``（模板 bbox，必填）
或 ``objectIds``；``mapBounds``（搜索范围）、``score``(0.6)、
``maxCount``(100)、``canOverlap``(false)、``maxOverlap``(0.2)、
``toleranceAngle``(0)、``minReduceArea``(256)、``method``(matchTemplate)。
返回 ``{status, match: [...]}``。
"""
return self._post_cmd_null("objectMatch", params)
def cmd_image_svg_to_dwg(self, **params: Any) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/imageSvgToDwg/_null/v1``。把 SVG / 位图转为 DWG。
常用参数: ``svg``（SVG 字符串）或 ``image``（base64 data-URL），
``layer``、``flipY``(true)、``offsetX``/``offsetY``、``lineWidthScale``、
``retDocument``（true 时返回 ``doc`` DWG JSON；false 时落盘并返回
``fileid``——是响应字段不是入参）。
"""
return self._post_cmd_null("imageSvgToDwg", params)
# ========================================================================
# 分类 5：组合 / 拆分 / 导出（Compose & Export）
# ========================================================================
def cmd_compose_new_map(self, **params: Any) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/composeNewMap/_null/v1``。组合多张 DWG 成新图。
参数为后缀编号风格：``count`` 声明张数；第 1 张图用无后缀键
（``mapid``/``version``/``clipBounds``/``selByCrossing``/``layers``/
``fourParameter``/``isInverseFourParamter``/``layerPrefix``/``layerSuffix``/
``savefilename``/``purge``/``isFullExtent``），第 2 张起加序号
（``mapid2``/``clipBounds2``…）。``isRenameBlockNameIfExist`` 默认 true。
"""
return self._post_cmd_null("composeNewMap", params)
def cmd_split_map(self, **params: Any) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/cmdSplitMap/_null/v1``。按 bbox 列表拆出多张子图。
关键参数: ``mapid``/``version`` 或 ``fileid``；``clipBounds``（bbox 数组，
必填）；``isFullExtent``；``method``（``cloneObjects``/``cloneDb``/``cloneMap``）。
返回 ``{fileids: [...]}``。
"""
return self._post_cmd_null("cmdSplitMap", params)
def cmd_export_layout(self, **params: Any) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/exportLayout/_null/v1``。从布局空间导出新 DWG。
关键参数: ``mapid``, ``version``, ``layoutIndex`` 或 ``layoutName``
（同时给时 ``layoutIndex`` 优先）。新图 mapid 由后端按请求 hash 生成，
返回 ``{status, fileid}``。
"""
return self._post_cmd_null("exportLayout", params)
def _cmd_post_or_get(
self,
cmd: str,
mapid: Optional[str],
version: Optional[str],
params: Mapping[str, Any],
) -> Any:
"""cmd 路由对**空 JSON body 的 POST** 直接回 500 ``param error``，GET 不校验。
参数为空时改走 GET（无 query），否则 POST JSON。
"""
url = self.cmd_url(cmd, mapid or "", version or "")
if not params:
return self._request("GET", url)
return self._request("POST", url, json_body=dict(params))
def cmd_export_pdf(
self,
mapid: Optional[str] = None,
version: Optional[str] = None,
**params: Any,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/exportPdf/{mapid}/{version}``（无参数时走 GET）。
参数为 ODA 导出选项：``pageWidth``/``pageHeight``（mm）、页边距
``pageLeftMargin`` 等、``bZoomToExtents``(true)、``bActiveLayout``(true)、
``geomDPI``/``colorImagesDPI``/``bwImagesDPI``、``isBlackWhiteMode``、
``bEmbededTTF``、``title``/``author``/``creator``/``producer``、``pdfPath``。
返回 ``{status, path}``（``download/xxx.pdf``）。
"""
return self._cmd_post_or_get("exportPdf", mapid, version, params)
def cmd_export_dxf(
self,
mapid: Optional[str] = None,
version: Optional[str] = None,
**params: Any,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/exportDxf/{mapid}/{version}``（无参数时走 GET）。整库导出 DXF。
常用参数: ``filename``（输出名，自动补 ``.dxf``）、``cadVersion``
（``2000``/``2004``/``2007``/``2010``/``2013``/``2018``，``*``=沿用源图）。
返回 ``{status, fileid}``。
"""
return self._cmd_post_or_get("exportDxf", mapid, version, params)
def cmd_save_offline_tile_db(self, **params: Any) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/saveOfflineTileDb/_null/v1``。导出离线瓦片包。
关键参数: ``mapid``/``version``、``layers``（style 名逗号分隔，必填）、
``rasterMinZoom``/``rasterMaxZoom``、``vectorMinZoom``/``vectorMaxZoom``。
复制既有瓦片到新 sqlite 包，返回 ``{status, dbfilename}``。
"""
return self._post_cmd_null("saveOfflineTileDb", params)
# ========================================================================
# 分类 6：WebCAD（cmd 始终注册；/service/webcad 是前端 license 握手，不是启用探针）
# ========================================================================
def list_webcad_draws(self) -> Dict[str, Any]:
"""``GET /api/v1/map/cmd/listWebcadDraws/_/v1``。列出所有 WebCAD 图。"""
return self._request("GET", self.cmd_url("listWebcadDraws", "_", "v1"))
def get_webcad_data(
self,
mapid: str,
version: str = "v1",
**params: Any,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/getWebcadData/{mapid}/{version}``。
常用参数: ``patchId``（必填，``base`` 或 ``pN``）, ``branch``（默认 ``main``）,
``designPath``, ``type``（默认 ``imports``）, ``editAreas``/``editLayers``,
``selByCrossing``, ``clipbounds``。
"""
return self._request(
"POST",
self.cmd_url("getWebcadData", mapid, version),
json_body=dict(params),
)
def save_webcad_patch(
self,
mapid: str,
version: str = "v1",
**params: Any,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/saveWebcadPatch/{mapid}/{version}``。
常用参数: ``patch``（JSON 增量，必填）, ``patchCompressed``, ``branch``,
``parentId``（冲突检测基线，``base``/``pN``，默认 ``base``）,
``author``/``userId``/``userName``, ``remark``（提交说明）,
``designPath``, ``type``, ``drawingName``, ``editAreas``/``editLayers``,
``tileAlpha``, ``conflictResolution``(+``Compressed``)。
返回 ``{status, patchId, branch}``；冲突时 ``{error:"conflict", conflict:{...}}``。
"""
return self._request(
"POST",
self.cmd_url("saveWebcadPatch", mapid, version),
json_body=dict(params),
)
def delete_webcad_draw(
self,
mapid: str,
version: str = "v1",
**params: Any,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/deleteWebcadDraw/{mapid}/{version}``（无参数时走 GET）。
常用参数: ``type``（默认 ``imports``）, ``designPath``, ``branch``
（默认 ``main``）, ``patchId``（只删指定补丁）。
"""
return self._cmd_post_or_get("deleteWebcadDraw", mapid, version, params)
def convert_webcad(
self,
mapid: str,
version: str = "v1",
**params: Any,
) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/convertWebCAD/{mapid}/{version}``（无参数时走 GET）。
把一张已上传的 DWG 转成 WebCAD 可用的结构（初始化分支、预计算）。
"""
return self._cmd_post_or_get("convertWebCAD", mapid, version, params)
def export_webcad(self, **params: Any) -> Dict[str, Any]:
"""``POST /api/v1/map/cmd/exportWebCAD/_null/v1``。
把 WebCAD 图导出为新的 DWG 文件。关键参数:
``mapid``, ``version``, ``designPath``, ``branch``（默认 ``main``）,
``filename``（输出名）, ``type``, ``isTileMode``, ``editAreas``/``editLayers``,
``unGroup``, ``exportDimAsNative``, ``useCache``, ``vcadData``。
返回 ``{status, path, filename, cached}``。
"""
return self._post_cmd_null("exportWebCAD", params)
# ========================================================================
# 分类 7：工作区（Workspace）
# ========================================================================
def get_workspaces(self) -> Any:
"""``GET /api/v1/service/getWorkspace``。返回当前账户可见的工作区列表。
非 root 用户下：``isPublic=true`` 的工作区返回完整字段；``isPublic=false``
的条目不脱敏删除，而是退化为仅 ``{"isPublic": false}`` 的占位项。
"""
return self._request("GET", self.service_url("getWorkspace"))
def workspace_create(
self,
*,
name: str,
alias: str = "",
work_dir: str = "",
is_public: bool = False,
disable_full_search: bool = False,
) -> Dict[str, Any]:
"""``POST /api/v1/service/workspaceCreate``。
返回 ``{"status":"ok","data":name}``。
"""
return self._request(
"POST",
self.service_url("workspaceCreate"),
json_body={
"name": name,
"alias": alias,
"workDir": work_dir,
"isPublic": bool(is_public),
"disableFullSearch": bool(disable_full_search),
},
)
def workspace_modify(self, workspace: Mapping[str, Any]) -> Dict[str, Any]:
"""``POST /api/v1/service/workspaceModify``。
后端按 ``name`` 定位，只更新 ``alias`` / ``isPublic`` / ``disableFullSearch``
三项——未给的会被重置为零值（``""``/``False``），故建议传 :meth:`get_workspaces`
里的完整 obj 再改。默认 casbin 策略下仅 root 可调（admin/guest 403）。
"""
return self._request(
"POST",
self.service_url("workspaceModify"),
json_body=dict(workspace),
)
def workspace_delete(self, name: str) -> Dict[str, Any]:
"""``DELETE /api/v1/service/workspaceDelete/{name}``。
⚠️ 不可逆：后端 ``os.RemoveAll`` 整个工作目录（该工作区全部图/瓦片/缓存）。
"""
url = self.service_url(f"workspaceDelete/{quote(name, safe='')}")
return self._request("DELETE", url)
def workspace_status(self) -> Any:
"""``GET /api/v1/service/workspaceStatus``。
返回**全部**工作区的状态数组（未起进程的 ``ProcIsNull: true``）；一个工作区都没有时为 ``None``。
"""
return self._request("GET", self.service_url("workspaceStatus"))
# ========================================================================
# 分类 8：KV 键值存储（Data Storage）
# ========================================================================
def get_custom_data(
self,
key: Union[str, Sequence[str]],
*,
ret_data_type: Optional[str] = None,
content_type: Optional[str] = None,
) -> Dict[str, Any]:
"""``GET /api/v1/service/data/{key}``。
``key`` 可以是字符串或字符串数组（批量，走 ``;;`` 拼接）。
``ret_data_type`` 可选 ``"value"`` / ``"prop"``，默认同时返回两个字段。
``content_type="image"`` 时后端会按 base64 dataURL 解出 PNG 返回；SDK
直接返回响应字节，调用方自己再处理。
"""
if isinstance(key, (list, tuple)):
key_seg = ";;".join(quote(str(k), safe="") for k in key)
else:
key_seg = quote(str(key), safe="")
params: Dict[str, Any] = {}
if ret_data_type is not None:
params["retDataType"] = ret_data_type
if content_type is not None:
params["contentType"] = content_type
return self._request(
"GET", self.service_url(f"data/{key_seg}", params)
)
def save_custom_data(
self,
key: str,
*,
value: Optional[str] = None,
prop: Optional[str] = None,
ttl: Optional[int] = None,
binary: Optional[bytes] = None,
mime: Optional[str] = None,
) -> Dict[str, Any]:
"""``POST /api/v1/service/data/{key}``，body 为 JSON ``{value, prop}``（都必须是字符串）。
二进制传 ``binary``：SDK 转 base64 存进 ``value``，``prop`` 缺省记为
``base64;<mime>``（``mime`` 缺省 ``application/octet-stream``）；回读用
``base64.b64decode(d["data"])``。后端 raw body 分支不可用（ShouldBindJSON 已先读走
body，小 body 存成 ``""``、大 body 被截断），SDK 不提供。
``ttl`` 单位秒，``None`` 或 0 = 长期有效。
"""
params: Dict[str, Any] = {}
if ttl is not None and ttl > 0:
params["ttl"] = int(ttl)
url = self.service_url(f"data/{quote(key, safe='')}", params)
body: Dict[str, Any] = {}
if binary is not None:
body["value"] = base64.b64encode(bytes(binary)).decode("ascii")
body["prop"] = prop if prop is not None else f"base64;{mime or 'application/octet-stream'}"
else:
if value is not None:
body["value"] = value
if prop is not None:
body["prop"] = prop
return self._request("POST", url, json_body=body)
def delete_custom_data(
self,
key: Union[str, Sequence[str]],
*,
is_prefix: bool = False,
) -> Dict[str, Any]:
"""``DELETE /api/v1/service/data/{key}``。
传字符串数组会批量删；``is_prefix=True`` 则把 ``key`` 作为前缀批量清。
"""
if isinstance(key, (list, tuple)):
key_seg = ";;".join(quote(str(k), safe="") for k in key)
else:
key_seg = quote(str(key), safe="")
params: Dict[str, Any] = {}
if is_prefix:
params["isPrefix"] = "true"
return self._request(
"DELETE", self.service_url(f"data/{key_seg}", params)
)
def save_multiple_custom_data(
self,
entries: Sequence[Mapping[str, Any]],
) -> Dict[str, Any]:
"""``POST /api/v1/service/datas``。
``entries`` 每个元素形如 ``{"key":..,"value":..,"prop":..,"ttl":"60"}``；
``ttl`` 字段是字符串形式的秒数。
"""
return self._request(
"POST",
self.service_url("datas"),
json_body=[dict(e) for e in entries],
)
def get_custom_data_keys(
self,
prefix: str = "",
*,
ret_data_type: Optional[str] = None,
is_sys: bool = False,
) -> Dict[str, Any]:
"""``GET /api/v1/service/datakeys``。
- ``prefix``: 键前缀（后端自动补 ``data_``，返回时去掉）。后端是 SQL
``key like '<prefix>%'``：``_``/``%`` 是通配符、ASCII 不分大小写、``'`` 会破坏 SQL
- ``ret_data_type``: ``value`` / ``prop`` / ``valueprop``（同时带出值）
- ``is_sys=True``: prefix 不补 ``data_``，返回整行对象数组
``{"keys":[{"Key","Value","Prop","Ttl"}]}``（Key 含 ``data_`` 前缀，Ttl 为到期 Unix 秒/0）
"""
params: Dict[str, Any] = {"prefix": prefix}
if ret_data_type is not None:
params["retDataType"] = ret_data_type
if is_sys:
params["isSys"] = "true"
return self._request("GET", self.service_url("datakeys", params))
# ========================================================================
# 分类 9：全文搜索（Full-text search）
# ========================================================================
def full_search(
self,
query: str = "",
*,
map_ver: Optional[str] = None,
type_: Optional[str] = None,
workspace: Optional[str] = None,
limit: int = 10,
offset: int = 0,
fields: Optional[str] = None,
facet: Optional[bool] = None,
bounding_box: Optional[str] = None,
time_range: Optional[str] = None,
use_get: bool = False,
) -> Dict[str, Any]:
"""``POST /api/v1/service/fullsearch/search`` (默认) 或 ``GET``。
Parameters
----------
query:
搜索关键词。
map_ver:
限定某张地图版本（``mapid_version``），多个用逗号分隔。
type_:
图元类型过滤：``"1,2,5"`` / ``"1-3"`` / ``"1-3,6"``。后端按 bleve
类型索引拆分。
workspace:
工作区名（多个逗号分隔）。
bounding_box:
``"minx,miny,maxx,maxy"`` 或 ``"[...]"``。
time_range:
``"2023-01-01~2023-12-31"`` 或 ``"7"``（最近 7 天）。
"""
body: Dict[str, Any] = {"query": query, "limit": limit, "offset": offset}
if map_ver is not None: body["map_ver"] = map_ver
if type_ is not None: body["type"] = type_
if workspace is not None: body["workspace"] = workspace
if fields is not None: body["fields"] = fields
if facet is not None: body["facet"] = bool(facet)
if bounding_box is not None: body["bounding_box"] = bounding_box
if time_range is not None: body["time_range"] = time_range
if use_get:
return self._request("GET", self.service_url("fullsearch/search", body))
return self._request(
"POST",
self.service_url("fullsearch/search"),
json_body=body,
)
def full_search_add(
self,
document: Mapping[str, Any],
) -> Dict[str, Any]:
"""``POST /api/v1/service/fullsearch/add``。
``document`` 至少包含 ``id`` / ``content`` / ``map_ver``，其余字段参见
:class:`Document` 定义。
"""
return self._request(
"POST",
self.service_url("fullsearch/add"),
json_body={"document": dict(document)},
)
def full_search_delete(
self,
doc_id: str,
) -> Dict[str, Any]:
"""``DELETE /api/v1/service/fullsearch/delete``。body 里放 ``{id}``。"""
return self._request(
"DELETE",
self.service_url("fullsearch/delete"),
json_body={"id": doc_id},
)
# ========================================================================
# 分类 10：符号库（Symbol Library）
# ========================================================================
def symbol_categories(self) -> Dict[str, Any]:
"""``GET /api/v1/service/symbol/categories``。返回分类列表。"""
return self._request("GET", self.service_url("symbol/categories"))
def symbol_list(
self,
category_id: str,
*,
page: int = 1,
page_size: int = 20,
keyword: Optional[str] = None,
) -> Dict[str, Any]:
"""``GET /api/v1/service/symbol/list/{categoryId}``。"""
params: Dict[str, Any] = {"page": page, "pageSize": page_size}
if keyword:
params["keyword"] = keyword
return self._request(
"GET",
self.service_url(f"symbol/list/{quote(category_id, safe='')}", params),
)
def symbol_get(self, symbol_id: str) -> Dict[str, Any]:
"""``GET /api/v1/service/symbol/{id}``。取单个符号详情。"""
return self._request(
"GET",
self.service_url(f"symbol/{quote(symbol_id, safe='')}"),
)
# ========================================================================
# 分类 12：/service 杂项（Service Misc）
# ========================================================================
def qrcode_url(self, content: str, size: int = 256) -> str:
"""``GET /api/v1/service/qrcode?content=&size=`` 的 URL 构造器。
直接把返回 URL 塞到 ``<img>`` 即可；后端回 PNG 字节。
"""
return self.service_url("qrcode", {"content": content, "size": int(size)})
def qrcode_bytes(self, content: str, size: int = 256) -> bytes:
"""实际下载 QR Code PNG 字节。"""
resp = self._request(
"GET",
self.qrcode_url(content, size),
stream=True,
)
if resp.status_code >= 400:
raise VjmapError(
f"HTTP {resp.status_code} for {resp.url}",
status=resp.status_code, url=resp.url, body=None,
)
return resp.content
def glyphs_url(self, fontstack: str, range_: str = "0-255") -> str:
"""``GET /api/v1/service/fonts/{fontstack}/{range}.pbf`` 的 URL。
``fontstack`` 可以逗号分隔多个字体名，后端会依次 fallback。
``range_`` 是 PBF 字形范围，常见 ``0-255`` / ``256-511``。
"""
sub = f"fonts/{quote(fontstack, safe=',')}/{quote(range_, safe='-')}"
if not sub.endswith(".pbf"):
sub += ".pbf"
return self.service_url(sub)
def glyphs_bytes(self, fontstack: str, range_: str = "0-255") -> bytes:
"""实际下载 PBF 字形二进制。"""
resp = self._request("GET", self.glyphs_url(fontstack, range_), stream=True)
if resp.status_code >= 400:
raise VjmapError(
f"HTTP {resp.status_code} for {resp.url}",
status=resp.status_code, url=resp.url, body=None,
)
return resp.content
# 后端硬编码的 Authorization（Basic base64("_vjmap@_")）
_UPLOAD_IMG_AUTH = "Basic X3ZqbWFwQF8="
# ========================================================================
# 分类 13：AI（条件启用）
# ========================================================================
def ai_enabled(self) -> bool:
"""调 :meth:`chat_faq` 探测 AI 子服务是否启用。
成功（任何非异常响应）视为可用；VjmapError 视为未启用。
"""
try:
self.chat_faq()
return True
except VjmapError:
return False
def vector_search(
self,
query: str,
*,
num_documents: int = 20,
collection: str = "vj",
where: Optional[str] = None,
where_document: Optional[str] = None,
search_description: Optional[str] = None,
disable_split_sentence: bool = False,
disable_search_knowledge: bool = False,
use_get: bool = False,
documents: Optional[Sequence[Mapping[str, Any]]] = None,
) -> Any:
"""``GET/POST /api/v1/service/vectorSearch``。
Parameters
----------
query:
查询文本。
num_documents:
要返回多少条相似文档。
collection:
向量集合名，默认 ``"vj"``。
where / where_document:
metadata 过滤表达式（chromem-go 语法）。
search_description:
附加的搜索描述，让模型更精准。
disable_split_sentence:
关掉"先分句再各自 embed"的默认行为。
disable_search_knowledge:
只搜用户自带 ``documents``，不搜向量库里的知识。
use_get:
True 时走 GET，仅支持简单参数。
documents:
可选：临时挂入的用户文档（不会持久化），每项形如
``{"content":"...","metadata":{...}}``。
"""
body: Dict[str, Any] = {
"query": query,
"numDocuments": int(num_documents),
"collection": collection,
"disableSplitSentence": bool(disable_split_sentence),
"disableSearchKnowledge": bool(disable_search_knowledge),
}
if where is not None: body["where"] = where
if where_document is not None: body["whereDocument"] = where_document
if search_description is not None: body["searchDescription"] = search_description
if documents is not None:
body["documents"] = [dict(d) for d in documents]
if use_get:
params = {
"query": query,
"numDocuments": int(num_documents),
"collection": collection,
}
if where is not None: params["where"] = where
if where_document is not None: params["whereDocument"] = where_document
return self._request("GET", self.service_url("vectorSearch", params))
return self._request(
"POST",
self.service_url("vectorSearch"),
json_body=body,
)
def chat_faq(self) -> Any:
"""``GET /api/v1/service/chatFaq``。返回所有 AI FAQ markdown 元数据。"""
return self._request("GET", self.service_url("chatFaq"))
def vector_docs(self) -> Any:
"""``GET /api/v1/service/vectorDocs``。返回向量库里已有文档的元数据列表。"""
return self._request("GET", self.service_url("vectorDocs"))
def upload_img(
self,
file: Union[str, io.IOBase, Tuple[str, bytes]],
*,
uri: str,
filename: Optional[str] = None,
watermark: bool = False,
authorization: Optional[str] = None,
) -> Any:
"""``POST /api/v1/service/uploadImg`` —— multipart 上传图片到服务器指定目录。
路由在全局 JWT 之后注册：普通 token 照常需要，另加固定 Basic ``Authorization`` 头。
响应 ``{"code":0,"path":<Uri+文件名>,"url":<请求头 ImgHost + 文件名>}``（SDK 不发 ImgHost，url 即文件名）。
Parameters
----------
file:
本地路径 / 已打开的文件对象 / ``(filename, bytes)``。
uri:
服务器端保存目录，**以 ``/`` 结尾**；会拼接 filename 作为最终路径。
filename:
可选指定文件名（默认用 ``file`` 原名）。
watermark:
是否让后端自动叠加水印。
authorization:
自定义 Authorization 头；默认用后端硬编码的
``Basic X3ZqbWFwQF8=``（对应 ``_vjmap@_``）。
"""
close_after = False
if isinstance(file, str):
fp = open(file, "rb")
close_after = True
fname = filename or os.path.basename(file)
body = (fname, fp)
elif hasattr(file, "read"):
fname = filename or getattr(file, "name", "upload.bin")
body = (os.path.basename(fname), file)
elif isinstance(file, tuple) and len(file) == 2:
body = (filename or file[0], file[1])
else:
raise VjmapError("upload_img: unsupported file type")
headers: Dict[str, str] = {
"Authorization": authorization or self._UPLOAD_IMG_AUTH,
"Uri": uri,
}
if watermark:
headers["watermark"] = "1"
try:
return self._request(
"POST",
self.service_url("uploadImg"),
files={"upload": body},
headers=headers,
)
finally:
if close_after:
try:
body[1].close()
except Exception:
pass
