实现pagetype类自动继承Page类，测试成功

apps/jingrow/jingrow/__init__.py

@@ -7,7 +7,7 @@ import inspect
 import logging
 import os
 from contextvars import ContextVar
-from jingrow.model.page import Page
+from jingrow.model.page import Page, get_page_instance
 from jingrow.config import Config
 from jingrow.utils.jingrow_api import upload_file_to_jingrow, get_record_count
 
@@ -36,7 +36,7 @@ for _lvl in list(_ANSI.keys()):
 
 def get_pg(pagetype: str, name: str):
     """获取单条记录并转为可属性访问的对象，失败返回 None。"""
-    pg = Page(pagetype)
+    pg = get_page_instance(pagetype)
     res = pg.get(name)
     if not isinstance(res, dict) or not res.get('success'):
         return None
@@ -46,7 +46,7 @@ def get_pg(pagetype: str, name: str):
 
 def create_pg(pagetype: str, data: Dict[str, Any]):
     """创建记录，返回创建后的数据对象或 None。"""
-    pg = Page(pagetype)
+    pg = get_page_instance(pagetype)
     res = pg.create(data)
     if not isinstance(res, dict) or not res.get('success'):
         return None
@@ -56,7 +56,7 @@ def create_pg(pagetype: str, data: Dict[str, Any]):
 
 def update_pg(pagetype: str, name: str, data: Dict[str, Any]):
     """更新记录，成功返回更新后的数据对象或 True，失败返回 False。"""
-    pg = Page(pagetype)
+    pg = get_page_instance(pagetype)
     res = pg.update(name, data)
     if not isinstance(res, dict) or not res.get('success'):
         return False
@@ -101,14 +101,14 @@ def get_hook_source() -> Optional[str]:
 
 
 def delete_pg(pagetype: str, name: str) -> bool:
-    pg = Page(pagetype)
+    pg = get_page_instance(pagetype)
     res = pg.delete(name)
     return bool(isinstance(res, dict) and res.get('success'))
 
 
 def get_list(pagetype: str, filters: List[List[Any]] = None, fields: List[str] = None, limit: int = None):
     """获取记录列表，返回对象列表；失败返回空列表。"""
-    pg = Page(pagetype)
+    pg = get_page_instance(pagetype)
     res = pg.list(filters=filters, fields=fields, limit=limit)
     if not isinstance(res, dict) or not res.get('success'):
         return []
@@ -135,7 +135,7 @@ def get_single(pagetype: str):
 
 def get_module_app(pagetype: str):
     """获取指定 pagetype 的模块应用信息，返回后端适配器的原始结果结构。"""
-    return Page(pagetype).get_module_app()
+    return get_page_instance(pagetype).get_module_app()
 
 
 def get_pg_id(
@@ -146,22 +146,22 @@ def get_pg_id(
     site: Optional[str] = None,
 ):
     """根据过滤条件或字段值获取文档 ID，返回后端适配器的原始结果结构。"""
-    return Page(pagetype).get_pg_id(filters=filters, field=field, value=value, site=site)
+    return get_page_instance(pagetype).get_pg_id(filters=filters, field=field, value=value, site=site)
 
 
 def get_meta(pagetype: str):
     """获取 pagetype 的元数据，返回后端适配器的原始结果结构。"""
-    return Page(pagetype).get_meta()
+    return get_page_instance(pagetype).get_meta()
 
 
 def get_field_mapping_from_jingrow(pagetype: str):
     """获取字段的label到fieldname的映射，返回 {label: fieldname} 的映射字典。"""
-    return Page(pagetype).get_field_mapping_from_jingrow()
+    return get_page_instance(pagetype).get_field_mapping_from_jingrow()
 
 
 def get_field_value_from_jingrow(pagetype: str, name: str, fieldname: str):
     """从Jingrow获取字段的当前值，返回字段的当前值，如果为空则返回None。"""
-    return Page(pagetype).get_field_value_from_jingrow(name, fieldname)
+    return get_page_instance(pagetype).get_field_value_from_jingrow(name, fieldname)
 
 
 def upload_file(file_data: bytes, filename: str, attached_to_pagetype: Optional[str] = None, attached_to_name: Optional[str] = None, attached_to_field: Optional[str] = None):

apps/jingrow/jingrow/ai/pagetype/local_ai_node/local_ai_node.py

@@ -0,0 +1,123 @@
+# Copyright (c) 2025, jingrow and contributors
+# For license information, please see license.txt
+
+from typing import Dict, Any
+from pathlib import Path
+import json
+import jingrow
+from jingrow.model.page import Page
+from jingrow.utils.path import get_jingrow_root
+from jingrow.utils.fs import atomic_write_json
+
+
+class LocalAiNode(Page):
+	"""
+	Local Ai Node 页面类型
+	在保存时自动同步节点数据到 JSON 文件
+	"""
+	
+	def create(self, data: Dict[str, Any]) -> Dict[str, Any]:
+		"""创建节点记录，并同步到 JSON 文件"""
+		result = super().create(data)
+		if result.get('success'):
+			created_data = result.get('data', {})
+			if created_data.get('node_type'):
+				_sync_node_to_json_file(created_data)
+		return result
+	
+	def update(self, name: str, data: Dict[str, Any]) -> Dict[str, Any]:
+		"""更新节点记录，并同步到 JSON 文件"""
+		result = super().update(name, data)
+		if result.get('success'):
+			# 获取更新后的完整数据
+			updated_res = self.get(name)
+			if updated_res.get('success'):
+				updated_data = updated_res.get('data', {})
+				if updated_data.get('node_type'):
+					_sync_node_to_json_file(updated_data)
+		return result
+
+
+def _sync_node_to_json_file(node_data: Dict[str, Any]) -> bool:
+	"""
+	将节点数据同步到 JSON 文件（LocalAiNode 类的辅助函数）
+	当保存 Local Ai Node 时，自动更新对应的 JSON 文件
+	
+	Args:
+		node_data: 节点数据字典，包含 node_type, node_label, node_icon 等字段
+		
+	Returns:
+		bool: 同步是否成功
+	"""
+	try:
+		node_type = node_data.get('node_type')
+		if not node_type:
+			return False
+		
+		# 获取节点目录路径
+		jingrow_root = get_jingrow_root()
+		nodes_root = jingrow_root / "ai" / "nodes"
+		node_dir = nodes_root / node_type
+		json_file = node_dir / f"{node_type}.json"
+		
+		# 确保目录存在
+		node_dir.mkdir(parents=True, exist_ok=True)
+		
+		# 解析 node_schema（可能是字典或 JSON 字符串）
+		node_schema = node_data.get('node_schema') or {}
+		if isinstance(node_schema, str):
+			try:
+				node_schema = json.loads(node_schema)
+			except (json.JSONDecodeError, TypeError):
+				node_schema = {}
+		if not isinstance(node_schema, dict):
+			node_schema = {}
+		
+		# 确保 node_schema 中不包含 metadata（如果存在则移除）
+		node_schema = {k: v for k, v in node_schema.items() if k != 'metadata'}
+		
+		# 构建完整的 JSON 文件内容
+		json_content = {
+			"metadata": {
+				"type": node_type,
+				"label": node_data.get('node_label') or node_type,
+				"icon": node_data.get('node_icon') or "fa-cube",
+				"color": node_data.get('node_color') or "#6b7280",
+				"description": node_data.get('node_description') or "",
+				"group": node_data.get('node_group') or "其他",
+				"component_type": node_data.get('node_component') or "GenericNode"
+			}
+		}
+		
+		# 合并 node_schema 的所有字段（properties, required, _layout 等）
+		# 确保包含完整的 schema 配置
+		json_content.update(node_schema)
+		
+		# 确保必需的字段存在（即使 node_schema 中没有）
+		json_content.setdefault("properties", {})
+		json_content.setdefault("required", [])
+		json_content.setdefault("_layout", {
+			"tabs": [{
+				"id": "tab_1",
+				"label": "基本设置",
+				"sections": [{
+					"id": "section_1",
+					"label": "",
+					"columns": [{
+						"id": "column_1",
+						"label": "",
+						"fields": []
+					}]
+				}]
+			}],
+			"activeTab": "tab_1"
+		})
+		
+		# 使用原子写入确保数据完整性
+		atomic_write_json(json_file, json_content)
+		
+		return True
+	except Exception as e:
+		# 静默失败，不影响主流程
+		jingrow.log_error("同步节点到 JSON 文件失败", str(e), exc=e)
+		return False

apps/jingrow/jingrow/model/page.py

@@ -7,6 +7,39 @@ from jingrow.core.hooks import execute_hook, execute_hook_async
 import asyncio
 
 
+def get_page_instance(pagetype: str) -> 'Page':
+    """
+    获取 Page 实例，支持使用自定义 Page 子类
+    
+    自动发现并加载 pagetype 对应的 Page 子类：
+    - 命名规则：pagetype 转换为类名，如 "Local Ai Node" -> "LocalAiNode"
+    - 模块路径：jingrow.ai.pagetype.{pagetype_snake_case}.{pagetype_snake_case}
+    - 如果找不到子类，回退到默认 Page 类
+    
+    Args:
+        pagetype: 页面类型名称
+        
+    Returns:
+        Page 实例（可能是自定义子类或默认 Page 类）
+    """
+    # 转换命名： "Local Ai Node" -> "LocalAiNode"
+    class_name = pagetype.replace(" ", "").replace("_", "")
+    
+    # 尝试从 pagetype 目录导入
+    try:
+        pagetype_snake = pagetype.lower().replace(' ', '_')
+        module_path = f"jingrow.ai.pagetype.{pagetype_snake}.{pagetype_snake}"
+        module = __import__(module_path, fromlist=[class_name])
+        page_class = getattr(module, class_name, None)
+        if page_class and issubclass(page_class, Page):
+            return page_class(pagetype)
+    except (ImportError, AttributeError, TypeError):
+        pass
+    
+    # 如果找不到子类，使用默认 Page
+    return Page(pagetype)
+
+
 class Page:
     """基于适配器的通用 Page 模型，提供常用 CRUD 与钩子机制。
     钩子事件命名与 API 对齐：after_insert, on_update, on_trash 等。