391 lines
19 KiB
Python
391 lines
19 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""pbl_evidence.api —— 产出物 CRUD + 证据幂等采集（第18/19章，M5a/M5b）。
|
||
|
||
幂等硬线：``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
|
||
|
||
from pbl_common.api import PblError, actor_id, json_dump, now_str, tenant_id
|
||
|
||
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 _require_tenant():
|
||
tid = tenant_id()
|
||
if not tid:
|
||
raise PblError('PBL_E_TENANT_MISSING',
|
||
'tenant_id 缺失：pbl_evidence 所有读写强制带租户（fail-closed）')
|
||
return str(tid)
|
||
|
||
|
||
def _blank(value):
|
||
return value is None or (isinstance(value, str) and value.strip() == '')
|
||
|
||
|
||
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):
|
||
"""新建产出物：自动生成 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):
|
||
"""按 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):
|
||
"""更新产出物：改 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):
|
||
"""删除产出物：先级联删其证据（避免孤儿证据污染 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):
|
||
"""产出物分页列表（强制租户；支持 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):
|
||
"""单事件幂等落证据（服务端权威；客户端不能直接写证据表）。
|
||
|
||
幂等：先按唯一索引三元组预查，命中即返回 ``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):
|
||
"""证据分页查询（强制租户；支持 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:
|
||
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
|