402 lines
17 KiB
Python
402 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
宜搭自定义页面 schema 生成/修改（编排：get-schema → 编译 + 构建 → update-schema）
|
||
|
||
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
||
【调用前必读】references/yida-custom-page-codegen.md
|
||
（JSX 入口签名 / Hooks 限制 / 行内样式 / 跨表单联动 5 种模式 / SEARCH/REPLACE
|
||
增量改写 / 常见坑速查表）。本脚本 --help 仅给出基本用法，**不要**只看
|
||
--help 就直接拼 JSX，几乎必踩坑。
|
||
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
||
用法:
|
||
python yida_custom_page_update.py --app APP_X --form FORM-XXX --code-file page.jsx --yes
|
||
python yida_custom_page_update.py --app APP_X --form FORM-XXX --code 'import ...' --yes
|
||
python yida_custom_page_update.py --app APP_X --form FORM-XXX --show-current
|
||
|
||
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
||
--show-current 模式：
|
||
只拉取现有 schema 并输出当前代码，不做修改。
|
||
|
||
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
||
写入模式：
|
||
全量替换代码。脚本调用纯 Python 编译管线（JSX→createElement 转换 +
|
||
Hooks 兼容层 _customState/didMount），构建标准 Jsx 组件 schema，并保留
|
||
page_id 和已有 dataSource。**零第三方依赖**（仅需 Python 3.7+ 标准库，
|
||
无需 pip install、无需 Node.js）。空页面和已有代码的页面均可使用。
|
||
|
||
跨表单联动场景：JSX 内可通过 Yida.api.form.* 直接读写同应用内任意表单，
|
||
搭配 `yida_form_inspector.py --action fields-snippet` 取目标表的字段常量片段。
|
||
详见 references/yida-custom-page-codegen.md §9。
|
||
|
||
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
_SCRIPT_DIR = Path(__file__).resolve().parent
|
||
if str(_SCRIPT_DIR) not in sys.path:
|
||
sys.path.insert(0, str(_SCRIPT_DIR))
|
||
|
||
from yida_page_compiler import compile_jsx_to_schema # noqa: E402
|
||
from yida_page_schema import extract_source_code # noqa: E402
|
||
from yida_jsx_pipeline import field_check, lint_check # noqa: E402
|
||
|
||
MAX_CODE_FILE_SIZE = 1 * 1024 * 1024
|
||
MAX_INLINE_CODE = 200 * 1024
|
||
|
||
|
||
def _gather_allowed_roots() -> list[Path]:
|
||
"""收集所有允许的路径根目录。任一命中即放行，详见 _resolve_safe_path。
|
||
|
||
优先级（靠前的优先，仅影响报错提示顺序）：
|
||
1. OPENYIDA_ALLOWED_ROOTS：显式多根，以 os.pathsep / ':' / ',' 分隔。
|
||
2. OPENCLAW_WORKSPACE：老环境变量，向后兼容。
|
||
3. 当前工作目录 cwd：兼容原行为。
|
||
4. 临时目录 tempdir：供脚本中转使用。
|
||
"""
|
||
roots: list[Path] = []
|
||
extra = os.environ.get("OPENYIDA_ALLOWED_ROOTS", "")
|
||
if extra:
|
||
seps = [os.pathsep, ":", ","]
|
||
parts: list[str] = [extra]
|
||
for sep in seps:
|
||
parts = [seg for chunk in parts for seg in chunk.split(sep)]
|
||
for part in parts:
|
||
part = part.strip()
|
||
if part:
|
||
roots.append(Path(part).expanduser().resolve())
|
||
legacy = os.environ.get("OPENCLAW_WORKSPACE")
|
||
if legacy:
|
||
roots.append(Path(legacy).expanduser().resolve())
|
||
roots.append(Path.cwd().resolve())
|
||
import tempfile as _tempfile
|
||
roots.append(Path(_tempfile.gettempdir()).resolve())
|
||
roots.append(Path("/tmp").resolve())
|
||
roots.append(Path("/private/tmp").resolve())
|
||
# 去重保序
|
||
seen: set[str] = set()
|
||
uniq: list[Path] = []
|
||
for r in roots:
|
||
s = str(r)
|
||
if s not in seen:
|
||
seen.add(s)
|
||
uniq.append(r)
|
||
return uniq
|
||
|
||
|
||
def _resolve_safe_path(path_str: str) -> Path:
|
||
target = Path(path_str).expanduser()
|
||
target = target.resolve() if target.is_absolute() else (Path.cwd() / target).resolve()
|
||
roots = _gather_allowed_roots()
|
||
for root in roots:
|
||
try:
|
||
target.relative_to(root)
|
||
return target
|
||
except ValueError:
|
||
continue
|
||
listing = "\n - ".join(str(r) for r in roots)
|
||
raise ValueError(
|
||
f"路径超出允许范围：{path_str}\n"
|
||
f"已尝试的允许根目录：\n - {listing}\n"
|
||
f"提示：设置 OPENYIDA_ALLOWED_ROOTS（允许多根，冒号/逗号分隔）或 OPENCLAW_WORKSPACE 扩展允许范围。"
|
||
)
|
||
|
||
|
||
def _run_dws(args: list[str], dry_run: bool = False) -> Any | None:
|
||
cmd = ["dws"] + args
|
||
if dry_run:
|
||
print(f" [dry-run] {' '.join(cmd)}")
|
||
return {"dry_run": True}
|
||
try:
|
||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
|
||
except FileNotFoundError:
|
||
print(" [FAIL] 找不到 'dws' 命令", file=sys.stderr)
|
||
return None
|
||
except subprocess.TimeoutExpired:
|
||
print(" [FAIL] dws 超时", file=sys.stderr)
|
||
return None
|
||
if result.returncode != 0:
|
||
err = result.stderr.strip() or result.stdout.strip()
|
||
print(f" [FAIL] dws 失败 (exit {result.returncode}): {err}", file=sys.stderr)
|
||
return None
|
||
try:
|
||
return json.loads(result.stdout)
|
||
except json.JSONDecodeError as e:
|
||
print(f" [FAIL] 非 JSON: {e}\n 输出: {result.stdout[:300]}", file=sys.stderr)
|
||
return None
|
||
|
||
|
||
def _unwrap_content(data: Any) -> Any:
|
||
"""兼容 dws JSON 输出的常见包裹层。"""
|
||
current = data
|
||
for _ in range(4):
|
||
if not isinstance(current, dict):
|
||
return current
|
||
if "content" in current:
|
||
current = current["content"]
|
||
continue
|
||
if "data" in current and isinstance(current["data"], dict):
|
||
current = current["data"]
|
||
continue
|
||
return current
|
||
return current
|
||
|
||
|
||
def _extract_form_type(info: Any) -> str:
|
||
"""从 get-info 的不同返回形态中提取 formType/type。"""
|
||
candidates: list[Any] = []
|
||
current = info
|
||
for _ in range(4):
|
||
if not isinstance(current, dict):
|
||
break
|
||
candidates.append(current)
|
||
next_obj = None
|
||
for key in ("content", "data", "result"):
|
||
value = current.get(key)
|
||
if isinstance(value, dict):
|
||
next_obj = value
|
||
break
|
||
if next_obj is None:
|
||
break
|
||
current = next_obj
|
||
|
||
for item in candidates:
|
||
value = item.get("formType") or item.get("type") or item.get("pageType")
|
||
if isinstance(value, str) and value.strip():
|
||
return value.strip().lower()
|
||
return ""
|
||
|
||
|
||
def _check_display_target(app: str, form: str, force: bool = False) -> bool:
|
||
"""发布前确认目标是自定义展示页，避免覆盖普通表单/流程表单。"""
|
||
print("Step 0: 校验发布目标")
|
||
info = _run_dws(["yida", "design", "form", "get-info", "--app", app,
|
||
"--form", form, "--format", "json"])
|
||
form_type = _extract_form_type(info)
|
||
if form_type == "display":
|
||
print(" [OK] 目标类型 display")
|
||
return True
|
||
|
||
if force:
|
||
reason = form_type or "unknown"
|
||
print(f" [WARN] 目标类型为 {reason}，已按 --force 跳过保护")
|
||
return True
|
||
|
||
if not info:
|
||
print(" [FAIL] 无法获取目标页面类型，已拒绝写入", file=sys.stderr)
|
||
elif form_type:
|
||
print(f" [FAIL] 目标 formType={form_type}，不是 display 自定义页面，已拒绝写入",
|
||
file=sys.stderr)
|
||
else:
|
||
print(" [FAIL] get-info 返回中未找到 formType，已拒绝写入", file=sys.stderr)
|
||
print(" [HINT] 请确认 --form 是 display 页面；确认无误时可加 --force 显式绕过",
|
||
file=sys.stderr)
|
||
return False
|
||
|
||
|
||
def _load_code(args: argparse.Namespace) -> str:
|
||
if args.code_file:
|
||
safe = _resolve_safe_path(args.code_file)
|
||
if not safe.exists():
|
||
raise ValueError(f"文件不存在: {safe}")
|
||
if safe.stat().st_size > MAX_CODE_FILE_SIZE:
|
||
raise ValueError(f"文件过大 (限制 {MAX_CODE_FILE_SIZE:,} 字节)")
|
||
return safe.read_text(encoding="utf-8")
|
||
elif args.code:
|
||
if len(args.code.encode("utf-8")) > MAX_INLINE_CODE:
|
||
raise ValueError(f"--code 过长 (限制 {MAX_INLINE_CODE:,} 字节)")
|
||
return args.code
|
||
else:
|
||
raise ValueError("必须提供 --code-file 或 --code")
|
||
|
||
|
||
def _extract_existing_data_source(schema: dict) -> dict | None:
|
||
"""从已有 schema 中提取 Page 组件的 dataSource，用于 merge 保留用户自定义数据源。"""
|
||
try:
|
||
return schema["pages"][0]["componentsTree"][0].get("dataSource")
|
||
except (KeyError, IndexError, TypeError):
|
||
return None
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(
|
||
description=(
|
||
"宜搭自定义页面 schema 生成/修改 "
|
||
"【调用前必读】references/yida-custom-page-codegen.md"
|
||
"（JSX 写法 / Hooks 限制 / 跨表单联动 / 常见坑），不要只看 --help 就拼 JSX"
|
||
),
|
||
formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__)
|
||
ap.add_argument("--app", required=True, help="应用编码 appType")
|
||
ap.add_argument("--form", required=True, help="页面 formUuid")
|
||
ap.add_argument("--code-file", help="新代码文件路径")
|
||
ap.add_argument("--code", help="新代码内联字符串")
|
||
ap.add_argument("--show-current", action="store_true", help="只输出当前代码不修改")
|
||
ap.add_argument("--yes", action="store_true", help="确认写入")
|
||
ap.add_argument("--dry-run", action="store_true", help="只编译不写入")
|
||
ap.add_argument("--skip-field-check", action="store_true",
|
||
help="跳过字段 ID 对账预检（不推荐）")
|
||
ap.add_argument("--skip-lint", action="store_true",
|
||
help="跳过 JSX 静态检查（30 条宜搭专属陷阱，不推荐）")
|
||
ap.add_argument("--force", action="store_true",
|
||
help="跳过发布目标 formType=display 保护（仅确认目标无误时使用）")
|
||
args = ap.parse_args()
|
||
|
||
if not args.show_current and not args.code_file and not args.code:
|
||
print("错误: 必须提供 --code-file / --code 或 --show-current", file=sys.stderr)
|
||
return 1
|
||
|
||
if not args.show_current and not args.dry_run:
|
||
if not _check_display_target(args.app, args.form, force=args.force):
|
||
return 1
|
||
|
||
# Step 1: 拉取现有 schema
|
||
print("Step 1: 获取现有 schema")
|
||
resp = _run_dws(["yida", "design", "form", "get-schema", "--app", args.app,
|
||
"--form", args.form, "--format", "json"], dry_run=args.dry_run)
|
||
if args.dry_run and not args.show_current:
|
||
try:
|
||
new_code = _load_code(args)
|
||
except ValueError as e:
|
||
print(f"错误: {e}", file=sys.stderr)
|
||
return 1
|
||
result = compile_jsx_to_schema(new_code, form_uuid=args.form)
|
||
if not result.get("ok"):
|
||
errors = result.get("errors", [])
|
||
err_msgs = "; ".join(e.get("message", "") for e in errors)
|
||
print(f" [FAIL] 编译失败: {err_msgs}", file=sys.stderr)
|
||
lint = result.get("lint", {})
|
||
if lint.get("warnings"):
|
||
for w in lint["warnings"]:
|
||
print(f" [WARN] {w.get('message', w)}", file=sys.stderr)
|
||
return 1
|
||
schema_json = result["schema"]
|
||
print(json.dumps({"ok": True, "dry_run": True, "formUuid": args.form,
|
||
"codeSize": len(new_code),
|
||
"schemaSize": len(schema_json)}, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
if not resp:
|
||
return 1
|
||
schema = resp
|
||
print(" [OK] 拿到 schema")
|
||
|
||
# --show-current 模式
|
||
if args.show_current:
|
||
try:
|
||
current_code = extract_source_code(schema)
|
||
except (ValueError, TypeError):
|
||
current_code = None
|
||
if current_code is None:
|
||
print(" [WARN] schema 中没有可提取的自定义页面代码")
|
||
print(json.dumps({"ok": False, "error": "not_a_custom_page"}, ensure_ascii=False))
|
||
return 1
|
||
print(json.dumps({"ok": True, "formUuid": args.form,
|
||
"codeSize": len(current_code),
|
||
"currentCode": current_code}, ensure_ascii=False, indent=2))
|
||
return 0
|
||
|
||
# 获取 page_id
|
||
page_id = args.form
|
||
pages = schema.get("pages", [])
|
||
if pages:
|
||
page_id = pages[0].get("id", args.form) or args.form
|
||
|
||
# 提取已有 dataSource（用于 merge）
|
||
existing_ds = _extract_existing_data_source(schema)
|
||
|
||
# Step 2: 加载新代码、编译并构建 schema
|
||
try:
|
||
new_code = _load_code(args)
|
||
except ValueError as e:
|
||
print(f"错误: {e}", file=sys.stderr)
|
||
return 1
|
||
|
||
if not new_code.strip():
|
||
print("错误: 代码不能为空", file=sys.stderr)
|
||
return 1
|
||
|
||
try:
|
||
current_code = extract_source_code(schema)
|
||
except (ValueError, TypeError):
|
||
current_code = None
|
||
previous_size = len(current_code) if current_code else 0
|
||
|
||
print(f"Step 2: 编译 + 构建 schema (新代码 {len(new_code):,} 字节)")
|
||
|
||
result = compile_jsx_to_schema(new_code, form_uuid=page_id, existing_data_source=existing_ds)
|
||
if not result.get("ok"):
|
||
errors = result.get("errors", [])
|
||
err_msgs = "; ".join(e.get("message", "") for e in errors)
|
||
print(f" [FAIL] 编译失败: {err_msgs}", file=sys.stderr)
|
||
lint = result.get("lint", {})
|
||
if lint.get("warnings"):
|
||
for w in lint["warnings"]:
|
||
print(f" [WARN] {w.get('message', w)}", file=sys.stderr)
|
||
return 1
|
||
|
||
schema_json = result["schema"]
|
||
lint = result.get("lint", {})
|
||
if lint.get("warnings"):
|
||
for w in lint["warnings"]:
|
||
print(f" [WARN] lint: {w.get('message', w)}")
|
||
|
||
print(f" [OK] 编译成功, schema 大小: {len(schema_json):,} 字节")
|
||
|
||
# Step 2.5: 字段 ID 对账预检（避免发布后运行时才报 fieldId 不存在）
|
||
if not args.skip_field_check:
|
||
print("Step 2.5: 字段 ID 对账")
|
||
chk = field_check(new_code, args.app)
|
||
for w in chk.get("warnings", []):
|
||
print(f" [WARN] {w.get('message', w)}")
|
||
if not chk.get("ok"):
|
||
print(" [FAIL] 字段对账未通过，为避免发布后页面报错，拒绝写入：", file=sys.stderr)
|
||
for e in chk.get("errors", []):
|
||
print(f" - {e.get('message', e)}", file=sys.stderr)
|
||
print(" [HINT] 修复后重试；确认需要忽略可加 --skip-field-check（不推荐）", file=sys.stderr)
|
||
return 1
|
||
info = chk.get("info", {})
|
||
if info.get("skipped"):
|
||
print(f" [OK] 跳过（{info['skipped']}）")
|
||
else:
|
||
print(f" [OK] 已校验 {info.get('referencedFieldCount', 0)} 个字段引用，"
|
||
f"覆盖 {len(info.get('checkedForms', []))} 张表单")
|
||
|
||
# Step 2.7: JSX 静态检查（避免发布后运行时才报错）
|
||
if not args.skip_lint:
|
||
print("Step 2.7: JSX 静态检查")
|
||
lr = lint_check(new_code, filename=args.code_file or "page.jsx")
|
||
for w in lr.get("warnings", []):
|
||
print(f" [WARN] L{w['line']} [{w['rule']}] {w['message']}")
|
||
if not lr.get("ok"):
|
||
print(" [FAIL] JSX 静态检查未通过，为避免发布后页面报错，拒绝写入：", file=sys.stderr)
|
||
for e in lr.get("errors", []):
|
||
print(f" L{e['line']} [{e['rule']}] {e['message']}", file=sys.stderr)
|
||
print(" [HINT] 修复后重试；确认需要忽略可加 --skip-lint（不推荐）", file=sys.stderr)
|
||
print(" [HINT] 或在 JSX 中加 // dws-lint-disable-line [rule] 关闭单行检查", file=sys.stderr)
|
||
return 1
|
||
info = lr.get("info", {})
|
||
print(f" [OK] 检查通过（错误 {info.get('errorCount', 0)} / 警告 {info.get('warningCount', 0)}）")
|
||
|
||
# Step 3: 写回
|
||
print("Step 3: 写入 schema")
|
||
resp = _run_dws(["yida", "design", "form", "update-schema", "--app", args.app,
|
||
"--form", args.form, "--form-type", "display",
|
||
"--content", schema_json, "--yes", "--format", "json"])
|
||
if not resp:
|
||
return 1
|
||
print(" [OK] 写入成功")
|
||
print(json.dumps({"ok": True, "formUuid": args.form, "codeSize": len(new_code),
|
||
"previousCodeSize": previous_size}, ensure_ascii=False))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|