Files
playbook/tools/tsl-codegen/scripts/generate.py
T

1042 lines
36 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""Generate compliant TSL codegen markdown from a YAML/JSON entry file.
The recording format is one leaf page: a `module` title, a relative `path`, and
an ordered `declarations` list containing function, class, or unit entries. This
script renders it to the markdown the codegen tree stores, matching
tools/tsl-codegen/STANDARD.md.
Rendered Markdown is passed through the repository-pinned Prettier before it is
written, keeping generated pages consistent with the existing codegen tree.
Input dispatch is by extension: .json parses with the stdlib (keeping the
toolchain dependency-free); .yml/.yaml needs pyyaml. If pyyaml is missing the
script says so and points at the JSON path.
Function declaration fields:
kind required `function`
name required must match the signature name case-insensitively
signature required verbatim, underscores/case untouched
desc required description; may contain multiple lines
tags optional list of Chinese keywords -> `<!-- tags: ... -->`
params required when the signature takes args; omit for nullary
returns required return type
example optional fenced tsl block, pasted verbatim
Each param: name/type/desc required; optional (bool) -> `可选。` prefix;
values (list of {value, desc}) -> a `**name 取值**` enum section.
Usage (run from repo root):
python tools/tsl-codegen/scripts/generate.py --file entry.yml
python tools/tsl-codegen/scripts/generate.py --file entry.json \
--scope my-project
python tools/tsl-codegen/scripts/generate.py --dir recordings
"""
import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path, PureWindowsPath
REPO_ROOT = Path(__file__).resolve().parents[3]
PRETTIER_CONFIG = REPO_ROOT / ".prettierrc.json"
FORMAT_SUFFIXES = {
None: (".json", ".yaml", ".yml"),
"json": (".json",),
"yaml": (".yaml", ".yml"),
}
class ChineseArgumentParser(argparse.ArgumentParser):
def error(self, message):
replacements = (
("unrecognized arguments:", "无法识别的参数:"),
("the following arguments are required:", "缺少必需参数:"),
("expected one argument", "需要一个参数值"),
("invalid choice: ", "取值无效:"),
)
if message.startswith("argument "):
message = "参数 " + message[len("argument ") :]
for source, target in replacements:
message = message.replace(source, target)
choice_marker = " (choose from "
if choice_marker in message and message.endswith(")"):
message = message[:-1].replace(choice_marker, "(可选值:", 1) + ""
self.print_usage(sys.stderr)
self.exit(2, f"{self.prog}: 错误:{message}\n")
class GenerationError(Exception):
pass
def die(msg):
raise GenerationError(msg)
def scope_name(value):
"""Validate a user-defined single directory name."""
if not value or value in {".", ".."} or "/" in value or "\\" in value:
raise argparse.ArgumentTypeError("scope 必须是单级目录名")
return value
def resolve_format(path, fmt):
"""Pick the parser: explicit --format wins, else derive from extension."""
if fmt:
return fmt
suffix = path.suffix.lower()
if suffix == ".json":
return "json"
if suffix in (".yml", ".yaml"):
return "yaml"
die(
f"无法根据扩展名 {suffix or '(无扩展名)'} 判断录入格式;"
"请使用 --format json 或 --format yaml"
)
def describe_json_error(exc):
translations = (
(
"Expecting property name enclosed in double quotes",
"对象属性名必须使用双引号",
),
("Expecting value", "此处缺少有效值"),
("Expecting ',' delimiter", "此处缺少逗号分隔符"),
("Extra data", "根值之后存在多余内容"),
("Unterminated string", "字符串未结束"),
("Invalid \\escape", "字符串中包含无效转义"),
("Invalid control character", "字符串中包含无效控制字符"),
)
for prefix, message in translations:
if exc.msg.startswith(prefix):
return message
return "语法无效"
def load_entries(path, fmt=None):
"""Parse a recording file as JSON (stdlib) or YAML (pyyaml).
Format is chosen by --format when given, else by file extension.
"""
try:
text = path.read_text(encoding="utf-8")
except UnicodeDecodeError:
die("文件不是有效的 UTF-8 编码")
except OSError as exc:
die(f"读取文件失败:{exc}")
fmt = resolve_format(path, fmt)
if fmt == "json":
try:
return json.loads(text)
except json.JSONDecodeError as exc:
die(
f"JSON 格式错误:第 {exc.lineno} 行,第 {exc.colno} 列:"
f"{describe_json_error(exc)}"
)
if fmt == "yaml":
try:
import yaml
except ImportError:
die(
"未安装 pyyaml;请运行 `python -m pip install pyyaml`"
"或改用无需额外依赖的 JSON 录入文件"
)
try:
return yaml.safe_load(text)
except yaml.YAMLError as exc:
mark = getattr(exc, "problem_mark", None)
if mark is None:
die("YAML 格式错误:语法无效")
die(
f"YAML 格式错误:第 {mark.line + 1} 行,第 {mark.column + 1} 列:"
"语法无效"
)
die(f"不支持的录入格式:{fmt};只能使用 json 或 yaml")
def escape_cell(text):
"""Escape `|` so a value stays inside one markdown table cell."""
return str(text).replace("|", "\\|")
def require(cond, msg):
if not cond:
die(msg)
def require_mapping(value, where):
require(isinstance(value, dict), f"{where}:必须是对象")
def reject_unknown(mapping, allowed, where):
unknown = sorted(set(mapping) - set(allowed))
require(not unknown, f"{where}:存在未知字段:{', '.join(unknown)}")
def non_empty_string(value, where):
require(isinstance(value, str) and value.strip(), f"{where}:不能为空")
def optional_draft_string(value, where):
if value == "":
return
non_empty_string(value, where)
def validate_tags(tags, where):
if tags is None:
return
require(isinstance(tags, list) and tags, f"{where}tags 必须是非空列表")
for index, tag in enumerate(tags):
non_empty_string(tag, f"{where}: tags[{index}]")
def signature_names(signature, where):
non_empty_string(signature, f"{where}: signature")
left = signature.find("(")
right = signature.rfind(")")
require(left > 0 and right == len(signature) - 1, f"{where}signature 格式无效")
name = signature[:left]
non_empty_string(name, f"{where}: signature name")
raw = signature[left + 1 : right].strip()
if not raw:
return name, []
names = [item.strip() for item in raw.split(",")]
require(all(names), f"{where}signature 中存在空参数名")
require(
len({item.casefold() for item in names}) == len(names),
f"{where}signature 中存在重复参数名",
)
return name, names
def validate_values(values, where):
require(isinstance(values, list) and values, f"{where}values 必须是非空列表")
for index, item in enumerate(values):
item_where = f"{where}[{index}]"
require_mapping(item, item_where)
reject_unknown(item, {"value", "desc"}, item_where)
require("value" in item, f"{item_where}:缺少 value")
non_empty_string(item.get("desc"), f"{item_where}: desc")
def validate_params(params, expected_names, where):
if not expected_names:
require(not params, f"{where}:无参数 signature 不能包含 params")
return
require(isinstance(params, list), f"{where}params 必须是列表")
require(len(params) == len(expected_names), f"{where}params 与 signature 不匹配")
actual_names = []
for index, param in enumerate(params):
param_where = f"{where}: params[{index}]"
require_mapping(param, param_where)
reject_unknown(
param, {"name", "type", "desc", "optional", "values"}, param_where
)
name = param.get("name")
non_empty_string(name, f"{param_where}: name")
non_empty_string(param.get("type"), f"{param_where}: type")
non_empty_string(param.get("desc"), f"{param_where}: desc")
if "optional" in param:
require(
isinstance(param["optional"], bool),
f"{param_where}optional 必须是布尔值",
)
if "values" in param:
validate_values(param["values"], f"{param_where}: values")
actual_names.append(name)
require(
[name.casefold() for name in actual_names]
== [name.casefold() for name in expected_names],
f"{where}params 必须与 signature 中的参数顺序一致",
)
def validate_examples(examples, where):
require(
isinstance(examples, list) and examples, f"{where}examples 必须是非空列表"
)
for index, example in enumerate(examples):
example_where = f"{where}: examples[{index}]"
require_mapping(example, example_where)
reject_unknown(example, {"desc", "code", "output"}, example_where)
non_empty_string(example.get("desc"), f"{example_where}: desc")
non_empty_string(example.get("code"), f"{example_where}: code")
if "output" in example:
non_empty_string(example["output"], f"{example_where}: output")
def validate_function(fn, where, *, returns_required, extra_fields=()):
require_mapping(fn, where)
allowed = {
"signature",
"desc",
"tags",
"params",
"returns",
"example",
"examples",
*extra_fields,
}
reject_unknown(fn, allowed, where)
name, names = signature_names(fn.get("signature"), where)
non_empty_string(fn.get("desc"), f"{where}: desc")
validate_tags(fn.get("tags"), where)
validate_params(fn.get("params"), names, where)
if returns_required:
non_empty_string(fn.get("returns"), f"{where}:缺少 returns")
elif "returns" in fn:
optional_draft_string(fn["returns"], f"{where}: returns")
require(
not ("example" in fn and "examples" in fn),
f"{where}example 和 examples 不能同时存在",
)
if "example" in fn:
non_empty_string(fn["example"], f"{where}: example")
if "examples" in fn:
validate_examples(fn["examples"], where)
return name
def validate_class_member(member, where):
require_mapping(member, where)
kind = member.get("kind")
require(
kind in {"method", "property", "field", "constant"},
f"{where}:未知 kind{kind}",
)
non_empty_string(member.get("name"), f"{where}: name")
visibility = member.get("visibility")
require(
visibility in {"public", "protected"},
f"{where}visibility 只能是 public 或 protected",
)
non_empty_string(member.get("desc"), f"{where}: desc")
validate_tags(member.get("tags"), where)
if kind == "method":
reject_unknown(
member,
{
"kind",
"name",
"visibility",
"binding",
"signature",
"desc",
"tags",
"params",
"returns",
"modifiers",
"example",
"examples",
},
where,
)
require(
member.get("binding") in {"instance", "class"},
f"{where}binding 无效",
)
parsed_name = validate_function(
member,
where,
returns_required=False,
extra_fields={"kind", "name", "visibility", "binding", "modifiers"},
)
require(
parsed_name.casefold() == member["name"].casefold(),
f"{where}name 与 signature 中的名称不一致",
)
if "modifiers" in member:
modifiers = member["modifiers"]
require(isinstance(modifiers, list), f"{where}modifiers 必须是列表")
allowed = {"overload", "virtual", "override"}
require(
all(item in allowed for item in modifiers),
f"{where}:存在无效 modifier",
)
require(
len(set(modifiers)) == len(modifiers), f"{where}:存在重复 modifier"
)
return
common = {"kind", "name", "visibility", "desc", "tags"}
if kind == "property":
reject_unknown(member, common | {"type", "params", "access"}, where)
if "type" in member:
optional_draft_string(member["type"], f"{where}: type")
require(
member.get("access") in {"read", "write", "readwrite"},
f"{where}access 无效",
)
params = member.get("params")
if params:
expected = [param.get("name") for param in params]
validate_params(params, expected, where)
return
if kind == "field":
reject_unknown(member, common | {"type", "static"}, where)
non_empty_string(member.get("type"), f"{where}: type")
if "static" in member:
require(isinstance(member["static"], bool), f"{where}static 必须是布尔值")
return
reject_unknown(member, common | {"type", "value", "static"}, where)
require("value" in member and member["value"] is not None, f"{where}:缺少 value")
if isinstance(member["value"], str):
non_empty_string(member["value"], f"{where}: value")
if "type" in member:
optional_draft_string(member["type"], f"{where}: type")
if "static" in member:
require(isinstance(member["static"], bool), f"{where}static 必须是布尔值")
def validate_class(cls, where):
require_mapping(cls, where)
reject_unknown(cls, {"kind", "name", "desc", "tags", "bases", "members"}, where)
require(cls.get("kind") == "class", f"{where}kind 必须是 class")
non_empty_string(cls.get("name"), f"{where}: name")
non_empty_string(cls.get("desc"), f"{where}: desc")
validate_tags(cls.get("tags"), where)
if "bases" in cls:
require(isinstance(cls["bases"], list), f"{where}bases 必须是列表")
for index, base in enumerate(cls["bases"]):
non_empty_string(base, f"{where}: bases[{index}]")
require(isinstance(cls.get("members"), list), f"{where}members 必须是列表")
for index, member in enumerate(cls["members"]):
validate_class_member(member, f"{where}: members[{index}]")
def validate_unit_member(member, where):
require_mapping(member, where)
kind = member.get("kind")
require(
kind in {"function", "variable", "constant", "class"},
f"{where}:未知 kind{kind}",
)
if kind == "class":
validate_class(member, where)
return
non_empty_string(member.get("name"), f"{where}: name")
non_empty_string(member.get("desc"), f"{where}: desc")
validate_tags(member.get("tags"), where)
if kind == "function":
parsed_name = validate_function(
member,
where,
returns_required=True,
extra_fields={"kind", "name"},
)
require(
parsed_name.casefold() == member["name"].casefold(),
f"{where}name 与 signature 中的名称不一致",
)
return
common = {"kind", "name", "desc", "tags", "type"}
if kind == "variable":
reject_unknown(member, common, where)
non_empty_string(member.get("type"), f"{where}: type")
return
reject_unknown(member, common | {"value"}, where)
require("value" in member and member["value"] is not None, f"{where}:缺少 value")
if isinstance(member["value"], str):
non_empty_string(member["value"], f"{where}: value")
if "type" in member:
optional_draft_string(member["type"], f"{where}: type")
def validate_unit(unit, where):
require_mapping(unit, where)
reject_unknown(unit, {"kind", "name", "desc", "tags", "members"}, where)
require(unit.get("kind") == "unit", f"{where}kind 必须是 unit")
non_empty_string(unit.get("name"), f"{where}: name")
non_empty_string(unit.get("desc"), f"{where}: desc")
validate_tags(unit.get("tags"), where)
require(
isinstance(unit.get("members"), list),
f"{where}members 必须是列表",
)
for index, member in enumerate(unit["members"]):
validate_unit_member(member, f"{where}: members[{index}]")
def validate_top_level_function(declaration, where):
require(
declaration.get("kind") == "function",
f"{where}kind 必须是 function",
)
non_empty_string(declaration.get("name"), f"{where}: name")
parsed_name = validate_function(
declaration,
where,
returns_required=True,
extra_fields={"kind", "name"},
)
require(
parsed_name.casefold() == declaration["name"].casefold(),
f"{where}name 与 signature 中的名称不一致",
)
def validate_declaration(declaration, where):
require_mapping(declaration, where)
kind = declaration.get("kind")
require(
kind in {"function", "class", "unit"},
f"{where}:未知 kind{kind}",
)
if kind == "function":
validate_top_level_function(declaration, where)
elif kind == "class":
validate_class(declaration, where)
else:
validate_unit(declaration, where)
def validate_declaration_uniqueness(declarations):
function_signatures = set()
class_names = set()
unit_names = set()
for index, declaration in enumerate(declarations):
where = f"declarations[{index}]"
kind = declaration["kind"]
if kind == "function":
key = declaration["signature"].casefold()
require(
key not in function_signatures,
f"{where}function signature 重复",
)
function_signatures.add(key)
continue
names = class_names if kind == "class" else unit_names
key = declaration["name"].casefold()
require(key not in names, f"{where}{kind} 名称重复")
names.add(key)
def validate_page(data):
require_mapping(data, "录入根")
reject_unknown(data, {"module", "path", "declarations"}, "录入根")
non_empty_string(data.get("module"), "录入根:module")
non_empty_string(data.get("path"), "录入根:path")
declarations = data.get("declarations")
require(
isinstance(declarations, list) and declarations,
"录入根:declarations 必须是非空列表",
)
for index, declaration in enumerate(declarations):
validate_declaration(declaration, f"declarations[{index}]")
validate_declaration_uniqueness(declarations)
return declarations
def param_desc(param, where):
"""Description column text: prepend `可选。` for optional params."""
desc = param.get("desc")
require(desc, f"{where}:参数 {param.get('name', '?')} 缺少 desc")
if param.get("optional") and not desc.startswith("可选。"):
return "可选。" + desc
return desc
def render_param_table(params, where):
"""Three-column 参数/类型/说明 table with single-space padding."""
lines = ["| 参数 | 类型 | 说明 |", "| --- | --- | --- |"]
for param in params:
name = param.get("name")
ptype = param.get("type")
require(name, f"{where}:存在缺少 name 的参数")
require(ptype, f"{where}:参数 {name} 缺少 type")
desc = param_desc(param, where)
lines.append(
f"| `{escape_cell(name)}` | {escape_cell(ptype)} | {escape_cell(desc)} |"
)
return lines
def render_enum_sections(params, heading_level=None):
"""Render value sections, preserving legacy function-page headings."""
lines = []
for param in params:
values = param.get("values")
if not values:
continue
if heading_level is None:
lines.append(f"**{param['name']} 取值**")
else:
lines.append(f"{'#' * heading_level} `{param['name']}` 取值")
lines.append("")
for item in values:
lines.append(f"- `{item['value']}` — {item['desc']}")
lines.append("")
return lines
def render_examples(fn, heading_level):
lines = []
if "examples" in fn:
lines.extend([f"{'#' * heading_level} 示例", ""])
for index, example in enumerate(fn["examples"], start=1):
lines.append(f"范例{index:02d}{example['desc']}")
lines.append("")
lines.append("```tsl")
lines.extend(example["code"].rstrip("\n").split("\n"))
output = example.get("output")
if output is not None:
output_lines = output.split("\n")
if len(output_lines) == 1:
lines.append(f"// 输出:{output_lines[0]}")
else:
lines.append("// 输出:")
lines.extend(
f"// {line}" if line else "//" for line in output_lines
)
lines.append("```")
lines.append("")
while lines and not lines[-1]:
lines.pop()
return lines
if fn.get("example"):
return [
f"{'#' * heading_level} 示例",
"",
"```tsl",
*fn["example"].rstrip("\n").split("\n"),
"```",
]
return []
def render_intro(heading, declaration, item):
lines = [heading, "", f"声明:{declaration}", "", item["desc"], ""]
tags = item.get("tags")
if tags:
lines.extend([f"<!-- tags: {' '.join(tags)} -->", ""])
return lines
def render_callable(
fn,
signature,
declaration,
where,
*,
level,
returns_required,
show_visibility,
):
validate_function(
fn,
where,
returns_required=returns_required,
extra_fields={"kind", "name", "visibility", "binding", "modifiers"},
)
lines = render_intro(f"{'#' * level} `{signature}`", declaration, fn)
if show_visibility:
lines.extend([f"可见性:`{fn['visibility']}`", ""])
if fn.get("modifiers"):
rendered = "、".join(f"`{item}`" for item in fn["modifiers"])
lines.extend([f"修饰符:{rendered}", ""])
params = fn.get("params") or []
if params:
lines.extend(render_param_table(params, where))
lines.append("")
lines.extend(render_enum_sections(params, level + 1))
if fn.get("returns"):
lines.append(f"返回:{fn['returns']}")
example_lines = render_examples(fn, level + 1)
if example_lines:
lines.extend(["", *example_lines])
return lines
def render_top_level_function(fn, index):
"""Render one function entry to a list of lines (no trailing blank)."""
where = f"declarations[{index}]"
sig = fn["signature"]
lines = render_intro(f"## `{sig}`", "function", fn)
params = fn.get("params") or []
if params:
lines.extend(render_param_table(params, f"{where} ({sig})"))
lines.append("")
lines.extend(render_enum_sections(params))
lines.append(f"返回:{fn['returns']}")
examples = render_examples(fn, 3)
if examples:
lines.extend(["", *examples])
return lines
def render_class_member(member, level, where):
kind = member["kind"]
if kind == "method":
declaration = "class function" if member["binding"] == "class" else "function"
return render_callable(
member,
member["signature"],
declaration,
where,
level=level,
returns_required=False,
show_visibility=True,
)
static_prefix = "static " if member.get("static") else ""
declaration = {
"property": "property",
"field": f"{static_prefix}field",
"constant": f"{static_prefix}const",
}[kind]
signature = member["name"]
if kind == "property" and member.get("params"):
signature += "(" + ", ".join(param["name"] for param in member["params"]) + ")"
lines = render_intro(f"{'#' * level} `{signature}`", declaration, member)
lines.extend([f"可见性:`{member['visibility']}`", ""])
if kind == "property":
if member.get("type"):
lines.extend([f"类型:{member['type']}", ""])
access = {"read": "read", "write": "write", "readwrite": "read / write"}[
member["access"]
]
lines.append(f"访问:{access}")
params = member.get("params") or []
if params:
lines.extend(["", *render_param_table(params, where), ""])
lines.extend(render_enum_sections(params, level + 1))
elif kind == "field":
lines.append(f"类型:{member['type']}")
else:
if member.get("type"):
lines.extend([f"类型:{member['type']}", ""])
lines.append(f"值:`{render_scalar(member['value'])}`")
return lines
def render_class(cls, level, where, *, page_root):
lines = render_intro(f"{'#' * level} `{cls['name']}`", "class", cls)
if cls.get("bases"):
lines.extend(["父类:" + "、".join(f"`{base}`" for base in cls["bases"]), ""])
for index, member in enumerate(cls["members"]):
lines.extend(
render_class_member(member, level + 1, f"{where}: members[{index}]")
)
lines.append("")
while lines and not lines[-1]:
lines.pop()
return lines
def render_unit_member(member, index, unit_where):
where = f"{unit_where}: members[{index}]"
kind = member["kind"]
if kind == "function":
return render_callable(
member,
member["signature"],
"function",
where,
level=3,
returns_required=True,
show_visibility=False,
)
if kind == "class":
return render_class(member, 3, where, page_root=False)
declaration = "var" if kind == "variable" else "const"
lines = render_intro(f"### `{member['name']}`", declaration, member)
if kind == "variable":
lines.append(f"类型:{member['type']}")
else:
if member.get("type"):
lines.extend([f"类型:{member['type']}", ""])
lines.append(f"值:`{render_scalar(member['value'])}`")
return lines
def render_unit(unit, where):
lines = render_intro(f"## `{unit['name']}`", "unit", unit)
for index, member in enumerate(unit["members"]):
lines.extend(render_unit_member(member, index, where))
lines.append("")
while lines and not lines[-1]:
lines.pop()
return lines
def render_scalar(value):
if isinstance(value, str):
return value
return json.dumps(value, ensure_ascii=False)
def render_declaration(declaration, index):
kind = declaration["kind"]
where = f"declarations[{index}]"
if kind == "function":
return render_top_level_function(declaration, index)
if kind == "class":
return render_class(declaration, 2, where, page_root=True)
return render_unit(declaration, where)
def render_page(data):
"""Validate and render a whole leaf page."""
declarations = validate_page(data)
out = [f"# {data['module']}", ""]
for index, declaration in enumerate(declarations):
out.extend(render_declaration(declaration, index))
out.append("")
return "\n".join(out).rstrip("\n") + "\n"
def format_markdown(text):
"""Format generated Markdown with the repository-pinned Prettier."""
npx = shutil.which("npx")
if not npx:
die("未找到 Prettier;请先在仓库根目录运行 `npm install`")
result = subprocess.run(
[
npx,
"--no-install",
"prettier",
"--config",
str(PRETTIER_CONFIG),
"--parser",
"markdown",
],
input=text,
capture_output=True,
text=True,
encoding="utf-8",
cwd=REPO_ROOT,
check=False,
)
if result.returncode != 0:
detail = result.stderr.strip() or "未知错误"
die(f"Prettier 格式化失败:{detail}")
return result.stdout
def output_path(data, scope):
"""Build the leaf-page destination from the recording file's relative path."""
relative = data.get("path")
require(relative, "录入数据缺少 path")
require(isinstance(relative, str), "path 必须是字符串")
recorded_path = PureWindowsPath(relative)
require(
not recorded_path.drive.startswith("\\\\"),
"path 必须是相对路径,不允许 UNC 路径",
)
require(
not recorded_path.drive,
"path 必须是相对路径,不允许 Windows 盘符路径",
)
require(
not recorded_path.root,
"path 必须是相对路径,不能以斜杠或反斜杠开头",
)
require(
bool(recorded_path.parts),
"path 必须指向具体文档,不能只表示当前目录",
)
require(
".." not in recorded_path.parts,
"path 不能包含 '..',不允许跳转到父目录",
)
require(
recorded_path.suffix == "",
"path 不能包含 .md 等文件扩展名",
)
relative_path = Path(*recorded_path.parts)
return (
Path("skills/tsl-api-reference/references/codegen")
/ scope
/ relative_path.with_suffix(".md")
)
def atomic_write(path, text):
"""Atomically replace path and remove the temporary file on failure."""
path.parent.mkdir(parents=True, exist_ok=True)
descriptor, temporary_name = tempfile.mkstemp(
prefix=f".{path.name}.", suffix=".tmp", dir=path.parent
)
temporary_path = Path(temporary_name)
try:
with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as handle:
handle.write(text)
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary_path, path)
finally:
if temporary_path.exists():
temporary_path.unlink()
def gather_directory_inputs(directory, fmt):
try:
if not directory.is_dir():
die(f"输入目录不存在或不是目录:{directory}")
suffixes = FORMAT_SUFFIXES[fmt]
inputs = sorted(
(
path
for path in directory.iterdir()
if path.is_file() and path.suffix.lower() in suffixes
),
key=lambda path: (path.name.casefold(), path.name),
)
except OSError as exc:
die(f"读取输入目录失败:{directory}{exc}")
if not inputs:
displayed_suffixes = "、".join(suffixes)
die(
f"目录 {directory} 中未找到符合条件的直属录入文件"
f"{displayed_suffixes}"
)
return inputs
def prepare_input(in_path, input_format, scope):
data = load_entries(in_path, input_format)
try:
rendered = render_page(data)
except GenerationError as exc:
raise GenerationError(f"录入数据校验失败:{exc}") from exc
try:
out_path = output_path(data, scope)
except GenerationError as exc:
raise GenerationError(f"输出路径无效:{exc}") from exc
try:
formatted = format_markdown(rendered)
except OSError as exc:
raise GenerationError(f"运行 Prettier 失败:{exc}") from exc
return in_path, out_path, formatted
def report_input_error(in_path, error):
print(f"错误:{in_path}{error}", file=sys.stderr)
def find_output_collisions(prepared):
owners = {}
collisions = []
for in_path, out_path, _ in prepared:
resolved_output = out_path.resolve()
first_input = owners.get(resolved_output)
if first_input is None:
owners[resolved_output] = in_path
continue
collisions.append(
f"输出目标冲突:{first_input}{in_path} 都会写入 {resolved_output}"
)
return collisions
def main(argv=None):
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
parser = ChineseArgumentParser(
description="从 YAML/JSON 录入文件生成 TSL API 文档",
usage=(
"%(prog)s [--help] [--scope SCOPE] [--format {json,yaml}] "
"(--file INPUT_FILE | --dir INPUT_DIR | INPUT_FILE)"
),
add_help=False,
allow_abbrev=False,
)
parser.add_argument(
"--help",
action="help",
help="显示本帮助并退出(不提供 -h 短选项)",
)
parser.add_argument(
"legacy_input",
nargs="?",
metavar="INPUT_FILE",
help="已废弃,请使用 --file;暂时兼容 YAML/JSON 录入文件路径",
)
parser.add_argument(
"--file",
dest="input_file",
metavar="INPUT_FILE",
help="要生成的单个 YAML/JSON 录入文件",
)
parser.add_argument(
"--dir",
dest="input_dir",
metavar="INPUT_DIR",
help="批量生成目录中的直属 YAML/JSON 录入文件",
)
parser.add_argument(
"--scope",
type=scope_name,
default="project",
help="codegen 下的一级目录,可自定义(默认:project)",
)
parser.add_argument(
"--format",
choices=["json", "yaml"],
help="录入文件格式;默认根据文件扩展名判断",
)
args = parser.parse_args(argv)
input_modes = (args.legacy_input, args.input_file, args.input_dir)
if sum(value is not None for value in input_modes) != 1:
parser.error("必须且只能指定一种输入方式:INPUT_FILE、--file 或 --dir")
is_batch = args.input_dir is not None
try:
if is_batch:
input_paths = gather_directory_inputs(Path(args.input_dir), args.format)
input_format = None
else:
in_path = Path(args.input_file or args.legacy_input)
if not in_path.is_file():
die(f"输入文件不存在或不是普通文件:{in_path}")
input_paths = [in_path]
input_format = args.format
except GenerationError as exc:
print(f"错误:{exc}", file=sys.stderr)
return 1
prepared = []
input_errors = []
for in_path in input_paths:
try:
prepared.append(prepare_input(in_path, input_format, args.scope))
except GenerationError as exc:
input_errors.append((in_path, exc))
collisions = find_output_collisions(prepared)
if input_errors or collisions:
for in_path, error in input_errors:
report_input_error(in_path, error)
for collision in collisions:
print(f"错误:{collision}", file=sys.stderr)
if is_batch:
issue_count = len(input_errors) + len(collisions)
print(
f"错误:批量生成已中止:发现 {issue_count} 个问题,"
f"共检查 {len(input_paths)} 个文件;未写入任何 Markdown 文件",
file=sys.stderr,
)
return 1
for in_path, out_path, text in prepared:
try:
atomic_write(out_path, text)
except OSError as exc:
report_input_error(in_path, f"写入 Markdown 失败:{out_path}{exc}")
return 1
print(f"已生成:{in_path} -> {out_path}", file=sys.stderr)
return 0
if __name__ == "__main__":
raise SystemExit(main())