chore: sync from OpenClaw workspace (2026-04-05 09:17)

examples/flask_skill_update_or_create.py

@@ -0,0 +1,84 @@
+# 将此路由合并进你的 skill 蓝图。
+# CI 不再写入 skill_type / monthly_price / yearly_price，避免每次发布覆盖后台手工配置。
+#
+# 要求：SkillModel.update_or_create 在「更新」时对 data 中未出现的列应保留数据库原值；
+# 若当前是整行覆盖，请在 Model 层改为按字段合并或白名单更新。
+# 若表对这三列 NOT NULL 且无默认值，仅在「首次插入」时在 Model 内写死默认即可。
+
+import json
+import re
+from datetime import datetime
+
+from flask import jsonify, request
+
+# from your_app import skill_bp, SkillModel  # 按你项目实际导入
+
+
+@skill_bp.route("/api/skill/update", methods=["POST"])
+def update_or_create_skill():
+    """CI/CD 自动化注册接口（可选携带 readme_md = SKILL.md 正文 Markdown）"""
+    try:
+        data = request.get_json(silent=True) or {}
+        if not data:
+            return jsonify({"code": 400, "msg": "请求体为空", "data": None}), 400
+
+        openclaw_meta = data.get("metadata", {}).get("openclaw", {})
+
+        slug = (
+            (data.get("slug") or "").strip()
+            or (openclaw_meta.get("slug") or "").strip()
+            or (data.get("name") or "").strip()
+        )
+        name = (data.get("name") or slug).strip()
+        version = str(data.get("version") or "").strip()
+        category = (openclaw_meta.get("category") or "").strip()
+
+        if not all([slug, name, version, category]):
+            return jsonify(
+                {
+                    "code": 400,
+                    "msg": "元数据不完整（需包含 slug/name/version/category）",
+                    "data": None,
+                }
+            ), 400
+
+        slug = slug.lower()
+
+        if not re.match(r"^[a-z0-9-]+$", slug):
+            return jsonify(
+                {
+                    "code": 400,
+                    "msg": "slug 格式非法，仅允许小写字母、数字和中划线",
+                    "data": None,
+                }
+            ), 400
+
+        skill_data = {
+            "slug": slug,
+            "name": name,
+            "description": data.get("description"),
+            "version": version,
+            "category": category,
+            "developer_name": data.get("author", "匠厂开发者"),
+            "tags": json.dumps(data.get("tags", []), ensure_ascii=False),
+            "status": 2,
+            "updated_at": datetime.now(),
+        }
+
+        if "readme_md" in data:
+            rm = data.get("readme_md")
+            skill_data["readme_md"] = "" if rm is None else str(rm)
+
+        success = SkillModel.update_or_create(slug=slug, data=skill_data)
+        if success:
+            return jsonify(
+                {
+                    "code": 200,
+                    "msg": "注册成功",
+                    "data": {"slug": slug, "name": name, "version": version},
+                }
+            ), 200
+        return jsonify({"code": 500, "msg": "数据持久化失败", "data": None}), 500
+
+    except Exception as e:
+        return jsonify({"code": 500, "msg": str(e), "data": None}), 500