jingrowlocal/jingrow-framework
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构create_record.py和update_record.py

Browse Source
This commit is contained in:
jingrow 2025-10-31 23:13:55 +08:00
parent ede503b863
commit ca0ca0b0b8
3 changed files with 16 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
apps/jingrow/jingrow/__init__.py
View File

@ -7,7 +7,6 @@ import inspect
import logging
import os
from jingrow.model.page import Page
from jingrow.services.runtime import get_adapter
from jingrow.config import Config
_local = {}
@ -113,10 +112,9 @@ def get_field_mapping(pagetype: str) -> Dict[str, str]:
return Page(pagetype).get_field_mapping_from_jingrow()
def map_fields(field_map: List[Dict[str, Any]], ai_outputs: Dict[str, Any], label_to_fieldname: Dict[str, str]) -> Dict[str, Any]:
def map_fields(pagetype: str, field_map: List[Dict[str, Any]], ai_outputs: Dict[str, Any], label_to_fieldname: Dict[str, str]) -> Dict[str, Any]:
"""依据字段映射与标签->字段名映射，将 AI 输出转为文档字段字典。"""
adapter = get_adapter()
return adapter.map_fields_by_labels(field_map, ai_outputs, label_to_fieldname)
return Page(pagetype).map_fields_by_labels(field_map, ai_outputs, label_to_fieldname)
def get_field_value(pagetype: str, name: str, fieldname: str) -> Optional[Any]:

20
apps/jingrow/jingrow/ai/nodes/create_record/create_record.py
View File

@ -1,11 +1,7 @@
import json
from typing import Dict, Any, Optional
from jingrow.utils.jingrow_api import (
get_field_mapping_from_jingrow,
create_record,
map_fields_by_labels,
)
import jingrow
def execute(context=None, inputs=None, config=None):
"""
@ -38,7 +34,7 @@ def execute(context=None, inputs=None, config=None):
field_map = config.get("field_map", [])
# 从 Jingrow 获取字段 label->fieldname 映射
label2field = get_field_mapping_from_jingrow(record_type) or {}
label2field = jingrow.get_field_mapping(record_type) or {}
# 收集AI输出：只平铺同一分支的所有上游节点数据
ai_outputs = {}
@ -82,7 +78,7 @@ def execute(context=None, inputs=None, config=None):
# 1. 优先用field_map（from→to）
if field_map:
record_data.update(map_fields_by_labels(field_map, ai_outputs, label2field))
record_data.update(jingrow.map_fields(field_map, ai_outputs, label2field))
mapped_fields.update([m.get("from") for m in field_map if m.get("from")])
# 同时标记目标字段为已映射，防止被自动映射覆盖
for mapping in field_map:
@ -100,14 +96,12 @@ def execute(context=None, inputs=None, config=None):
record_data[fieldname] = value
mapped_fields.add(label)
# 调用 Jingrow 创建记录
api_res = create_record(record_type, record_data)
if not api_res.get("success"):
return {"success": False, "error": api_res.get("error", "创建记录失败")}
# 调用 Jingrow 创建记录（使用 jingrow.create_pg 返回创建后的数据或 None）
created = jingrow.create_pg(record_type, record_data) or {}
if not created:
return {"success": False, "error": "创建记录失败"}
created = api_res.get("data", {})
created_name = created.get("name") or record_data.get("name")
created_id = created.get("name") or created.get("id") # 兼容不同返回
return {
"success": True,

18
apps/jingrow/jingrow/ai/nodes/update_record/update_record.py
View File

@ -1,11 +1,7 @@
import json
from typing import Dict, Any, Optional
from jingrow.utils.jingrow_api import (
get_field_mapping_from_jingrow,
get_field_value_from_jingrow,
update_record,
)
import jingrow
def execute(context=None, inputs=None, config=None):
"""
@ -80,7 +76,7 @@ def execute(context=None, inputs=None, config=None):
}
# 获取 label->fieldname 映射
label_to_fieldname = get_field_mapping_from_jingrow(record_type) or {}
label_to_fieldname = jingrow.get_field_mapping(record_type) or {}
fieldname_set = set(label_to_fieldname.values())
updated = False
@ -139,7 +135,7 @@ def execute(context=None, inputs=None, config=None):
# 在追加模式下，先检查是否重复
if write_mode == "append":
# 从服务端查询当前值，避免并发覆盖
current_value = get_field_value_from_jingrow(record_type, record_name, to_field) or ""
current_value = jingrow.get_field_value(record_type, record_name, to_field) or ""
current_str = str(current_value).strip()
new_value_str = str(value).strip()
@ -164,10 +160,10 @@ def execute(context=None, inputs=None, config=None):
# 调用 Jingrow 更新记录（仅当有更新）
if updated:
api_res = update_record(record_type, record_name, mock_record)
if not api_res.get("success"):
return {"success": False, "error": api_res.get("error", "更新记录失败")}
server_record = api_res.get("data", {})
upd_res = jingrow.update_pg(record_type, record_name, mock_record)
if not upd_res:
return {"success": False, "error": "更新记录失败"}
server_record = upd_res if isinstance(upd_res, dict) else {}
else:
server_record = {}