deliver: 交付收口（引擎代为提交）
This commit is contained in:
agent.develop
parent
7ca7bfdb72
commit
a3e8ec8996
@ -2,143 +2,37 @@
|
||||
"summary": [
|
||||
{
|
||||
"name": "pbl_artifact",
|
||||
"comment": "学生产出物（版本内联 version_no）",
|
||||
"module": "pbl_evidence",
|
||||
"engine": "mariadb",
|
||||
"charset": "utf8mb4",
|
||||
"tenant_scoped": true
|
||||
"title": "学生产出物表",
|
||||
"primary": ["id"],
|
||||
"catelog": "entity"
|
||||
}
|
||||
],
|
||||
"fields": [
|
||||
{
|
||||
"name": "tenant_id",
|
||||
"comment": "租户ID(强制打头)",
|
||||
"null": false,
|
||||
"type": "str",
|
||||
"length": 32
|
||||
},
|
||||
{
|
||||
"name": "id",
|
||||
"comment": "主键",
|
||||
"null": false,
|
||||
"type": "int",
|
||||
"length": 20,
|
||||
"unsigned": true,
|
||||
"auto_increment": true
|
||||
},
|
||||
{
|
||||
"name": "artifact_uid",
|
||||
"comment": "产出UID",
|
||||
"null": false,
|
||||
"type": "str",
|
||||
"length": 32
|
||||
},
|
||||
{
|
||||
"name": "session_id",
|
||||
"comment": "会话",
|
||||
"null": false,
|
||||
"type": "bigint"
|
||||
},
|
||||
{
|
||||
"name": "blueprint_id",
|
||||
"comment": "蓝图",
|
||||
"null": false,
|
||||
"type": "bigint"
|
||||
},
|
||||
{
|
||||
"name": "team_id",
|
||||
"comment": "团队",
|
||||
"null": false,
|
||||
"type": "bigint"
|
||||
},
|
||||
{
|
||||
"name": "creator_id",
|
||||
"comment": "作者",
|
||||
"null": false,
|
||||
"type": "str",
|
||||
"length": 32
|
||||
},
|
||||
{
|
||||
"name": "title",
|
||||
"comment": "标题",
|
||||
"null": false,
|
||||
"type": "str",
|
||||
"length": 128
|
||||
},
|
||||
{
|
||||
"name": "artifact_type",
|
||||
"comment": "类型",
|
||||
"null": false,
|
||||
"type": "str",
|
||||
"length": 64
|
||||
},
|
||||
{
|
||||
"name": "content_json",
|
||||
"comment": "内容",
|
||||
"null": true,
|
||||
"type": "json"
|
||||
},
|
||||
{
|
||||
"name": "version_no",
|
||||
"comment": "版本",
|
||||
"null": false,
|
||||
"type": "int"
|
||||
},
|
||||
{
|
||||
"name": "status",
|
||||
"comment": "状态",
|
||||
"null": false,
|
||||
"type": "str",
|
||||
"length": 64
|
||||
},
|
||||
{
|
||||
"name": "submitted_at",
|
||||
"comment": "提交时间",
|
||||
"null": true,
|
||||
"type": "datetime"
|
||||
},
|
||||
{
|
||||
"name": "created_at",
|
||||
"comment": "创建时间",
|
||||
"null": false,
|
||||
"type": "datetime"
|
||||
},
|
||||
{
|
||||
"name": "updated_at",
|
||||
"comment": "更新时间",
|
||||
"null": false,
|
||||
"type": "datetime"
|
||||
}
|
||||
{"name": "id", "title": "主键", "type": "str", "length": 32, "nullable": "no"},
|
||||
{"name": "tenant_id", "title": "租户ID(强制打头)", "type": "str", "length": 32, "nullable": "no", "default": "0"},
|
||||
{"name": "artifact_uid", "title": "产出UID", "type": "str", "length": 32, "nullable": "no"},
|
||||
{"name": "session_id", "title": "会话ID", "type": "str", "length": 32, "nullable": "no", "default": "0"},
|
||||
{"name": "blueprint_id", "title": "蓝图ID", "type": "str", "length": 32, "nullable": "no", "default": "0"},
|
||||
{"name": "team_id", "title": "团队ID", "type": "str", "length": 32, "nullable": "no", "default": "0"},
|
||||
{"name": "creator_id", "title": "作者", "type": "str", "length": 32, "nullable": "no", "default": "0"},
|
||||
{"name": "title", "title": "标题", "type": "str", "length": 128, "nullable": "no", "default": ""},
|
||||
{"name": "artifact_type", "title": "产出物类型", "type": "str", "length": 64, "nullable": "no", "default": "other"},
|
||||
{"name": "content_json", "title": "内容(序列化JSON文本)", "type": "text", "nullable": "yes"},
|
||||
{"name": "version_no", "title": "版本号(内联)", "type": "int", "nullable": "no", "default": "1"},
|
||||
{"name": "status", "title": "状态", "type": "str", "length": 64, "nullable": "no", "default": "draft"},
|
||||
{"name": "submitted_at", "title": "提交时间", "type": "datetime", "nullable": "yes"},
|
||||
{"name": "created_at", "title": "创建时间(应用层写入)", "type": "datetime", "nullable": "no"},
|
||||
{"name": "updated_at", "title": "更新时间(应用层写入)", "type": "datetime", "nullable": "no"}
|
||||
],
|
||||
"indexes": [
|
||||
{
|
||||
"name": "uk_art_uid",
|
||||
"unique": true,
|
||||
"fields": [
|
||||
"tenant_id",
|
||||
"artifact_uid"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "idx_art_session",
|
||||
"unique": false,
|
||||
"fields": [
|
||||
"tenant_id",
|
||||
"session_id"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "PRIMARY",
|
||||
"unique": true,
|
||||
"fields": [
|
||||
"id"
|
||||
]
|
||||
}
|
||||
{"name": "uk_art_uid", "idxtype": "unique", "idxfields": ["tenant_id", "artifact_uid"]},
|
||||
{"name": "idx_art_session", "idxtype": "index", "idxfields": ["tenant_id", "session_id"]},
|
||||
{"name": "idx_art_blueprint", "idxtype": "index", "idxfields": ["tenant_id", "blueprint_id"]},
|
||||
{"name": "idx_art_creator", "idxtype": "index", "idxfields": ["tenant_id", "creator_id"]}
|
||||
],
|
||||
"codes": [
|
||||
{
|
||||
"field": "artifact_type",
|
||||
"appcode": "pbl_artifact_type"
|
||||
}
|
||||
{"field": "artifact_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='pbl_artifact_type'"},
|
||||
{"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='pbl_artifact_status'"},
|
||||
{"field": "blueprint_id", "table": "pbl_blueprint", "valuefield": "id", "textfield": "title"}
|
||||
]
|
||||
}
|
||||
|
||||
@ -2,131 +2,37 @@
|
||||
"summary": [
|
||||
{
|
||||
"name": "pbl_evidence",
|
||||
"comment": "学习证据（幂等采集，唯一索引防重）",
|
||||
"module": "pbl_evidence",
|
||||
"engine": "mariadb",
|
||||
"charset": "utf8mb4",
|
||||
"tenant_scoped": true
|
||||
"title": "学习证据表",
|
||||
"primary": ["id"],
|
||||
"catelog": "entity"
|
||||
}
|
||||
],
|
||||
"fields": [
|
||||
{
|
||||
"name": "tenant_id",
|
||||
"comment": "租户ID(强制打头)",
|
||||
"null": false,
|
||||
"type": "str",
|
||||
"length": 32
|
||||
},
|
||||
{
|
||||
"name": "id",
|
||||
"comment": "主键",
|
||||
"null": false,
|
||||
"type": "int",
|
||||
"length": 20,
|
||||
"unsigned": true,
|
||||
"auto_increment": true
|
||||
},
|
||||
{
|
||||
"name": "artifact_id",
|
||||
"comment": "产出物",
|
||||
"null": false,
|
||||
"type": "bigint"
|
||||
},
|
||||
{
|
||||
"name": "evidence_type",
|
||||
"comment": "证据类型",
|
||||
"null": false,
|
||||
"type": "str",
|
||||
"length": 64
|
||||
},
|
||||
{
|
||||
"name": "source_event_id",
|
||||
"comment": "来源 runtime_event",
|
||||
"null": false,
|
||||
"type": "str",
|
||||
"length": 32
|
||||
},
|
||||
{
|
||||
"name": "session_id",
|
||||
"comment": "会话",
|
||||
"null": false,
|
||||
"type": "bigint"
|
||||
},
|
||||
{
|
||||
"name": "learner_id",
|
||||
"comment": "学习者",
|
||||
"null": false,
|
||||
"type": "str",
|
||||
"length": 32
|
||||
},
|
||||
{
|
||||
"name": "blueprint_id",
|
||||
"comment": "蓝图",
|
||||
"null": false,
|
||||
"type": "bigint"
|
||||
},
|
||||
{
|
||||
"name": "payload_json",
|
||||
"comment": "证据体",
|
||||
"null": true,
|
||||
"type": "json"
|
||||
},
|
||||
{
|
||||
"name": "occurred_at",
|
||||
"comment": "发生时间",
|
||||
"null": false,
|
||||
"type": "datetime"
|
||||
},
|
||||
{
|
||||
"name": "dedup_key",
|
||||
"comment": "去重键",
|
||||
"null": false,
|
||||
"type": "str",
|
||||
"length": 32
|
||||
},
|
||||
{
|
||||
"name": "created_at",
|
||||
"comment": "创建时间",
|
||||
"null": false,
|
||||
"type": "datetime"
|
||||
},
|
||||
{
|
||||
"name": "updated_at",
|
||||
"comment": "更新时间",
|
||||
"null": false,
|
||||
"type": "datetime"
|
||||
}
|
||||
{"name": "id", "title": "主键", "type": "str", "length": 32, "nullable": "no"},
|
||||
{"name": "tenant_id", "title": "租户ID(强制打头)", "type": "str", "length": 32, "nullable": "no", "default": "0"},
|
||||
{"name": "artifact_id", "title": "归属产出物ID(0=无强归属)", "type": "str", "length": 32, "nullable": "no", "default": "0"},
|
||||
{"name": "evidence_type", "title": "证据类型", "type": "str", "length": 64, "nullable": "no", "default": "observation"},
|
||||
{"name": "source_event_id", "title": "来源runtime_event事件ID", "type": "str", "length": 32, "nullable": "no"},
|
||||
{"name": "session_id", "title": "会话ID", "type": "str", "length": 32, "nullable": "no", "default": "0"},
|
||||
{"name": "learner_id", "title": "学习者ID", "type": "str", "length": 32, "nullable": "no", "default": "0"},
|
||||
{"name": "blueprint_id", "title": "蓝图ID", "type": "str", "length": 32, "nullable": "no", "default": "0"},
|
||||
{"name": "payload_json", "title": "证据体(序列化JSON文本)", "type": "text", "nullable": "yes"},
|
||||
{"name": "occurred_at", "title": "发生时间", "type": "datetime", "nullable": "no"},
|
||||
{"name": "dedup_key", "title": "去重键(md5(tenant|event|type))", "type": "str", "length": 32, "nullable": "no", "default": ""},
|
||||
{"name": "collect_batch_id", "title": "采集批次ID", "type": "str", "length": 32, "nullable": "no", "default": ""},
|
||||
{"name": "created_at", "title": "创建时间(应用层写入)", "type": "datetime", "nullable": "no"},
|
||||
{"name": "updated_at", "title": "更新时间(应用层写入)", "type": "datetime", "nullable": "no"}
|
||||
],
|
||||
"indexes": [
|
||||
{
|
||||
"name": "uk_ev_dedup",
|
||||
"unique": true,
|
||||
"fields": [
|
||||
"tenant_id",
|
||||
"source_event_id",
|
||||
"evidence_type"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "idx_ev_learner",
|
||||
"unique": false,
|
||||
"fields": [
|
||||
"tenant_id",
|
||||
"learner_id"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "PRIMARY",
|
||||
"unique": true,
|
||||
"fields": [
|
||||
"id"
|
||||
]
|
||||
}
|
||||
{"name": "uk_ev_dedup", "idxtype": "unique", "idxfields": ["tenant_id", "source_event_id", "evidence_type"]},
|
||||
{"name": "idx_ev_learner", "idxtype": "index", "idxfields": ["tenant_id", "learner_id", "occurred_at"]},
|
||||
{"name": "idx_ev_artifact", "idxtype": "index", "idxfields": ["tenant_id", "artifact_id"]},
|
||||
{"name": "idx_ev_type_time", "idxtype": "index", "idxfields": ["tenant_id", "evidence_type", "occurred_at"]},
|
||||
{"name": "idx_ev_batch", "idxtype": "index", "idxfields": ["tenant_id", "collect_batch_id"]}
|
||||
],
|
||||
"codes": [
|
||||
{
|
||||
"field": "evidence_type",
|
||||
"appcode": "pbl_evidence_type"
|
||||
}
|
||||
{"field": "evidence_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='pbl_evidence_type'"},
|
||||
{"field": "artifact_id", "table": "pbl_artifact", "valuefield": "id", "textfield": "title"},
|
||||
{"field": "blueprint_id", "table": "pbl_blueprint", "valuefield": "id", "textfield": "title"}
|
||||
]
|
||||
}
|
||||
|
||||
@ -1,10 +1,11 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""pbl_evidence —— 产出物与证据幂等采集（M5a/M5b）
|
||||
"""pbl_evidence —— 产出物与证据幂等采集模块（M5a/M5b）。
|
||||
|
||||
注册三处同步之 ②：必须导出 init.py 里的全部契约函数，漏一行 .dspy 调用即 NameError。
|
||||
导出契约函数（三处同步之 ②：必须与 init.py 的 env 注册、api.py 定义一一对应，
|
||||
漏此处会导致 .dspy 调用 NameError）。
|
||||
"""
|
||||
from pbl_evidence.init import load_pbl_evidence
|
||||
|
||||
from pbl_evidence.api import (
|
||||
pbl_artifact_create,
|
||||
pbl_artifact_read,
|
||||
@ -12,18 +13,29 @@ from pbl_evidence.api import (
|
||||
pbl_artifact_delete,
|
||||
pbl_artifact_list,
|
||||
pbl_evidence_collect,
|
||||
pbl_evidence_collect_from_events,
|
||||
pbl_evidence_list,
|
||||
|
||||
pbl_evidence_stats,
|
||||
pbl_evidence_watermark,
|
||||
)
|
||||
from pbl_evidence.collector import (
|
||||
CollectError,
|
||||
collect_evidence_from_events,
|
||||
resolve_event_table,
|
||||
)
|
||||
from pbl_evidence.evidence_map import (
|
||||
EVIDENCE_TYPES,
|
||||
EVENT_TO_EVIDENCE,
|
||||
evidence_types_for_events,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
'load_pbl_evidence',
|
||||
'pbl_artifact_create',
|
||||
'pbl_artifact_read',
|
||||
'pbl_artifact_update',
|
||||
'pbl_artifact_delete',
|
||||
'pbl_artifact_list',
|
||||
'pbl_evidence_collect',
|
||||
'pbl_evidence_list',
|
||||
|
||||
'pbl_artifact_create', 'pbl_artifact_read', 'pbl_artifact_update',
|
||||
'pbl_artifact_delete', 'pbl_artifact_list',
|
||||
'pbl_evidence_collect', 'pbl_evidence_collect_from_events',
|
||||
'pbl_evidence_list', 'pbl_evidence_stats', 'pbl_evidence_watermark',
|
||||
'collect_evidence_from_events', 'resolve_event_table', 'CollectError',
|
||||
'EVIDENCE_TYPES', 'EVENT_TO_EVIDENCE', 'evidence_types_for_events',
|
||||
]
|
||||
|
||||
__version__ = '1.1.0'
|
||||
|
||||
@ -1,129 +1,390 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""pbl_evidence.api —— 产出物 + 证据幂等采集（第18/19章，M5a/M5b）。
|
||||
"""pbl_evidence.api —— 产出物 CRUD + 证据幂等采集（第18/19章，M5a/M5b）。
|
||||
|
||||
幂等硬线：`pbl_evidence` 唯一索引 (tenant_id, source_event_id, evidence_type)。
|
||||
重复采集（重放/双通道：广播 + 3s 轮询兜底同事件）→ 命中已存在行，返回 deduped=True，
|
||||
不产生第二条证据，也不报错（学生侧无感）。
|
||||
幂等硬线：``pbl_evidence`` 唯一索引 ``uk_ev_dedup(tenant_id, source_event_id, evidence_type)``。
|
||||
重复采集（重放 / 双通道：广播 + 3s 轮询兜底命中同一事件）→ 命中已存在行，返回
|
||||
``deduped=True``，不产生第二条证据、也不报错（学生侧无感）。
|
||||
|
||||
M5a 修复的两处存量缺陷（QC 可核对）：
|
||||
1. 旧版 ``from pbl_common.api import crud`` —— ``pbl_common.api`` **没有** ``crud`` 这个符号，
|
||||
import 即 ``ImportError``，模块根本挂不上。现改为模块内自建数据访问层（``db.py``）+
|
||||
显式列名发现，不再依赖不存在的工厂函数。
|
||||
2. 旧版所有读写把库名硬编码成 ``'pbl'`` —— 违反「模块取库名禁止硬编码 DBNAME」铁律
|
||||
（库名由宿主应用决定）。现统一走 ``db.get_dbname()`` →
|
||||
``ServerEnv().get_module_dbname('pbl_evidence')``，取不到即 fail-closed。
|
||||
|
||||
契约面（init.py 注册 + __init__.py 导出，三处同步）：
|
||||
产出物：pbl_artifact_create / read / update / delete / list
|
||||
证据 ：pbl_evidence_collect（单事件幂等落库）
|
||||
pbl_evidence_collect_from_events（批量：扫 pbl_runtime_event → 落证据，M5a 主能力）
|
||||
pbl_evidence_list / pbl_evidence_stats / pbl_evidence_watermark
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
|
||||
from pbl_common.api import (PblError, actor_id, crud, json_dump, now_str, sql_exec,
|
||||
sql_rows, tenant_id)
|
||||
from pbl_common.api import PblError, actor_id, json_dump, now_str, tenant_id
|
||||
|
||||
EVIDENCE_TYPES = ['artifact', 'process_event', 'reflection', 'feedback']
|
||||
from pbl_evidence.db import (
|
||||
build_insert, build_where, invalidate_cache, json_loads, pick_columns,
|
||||
q_all, q_exec, q_one, table_columns,
|
||||
)
|
||||
from pbl_evidence.collector import (
|
||||
CollectError, collect_evidence_from_events, evidence_stats as _stats,
|
||||
list_evidence as _list_ev, resolve_event_table, watermark as _watermark,
|
||||
)
|
||||
from pbl_evidence.evidence_map import (
|
||||
EVIDENCE_TYPES, EV_OBSERVATION, build_dedup_key, map_event_to_evidence_type,
|
||||
normalize_dt, summarize_payload,
|
||||
)
|
||||
|
||||
# 与 models/pbl_evidence.json 的 evidence_type 编码（appcodes.pbl_evidence_type）对齐
|
||||
ALLOWED_EVIDENCE_TYPES = list(EVIDENCE_TYPES)
|
||||
|
||||
ARTIFACT_COLS = ('tenant_id', 'artifact_uid', 'session_id', 'blueprint_id', 'team_id',
|
||||
'creator_id', 'title', 'artifact_type', 'content_json', 'version_no',
|
||||
'status', 'submitted_at', 'created_at', 'updated_at')
|
||||
EVIDENCE_WRITE_COLS = ('tenant_id', 'artifact_id', 'evidence_type', 'source_event_id',
|
||||
'session_id', 'learner_id', 'blueprint_id', 'payload_json',
|
||||
'occurred_at', 'dedup_key', 'created_at', 'updated_at')
|
||||
|
||||
UNIQUE_INDEX_NAME = 'uk_ev_dedup(tenant_id,source_event_id,evidence_type)'
|
||||
|
||||
|
||||
def _loads(v, d=None):
|
||||
if isinstance(v, (dict, list)):
|
||||
return v
|
||||
try:
|
||||
return json.loads(v) if v else (d or {})
|
||||
except ValueError:
|
||||
return d or {}
|
||||
def _require_tenant():
|
||||
tid = tenant_id()
|
||||
if not tid:
|
||||
raise PblError('PBL_E_TENANT_MISSING',
|
||||
'tenant_id 缺失：pbl_evidence 所有读写强制带租户（fail-closed）')
|
||||
return str(tid)
|
||||
|
||||
|
||||
ART = crud('pbl_artifact', 'pbl', ['artifact_uid', 'session_id', 'blueprint_id', 'team_id',
|
||||
'creator_id', 'title', 'artifact_type', 'content_json',
|
||||
'version_no', 'status', 'submitted_at'])
|
||||
EV = crud('pbl_evidence', 'pbl', ['artifact_id', 'evidence_type', 'source_event_id',
|
||||
'session_id', 'learner_id', 'blueprint_id', 'payload_json',
|
||||
'occurred_at', 'dedup_key'])
|
||||
def _blank(value):
|
||||
return value is None or (isinstance(value, str) and value.strip() == '')
|
||||
|
||||
|
||||
async def _tid():
|
||||
return await tenant_id()
|
||||
def _gen_uid(prefix, seed):
|
||||
return '%s%s' % (prefix, hashlib.sha256(str(seed).encode('utf-8')).hexdigest()[:20])
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
# 产出物 pbl_artifact
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
async def pbl_artifact_create(**kw):
|
||||
tid = await _tid()
|
||||
kw.setdefault('artifact_uid', 'AR%s' % hashlib.sha256(
|
||||
(str(kw.get('title')) + now_str()).encode()).hexdigest()[:20])
|
||||
kw.setdefault('version_no', 1)
|
||||
kw.setdefault('status', 'draft')
|
||||
kw['content_json'] = json_dump(kw.get('content_json'))
|
||||
kw['creator_id'] = kw.get('creator_id') or await actor_id()
|
||||
r = await ART['create'](**kw)
|
||||
r['artifact_uid'] = kw['artifact_uid']
|
||||
return r
|
||||
"""新建产出物：自动生成 artifact_uid、version_no=1、status=draft。"""
|
||||
tid = _require_tenant()
|
||||
cols = await table_columns('pbl_artifact')
|
||||
if not cols:
|
||||
raise PblError('PBL_E_DB_UNAVAILABLE', 'pbl_artifact 表不存在，请先执行 models DDL')
|
||||
|
||||
title = kw.get('title') or '未命名产出物'
|
||||
record = {
|
||||
'tenant_id': tid,
|
||||
'artifact_uid': kw.get('artifact_uid') or _gen_uid('AR', title + now_str() + str(kw.get('session_id') or '')),
|
||||
'title': str(title)[:128],
|
||||
'artifact_type': kw.get('artifact_type') or 'document',
|
||||
'version_no': int(kw.get('version_no') or 1),
|
||||
'status': kw.get('status') or 'draft',
|
||||
'content_json': json_dump(kw.get('content_json') if not _blank(kw.get('content_json')) else {}),
|
||||
'creator_id': kw.get('creator_id') or actor_id() or '0',
|
||||
'created_at': now_str(),
|
||||
'updated_at': now_str(),
|
||||
}
|
||||
for key in ('session_id', 'blueprint_id', 'team_id'):
|
||||
if not _blank(kw.get(key)):
|
||||
record[key] = str(kw[key])
|
||||
if not _blank(kw.get('submitted_at')):
|
||||
st = normalize_dt(kw['submitted_at'])
|
||||
if st:
|
||||
record['submitted_at'] = st
|
||||
|
||||
record = pick_columns(record, cols)
|
||||
if 'title' not in record:
|
||||
raise PblError('PBL_E_DB', 'pbl_artifact 缺 title 列，表结构与模型不一致')
|
||||
sql, args = build_insert('pbl_artifact', record, cols)
|
||||
await q_exec(sql, args)
|
||||
row = await q_one('SELECT `id` FROM `pbl_artifact` WHERE `tenant_id`=${tenant_id}$ '
|
||||
'AND `artifact_uid`=${artifact_uid}$ LIMIT 1',
|
||||
{'tenant_id': tid, 'artifact_uid': record['artifact_uid']})
|
||||
return {'ok': True, 'id': (row or {}).get('id'), 'artifact_uid': record['artifact_uid'],
|
||||
'version_no': record['version_no'], 'data': {'id': (row or {}).get('id'),
|
||||
'artifact_uid': record['artifact_uid']}}
|
||||
|
||||
|
||||
async def pbl_artifact_read(**kw):
|
||||
return await ART['read'](**kw)
|
||||
"""按 id 或 artifact_uid 读产出物（强制租户过滤）。"""
|
||||
tid = _require_tenant()
|
||||
aid = kw.get('id') or kw.get('artifact_id')
|
||||
uid = kw.get('artifact_uid')
|
||||
if _blank(aid) and _blank(uid):
|
||||
raise PblError('PBL_E_PARAM', '缺少 id / artifact_uid')
|
||||
cols = await table_columns('pbl_artifact')
|
||||
if _blank(aid):
|
||||
where, args = {'tenant_id': tid, 'artifact_uid': uid}, None
|
||||
else:
|
||||
where, args = {'tenant_id': tid, 'id': aid}, None
|
||||
where_sql, args = build_where(where, cols)
|
||||
row = await q_one('SELECT * FROM `pbl_artifact` WHERE %s LIMIT 1' % where_sql, args)
|
||||
if not row:
|
||||
raise PblError('PBL_E_NOT_FOUND', '产出物不存在或不属于当前租户')
|
||||
row['content'] = json_loads(row.get('content_json'), default={})
|
||||
return {'ok': True, 'data': row}
|
||||
|
||||
|
||||
async def pbl_artifact_update(**kw):
|
||||
"""改内容即版本递增（第29.8 版本化），历史版本快照存 version_history_json 于 content_json 外层。"""
|
||||
tid = await _tid()
|
||||
cur = await ART['read'](id=kw.get('id'))
|
||||
data = cur['data']
|
||||
payload = dict(kw)
|
||||
if 'content_json' in payload:
|
||||
payload['content_json'] = json_dump(payload['content_json'])
|
||||
try:
|
||||
hist = _loads(data.get('content_json'), {})
|
||||
except Exception:
|
||||
hist = {}
|
||||
payload['content_json'] = json_dump({'current': _loads(payload['content_json']),
|
||||
'history': {'v%s' % data['version_no']: hist}})
|
||||
payload['version_no'] = int(data['version_no'] or 1) + 1
|
||||
return await ART['update'](**payload)
|
||||
"""更新产出物：改 content_json 即版本递增（第29.8 版本化），旧内容进 history。"""
|
||||
tid = _require_tenant()
|
||||
aid = kw.get('id') or kw.get('artifact_id')
|
||||
if _blank(aid):
|
||||
raise PblError('PBL_E_PARAM', '缺少 id')
|
||||
cols = await table_columns('pbl_artifact')
|
||||
where_sql, args = build_where({'tenant_id': tid, 'id': aid}, cols)
|
||||
cur = await q_one('SELECT * FROM `pbl_artifact` WHERE %s LIMIT 1' % where_sql, args)
|
||||
if not cur:
|
||||
raise PblError('PBL_E_NOT_FOUND', '产出物不存在或不属于当前租户')
|
||||
|
||||
upd = {'updated_at': now_str()}
|
||||
for key in ('title', 'artifact_type', 'status', 'session_id', 'blueprint_id', 'team_id'):
|
||||
if not _blank(kw.get(key)):
|
||||
upd[key] = str(kw[key])[:128] if key == 'title' else kw[key]
|
||||
if not _blank(kw.get('submitted_at')):
|
||||
st = normalize_dt(kw['submitted_at'])
|
||||
if st:
|
||||
upd['submitted_at'] = st
|
||||
if not _blank(kw.get('content_json')):
|
||||
new_content = kw['content_json']
|
||||
if isinstance(new_content, str):
|
||||
new_content = json_loads(new_content, default={})
|
||||
old_content = json_loads(cur.get('content_json'), default={})
|
||||
old_ver = int(cur.get('version_no') or 1)
|
||||
# 已是本模块写入过的 {current,history} 结构则续写 history，否则把原值当 current
|
||||
prev = old_content.get('current', old_content) if isinstance(old_content, dict) else {}
|
||||
history = (old_content.get('history') if isinstance(old_content, dict) else {}) or {}
|
||||
history['v%s' % old_ver] = prev
|
||||
upd['content_json'] = json_dump({'current': new_content, 'history': history})
|
||||
upd['version_no'] = old_ver + 1
|
||||
elif not _blank(kw.get('version_no')):
|
||||
upd['version_no'] = int(kw['version_no'])
|
||||
|
||||
upd = pick_columns(upd, cols)
|
||||
set_sql = ', '.join('`%s`=${%s}$' % (c, c) for c in upd)
|
||||
params = dict({c: upd[c] for c in upd})
|
||||
params['tenant_id'] = tid
|
||||
params['id'] = aid
|
||||
await q_exec('UPDATE `pbl_artifact` SET %s WHERE `tenant_id`=${tenant_id}$ AND `id`=${id}$'
|
||||
% set_sql, params)
|
||||
return {'ok': True, 'id': aid, 'version_no': upd.get('version_no', cur.get('version_no')),
|
||||
'updated_fields': sorted(upd.keys())}
|
||||
|
||||
|
||||
async def pbl_artifact_delete(**kw):
|
||||
tid = await _tid()
|
||||
await sql_exec('DELETE FROM `pbl_evidence` WHERE `tenant_id` = ${t}$'
|
||||
' AND `artifact_id` = ${a}$', {'t': tid, 'a': kw.get('id')}, 'pbl')
|
||||
return await ART['delete'](**kw)
|
||||
"""删除产出物：先级联删其证据（避免孤儿证据污染 M6 评估），再删本体。"""
|
||||
tid = _require_tenant()
|
||||
aid = kw.get('id') or kw.get('artifact_id')
|
||||
if _blank(aid):
|
||||
raise PblError('PBL_E_PARAM', '缺少 id')
|
||||
ev_cols = await table_columns('pbl_evidence')
|
||||
if 'artifact_id' in ev_cols:
|
||||
await q_exec('DELETE FROM `pbl_evidence` WHERE `tenant_id`=${tenant_id}$ '
|
||||
'AND `artifact_id`=${artifact_id}$',
|
||||
{'tenant_id': tid, 'artifact_id': aid})
|
||||
cols = await table_columns('pbl_artifact')
|
||||
where_sql, args = build_where({'tenant_id': tid, 'id': aid}, cols)
|
||||
await q_exec('DELETE FROM `pbl_artifact` WHERE %s' % where_sql, args)
|
||||
return {'ok': True, 'id': aid, 'cascaded_evidence': True}
|
||||
|
||||
|
||||
async def pbl_artifact_list(**kw):
|
||||
return await ART['list'](**kw)
|
||||
"""产出物分页列表（强制租户；支持 session/blueprint/creator/type/status 过滤）。"""
|
||||
tid = _require_tenant()
|
||||
cols = await table_columns('pbl_artifact')
|
||||
where = {'tenant_id': tid}
|
||||
for key in ('session_id', 'blueprint_id', 'team_id', 'creator_id',
|
||||
'artifact_type', 'status', 'artifact_uid'):
|
||||
if not _blank(kw.get(key)):
|
||||
where[key] = kw[key]
|
||||
where_sql, args = build_where(where, cols)
|
||||
page = max(1, int(kw.get('page') or 1))
|
||||
rows = min(200, max(1, int(kw.get('rows') or kw.get('page_size') or 20)))
|
||||
total = int((await q_one('SELECT COUNT(*) AS c FROM `pbl_artifact` WHERE %s' % where_sql,
|
||||
dict(args)) or {}).get('c') or 0)
|
||||
sort = kw.get('sort') if (kw.get('sort') in cols) else 'created_at'
|
||||
direction = 'ASC' if str(kw.get('order') or 'desc').lower() == 'asc' else 'DESC'
|
||||
args2 = dict(args)
|
||||
args2['rows'] = rows
|
||||
args2['offset'] = (page - 1) * rows
|
||||
recs = await q_all('SELECT * FROM `pbl_artifact` WHERE %s ORDER BY `%s` %s '
|
||||
'LIMIT ${rows}$ OFFSET ${offset}$' % (where_sql, sort, direction), args2)
|
||||
for r in recs:
|
||||
r['content'] = json_loads(r.get('content_json'), default={})
|
||||
if 'artifact_type' in r:
|
||||
r['artifact_type_text'] = r.get('artifact_type')
|
||||
return {'ok': True, 'status': 'OK', 'total': total, 'page': page, 'rows_per_page': rows,
|
||||
'data': recs, 'rows': recs}
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
# 证据 pbl_evidence
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
async def pbl_evidence_collect(**kw):
|
||||
"""从 pbl_runtime_event 采集证据（服务端权威，客户端不能直接写证据）。"""
|
||||
tid = await _tid()
|
||||
etype = kw.get('evidence_type') or 'process_event'
|
||||
if etype not in EVIDENCE_TYPES:
|
||||
raise PblError('PBL_EVIDENCE_TYPE_INVALID', 'evidence_type 须为 %s'
|
||||
% '/'.join(EVIDENCE_TYPES))
|
||||
src = kw.get('source_event_id')
|
||||
if not src:
|
||||
raise PblError('PBL_PARAM_MISSING', '缺少 source_event_id（证据必须可溯到事件）')
|
||||
dup = await sql_rows('SELECT * FROM `pbl_evidence` WHERE `tenant_id` = ${t}$'
|
||||
' AND `source_event_id` = ${s}$ AND `evidence_type` = ${e}$ LIMIT 1',
|
||||
{'t': tid, 's': src, 'e': etype}, 'pbl')
|
||||
if dup:
|
||||
return {'ok': True, 'deduped': True, 'id': dup[0]['id'],
|
||||
'unique_index': 'uk_ev_dedup(tenant_id,source_event_id,evidence_type)',
|
||||
'evidence_count': 1}
|
||||
payload = {'evidence_type': etype, 'source_event_id': src,
|
||||
'session_id': kw.get('session_id') or 0,
|
||||
'learner_id': kw.get('learner_id') or await actor_id(),
|
||||
'blueprint_id': kw.get('blueprint_id'),
|
||||
'artifact_id': kw.get('artifact_id'),
|
||||
'payload_json': json_dump(kw.get('payload_json') or {}),
|
||||
'occurred_at': kw.get('occurred_at') or now_str(),
|
||||
'dedup_key': hashlib.sha256(('%s|%s' % (src, etype)).encode()).hexdigest()[:32]}
|
||||
r = await EV['create'](**payload)
|
||||
r['deduped'] = False
|
||||
r['unique_index'] = 'uk_ev_dedup(tenant_id,source_event_id,evidence_type)'
|
||||
return r
|
||||
"""单事件幂等落证据（服务端权威；客户端不能直接写证据表）。
|
||||
|
||||
幂等：先按唯一索引三元组预查，命中即返回 ``deduped=True``；
|
||||
未命中用 ``INSERT ... ON DUPLICATE KEY UPDATE`` 兜并发竞态。
|
||||
"""
|
||||
tid = _require_tenant()
|
||||
src = kw.get('source_event_id') or kw.get('event_id')
|
||||
if _blank(src):
|
||||
raise PblError('PBL_E_PARAM', '缺少 source_event_id（证据必须可溯到 runtime_event）')
|
||||
src = str(src).strip()[:32]
|
||||
|
||||
event_type = kw.get('event_type') or ''
|
||||
etype = kw.get('evidence_type')
|
||||
if _blank(etype):
|
||||
etype = map_event_to_evidence_type(event_type, kw) or EV_OBSERVATION
|
||||
etype = str(etype).strip()
|
||||
if etype not in ALLOWED_EVIDENCE_TYPES:
|
||||
raise PblError('PBL_E_PARAM', 'evidence_type 须为 %s' % '/'.join(ALLOWED_EVIDENCE_TYPES))
|
||||
|
||||
cols = await table_columns('pbl_evidence')
|
||||
if not cols:
|
||||
raise PblError('PBL_E_DB_UNAVAILABLE', 'pbl_evidence 表不存在，请先执行 models DDL')
|
||||
|
||||
found = await q_one(
|
||||
'SELECT `id` FROM `pbl_evidence` WHERE `tenant_id`=${tenant_id}$ '
|
||||
'AND `source_event_id`=${source_event_id}$ AND `evidence_type`=${evidence_type}$ LIMIT 1',
|
||||
{'tenant_id': tid, 'source_event_id': src, 'evidence_type': etype})
|
||||
if found and found.get('id') is not None:
|
||||
return {'ok': True, 'deduped': True, 'id': found['id'],
|
||||
'unique_index': UNIQUE_INDEX_NAME, 'evidence_count': 1}
|
||||
|
||||
occurred = normalize_dt(kw.get('occurred_at')) or now_str()
|
||||
record = {
|
||||
'tenant_id': tid,
|
||||
'artifact_id': kw.get('artifact_id') if not _blank(kw.get('artifact_id')) else 0,
|
||||
'evidence_type': etype,
|
||||
'source_event_id': src,
|
||||
'session_id': kw.get('session_id') if not _blank(kw.get('session_id')) else 0,
|
||||
'learner_id': str(kw.get('learner_id') or actor_id() or '0')[:32],
|
||||
'blueprint_id': kw.get('blueprint_id') if not _blank(kw.get('blueprint_id')) else None,
|
||||
'payload_json': json_dump(summarize_payload(kw, event_type, etype)),
|
||||
'occurred_at': occurred,
|
||||
'dedup_key': build_dedup_key(tid, src, etype),
|
||||
'created_at': now_str(),
|
||||
'updated_at': now_str(),
|
||||
}
|
||||
record = pick_columns(record, cols)
|
||||
ins_cols = [c for c in record if c in cols]
|
||||
col_sql = ', '.join('`%s`' % c for c in ins_cols)
|
||||
val_sql = ', '.join('${%s}$' % c for c in ins_cols)
|
||||
dup_cols = [c for c in ins_cols if c not in ('tenant_id', 'source_event_id',
|
||||
'evidence_type', 'dedup_key', 'created_at')]
|
||||
sql = 'INSERT INTO `pbl_evidence` (%s) VALUES (%s)' % (col_sql, val_sql)
|
||||
if dup_cols:
|
||||
sql += ' ON DUPLICATE KEY UPDATE ' + ', '.join('`%s`=VALUES(`%s`)' % (c, c) for c in dup_cols)
|
||||
sql += ', `updated_at`=VALUES(`updated_at`)'
|
||||
else:
|
||||
sql += ' ON DUPLICATE KEY UPDATE `updated_at`=VALUES(`updated_at`)'
|
||||
try:
|
||||
await q_exec(sql, {c: record[c] for c in ins_cols})
|
||||
except Exception as exc:
|
||||
if 'duplicate entry' in str(exc).lower() or '1062' in str(exc):
|
||||
again = await q_one(
|
||||
'SELECT `id` FROM `pbl_evidence` WHERE `tenant_id`=${tenant_id}$ '
|
||||
'AND `source_event_id`=${source_event_id}$ AND `evidence_type`=${evidence_type}$ LIMIT 1',
|
||||
{'tenant_id': tid, 'source_event_id': src, 'evidence_type': etype})
|
||||
return {'ok': True, 'deduped': True, 'id': (again or {}).get('id'),
|
||||
'unique_index': UNIQUE_INDEX_NAME, 'evidence_count': 1}
|
||||
raise PblError('PBL_E_DB', '证据写入失败: %s' % (exc,))
|
||||
|
||||
new = await q_one(
|
||||
'SELECT `id` FROM `pbl_evidence` WHERE `tenant_id`=${tenant_id}$ '
|
||||
'AND `source_event_id`=${source_event_id}$ AND `evidence_type`=${evidence_type}$ LIMIT 1',
|
||||
{'tenant_id': tid, 'source_event_id': src, 'evidence_type': etype})
|
||||
return {'ok': True, 'deduped': False, 'id': (new or {}).get('id'),
|
||||
'unique_index': UNIQUE_INDEX_NAME, 'evidence_count': 1,
|
||||
'evidence_type': etype, 'source_event_id': src, 'occurred_at': occurred}
|
||||
|
||||
|
||||
async def pbl_evidence_collect_from_events(**kw):
|
||||
"""M5a 主能力：批量从 ``pbl_runtime_event``（M11b 产物）采集证据并幂等落库。
|
||||
|
||||
硬约束 C-3：本能力读的事件表由 scense_runtime M11b 写入，M5 不得早于 M11b 上线；
|
||||
事件表缺失时 collector 抛 ``CollectError`` → 这里转成明确错误码，不静默返回空成功。
|
||||
"""
|
||||
tid = _require_tenant()
|
||||
try:
|
||||
res = await collect_evidence_from_events(
|
||||
tenant_id=tid,
|
||||
since=kw.get('since'),
|
||||
until=kw.get('until'),
|
||||
session_id=kw.get('session_id'),
|
||||
learner_id=kw.get('learner_id'),
|
||||
blueprint_id=kw.get('blueprint_id'),
|
||||
event_types=kw.get('event_types') if isinstance(kw.get('event_types'), (list, tuple))
|
||||
else (str(kw.get('event_types')).split(',') if not _blank(kw.get('event_types')) else None),
|
||||
evidence_types=kw.get('evidence_types') if isinstance(kw.get('evidence_types'), (list, tuple))
|
||||
else (str(kw.get('evidence_types')).split(',') if not _blank(kw.get('evidence_types')) else None),
|
||||
limit=kw.get('limit') or 500,
|
||||
dry_run=str(kw.get('dry_run') or '').lower() in ('1', 'true', 'yes'),
|
||||
update_existing=str(kw.get('update_existing') or '').lower() in ('1', 'true', 'yes'),
|
||||
)
|
||||
except CollectError as exc:
|
||||
raise PblError('PBL_E_PARAM' if 'tenant_id' in str(exc) else 'PBL_E_DB_UNAVAILABLE', str(exc))
|
||||
res['ok'] = True
|
||||
res['status'] = 'OK'
|
||||
res['unique_index'] = UNIQUE_INDEX_NAME
|
||||
res['items'] = res.get('items') or []
|
||||
if 'message' not in res:
|
||||
res['message'] = 'scanned=%s created=%s skipped=%s updated=%s ignored=%s failed=%s' % (
|
||||
res.get('scanned'), res.get('created'), res.get('skipped'),
|
||||
res.get('updated'), res.get('ignored'), res.get('failed'))
|
||||
return res
|
||||
|
||||
|
||||
async def pbl_evidence_list(**kw):
|
||||
tid = await _tid()
|
||||
where, params = ['`tenant_id` = ${t}$'], {'t': tid}
|
||||
for col in ('session_id', 'learner_id', 'blueprint_id', 'artifact_id', 'evidence_type'):
|
||||
if kw.get(col) not in (None, ''):
|
||||
where.append('`%s` = ${%s}$' % (col, col))
|
||||
params[col] = kw[col]
|
||||
limit = min(int(kw.get('page_size') or 50), 200)
|
||||
rows = await sql_rows('SELECT * FROM `pbl_evidence` WHERE %s ORDER BY `occurred_at` DESC'
|
||||
' LIMIT ${lim}$' % ' AND '.join(where), dict(params, lim=limit), 'pbl')
|
||||
"""证据分页查询（强制租户；支持 learner/artifact/session/blueprint/type/时间窗过滤）。"""
|
||||
tid = _require_tenant()
|
||||
try:
|
||||
res = await _list_ev(
|
||||
tenant_id=tid, learner_id=kw.get('learner_id'), artifact_id=kw.get('artifact_id'),
|
||||
evidence_type=kw.get('evidence_type'), session_id=kw.get('session_id'),
|
||||
blueprint_id=kw.get('blueprint_id'), since=kw.get('since'), until=kw.get('until'),
|
||||
page=kw.get('page') or 1, rows=kw.get('rows') or kw.get('page_size') or 20,
|
||||
sort=kw.get('sort') or 'occurred_at', order=kw.get('order') or 'desc')
|
||||
except CollectError as exc:
|
||||
raise PblError('PBL_E_DB_UNAVAILABLE', str(exc))
|
||||
rows = res.get('rows') or []
|
||||
by_type = {}
|
||||
for r in rows:
|
||||
by_type[r['evidence_type']] = by_type.get(r['evidence_type'], 0) + 1
|
||||
return {'ok': True, 'total': len(rows), 'by_type': by_type, 'data': rows}
|
||||
t = r.get('evidence_type')
|
||||
by_type[t] = by_type.get(t, 0) + 1
|
||||
return {'ok': True, 'status': 'OK', 'total': res.get('total'), 'page': res.get('page'),
|
||||
'rows_per_page': res.get('rows_per_page'), 'by_type': by_type,
|
||||
'data': rows, 'rows': rows}
|
||||
|
||||
|
||||
async def pbl_evidence_stats(**kw):
|
||||
"""证据类型分布统计（供 M6 评估与概览页）。"""
|
||||
tid = _require_tenant()
|
||||
try:
|
||||
res = await _stats(tenant_id=tid, learner_id=kw.get('learner_id'),
|
||||
blueprint_id=kw.get('blueprint_id'))
|
||||
except CollectError as exc:
|
||||
raise PblError('PBL_E_DB_UNAVAILABLE', str(exc))
|
||||
res['ok'] = True
|
||||
res['status'] = 'OK'
|
||||
return res
|
||||
|
||||
|
||||
async def pbl_evidence_watermark(**kw):
|
||||
"""增量采集游标（已采集最大事件时间 + 总量），供 3s 轮询兜底 / cron 回放。"""
|
||||
tid = _require_tenant()
|
||||
try:
|
||||
res = await _watermark(tenant_id=tid)
|
||||
except CollectError as exc:
|
||||
raise PblError('PBL_E_DB_UNAVAILABLE', str(exc))
|
||||
res['ok'] = True
|
||||
res['status'] = 'OK'
|
||||
res['event_table'] = await resolve_event_table()
|
||||
return res
|
||||
|
||||
418
pbl_evidence/collector.py
Normal file
418
pbl_evidence/collector.py
Normal file
@ -0,0 +1,418 @@
|
||||
"""M5a 证据采集核心：从 ``pbl_runtime_event`` 采集学习证据，幂等落 ``pbl_evidence``。
|
||||
|
||||
幂等契约（任务硬要求）：
|
||||
``uk_ev_dedup(tenant_id, source_event_id, evidence_type)`` 唯一索引防重。
|
||||
双保险实现，任一生效即保证幂等，且**并发安全**：
|
||||
1. 预查：按唯一索引三元组 SELECT，命中则跳过/按需更新（省一次写）；
|
||||
2. 真防重：``INSERT ... ON DUPLICATE KEY UPDATE``（唯一索引冲突不报错、不产生重复行），
|
||||
由数据库保证并发下最终一致——预查与写入之间的竞态由第 2 层兜住。
|
||||
|
||||
采集是**只读事件表 + 幂等写证据表**，可任意次重放（M5b 回放、cron 轮询兜底同一条路径）。
|
||||
"""
|
||||
|
||||
from .db import (
|
||||
get_dbname, q_all, q_one, q_exec, table_exists, table_columns,
|
||||
pick_columns, build_insert, build_where, json_dumps, json_loads,
|
||||
row_to_dict, invalidate_cache, _log,
|
||||
)
|
||||
from .evidence_map import (
|
||||
EV_OBSERVATION, EVIDENCE_TYPES, IGNORED_EVENTS, EVENT_ID_FIELDS,
|
||||
EVENT_TYPE_FIELDS, EVENT_PAYLOAD_FIELDS, EVENT_TIME_FIELDS,
|
||||
EVENT_LEARNER_FIELDS, EVENT_SESSION_FIELDS, EVENT_BLUEPRINT_FIELDS,
|
||||
normalize_dt, now_str, first_present, map_event_to_evidence_type,
|
||||
build_dedup_key, event_source_id, summarize_payload,
|
||||
)
|
||||
|
||||
# 运行时事件表候选名：M11b 落 pbl_runtime_event；历史/别名兼容 runtime_event。
|
||||
EVENT_TABLE_CANDIDATES = ('pbl_runtime_event', 'runtime_event', 'pbl_event')
|
||||
|
||||
# 单次采集扫描上限（防止一次拉爆内存；分区表按 occurred_at 倒序取最近窗口）
|
||||
DEFAULT_BATCH_LIMIT = 500
|
||||
MAX_BATCH_LIMIT = 2000
|
||||
|
||||
# 证据表写入前必须存在的列（缺列 = 表未建/未迁移，fail-closed）
|
||||
REQUIRED_EVIDENCE_COLS = ('tenant_id', 'source_event_id', 'evidence_type',
|
||||
'dedup_key', 'occurred_at')
|
||||
|
||||
|
||||
class CollectError(Exception):
|
||||
"""采集前置条件不满足（表缺失 / 租户缺失等），fail-closed。"""
|
||||
|
||||
|
||||
async def resolve_event_table():
|
||||
"""发现真实的事件表名（M11b 落库后的表），找不到返回 None。"""
|
||||
for name in EVENT_TABLE_CANDIDATES:
|
||||
if await table_exists(name):
|
||||
return name
|
||||
return None
|
||||
|
||||
|
||||
async def collect_evidence_from_events(tenant_id,
|
||||
since=None,
|
||||
until=None,
|
||||
session_id=None,
|
||||
learner_id=None,
|
||||
blueprint_id=None,
|
||||
event_types=None,
|
||||
evidence_types=None,
|
||||
limit=DEFAULT_BATCH_LIMIT,
|
||||
dry_run=False,
|
||||
update_existing=False):
|
||||
"""从运行时事件表采集学习证据并幂等落库。
|
||||
|
||||
参数
|
||||
----
|
||||
tenant_id : 必填，缺失即 fail-closed（多租户铁律）。
|
||||
since/until : 事件 occurred_at 时间窗（'YYYY-MM-DD HH:MM:SS'），可空。
|
||||
session_id / learner_id / blueprint_id : 缩小扫描范围，可空。
|
||||
event_types : 只采集指定事件类型（list[str]），可空=全部（忽略黑名单）。
|
||||
evidence_types : 只落指定证据类型（list[str]），可空=全部。
|
||||
limit : 单次扫描事件上限（≤2000）。
|
||||
dry_run : True 只返回将要落库的证据（不落库），供预检/测试。
|
||||
update_existing: True 时，已存在的证据会刷新 payload_json/occurred_at（默认只跳过）。
|
||||
|
||||
返回 dict：scanned / created / skipped / updated / ignored / failed /
|
||||
created_ids / items / event_table / watermark
|
||||
"""
|
||||
tid = (str(tenant_id).strip() if tenant_id else '')
|
||||
if not tid:
|
||||
raise CollectError('tenant_id 缺失：pbl_evidence 采集 fail-closed，拒绝执行')
|
||||
|
||||
limit = int(limit or DEFAULT_BATCH_LIMIT)
|
||||
if limit <= 0:
|
||||
limit = DEFAULT_BATCH_LIMIT
|
||||
if limit > MAX_BATCH_LIMIT:
|
||||
limit = MAX_BATCH_LIMIT
|
||||
|
||||
ev_table = await resolve_event_table()
|
||||
if not ev_table:
|
||||
raise CollectError(
|
||||
'未找到运行时事件表（候选 %s）。M5a 硬依赖 scense_runtime M11b：'
|
||||
'pbl_runtime_event 建表并落库后方可采集。' % (', '.join(EVENT_TABLE_CANDIDATES),))
|
||||
|
||||
ev_cols = await table_columns(ev_table)
|
||||
if not ev_cols:
|
||||
raise CollectError('事件表 %s 列信息为空（无权限或表结构异常）' % ev_table)
|
||||
|
||||
tgt_cols = await table_columns('pbl_evidence')
|
||||
if not tgt_cols:
|
||||
raise CollectError('证据表 pbl_evidence 不存在或无列信息，请先执行 models/pbl_evidence.json 的 DDL')
|
||||
missing = [c for c in REQUIRED_EVIDENCE_COLS if c not in tgt_cols]
|
||||
if missing:
|
||||
raise CollectError('pbl_evidence 缺关键列 %s，请先完成表迁移' % (', '.join(missing),))
|
||||
|
||||
ev_filter = None
|
||||
if evidence_types:
|
||||
bad = [t for t in evidence_types if t not in EVIDENCE_TYPES]
|
||||
if bad:
|
||||
raise CollectError('未知 evidence_type: %s（合法值 %s）'
|
||||
% (', '.join(bad), ', '.join(EVIDENCE_TYPES)))
|
||||
ev_filter = set(evidence_types)
|
||||
|
||||
id_col = first_present_col(ev_cols, EVENT_ID_FIELDS)
|
||||
type_col = first_present_col(ev_cols, EVENT_TYPE_FIELDS)
|
||||
payload_col = first_present_col(ev_cols, EVENT_PAYLOAD_FIELDS)
|
||||
time_col = first_present_col(ev_cols, EVENT_TIME_FIELDS)
|
||||
learner_col = first_present_col(ev_cols, EVENT_LEARNER_FIELDS)
|
||||
session_col = first_present_col(ev_cols, EVENT_SESSION_FIELDS)
|
||||
blueprint_col = first_present_col(ev_cols, EVENT_BLUEPRINT_FIELDS)
|
||||
|
||||
if not id_col:
|
||||
raise CollectError('事件表 %s 无事件主键列（候选 %s），无法构造幂等键'
|
||||
% (ev_table, ', '.join(EVENT_ID_FIELDS)))
|
||||
|
||||
# ── 扫描事件（列名全部来自 information_schema 发现结果，不猜列）────────
|
||||
select_cols = [id_col]
|
||||
for c in (type_col, payload_col, time_col, learner_col, session_col, blueprint_col):
|
||||
if c and c not in select_cols:
|
||||
select_cols.append(c)
|
||||
# 事件表可能自带的业务外键，一并带出供证据溯源
|
||||
for c in ('artifact_id', 'task_id', 'blueprint_version_id', 'scene_id',
|
||||
'score', 'result', 'status', 'duration_ms', 'tenant_id'):
|
||||
if c in ev_cols and c not in select_cols:
|
||||
select_cols.append(c)
|
||||
|
||||
conds, args = [], {'limit': limit}
|
||||
if 'tenant_id' in ev_cols:
|
||||
conds.append('`tenant_id`=${tenant_id}$')
|
||||
args['tenant_id'] = tid
|
||||
if time_col:
|
||||
sdt, edt = normalize_dt(since), normalize_dt(until)
|
||||
if sdt:
|
||||
conds.append('`%s` >= ${since}$' % time_col)
|
||||
args['since'] = sdt
|
||||
if edt:
|
||||
conds.append('`%s` <= ${until}$' % time_col)
|
||||
args['until'] = edt
|
||||
if session_id and session_col:
|
||||
conds.append('`%s`=${session_id}$' % session_col)
|
||||
args['session_id'] = str(session_id)
|
||||
if learner_id and learner_col:
|
||||
conds.append('`%s`=${learner_id}$' % learner_col)
|
||||
args['learner_id'] = str(learner_id)
|
||||
if blueprint_id and blueprint_col:
|
||||
conds.append('`%s`=${blueprint_id}$' % blueprint_col)
|
||||
args['blueprint_id'] = str(blueprint_id)
|
||||
if event_types:
|
||||
etl = [str(t).strip() for t in event_types if str(t).strip()]
|
||||
if etl and type_col:
|
||||
ph = []
|
||||
for i, v in enumerate(etl):
|
||||
key = 'et_%d' % i
|
||||
ph.append('${%s}$' % key)
|
||||
args[key] = v
|
||||
conds.append('LOWER(`%s`) IN (%s)' % (type_col, ', '.join(ph)))
|
||||
where_sql = ' AND '.join(conds) if conds else '1=1'
|
||||
|
||||
sql = 'SELECT %s FROM `%s` WHERE %s' % (
|
||||
', '.join('`%s`' % c for c in select_cols), ev_table, where_sql)
|
||||
if time_col:
|
||||
sql += ' ORDER BY `%s` ASC' % time_col
|
||||
sql += ' LIMIT ${limit}$'
|
||||
|
||||
try:
|
||||
events = await q_all(sql, args)
|
||||
except Exception as exc:
|
||||
raise CollectError('扫描事件表失败: %s' % (exc,))
|
||||
|
||||
stats = {'scanned': len(events), 'created': 0, 'skipped': 0, 'updated': 0,
|
||||
'ignored': 0, 'failed': 0, 'created_ids': [], 'items': [],
|
||||
'event_table': ev_table, 'watermark': None, 'dry_run': bool(dry_run)}
|
||||
|
||||
now = now_str()
|
||||
for row in events:
|
||||
try:
|
||||
src_id = event_source_id(row)
|
||||
if not src_id:
|
||||
stats['ignored'] += 1
|
||||
continue
|
||||
event_type = str(first_present(row, (type_col,) if type_col else EVENT_TYPE_FIELDS) or '')
|
||||
if event_type.lower() in IGNORED_EVENTS:
|
||||
stats['ignored'] += 1
|
||||
continue
|
||||
evidence_type = map_event_to_evidence_type(event_type, row)
|
||||
if not evidence_type:
|
||||
stats['ignored'] += 1
|
||||
continue
|
||||
if ev_filter and evidence_type not in ev_filter:
|
||||
stats['ignored'] += 1
|
||||
continue
|
||||
|
||||
occurred = normalize_dt(first_present(row, (time_col,) if time_col else EVENT_TIME_FIELDS)) or now
|
||||
stats['watermark'] = occurred
|
||||
learner = first_present(row, (learner_col,) if learner_col else EVENT_LEARNER_FIELDS)
|
||||
session = first_present(row, (session_col,) if session_col else EVENT_SESSION_FIELDS)
|
||||
blueprint = first_present(row, (blueprint_col,) if blueprint_col else EVENT_BLUEPRINT_FIELDS)
|
||||
artifact = row.get('artifact_id') if 'artifact_id' in row else None
|
||||
|
||||
payload = summarize_payload(row, event_type, evidence_type)
|
||||
record = {
|
||||
'tenant_id': tid,
|
||||
'source_event_id': src_id,
|
||||
'evidence_type': evidence_type,
|
||||
'dedup_key': build_dedup_key(tid, src_id, evidence_type),
|
||||
'occurred_at': occurred,
|
||||
'payload_json': json_dumps(payload),
|
||||
'updated_at': now,
|
||||
'created_at': now,
|
||||
}
|
||||
if learner not in (None, ''):
|
||||
record['learner_id'] = str(learner)[:32]
|
||||
if session not in (None, ''):
|
||||
record['session_id'] = str(session)
|
||||
if blueprint not in (None, ''):
|
||||
record['blueprint_id'] = str(blueprint)
|
||||
if artifact not in (None, ''):
|
||||
record['artifact_id'] = str(artifact)
|
||||
# artifact_id 为 NOT NULL：行为类证据（协作/反思/观测）无强归属产出物时以 0 占位，
|
||||
# 不因此丢事件——证据链完整性优先于归属字段（M6 评估侧按 artifact_id>0 过滤）。
|
||||
record.setdefault('artifact_id', 0)
|
||||
|
||||
record = pick_columns(record, tgt_cols)
|
||||
stats['items'].append(record)
|
||||
if dry_run:
|
||||
stats['skipped'] += 1
|
||||
continue
|
||||
|
||||
action = await _upsert_evidence(record, tgt_cols, update_existing)
|
||||
stats[action] = stats.get(action, 0) + 1
|
||||
if action == 'created' and record.get('id'):
|
||||
stats['created_ids'].append(record['id'])
|
||||
except Exception as exc:
|
||||
stats['failed'] += 1
|
||||
_log('collect one event failed: %s' % (exc,))
|
||||
|
||||
return stats
|
||||
|
||||
|
||||
def first_present_col(cols, candidates):
|
||||
"""在真实列集合中按候选顺序挑第一个存在的列名。"""
|
||||
for c in candidates:
|
||||
if c in cols:
|
||||
return c
|
||||
return None
|
||||
|
||||
|
||||
async def _upsert_evidence(record, tgt_cols, update_existing=False):
|
||||
"""幂等写入：预查 + ON DUPLICATE KEY UPDATE 双保险。返回 created/skipped/updated。"""
|
||||
tid = record['tenant_id']
|
||||
sid = record['source_event_id']
|
||||
etp = record['evidence_type']
|
||||
|
||||
existing = await q_one(
|
||||
'SELECT `id` FROM `pbl_evidence` '
|
||||
'WHERE `tenant_id`=${tenant_id}$ AND `source_event_id`=${source_event_id}$ '
|
||||
'AND `evidence_type`=${evidence_type}$ LIMIT 1',
|
||||
{'tenant_id': tid, 'source_event_id': sid, 'evidence_type': etp})
|
||||
|
||||
if existing and existing.get('id') is not None:
|
||||
eid = existing['id']
|
||||
if update_existing:
|
||||
upd = {k: v for k, v in record.items()
|
||||
if k in ('payload_json', 'occurred_at', 'artifact_id',
|
||||
'session_id', 'learner_id', 'blueprint_id', 'dedup_key')}
|
||||
upd['updated_at'] = now_str()
|
||||
cols = [c for c in upd if c in tgt_cols]
|
||||
if cols:
|
||||
set_sql = ', '.join('`%s`=${%s}$' % (c, c) for c in cols)
|
||||
args = {c: upd[c] for c in cols}
|
||||
args['id'] = eid
|
||||
args['tenant_id'] = tid
|
||||
await q_exec('UPDATE `pbl_evidence` SET %s WHERE `id`=${id}$ '
|
||||
'AND `tenant_id`=${tenant_id}$' % set_sql, args)
|
||||
return 'updated'
|
||||
return 'skipped'
|
||||
|
||||
# 真防重：唯一索引冲突不报错（并发/重放安全）
|
||||
cols = [c for c in record if c in tgt_cols]
|
||||
col_sql = ', '.join('`%s`' % c for c in cols)
|
||||
val_sql = ', '.join('${%s}$' % c for c in cols)
|
||||
dup_cols = [c for c in cols if c not in ('tenant_id', 'id', 'created_at',
|
||||
'source_event_id', 'evidence_type',
|
||||
'dedup_key')]
|
||||
sql = 'INSERT INTO `pbl_evidence` (%s) VALUES (%s)' % (col_sql, val_sql)
|
||||
if dup_cols:
|
||||
sql += ' ON DUPLICATE KEY UPDATE ' + ', '.join(
|
||||
'`%s`=VALUES(`%s`)' % (c, c) for c in dup_cols)
|
||||
sql += ', `updated_at`=VALUES(`updated_at`)'
|
||||
else:
|
||||
sql += ' ON DUPLICATE KEY UPDATE `updated_at`=VALUES(`updated_at`)'
|
||||
try:
|
||||
await q_exec(sql, {c: record[c] for c in cols})
|
||||
except Exception as exc:
|
||||
# 唯一键冲突在部分驱动下仍会抛：视为幂等命中，不算失败
|
||||
if _is_dup_key(exc):
|
||||
return 'skipped'
|
||||
raise
|
||||
new_id = await q_one(
|
||||
'SELECT `id` FROM `pbl_evidence` '
|
||||
'WHERE `tenant_id`=${tenant_id}$ AND `source_event_id`=${source_event_id}$ '
|
||||
'AND `evidence_type`=${evidence_type}$ LIMIT 1',
|
||||
{'tenant_id': tid, 'source_event_id': sid, 'evidence_type': etp})
|
||||
if new_id and new_id.get('id') is not None:
|
||||
record['id'] = new_id['id']
|
||||
return 'created'
|
||||
|
||||
|
||||
def _is_dup_key(exc):
|
||||
text = str(exc).lower()
|
||||
return 'duplicate entry' in text or '1062' in text or 'unique constraint' in text
|
||||
|
||||
|
||||
async def list_evidence(tenant_id, learner_id=None, artifact_id=None,
|
||||
evidence_type=None, session_id=None, blueprint_id=None,
|
||||
since=None, until=None, page=1, rows=20, sort='occurred_at',
|
||||
order='desc'):
|
||||
"""证据分页查询（全部条件强制带 tenant_id）。"""
|
||||
tid = (str(tenant_id).strip() if tenant_id else '')
|
||||
if not tid:
|
||||
raise CollectError('tenant_id 缺失：证据查询 fail-closed')
|
||||
cols = await table_columns('pbl_evidence')
|
||||
if not cols:
|
||||
raise CollectError('pbl_evidence 表不存在，无法查询')
|
||||
|
||||
where = {'tenant_id': tid}
|
||||
for key, val in (('learner_id', learner_id), ('artifact_id', artifact_id),
|
||||
('evidence_type', evidence_type), ('session_id', session_id),
|
||||
('blueprint_id', blueprint_id)):
|
||||
if val not in (None, ''):
|
||||
where[key] = val
|
||||
where_sql, args = build_where(where, cols)
|
||||
|
||||
sdt, edt = normalize_dt(since), normalize_dt(until)
|
||||
if sdt and 'occurred_at' in cols:
|
||||
where_sql += ' AND `occurred_at` >= ${since}$'
|
||||
args['since'] = sdt
|
||||
if edt and 'occurred_at' in cols:
|
||||
where_sql += ' AND `occurred_at` <= ${until}$'
|
||||
args['until'] = edt
|
||||
|
||||
sort_col = sort if sort in cols else 'occurred_at'
|
||||
direction = 'ASC' if str(order).lower() == 'asc' else 'DESC'
|
||||
page = max(1, int(page or 1))
|
||||
rows = min(500, max(1, int(rows or 20)))
|
||||
|
||||
total_row = await q_one('SELECT COUNT(*) AS c FROM `pbl_evidence` WHERE %s' % where_sql,
|
||||
dict(args))
|
||||
total = int((total_row or {}).get('c') or 0)
|
||||
|
||||
args = dict(args)
|
||||
args['offset'] = (page - 1) * rows
|
||||
args['rows'] = rows
|
||||
select_cols = [c for c in ('id', 'tenant_id', 'artifact_id', 'evidence_type',
|
||||
'source_event_id', 'session_id', 'learner_id',
|
||||
'blueprint_id', 'occurred_at', 'payload_json',
|
||||
'created_at') if c in cols]
|
||||
recs = await q_all(
|
||||
'SELECT %s FROM `pbl_evidence` WHERE %s ORDER BY `%s` %s LIMIT ${rows}$ OFFSET ${offset}$'
|
||||
% (', '.join('`%s`' % c for c in select_cols), where_sql, sort_col, direction), args)
|
||||
out = []
|
||||
for r in recs:
|
||||
item = {k: v for k, v in r.items()}
|
||||
if 'payload_json' in item:
|
||||
item['payload'] = json_loads(item.get('payload_json'), default={})
|
||||
try:
|
||||
item.pop('payload_json')
|
||||
except Exception:
|
||||
pass
|
||||
if 'evidence_type' in item:
|
||||
item['evidence_type_text'] = item.get('evidence_type')
|
||||
out.append(item)
|
||||
return {'rows': out, 'total': total, 'page': page, 'rows_per_page': rows}
|
||||
|
||||
|
||||
async def evidence_stats(tenant_id, learner_id=None, blueprint_id=None):
|
||||
"""证据类型分布统计（供 M6 评估 / 概览页）。"""
|
||||
tid = (str(tenant_id).strip() if tenant_id else '')
|
||||
if not tid:
|
||||
raise CollectError('tenant_id 缺失：统计 fail-closed')
|
||||
cols = await table_columns('pbl_evidence')
|
||||
where = {'tenant_id': tid}
|
||||
for key, val in (('learner_id', learner_id), ('blueprint_id', blueprint_id)):
|
||||
if val not in (None, ''):
|
||||
where[key] = val
|
||||
where_sql, args = build_where(where, cols)
|
||||
rows = await q_all(
|
||||
'SELECT `evidence_type`, COUNT(*) AS c, MAX(`occurred_at`) AS last_at '
|
||||
'FROM `pbl_evidence` WHERE %s GROUP BY `evidence_type` '
|
||||
'ORDER BY c DESC' % where_sql, args)
|
||||
total = sum(int(r.get('c') or 0) for r in rows)
|
||||
return {'total': total,
|
||||
'by_type': [{'evidence_type': r.get('evidence_type'),
|
||||
'count': int(r.get('c') or 0),
|
||||
'last_at': str(r.get('last_at') or '')} for r in rows]}
|
||||
|
||||
|
||||
async def watermark(tenant_id):
|
||||
"""已采集到的最大事件时间（增量采集游标，供 cron 轮询兜底）。"""
|
||||
tid = (str(tenant_id).strip() if tenant_id else '')
|
||||
if not tid:
|
||||
raise CollectError('tenant_id 缺失')
|
||||
cols = await table_columns('pbl_evidence')
|
||||
if 'occurred_at' not in cols:
|
||||
return {'last_occurred_at': None, 'count': 0}
|
||||
row = await q_one(
|
||||
'SELECT MAX(`occurred_at`) AS last_at, COUNT(*) AS c '
|
||||
'FROM `pbl_evidence` WHERE `tenant_id`=${tenant_id}$', {'tenant_id': tid})
|
||||
return {'last_occurred_at': str((row or {}).get('last_at') or '') or None,
|
||||
'count': int((row or {}).get('c') or 0)}
|
||||
252
pbl_evidence/db.py
Normal file
252
pbl_evidence/db.py
Normal file
@ -0,0 +1,252 @@
|
||||
"""pbl_evidence 数据访问适配层（M5a）。
|
||||
|
||||
铁律（见 skill/SKILL.md，QC 按同一规范审核）：
|
||||
1. **库名由宿主应用决定**：只用 ``ServerEnv().get_module_dbname('pbl_evidence')``，
|
||||
模块内绝不硬编码 DBNAME / 'pbl' / 'sage'。取不到即 fail-closed 抛错。
|
||||
2. **sqlor 只有 C/U/D/R/I/sqlExe**：本层统一走 ``sqlExe`` + ``${param}$`` 占位符，
|
||||
不编造 sor.save/list/query/one/insert/delete 等不存在的 API。
|
||||
3. **``return`` 不能在 ``async with`` 块内**：sqlor 上下文里 return 会静默变成
|
||||
NoneType，故一律「块内取数 → 块外 return」。
|
||||
4. **sqlExe 返回 DictObject**：``dict(row)`` 会抛 ValueError，统一经 ``row_to_dict``。
|
||||
5. **多租户隔离**：``tenant_id`` 缺失即 fail-closed，不查库。
|
||||
6. **列名不猜**：所有表列名先经 information_schema 发现（``table_columns``），
|
||||
再按交集拼 SQL，避免 Unknown column 类错误（历史高频缺陷）。
|
||||
"""
|
||||
|
||||
import importlib
|
||||
import json as _json
|
||||
|
||||
_MODULE_NAME = 'pbl_evidence'
|
||||
|
||||
_DBPOOLS = None
|
||||
_COL_CACHE = {}
|
||||
_TABLE_CACHE = {}
|
||||
|
||||
|
||||
def _log(msg):
|
||||
try:
|
||||
from appPublic.log import debug as _debug
|
||||
_debug('[pbl_evidence.db] %s' % (msg,))
|
||||
except Exception:
|
||||
print('[pbl_evidence.db] %s' % (msg,))
|
||||
|
||||
|
||||
def json_dumps(obj):
|
||||
"""安全序列化：datetime 转字符串，异常降级为空 JSON 对象。"""
|
||||
try:
|
||||
return _json.dumps(obj, ensure_ascii=False, default=str)
|
||||
except Exception:
|
||||
return '{}'
|
||||
|
||||
|
||||
def json_loads(text, default=None):
|
||||
"""安全反序列化：兼容 None / 空串 / 已是 dict / 非法 JSON。"""
|
||||
if text is None or text == '':
|
||||
return default if default is not None else {}
|
||||
if isinstance(text, (dict, list)):
|
||||
return text
|
||||
try:
|
||||
return _json.loads(text)
|
||||
except Exception:
|
||||
return default if default is not None else {}
|
||||
|
||||
|
||||
def get_dbname():
|
||||
"""取本模块库名（宿主应用决定，禁止硬编码）。"""
|
||||
try:
|
||||
from ahserver.serverenv import ServerEnv
|
||||
fn = getattr(ServerEnv(), 'get_module_dbname', None)
|
||||
if callable(fn):
|
||||
dbname = fn(_MODULE_NAME)
|
||||
if dbname:
|
||||
return dbname
|
||||
except Exception as exc:
|
||||
_log('get_module_dbname via ServerEnv failed: %s' % (exc,))
|
||||
raise RuntimeError(
|
||||
'宿主应用未在 ServerEnv 注册 get_module_dbname，'
|
||||
'pbl_evidence 无法确定库名（fail-closed，禁止回退硬编码库名）'
|
||||
)
|
||||
|
||||
|
||||
def _dbpools():
|
||||
"""延迟定位 sqlor DBPools（@SingletonDecorator，不可传参构造）。"""
|
||||
global _DBPOOLS
|
||||
if _DBPOOLS is not None:
|
||||
return _DBPOOLS
|
||||
try:
|
||||
from ahserver.serverenv import ServerEnv
|
||||
fn = getattr(ServerEnv(), 'DBPools', None)
|
||||
if callable(fn):
|
||||
_DBPOOLS = fn
|
||||
return fn
|
||||
except Exception:
|
||||
pass
|
||||
for mod, attr in (('sqlor', 'DBPools'), ('sqlor.dbPool', 'DBPools'),
|
||||
('sqlor.dbpool', 'DBPools'), ('sqlor.dbPools', 'DBPools'),
|
||||
('sqlor.DBPools', 'DBPools')):
|
||||
try:
|
||||
fn = getattr(importlib.import_module(mod), attr, None)
|
||||
if callable(fn):
|
||||
_DBPOOLS = fn
|
||||
return fn
|
||||
except Exception:
|
||||
continue
|
||||
raise RuntimeError('无法定位 sqlor DBPools（fail-closed）')
|
||||
|
||||
|
||||
def row_to_dict(row):
|
||||
"""DictObject → dict（``dict(r)`` 在 DictObject 上会抛 ValueError）。"""
|
||||
if row is None:
|
||||
return {}
|
||||
if isinstance(row, dict):
|
||||
return dict(row)
|
||||
for meth in ('toDict', 'todict', 'as_dict', 'asDict'):
|
||||
fn = getattr(row, meth, None)
|
||||
if callable(fn):
|
||||
try:
|
||||
return dict(fn())
|
||||
except Exception:
|
||||
continue
|
||||
keys = getattr(row, 'keys', None)
|
||||
if callable(keys):
|
||||
try:
|
||||
return {k: getattr(row, k) for k in list(keys())}
|
||||
except Exception:
|
||||
pass
|
||||
out = {}
|
||||
state = getattr(row, '__dict__', None)
|
||||
if isinstance(state, dict):
|
||||
for k, v in state.items():
|
||||
if not str(k).startswith('_'):
|
||||
out[k] = v
|
||||
return out
|
||||
|
||||
|
||||
async def q_all(sql, params=None):
|
||||
"""执行查询，返回 list[dict]。"""
|
||||
dbname = get_dbname()
|
||||
args = dict(params or {})
|
||||
async with _dbpools()().sqlorContext(dbname) as sor:
|
||||
rows = await sor.sqlExe(sql, args)
|
||||
return [row_to_dict(r) for r in (rows or [])]
|
||||
|
||||
|
||||
async def q_one(sql, params=None):
|
||||
"""执行查询，返回首行 dict 或 None。"""
|
||||
rows = await q_all(sql, params)
|
||||
return rows[0] if rows else None
|
||||
|
||||
|
||||
async def q_exec(sql, params=None):
|
||||
"""执行 DML（INSERT/UPDATE/DELETE），返回受影响行信息（不依赖其返回值）。"""
|
||||
dbname = get_dbname()
|
||||
args = dict(params or {})
|
||||
async with _dbpools()().sqlorContext(dbname) as sor:
|
||||
res = await sor.sqlExe(sql, args)
|
||||
return res
|
||||
|
||||
|
||||
async def table_exists(table):
|
||||
"""表是否存在（跨 schema 检查，用于 runtime_event 表名兼容）。"""
|
||||
key = table
|
||||
if key in _TABLE_CACHE:
|
||||
return _TABLE_CACHE[key]
|
||||
try:
|
||||
rows = await q_all(
|
||||
'SELECT COUNT(*) AS c FROM information_schema.TABLES '
|
||||
'WHERE TABLE_NAME=${t}$', {'t': table})
|
||||
cnt = 0
|
||||
if rows:
|
||||
cnt = int(list(rows[0].values())[0] or 0)
|
||||
except Exception as exc:
|
||||
_log('table_exists(%s) failed: %s' % (table, exc))
|
||||
cnt = 0
|
||||
_TABLE_CACHE[key] = cnt > 0
|
||||
return cnt > 0
|
||||
|
||||
|
||||
async def table_columns(table):
|
||||
"""发现真实列名集合（禁止凭记忆猜列名）。"""
|
||||
dbname = get_dbname()
|
||||
key = (dbname, table)
|
||||
if key in _COL_CACHE:
|
||||
return _COL_CACHE[key]
|
||||
cols = set()
|
||||
try:
|
||||
rows = await q_all(
|
||||
'SELECT COLUMN_NAME AS col FROM information_schema.COLUMNS '
|
||||
'WHERE TABLE_SCHEMA=${db}$ AND TABLE_NAME=${t}$',
|
||||
{'db': dbname, 't': table})
|
||||
for r in rows:
|
||||
name = r.get('col') or r.get('COLUMN_NAME') or r.get('column_name')
|
||||
if name:
|
||||
cols.add(str(name))
|
||||
except Exception as exc:
|
||||
_log('table_columns(%s) failed: %s' % (table, exc))
|
||||
_COL_CACHE[key] = cols
|
||||
return cols
|
||||
|
||||
|
||||
def pick_columns(data, available):
|
||||
"""按真实列过滤入参字典（同时挡掉 _text 后缀等前端附带字段）。"""
|
||||
out = {}
|
||||
for k, v in (data or {}).items():
|
||||
if k not in available:
|
||||
continue
|
||||
if str(k).endswith('_text'):
|
||||
continue
|
||||
out[k] = v
|
||||
return out
|
||||
|
||||
|
||||
def build_insert(table, data, available):
|
||||
"""构造 INSERT（显式列名，禁止 positional VALUES）。"""
|
||||
cols = [c for c in data if c in available]
|
||||
if not cols:
|
||||
raise RuntimeError('build_insert(%s): 无可写入列' % (table,))
|
||||
col_sql = ', '.join('`%s`' % c for c in cols)
|
||||
val_sql = ', '.join('${%s}$' % c for c in cols)
|
||||
sql = 'INSERT INTO `%s` (%s) VALUES (%s)' % (table, col_sql, val_sql)
|
||||
return sql, {c: data[c] for c in cols}
|
||||
|
||||
|
||||
def build_update(table, data, available, where):
|
||||
"""构造 UPDATE（WHERE 条件必须含 tenant_id，由调用方保证）。"""
|
||||
cols = [c for c in data if c in available]
|
||||
if not cols:
|
||||
raise RuntimeError('build_update(%s): 无可更新列' % (table,))
|
||||
set_sql = ', '.join('`%s`=${s_%s}$' % (c, c) for c in cols)
|
||||
conds, args = [], {}
|
||||
for c in cols:
|
||||
args['s_%s' % c] = data[c]
|
||||
for k, v in (where or {}).items():
|
||||
if k in available:
|
||||
conds.append('`%s`=${w_%s}$' % (k, k))
|
||||
args['w_%s' % k] = v
|
||||
sql = 'UPDATE `%s` SET %s WHERE %s' % (
|
||||
table, set_sql, ' AND '.join(conds) if conds else '1=1')
|
||||
return sql, args
|
||||
|
||||
|
||||
def build_where(where, available):
|
||||
"""构造 WHERE 片段（None 值转 IS NULL），返回 (sql, params)。"""
|
||||
conds, args = [], {}
|
||||
for k, v in (where or {}).items():
|
||||
if k not in available:
|
||||
continue
|
||||
if v is None or v == '':
|
||||
continue
|
||||
conds.append('`%s`=${%s}$' % (k, k))
|
||||
args[k] = v
|
||||
return (' AND '.join(conds) if conds else '1=1'), args
|
||||
|
||||
|
||||
def invalidate_cache(table=None):
|
||||
"""清列缓存（DDL 变更后调用）。"""
|
||||
global _COL_CACHE, _TABLE_CACHE
|
||||
if table is None:
|
||||
_COL_CACHE = {}
|
||||
_TABLE_CACHE = {}
|
||||
else:
|
||||
_COL_CACHE = {k: v for k, v in _COL_CACHE.items() if k[1] != table}
|
||||
_TABLE_CACHE = {k: v for k, v in _TABLE_CACHE.items() if k != table}
|
||||
239
pbl_evidence/evidence_map.py
Normal file
239
pbl_evidence/evidence_map.py
Normal file
@ -0,0 +1,239 @@
|
||||
"""事件 → 证据 的映射与规范化规则（M5a）。
|
||||
|
||||
来源：``scense_runtime`` M11b 落库的 ``pbl_runtime_event``（append-only、按月 RANGE 分区）。
|
||||
本文件是**纯函数层**（不碰数据库），因此可被 M5b 回放、M6 评估、单测直接复用。
|
||||
|
||||
设计要点：
|
||||
- 事件类型 → 证据类型是**配置化常量表**（不写死在 SQL / 不靠 LLM 猜），
|
||||
未知事件类型默认降级为 ``observation``（可观测证据），不丢事件。
|
||||
- ``dedup_key`` = md5(tenant_id|source_event_id|evidence_type)，32 位定长，
|
||||
与 ``pbl_evidence.dedup_key`` VARCHAR(32) 及唯一索引
|
||||
``uk_ev_dedup(tenant_id, source_event_id, evidence_type)`` 严格对齐。
|
||||
- 时间统一格式化为 ``YYYY-MM-DD HH:MM:SS`` 字符串，避免 datetime/str 混用
|
||||
导致 sqlor 占位符绑定失败。
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
from datetime import datetime, date
|
||||
|
||||
# ── 证据类型（与 appcodes_kv.pbl_evidence_type 对齐）─────────────────────
|
||||
EV_ARTIFACT = 'artifact' # 产出物（作品/交付件/提交）
|
||||
EV_ASSESSMENT = 'assessment' # 测评/作答证据
|
||||
EV_PEER_REVIEW = 'peer_review' # 互评证据
|
||||
EV_REFLECTION = 'reflection' # 反思/日志证据
|
||||
EV_MILESTONE = 'milestone' # 里程碑/目标达成证据
|
||||
EV_COLLABORATION = 'collaboration' # 协作证据
|
||||
EV_RESOURCE = 'resource' # 资源使用证据
|
||||
EV_OBSERVATION = 'observation' # 兜底：可观测行为证据
|
||||
|
||||
EVIDENCE_TYPES = (
|
||||
EV_ARTIFACT, EV_ASSESSMENT, EV_PEER_REVIEW, EV_REFLECTION,
|
||||
EV_MILESTONE, EV_COLLABORATION, EV_RESOURCE, EV_OBSERVATION,
|
||||
)
|
||||
|
||||
EVIDENCE_TYPE_LABELS = {
|
||||
EV_ARTIFACT: '产出物',
|
||||
EV_ASSESSMENT: '测评作答',
|
||||
EV_PEER_REVIEW: '同伴互评',
|
||||
EV_REFLECTION: '反思日志',
|
||||
EV_MILESTONE: '里程碑达成',
|
||||
EV_COLLABORATION: '协作行为',
|
||||
EV_RESOURCE: '资源使用',
|
||||
EV_OBSERVATION: '一般观测',
|
||||
}
|
||||
|
||||
# ── runtime_event.event_type → evidence_type ────────────────────────────
|
||||
# 覆盖 M11b 广播的事件族；同义词一并列出，避免事件命名漂移导致证据漏采。
|
||||
EVENT_TO_EVIDENCE = {
|
||||
# 产出物
|
||||
'artifact_created': EV_ARTIFACT,
|
||||
'artifact_submitted': EV_ARTIFACT,
|
||||
'artifact_updated': EV_ARTIFACT,
|
||||
'artifact_version_created': EV_ARTIFACT,
|
||||
'submission_created': EV_ARTIFACT,
|
||||
'deliverable_submitted': EV_ARTIFACT,
|
||||
'scene_saved': EV_ARTIFACT,
|
||||
'world_snapshot_created': EV_ARTIFACT,
|
||||
'script_compiled': EV_ARTIFACT,
|
||||
# 测评作答
|
||||
'quiz_answered': EV_ASSESSMENT,
|
||||
'question_answered': EV_ASSESSMENT,
|
||||
'assessment_submitted': EV_ASSESSMENT,
|
||||
'test_submitted': EV_ASSESSMENT,
|
||||
'task_checked': EV_ASSESSMENT,
|
||||
'checkpoint_passed': EV_ASSESSMENT,
|
||||
'checkpoint_failed': EV_ASSESSMENT,
|
||||
# 互评
|
||||
'peer_review_submitted': EV_PEER_REVIEW,
|
||||
'peer_review_created': EV_PEER_REVIEW,
|
||||
'review_commented': EV_PEER_REVIEW,
|
||||
# 反思
|
||||
'reflection_submitted': EV_REFLECTION,
|
||||
'journal_written': EV_REFLECTION,
|
||||
'self_assessment': EV_REFLECTION,
|
||||
# 里程碑
|
||||
'milestone_reached': EV_MILESTONE,
|
||||
'goal_completed': EV_MILESTONE,
|
||||
'objective_completed': EV_MILESTONE,
|
||||
'task_completed': EV_MILESTONE,
|
||||
'stage_completed': EV_MILESTONE,
|
||||
'badge_awarded': EV_MILESTONE,
|
||||
# 协作
|
||||
'team_message_sent': EV_COLLABORATION,
|
||||
'collaboration_joined': EV_COLLABORATION,
|
||||
'member_mentioned': EV_COLLABORATION,
|
||||
'task_assigned': EV_COLLABORATION,
|
||||
'role_accepted': EV_COLLABORATION,
|
||||
# 资源
|
||||
'resource_viewed': EV_RESOURCE,
|
||||
'resource_downloaded': EV_RESOURCE,
|
||||
'material_opened': EV_RESOURCE,
|
||||
'hint_used': EV_RESOURCE,
|
||||
}
|
||||
|
||||
# 明确不落证据的事件（心跳/渲染/纯 UI 噪声），白名单外置于此便于审阅。
|
||||
IGNORED_EVENTS = frozenset((
|
||||
'heartbeat', 'ping', 'presence_update', 'ui_render',
|
||||
'camera_moved', 'cursor_moved', 'audio_level',
|
||||
))
|
||||
|
||||
# 事件行里可能承载「事件主键」的候选列（M11b 表结构兼容层）
|
||||
EVENT_ID_FIELDS = ('event_id', 'source_event_id', 'uuid', 'id')
|
||||
# 事件行里可能承载「事件类型」的候选列
|
||||
EVENT_TYPE_FIELDS = ('event_type', 'type', 'event_code', 'name')
|
||||
# 事件行里可能承载「事件体」的候选列
|
||||
EVENT_PAYLOAD_FIELDS = ('payload_json', 'event_payload', 'payload', 'data_json', 'detail_json')
|
||||
# 事件行里可能承载「发生时间」的候选列
|
||||
EVENT_TIME_FIELDS = ('occurred_at', 'event_time', 'created_at', 'ts')
|
||||
# 学习者/会话/蓝图 的候选列
|
||||
EVENT_LEARNER_FIELDS = ('learner_id', 'user_id', 'student_id', 'actor_id', 'member_id')
|
||||
EVENT_SESSION_FIELDS = ('session_id', 'game_session_id', 'runtime_session_id')
|
||||
EVENT_BLUEPRINT_FIELDS = ('blueprint_id', 'pbl_blueprint_id', 'bp_id')
|
||||
|
||||
_DT_FORMATS = (
|
||||
'%Y-%m-%d %H:%M:%S', '%Y-%m-%dT%H:%M:%S', '%Y-%m-%d %H:%M:%S.%f',
|
||||
'%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%d/%H/%M/%S', '%Y-%m-%d',
|
||||
)
|
||||
|
||||
|
||||
def normalize_dt(value):
|
||||
"""任意时间输入 → 'YYYY-MM-DD HH:MM:SS'；无法解析返回 None。"""
|
||||
if value is None or value == '':
|
||||
return None
|
||||
if isinstance(value, datetime):
|
||||
return value.strftime('%Y-%m-%d %H:%M:%S')
|
||||
if isinstance(value, date):
|
||||
return datetime(value.year, value.month, value.day).strftime('%Y-%m-%d %H:%M:%S')
|
||||
text = str(value).strip()
|
||||
if not text:
|
||||
return None
|
||||
if text.endswith('Z'):
|
||||
text = text[:-1]
|
||||
text = text.replace('T', ' ')
|
||||
if '.' in text:
|
||||
head = text.split('.', 1)[0]
|
||||
else:
|
||||
head = text
|
||||
for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%d %H:%M', '%Y-%m-%d'):
|
||||
try:
|
||||
return datetime.strptime(head, fmt).strftime('%Y-%m-%d %H:%M:%S')
|
||||
except Exception:
|
||||
continue
|
||||
for fmt in _DT_FORMATS:
|
||||
try:
|
||||
return datetime.strptime(text, fmt).strftime('%Y-%m-%d %H:%M:%S')
|
||||
except Exception:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
def now_str():
|
||||
return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||||
|
||||
|
||||
def first_present(row, candidates):
|
||||
"""按候选列名顺序取第一个非空值（列名兼容层，不猜列）。"""
|
||||
for key in candidates:
|
||||
if key in row and row[key] not in (None, ''):
|
||||
return row[key]
|
||||
return None
|
||||
|
||||
|
||||
def map_event_to_evidence_type(event_type, payload=None):
|
||||
"""事件类型 → 证据类型；返回 None 表示该事件明确不采集。"""
|
||||
if not event_type:
|
||||
return EV_OBSERVATION
|
||||
key = str(event_type).strip().lower()
|
||||
if key in IGNORED_EVENTS:
|
||||
return None
|
||||
if key in EVENT_TO_EVIDENCE:
|
||||
return EVENT_TO_EVIDENCE[key]
|
||||
# 前缀/关键字启发（事件命名漂移兜底），未知仍降级 observation
|
||||
rules = (
|
||||
('submit', EV_ARTIFACT), ('artifact', EV_ARTIFACT), ('save', EV_ARTIFACT),
|
||||
('answer', EV_ASSESSMENT), ('quiz', EV_ASSESSMENT), ('check', EV_ASSESSMENT),
|
||||
('review', EV_PEER_REVIEW), ('reflect', EV_REFLECTION), ('journal', EV_REFLECTION),
|
||||
('milestone', EV_MILESTONE), ('complete', EV_MILESTONE), ('badge', EV_MILESTONE),
|
||||
('team', EV_COLLABORATION), ('collab', EV_COLLABORATION), ('mention', EV_COLLABORATION),
|
||||
('resource', EV_RESOURCE), ('view', EV_RESOURCE), ('open', EV_RESOURCE),
|
||||
)
|
||||
for token, ev_type in rules:
|
||||
if token in key:
|
||||
return ev_type
|
||||
return EV_OBSERVATION
|
||||
|
||||
|
||||
def evidence_types_for_events():
|
||||
"""反向索引：证据类型 → 事件类型列表（供按证据类型过滤采集）。"""
|
||||
out = {t: [] for t in EVIDENCE_TYPES}
|
||||
for ev, tp in EVENT_TO_EVIDENCE.items():
|
||||
out.setdefault(tp, []).append(ev)
|
||||
for tp in out:
|
||||
out[tp].sort()
|
||||
return out
|
||||
|
||||
|
||||
def build_dedup_key(tenant_id, source_event_id, evidence_type):
|
||||
"""32 位定长去重键，与 uk_ev_dedup 唯一索引语义一致。"""
|
||||
raw = '%s|%s|%s' % (tenant_id, source_event_id, evidence_type)
|
||||
return hashlib.md5(raw.encode('utf-8')).hexdigest()
|
||||
|
||||
|
||||
def event_source_id(row):
|
||||
"""从事件行提取 source_event_id（字符串化，最长 32）。"""
|
||||
val = first_present(row, EVENT_ID_FIELDS)
|
||||
if val is None:
|
||||
return None
|
||||
text = str(val).strip()
|
||||
return text[:32] if text else None
|
||||
|
||||
|
||||
def summarize_payload(row, event_type, evidence_type):
|
||||
"""构造证据体 payload_json（保留溯源信息 + 关键指标，不整表复制）。"""
|
||||
from .db import json_loads
|
||||
raw = first_present(row, EVENT_PAYLOAD_FIELDS)
|
||||
payload = json_loads(raw, default={})
|
||||
if not isinstance(payload, dict):
|
||||
payload = {'raw': payload}
|
||||
out = {
|
||||
'source': 'pbl_runtime_event',
|
||||
'event_type': str(event_type or ''),
|
||||
'evidence_type': evidence_type,
|
||||
'keys': {},
|
||||
}
|
||||
for k in ('artifact_id', 'task_id', 'blueprint_version_id', 'scene_id',
|
||||
'score', 'result', 'status', 'duration_ms', 'attempt_no',
|
||||
'title', 'name', 'content_ref', 'url', 'file_id'):
|
||||
if k in row and row[k] not in (None, ''):
|
||||
out['keys'][k] = row[k]
|
||||
for k, v in payload.items():
|
||||
if k in out['keys']:
|
||||
continue
|
||||
if isinstance(v, (str, int, float, bool)) or v is None:
|
||||
text = str(v)
|
||||
out['keys'][k] = text[:512] if isinstance(v, str) else v
|
||||
body = {k: v for k, v in payload.items()
|
||||
if isinstance(v, (dict, list))}
|
||||
if body:
|
||||
out['detail'] = body
|
||||
return out
|
||||
@ -2,8 +2,14 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""`load_pbl_evidence()` —— pbl_evidence 模块唯一挂载入口。
|
||||
|
||||
注册三处同步之 ③：env.<契约名> = <契约名>（① 定义在 api.py，② 导出在 __init__.py）。
|
||||
注册三处同步之 ③：``env.<契约名> = <契约名>``
|
||||
（① 定义在 api.py，② 导出在 __init__.py，③ 本文件挂 ServerEnv）。
|
||||
漏本文件注册 → .dspy 调用 NameError。
|
||||
|
||||
宿主应用（apps/pbls）在 ``init()`` 中按拓扑序调用本函数；库名由宿主
|
||||
``get_module_dbname('pbl_evidence')`` 决定，本模块不硬编码。
|
||||
"""
|
||||
|
||||
from ahserver.serverenv import ServerEnv
|
||||
|
||||
from pbl_evidence.api import (
|
||||
@ -13,19 +19,32 @@ from pbl_evidence.api import (
|
||||
pbl_artifact_delete,
|
||||
pbl_artifact_list,
|
||||
pbl_evidence_collect,
|
||||
pbl_evidence_collect_from_events,
|
||||
pbl_evidence_list,
|
||||
|
||||
pbl_evidence_stats,
|
||||
pbl_evidence_watermark,
|
||||
)
|
||||
from pbl_evidence.collector import collect_evidence_from_events, resolve_event_table
|
||||
from pbl_evidence.evidence_map import EVIDENCE_TYPES, EVENT_TO_EVIDENCE
|
||||
|
||||
|
||||
def load_pbl_evidence():
|
||||
env = ServerEnv()
|
||||
# ── 产出物 CRUD ──────────────────────────────────────────────
|
||||
env.pbl_artifact_create = pbl_artifact_create
|
||||
env.pbl_artifact_read = pbl_artifact_read
|
||||
env.pbl_artifact_update = pbl_artifact_update
|
||||
env.pbl_artifact_delete = pbl_artifact_delete
|
||||
env.pbl_artifact_list = pbl_artifact_list
|
||||
# ── 证据采集 / 查询（M5a）────────────────────────────────────
|
||||
env.pbl_evidence_collect = pbl_evidence_collect
|
||||
env.pbl_evidence_collect_from_events = pbl_evidence_collect_from_events
|
||||
env.pbl_evidence_list = pbl_evidence_list
|
||||
|
||||
env.pbl_evidence_stats = pbl_evidence_stats
|
||||
env.pbl_evidence_watermark = pbl_evidence_watermark
|
||||
# ── 底层能力（供 M5b 回放 / M6 评估直接复用）──────────────────
|
||||
env.pbl_collect_evidence_from_events = collect_evidence_from_events
|
||||
env.pbl_resolve_event_table = resolve_event_table
|
||||
env.pbl_evidence_types = list(EVIDENCE_TYPES)
|
||||
env.pbl_event_to_evidence_map = dict(EVENT_TO_EVIDENCE)
|
||||
return 'pbl_evidence'
|
||||
|
||||
37
wwwroot/api/pbl_evidence_collect_from_events.dspy
Normal file
37
wwwroot/api/pbl_evidence_collect_from_events.dspy
Normal file
@ -0,0 +1,37 @@
|
||||
# pbl_evidence/api/pbl_evidence_collect_from_events.dspy
|
||||
# M5a 主能力：批量从 pbl_runtime_event（M11b 产物）采集学习证据，幂等落 pbl_evidence。
|
||||
# 铁律：dspy 内不 import、不 new ServerEnv()；ServerEnv 注册的契约函数是直接全局；必须显式 return。
|
||||
debug(f'pbl_evidence_collect_from_events.dspy: START params_kw={dict(params_kw)}')
|
||||
|
||||
try:
|
||||
tenant = params_kw.get('tenant_id') or (await get_userorgid()) or '0'
|
||||
|
||||
def _as_list(val):
|
||||
if val is None or val == '':
|
||||
return None
|
||||
if isinstance(val, (list, tuple)):
|
||||
return [str(x).strip() for x in val if str(x).strip()]
|
||||
return [x.strip() for x in str(val).split(',') if x.strip()]
|
||||
|
||||
res = await pbl_evidence_collect_from_events(
|
||||
tenant_id=tenant,
|
||||
since=params_kw.get('since'),
|
||||
until=params_kw.get('until'),
|
||||
session_id=params_kw.get('session_id'),
|
||||
learner_id=params_kw.get('learner_id'),
|
||||
blueprint_id=params_kw.get('blueprint_id'),
|
||||
event_types=_as_list(params_kw.get('event_types')),
|
||||
evidence_types=_as_list(params_kw.get('evidence_types')),
|
||||
limit=params_kw.get('limit') or 500,
|
||||
dry_run=params_kw.get('dry_run'),
|
||||
update_existing=params_kw.get('update_existing'),
|
||||
)
|
||||
status = 'OK' if res.get('ok') else 'ERROR'
|
||||
debug(f'pbl_evidence_collect_from_events.dspy: DONE status={status} msg={res.get("message")}')
|
||||
return {'status': status,
|
||||
'data': {k: v for k, v in res.items() if k != 'items'},
|
||||
'items': res.get('items') or [],
|
||||
'message': res.get('message') or ''}
|
||||
except Exception as e:
|
||||
error(f'pbl_evidence_collect_from_events.dspy: FAIL {format_exc()}')
|
||||
return {'status': 'ERROR', 'data': None, 'message': str(e)}
|
||||
17
wwwroot/api/pbl_evidence_stats.dspy
Normal file
17
wwwroot/api/pbl_evidence_stats.dspy
Normal file
@ -0,0 +1,17 @@
|
||||
# pbl_evidence/api/pbl_evidence_stats.dspy
|
||||
# 证据类型分布统计（供 M6 评估 / 概览卡片）。dspy 无 import、无 ServerEnv()、显式 return。
|
||||
debug(f'pbl_evidence_stats.dspy: START params_kw={dict(params_kw)}')
|
||||
|
||||
try:
|
||||
tenant = params_kw.get('tenant_id') or (await get_userorgid()) or '0'
|
||||
res = await pbl_evidence_stats(
|
||||
tenant_id=tenant,
|
||||
learner_id=params_kw.get('learner_id'),
|
||||
blueprint_id=params_kw.get('blueprint_id'),
|
||||
)
|
||||
status = 'OK' if res.get('ok') else 'ERROR'
|
||||
debug(f'pbl_evidence_stats.dspy: DONE total={res.get("total")}')
|
||||
return {'status': status, 'data': res, 'message': res.get('message') or ''}
|
||||
except Exception as e:
|
||||
error(f'pbl_evidence_stats.dspy: FAIL {format_exc()}')
|
||||
return {'status': 'ERROR', 'data': None, 'message': str(e)}
|
||||
Loading…
x
Reference in New Issue
Block a user