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

使用jingrow统一crud函数重构apps/jingrow/jingrow/ai/pagetype/local_ai_agent/__init__.py

Browse Source
This commit is contained in:
jingrow 2025-10-31 19:50:29 +08:00
parent 5018c4bead
commit 475a6dfab3
1 changed files with 9 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
apps/jingrow/jingrow/ai/pagetype/local_ai_agent/__init__.py
View File

@ -3,14 +3,13 @@
import jingrow import jingrow
import logging import logging
from jingrow.utils.jingrow_api import get_record_list, get_pagetype_module_app, get_record
from jingrow.utils.jinja import render_template from jingrow.utils.jinja import render_template
def _get_all_local_ai_agents(): def _get_all_local_ai_agents():
"""获取所有启用的事件触发 Local Ai Agent，支持按PageType和Module分组缓存返回。 """获取所有启用的事件触发 Local Ai Agent，支持按PageType和Module分组缓存返回。
""" """
res = get_record_list( data = jingrow.get_list(
"Local Ai Agent", "Local Ai Agent",
filters=[["enabled", "=", 1], ["trigger_mode", "=", "Event Trigger"]], filters=[["enabled", "=", 1], ["trigger_mode", "=", "Event Trigger"]],
fields=[ fields=[
@ -23,7 +22,6 @@ def _get_all_local_ai_agents():
"condition", "condition",
], ],
) )
data = res.get("data", []) if res.get("success") else []
agents: dict[str, list[dict]] = {} agents: dict[str, list[dict]] = {}
for agent in data: for agent in data:
@ -69,8 +67,8 @@ def run_agent(pg=None, method=None, event=None, page=None, **kwargs):
# 获取 pagetype 所属 module（通过 API 获取 module/app） # 获取 pagetype 所属 module（通过 API 获取 module/app）
module_name = None module_name = None
try: try:
ma = get_pagetype_module_app(pg_pagetype) ma = jingrow.get_module_app(pg_pagetype)
if ma.get("success"): if isinstance(ma, dict) and ma.get("success"):
module_name = ma.get("module") module_name = ma.get("module")
except Exception: except Exception:
module_name = None module_name = None
@ -120,9 +118,9 @@ def run_agent(pg=None, method=None, event=None, page=None, **kwargs):
_nm = base_ctx.get("name") or _get_pg_field(pg, "name") _nm = base_ctx.get("name") or _get_pg_field(pg, "name")
pg_ctx = base_ctx or {} pg_ctx = base_ctx or {}
if _pt and _nm: if _pt and _nm:
api_res = get_record(str(_pt), str(_nm)) api_res = jingrow.get_pg(str(_pt), str(_nm))
if api_res and api_res.get("success") and isinstance(api_res.get("data"), dict): if isinstance(api_res, dict):
pg_ctx = api_res.get("data") pg_ctx = api_res
result = render_template(cond_tpl, {"pg": pg_ctx}) result = render_template(cond_tpl, {"pg": pg_ctx})
if str(result).strip().lower() in ("true", "1", "yes"): if str(result).strip().lower() in ("true", "1", "yes"):
trigger = True trigger = True
@ -148,9 +146,9 @@ def enqueue_local_ai_agent(pg, agent_name):
此函数会读取本地智能体配置，调用本地任务创建函数以执行远端/本地智能体。 此函数会读取本地智能体配置，调用本地任务创建函数以执行远端/本地智能体。
""" """
try: try:
# 通过 API 获取 Local Ai Agent 详情 # 通过统一 CRUD 获取 Local Ai Agent 详情
res = get_record_list("Local Ai Agent", filters=[["name", "=", agent_name]], fields=["name", "agent_name"], limit=1) items = jingrow.get_list("Local Ai Agent", filters=[["name", "=", agent_name]], fields=["name", "agent_name"], limit=1)
agent = (res.get("data") or [None])[0] if res.get("success") else None agent = (items or [None])[0]
if not agent: if not agent:
return return