162 lines
5.6 KiB
Python
162 lines
5.6 KiB
Python
from __future__ import annotations
|
||
|
||
# JSON 校验直接走标准库。
|
||
import json
|
||
# XML 校验直接走标准库。
|
||
import xml.etree.ElementTree as element_tree
|
||
from typing import Any
|
||
|
||
|
||
def normalized_config_type(config_type: str, data_id: str = "") -> str:
|
||
# 统一把类型名压平，避免前后端大小写不一致。
|
||
lowered = (config_type or "").strip().lower()
|
||
if lowered in {"yml", "yaml"}:
|
||
return "yaml"
|
||
if lowered in {"json", "properties", "xml", "html", "text"}:
|
||
return lowered
|
||
|
||
# 没显式类型时，按常见后缀兜底。
|
||
suffix = data_id.strip().lower()
|
||
if suffix.endswith((".yml", ".yaml")):
|
||
return "yaml"
|
||
if suffix.endswith(".json"):
|
||
return "json"
|
||
if suffix.endswith(".properties"):
|
||
return "properties"
|
||
if suffix.endswith(".xml"):
|
||
return "xml"
|
||
if suffix.endswith(".html"):
|
||
return "html"
|
||
return "text"
|
||
|
||
|
||
def dependency_error(message: str) -> dict[str, Any]:
|
||
# 缺少解析依赖时也返回统一结构，前端方便直接展示。
|
||
return {
|
||
"ok": False,
|
||
"strict": False,
|
||
"type": "",
|
||
"message": message,
|
||
}
|
||
|
||
|
||
def validation_success(config_type: str, message: str, *, strict: bool = True, details: dict[str, Any] | None = None) -> dict[str, Any]:
|
||
# 成功结果也统一带上 strict 标记，告诉前端是不是强校验。
|
||
return {
|
||
"ok": True,
|
||
"strict": strict,
|
||
"type": config_type,
|
||
"message": message,
|
||
"details": details or {},
|
||
}
|
||
|
||
|
||
def validation_error(config_type: str, message: str, *, details: dict[str, Any] | None = None) -> dict[str, Any]:
|
||
# 失败结构保持和成功结构一致。
|
||
return {
|
||
"ok": False,
|
||
"strict": True,
|
||
"type": config_type,
|
||
"message": message,
|
||
"details": details or {},
|
||
}
|
||
|
||
|
||
def validate_yaml(content: str) -> dict[str, Any]:
|
||
try:
|
||
# 按需导入 PyYAML，避免依赖缺失时影响其他功能启动。
|
||
import yaml
|
||
except Exception as exc: # noqa: BLE001
|
||
return dependency_error(f"yaml validator unavailable: {exc}")
|
||
|
||
try:
|
||
# 空 YAML 允许解析成 null。
|
||
parsed = yaml.safe_load(content)
|
||
except yaml.YAMLError as exc:
|
||
# PyYAML 异常里通常会带问题行列。
|
||
problem_mark = getattr(exc, "problem_mark", None)
|
||
details = {}
|
||
if problem_mark is not None:
|
||
details = {"line": int(problem_mark.line) + 1, "column": int(problem_mark.column) + 1}
|
||
return validation_error("yaml", str(exc), details=details)
|
||
|
||
parsed_kind = type(parsed).__name__ if parsed is not None else "null"
|
||
return validation_success("yaml", f"yaml format ok ({parsed_kind})", details={"parsedKind": parsed_kind})
|
||
|
||
|
||
def validate_json_text(content: str) -> dict[str, Any]:
|
||
try:
|
||
# JSON 必须是合法文本，空串直接视为错误。
|
||
parsed = json.loads(content)
|
||
except json.JSONDecodeError as exc:
|
||
return validation_error(
|
||
"json",
|
||
exc.msg,
|
||
details={"line": int(exc.lineno), "column": int(exc.colno)},
|
||
)
|
||
|
||
parsed_kind = type(parsed).__name__
|
||
return validation_success("json", f"json format ok ({parsed_kind})", details={"parsedKind": parsed_kind})
|
||
|
||
|
||
def validate_properties_text(content: str) -> dict[str, Any]:
|
||
try:
|
||
# Java properties 校验直接走专用解析器，避免手写规则误判。
|
||
import javaproperties
|
||
except Exception as exc: # noqa: BLE001
|
||
return dependency_error(f"properties validator unavailable: {exc}")
|
||
|
||
try:
|
||
# loads 会在格式异常时抛出 ParseError。
|
||
parsed = javaproperties.loads(content)
|
||
except Exception as exc: # noqa: BLE001
|
||
line = getattr(exc, "lineno", None)
|
||
details = {"line": int(line)} if isinstance(line, int) else {}
|
||
return validation_error("properties", str(exc), details=details)
|
||
|
||
return validation_success("properties", f"properties format ok ({len(parsed)} keys)", details={"keysCount": len(parsed)})
|
||
|
||
|
||
def validate_xml_text(content: str) -> dict[str, Any]:
|
||
try:
|
||
# XML 用标准库直接做语法解析。
|
||
root = element_tree.fromstring(content)
|
||
except element_tree.ParseError as exc:
|
||
line, column = getattr(exc, "position", (0, 0))
|
||
return validation_error(
|
||
"xml",
|
||
str(exc),
|
||
details={"line": int(line), "column": int(column) + 1},
|
||
)
|
||
|
||
return validation_success("xml", f"xml format ok (<{root.tag}>)", details={"rootTag": root.tag})
|
||
|
||
|
||
def validate_text_like(config_type: str) -> dict[str, Any]:
|
||
# text/html 先不做严格语法校验，只给出明确提示。
|
||
return validation_success(
|
||
config_type,
|
||
f"{config_type} saved as plain text; strict syntax validation is not enabled",
|
||
strict=False,
|
||
)
|
||
|
||
|
||
def validate_config_content(content: str, *, config_type: str = "", data_id: str = "") -> dict[str, Any]:
|
||
# 统一解析类型。
|
||
normalized_type = normalized_config_type(config_type, data_id)
|
||
|
||
# 按类型路由到对应校验器。
|
||
if normalized_type == "yaml":
|
||
return validate_yaml(content)
|
||
if normalized_type == "json":
|
||
return validate_json_text(content)
|
||
if normalized_type == "properties":
|
||
return validate_properties_text(content)
|
||
if normalized_type == "xml":
|
||
return validate_xml_text(content)
|
||
if normalized_type in {"html", "text"}:
|
||
return validate_text_like(normalized_type)
|
||
|
||
# 理论上不会走到这里，这里只是兜底。
|
||
return validation_success(normalized_type, "no validator configured", strict=False)
|