diff --git a/models/pbl_artifact.json b/models/pbl_artifact.json index c98d45d..bac407d 100644 --- a/models/pbl_artifact.json +++ b/models/pbl_artifact.json @@ -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"} ] } diff --git a/models/pbl_evidence.json b/models/pbl_evidence.json index d159644..1162708 100644 --- a/models/pbl_evidence.json +++ b/models/pbl_evidence.json @@ -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"} ] } diff --git a/pbl_evidence/__init__.py b/pbl_evidence/__init__.py index da9cd67..43ccd96 100644 --- a/pbl_evidence/__init__.py +++ b/pbl_evidence/__init__.py @@ -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' diff --git a/pbl_evidence/api.py b/pbl_evidence/api.py index 8164c3e..201e502 100644 --- a/pbl_evidence/api.py +++ b/pbl_evidence/api.py @@ -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 diff --git a/pbl_evidence/collector.py b/pbl_evidence/collector.py new file mode 100644 index 0000000..8b51fe2 --- /dev/null +++ b/pbl_evidence/collector.py @@ -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)} diff --git a/pbl_evidence/db.py b/pbl_evidence/db.py new file mode 100644 index 0000000..ce8b94b --- /dev/null +++ b/pbl_evidence/db.py @@ -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} diff --git a/pbl_evidence/evidence_map.py b/pbl_evidence/evidence_map.py new file mode 100644 index 0000000..a3c05d9 --- /dev/null +++ b/pbl_evidence/evidence_map.py @@ -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 diff --git a/pbl_evidence/init.py b/pbl_evidence/init.py index 0a0b7d9..0f4481c 100644 --- a/pbl_evidence/init.py +++ b/pbl_evidence/init.py @@ -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' diff --git a/wwwroot/api/pbl_evidence_collect_from_events.dspy b/wwwroot/api/pbl_evidence_collect_from_events.dspy new file mode 100644 index 0000000..d7474ed --- /dev/null +++ b/wwwroot/api/pbl_evidence_collect_from_events.dspy @@ -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)} diff --git a/wwwroot/api/pbl_evidence_stats.dspy b/wwwroot/api/pbl_evidence_stats.dspy new file mode 100644 index 0000000..310f10e --- /dev/null +++ b/wwwroot/api/pbl_evidence_stats.dspy @@ -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)}