From 1d48d2c01211142c5602b079cbf91dc3c3c97611 Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Tue, 22 Sep 2026 14:13:03 +0800 Subject: [PATCH] =?UTF-8?q?deliver:=20=E4=BA=A4=E4=BB=98=E6=94=B6=E5=8F=A3?= =?UTF-8?q?=EF=BC=88=E5=BC=95=E6=93=8E=E4=BB=A3=E4=B8=BA=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pbl_evidence/__init__.py | 13 + pbl_evidence/aggregation.py | 569 ++++++++++++++++++ pbl_evidence/init.py | 12 + scripts/load_path.py | 6 + scripts/selftest_m5b.py | 408 +++++++++++++ wwwroot/api/pbl_evidence_aggregate.dspy | 20 + .../api/pbl_evidence_collect_idempotent.dspy | 15 + wwwroot/api/pbl_evidence_query.dspy | 21 + wwwroot/api/pbl_evidence_subject_summary.dspy | 22 + wwwroot/index.ui | 114 +++- 10 files changed, 1199 insertions(+), 1 deletion(-) create mode 100644 pbl_evidence/aggregation.py create mode 100644 scripts/selftest_m5b.py create mode 100644 wwwroot/api/pbl_evidence_aggregate.dspy create mode 100644 wwwroot/api/pbl_evidence_collect_idempotent.dspy create mode 100644 wwwroot/api/pbl_evidence_query.dspy create mode 100644 wwwroot/api/pbl_evidence_subject_summary.dspy diff --git a/pbl_evidence/__init__.py b/pbl_evidence/__init__.py index 72af496..8bcfa19 100644 --- a/pbl_evidence/__init__.py +++ b/pbl_evidence/__init__.py @@ -28,6 +28,15 @@ from pbl_evidence.crud_api import ( pbl_evidence_update, pbl_evidence_delete, ) +from pbl_evidence.aggregation import ( + pbl_evidence_collect_idempotent, + pbl_evidence_aggregate, + pbl_evidence_query, + pbl_evidence_subject_summary, + resolve_idempotency_key, + apply_default_filters, + aggregate_rows, +) from pbl_evidence.evidence_map import ( EVIDENCE_TYPES, EVENT_TO_EVIDENCE, @@ -43,6 +52,10 @@ __all__ = [ 'EVIDENCE_TYPES', 'EVENT_TO_EVIDENCE', 'evidence_types_for_events', # CRUD 框架适配端点后端（json/pbl_evidence.json 的 editable + alters.dataurl） 'pbl_artifact_options', 'pbl_evidence_update', 'pbl_evidence_delete', + + 'pbl_evidence_collect_idempotent', 'pbl_evidence_aggregate', + 'pbl_evidence_query', 'pbl_evidence_subject_summary', + 'resolve_idempotency_key', 'apply_default_filters', 'aggregate_rows', ] __version__ = '1.2.0' diff --git a/pbl_evidence/aggregation.py b/pbl_evidence/aggregation.py new file mode 100644 index 0000000..ea8e37a --- /dev/null +++ b/pbl_evidence/aggregation.py @@ -0,0 +1,569 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""pbl_evidence.aggregation —— M5b 证据聚合与查询契约（第 18 章 / US-17 / Q-OPEN-10）。 + +本文件是 **M5b 增量**，与 M5a（api.py：单事件幂等落库 + 批量采集）分层： + +| M5b 需求点 | 实现位置 | +|---|---| +| 四类证据（decision/action/observation/artifact_version）幂等采集契约对齐 | ``pbl_evidence_collect_idempotent``（包装 api.pbl_evidence_collect，补 idempotency_key 归一与 created 布尔出参） | +| 证据聚合（按 learner/blueprint/session/type 分组计数 + 四类覆盖度 + 水位） | ``pbl_evidence_aggregate`` + 纯函数 ``aggregate_rows`` | +| 证据查询契约（分页 + 时间范围 + 多类型 + 总数 + 水位） | ``pbl_evidence_query`` | +| 学生 / 团队证据档案（list_evidence_by_student / by_team 等价） | ``pbl_evidence_subject_summary`` | +| **Q-OPEN-10** 预览 / Playtest 数据默认过滤 STUDENT | 纯函数 ``is_preview_row`` / ``is_student_private`` / ``apply_default_filters``，被上面三个查询入口统一调用 | + +设计口径（docs/01-design/modules/pbl_evidence.md §3.2）： + +* 幂等：``idempotency_key`` 租户内唯一，落到 ``pbl_evidence.dedup_key``，唯一索引 + ``uk_ev_dedup(tenant_id, source_event_id, evidence_type)`` 兜并发竞态；重复采集 + **不报错**，返回已有记录 + ``created=False``（同时保留 M5a 的 ``deduped=True``）。 +* 只读：聚合与查询全部 SELECT，本模块**无任何 research 层写入路径**（Q6 硬约束）。 +* 租户：所有读写 ``tenant_id`` 强制打头，缺失即 fail-closed（``PBL_E_TENANT_MISSING``）。 +* 库名：经 ``db.get_dbname()`` → ``ServerEnv().get_module_dbname('pbl_evidence')``，不硬编码。 + +Q-OPEN-10 判定依据（表无独立 audience 列，故按可持久化证据判定）： + +1. **payload 预览标记**：``pbl_preview`` / ``preview`` / ``is_preview`` / ``playtest`` / + ``is_playtest`` / ``demo`` / ``is_demo`` / ``env_mode`` / ``mode`` 命中 preview/playtest/ + demo/trial/true/1 族； +2. **预览会话**：调用方传 ``preview_session_ids`` 显式声明哪些 session 属预览环境； +3. **学生私有可见性**：payload 的 ``visibility`` / ``audience`` / ``data_scope`` / + ``visible_to`` 为 STUDENT 族（学生侧私有，教师统计与正式分析默认不计入）。 + +默认（``include_preview`` 缺省 False）三类全部剔除；显式传 ``include_preview=True`` / +``include_student_private=True`` 才纳入。出参统一带 ``preview_filter`` 段回显本次扫描/ +剔除行数与命中的规则，保证口径可审计（不静默丢数）。 +""" + +import hashlib +import json + +from pbl_common.api import PblError, now_str, tenant_id + +from pbl_evidence.db import build_where, json_loads, q_all, q_one, table_columns +from pbl_evidence.evidence_map import EVIDENCE_TYPES, normalize_dt + +# ── 口径常量（集中定义，禁止散落在 SQL / 前端）───────────────────── +MAX_SCAN_ROWS = 5000 # 行级聚合扫描上限（超出置 truncated=True，不静默丢数） +DEFAULT_PAGE_ROWS = 20 +MAX_PAGE_ROWS = 200 + +PREVIEW_PAYLOAD_KEYS = ('pbl_preview', 'preview', 'is_preview', 'playtest', + 'is_playtest', 'demo', 'is_demo', 'env_mode', 'mode') +PREVIEW_VALUES = ('preview', 'playtest', 'demo', 'student_preview', 'trial', + 'true', '1', 'yes', 'y') +VISIBILITY_KEYS = ('visibility', 'audience', 'data_scope', 'visible_to') +STUDENT_PRIVATE_VALUES = ('student', 'student_only', 'student_private', 'private') + +GROUP_BY_KEYS = { + 'learner': 'learner_id', 'learner_id': 'learner_id', 'student': 'learner_id', + 'blueprint': 'blueprint_id', 'blueprint_id': 'blueprint_id', + 'session': 'session_id', 'session_id': 'session_id', 'team': 'session_id', + 'artifact': 'artifact_id', 'artifact_id': 'artifact_id', + 'type': 'evidence_type', 'evidence_type': 'evidence_type', +} + +# 查询/聚合只取必要列（payload_json 必须取——Q-OPEN-10 判定依赖它） +SCAN_COLUMNS = ('id', 'evidence_type', 'source_event_id', 'learner_id', 'blueprint_id', + 'session_id', 'artifact_id', 'occurred_at', 'created_at', 'dedup_key', + 'payload_json') + + +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 _as_int(value, default=0): + try: + return int(value) + except Exception: + return default + + +def _truthy_preview_value(value): + """payload 值是否表示「这是预览 / Playtest 环境产生的数据」。""" + if isinstance(value, bool): + return value + if value is None: + return False + if isinstance(value, (int, float)): + return bool(value) + return str(value).strip().lower() in PREVIEW_VALUES + + +# ══════════════════════════════════════════════════════════════════════ +# 纯函数层（无 IO，可离线单测） +# ══════════════════════════════════════════════════════════════════════ +def resolve_idempotency_key(kw): + """归一幂等键。 + + 优先级：显式 ``idempotency_key`` → ``dedup_key`` → 由 + ``tenant|source_event_id|evidence_type`` 派生（与唯一索引三元组同口径， + 保证「同一事件 + 同一证据类型」永远得到同一个键，重放不产生第二条证据）。 + + 返回 ``(key, derived_bool)``。 + """ + explicit = kw.get('idempotency_key') or kw.get('dedup_key') + if not _blank(explicit): + return str(explicit).strip()[:64], False + seed = '%s|%s|%s' % (str(kw.get('tenant_id') or ''), + str(kw.get('source_event_id') or kw.get('event_id') or ''), + str(kw.get('evidence_type') or '')) + return 'IK' + hashlib.sha256(seed.encode('utf-8')).hexdigest()[:24], True + + +def row_payload(row): + """取行 payload dict（payload_json 可能是 str / dict / None / 坏 JSON）。""" + payload = row.get('payload_json') + if payload is None: + payload = row.get('payload') + if isinstance(payload, str): + try: + payload = json.loads(payload) + except Exception: + payload = {} + if not isinstance(payload, dict): + payload = {} + return payload + + +def is_preview_row(row, preview_session_ids=None): + """Q-OPEN-10 规则 1/2：payload 预览标记，或属于显式声明的预览会话。""" + payload = row_payload(row) + for key in PREVIEW_PAYLOAD_KEYS: + if key in payload and _truthy_preview_value(payload[key]): + return True, 'payload:%s' % key + sid = str(row.get('session_id') or '') + if preview_session_ids and sid: + declared = set(str(s) for s in preview_session_ids) + if sid in declared: + return True, 'preview_session:%s' % sid + return False, '' + + +def is_student_private(row): + """Q-OPEN-10 规则 3：STUDENT 私有可见性（教师统计 / 正式分析默认不计入）。""" + payload = row_payload(row) + for key in VISIBILITY_KEYS: + val = payload.get(key) + if val is None: + continue + if str(val).strip().lower() in STUDENT_PRIVATE_VALUES: + return True, 'payload:%s=%s' % (key, val) + return False, '' + + +def apply_default_filters(rows, kw): + """统一过滤入口（纯函数，可单测）。 + + kw: + include_preview（默认 False）/ include_student_private（默认 False） + preview_session_ids（list 或逗号串） + subject_type + subject_id（student|learner|team 主体过滤） + 返回 ``(kept_rows, filter_stats)``；filter_stats 供出参回显口径。 + """ + include_preview = bool(kw.get('include_preview') or kw.get('all_env') or False) + include_student = bool(kw.get('include_student_private') or False) + preview_sessions = kw.get('preview_session_ids') or [] + if isinstance(preview_sessions, str): + preview_sessions = [s.strip() for s in preview_sessions.split(',') if s.strip()] + preview_sessions = list(preview_sessions) + + subject_type = str(kw.get('subject_type') or '').strip().lower() + subject_id = kw.get('subject_id') + + stats = { + 'applied': False, + 'scanned': len(rows or []), + 'excluded_preview': 0, + 'excluded_student': 0, + 'excluded_subject': 0, + 'preview_rules': [], + 'include_preview': include_preview, + 'include_student_private': include_student, + 'preview_session_count': len(preview_sessions), + } + + def _note(rule): + stats['applied'] = True + if rule and rule not in stats['preview_rules']: + stats['preview_rules'].append(rule) + + kept = [] + for row in rows or []: + if not include_preview: + hit, rule = is_preview_row(row, preview_sessions) + if hit: + stats['excluded_preview'] += 1 + _note(rule) + continue + if not include_student: + hit, rule = is_student_private(row) + if hit: + stats['excluded_student'] += 1 + _note(rule) + continue + if subject_id and subject_type in ('student', 'learner'): + if str(row.get('learner_id') or '') != str(subject_id): + stats['excluded_subject'] += 1 + continue + if subject_id and subject_type == 'team': + if str(row.get('session_id') or '') != str(subject_id): + stats['excluded_subject'] += 1 + continue + kept.append(row) + stats['kept'] = len(kept) + return kept, stats + + +def aggregate_rows(rows, group_by='learner'): + """纯聚合（无 IO，可单测）：四类计数 + 覆盖度 + 分组 + 时间水位。""" + gkey = GROUP_BY_KEYS.get(str(group_by or 'learner').strip().lower(), 'learner_id') + by_type = {t: 0 for t in EVIDENCE_TYPES} + groups = {} + max_occurred, min_occurred, max_id = '', '', 0 + + for row in rows or []: + etype = str(row.get('evidence_type') or '') + if etype in by_type: + by_type[etype] += 1 + gval = str(row.get(gkey) if row.get(gkey) not in (None, '') else '0') + bucket = groups.get(gval) + if bucket is None: + bucket = {'group_key': gval, 'total': 0, + 'by_type': {t: 0 for t in EVIDENCE_TYPES}, + 'first_occurred_at': None, 'last_occurred_at': None, + 'blueprint_ids': set(), 'learner_ids': set()} + groups[gval] = bucket + bucket['total'] += 1 + if etype in bucket['by_type']: + bucket['by_type'][etype] += 1 + if row.get('blueprint_id') not in (None, ''): + bucket['blueprint_ids'].add(str(row['blueprint_id'])) + if row.get('learner_id') not in (None, ''): + bucket['learner_ids'].add(str(row['learner_id'])) + occ = str(row.get('occurred_at') or '') + if occ: + if not bucket['first_occurred_at'] or occ < bucket['first_occurred_at']: + bucket['first_occurred_at'] = occ + if not bucket['last_occurred_at'] or occ > bucket['last_occurred_at']: + bucket['last_occurred_at'] = occ + if not max_occurred or occ > max_occurred: + max_occurred = occ + if not min_occurred or occ < min_occurred: + min_occurred = occ + max_id = max(max_id, _as_int(row.get('id'))) + + group_list = [] + for bucket in groups.values(): + group_list.append({ + 'group_key': bucket['group_key'], + 'total': bucket['total'], + 'by_type': bucket['by_type'], + 'blueprint_ids': sorted(bucket['blueprint_ids']), + 'learner_count': len(bucket['learner_ids']), + 'first_occurred_at': bucket['first_occurred_at'], + 'last_occurred_at': bucket['last_occurred_at'], + 'type_coverage': sum(1 for t in EVIDENCE_TYPES if bucket['by_type'][t] > 0), + }) + group_list.sort(key=lambda x: (-x['total'], x['group_key'])) + + covered = sum(1 for t in EVIDENCE_TYPES if by_type[t] > 0) + return { + 'total': len(rows or []), + 'by_type': by_type, + 'coverage': { + 'covered_types': covered, + 'total_types': len(EVIDENCE_TYPES), + 'missing_types': [t for t in EVIDENCE_TYPES if by_type[t] == 0], + }, + 'groups': group_list, + 'group_by': gkey, + 'watermark': { + 'max_occurred_at': max_occurred or None, + 'min_occurred_at': min_occurred or None, + 'max_id': max_id or None, + 'count': len(rows or []), + }, + } + + +def _time_range_params(kw): + """支持 time_from/time_to、start_time/end_time，以及 time_range:{from,to} / [from,to]。""" + rng = kw.get('time_range') + tf = tt = None + if isinstance(rng, dict): + tf = rng.get('from') or rng.get('start') + tt = rng.get('to') or rng.get('end') + elif isinstance(rng, (list, tuple)) and len(rng) >= 2: + tf, tt = rng[0], rng[1] + tf = kw.get('time_from') or kw.get('start_time') or tf + tt = kw.get('time_to') or kw.get('end_time') or tt + out = {} + if not _blank(tf): + d = normalize_dt(tf) + if d: + out['time_from'] = d + if not _blank(tt): + d = normalize_dt(tt) + if d: + out['time_to'] = d + return out + + +def parse_evidence_types(kw): + """解析 evidence_types（list 或逗号串），校验必须属于四类。""" + types = kw.get('evidence_types') or kw.get('types') + if isinstance(types, str): + types = [t.strip() for t in types.split(',') if t.strip()] + if not types: + return [] + if not isinstance(types, (list, tuple)): + raise PblError('PBL_E_PARAM', 'evidence_types 须为数组或逗号分隔串') + bad = [t for t in types if t not in EVIDENCE_TYPES] + if bad: + raise PblError('PBL_E_VALIDATION', + 'evidence_types 须属于四类 %s，非法值：%s' + % ('/'.join(EVIDENCE_TYPES), bad)) + return list(types) + + +async def scan_evidence_rows(where, kw, limit=MAX_SCAN_ROWS, cols=None): + """行级扫描（只读，SELECT 指定列）。返回 ``(rows, truncated)``。""" + cols = cols or await table_columns('pbl_evidence') + if not cols: + raise PblError('PBL_E_DB_UNAVAILABLE', 'pbl_evidence 表不存在，请先执行 models DDL') + sel = [c for c in SCAN_COLUMNS if c in cols] or ['id'] + where_sql, args = build_where({k: v for k, v in (where or {}).items() + if v not in (None, '')}, cols) + + tr = _time_range_params(kw) + if 'time_from' in tr and 'occurred_at' in cols: + where_sql += ' AND `occurred_at` >= ${time_from}$' + args['time_from'] = tr['time_from'] + if 'time_to' in tr and 'occurred_at' in cols: + where_sql += ' AND `occurred_at` <= ${time_to}$' + args['time_to'] = tr['time_to'] + + types = parse_evidence_types(kw) + if types and 'evidence_type' in cols: + holders = [] + for idx, t in enumerate(types): + holders.append('${etype%s}$' % idx) + args['etype%s' % idx] = t + where_sql += ' AND `evidence_type` IN (%s)' % ', '.join(holders) + + limit = min(MAX_SCAN_ROWS, max(1, _as_int(limit, MAX_SCAN_ROWS))) + args['scan_limit'] = limit + order_col = 'occurred_at' if 'occurred_at' in cols else 'id' + sql = ('SELECT `%s` FROM `pbl_evidence` WHERE %s ORDER BY `%s` DESC, `id` DESC ' + 'LIMIT ${scan_limit}$' % ('`, `'.join(sel), where_sql, order_col)) + rows = await q_all(sql, args) + out = [dict(r) if not isinstance(r, dict) else r for r in (rows or [])] + return out, len(out) >= limit + + +# ══════════════════════════════════════════════════════════════════════ +# M5b 契约 ①：四类证据幂等采集（对齐设计 §3.2 出参 created） +# ══════════════════════════════════════════════════════════════════════ +async def pbl_evidence_collect_idempotent(**kw): + """四类证据幂等采集（M5b 契约包装，复用 M5a 落库实现，不重复写 SQL）。 + + 入参：``source_event_id``(必填) / ``evidence_type``(四类之一，缺省按事件类型映射) + / ``idempotency_key``(可空，空则按唯一索引三元组派生) / ``payload`` + / ``learner_id`` / ``blueprint_id`` / ``session_id`` / ``artifact_id`` + / ``occurred_at`` / ``event_type`` + 出参：``{ok, created:bool, deduped:bool, id, evidence_type, idempotency_key, + idempotency_key_derived, unique_index}`` + 重复采集不报错：``created=False``（等价 M5a 的 ``deduped=True``）。 + """ + from pbl_evidence.api import pbl_evidence_collect # 延迟导入避免循环依赖 + + tid = _require_tenant() + etype = str(kw.get('evidence_type') or '').strip() + if etype and etype not in EVIDENCE_TYPES: + raise PblError('PBL_E_VALIDATION', + 'evidence_type 须为四类 %s，收到：%s' + % ('/'.join(EVIDENCE_TYPES), etype)) + + key, derived = resolve_idempotency_key(dict(kw, tenant_id=tid)) + call_kw = dict(kw) + call_kw['dedup_key'] = key + res = await pbl_evidence_collect(**call_kw) + if not isinstance(res, dict): + raise PblError('PBL_E_DB', '证据采集返回异常结构：%r' % (type(res),)) + + deduped = bool(res.get('deduped')) + out = dict(res) + out.update({ + 'created': not deduped, + 'deduped': deduped, + 'idempotency_key': key, + 'idempotency_key_derived': derived, + 'evidence_type': etype or res.get('evidence_type'), + 'tenant_id': tid, + }) + return out + + +# ══════════════════════════════════════════════════════════════════════ +# M5b 契约 ②：证据聚合（四类覆盖度 + 分组 + 水位，Q-OPEN-10 默认过滤） +# ══════════════════════════════════════════════════════════════════════ +async def pbl_evidence_aggregate(**kw): + """证据聚合统计（只读；供 M6 评估 / M8 分析 / 教师看板复用）。 + + 入参：``blueprint_id`` / ``learner_id``(或 ``student_id``) / ``session_id`` + / ``artifact_id`` / ``evidence_types`` / ``group_by`` + (learner|blueprint|session|team|artifact|type) / ``time_from,time_to`` + 或 ``time_range`` / ``include_preview``(默认 False) + / ``include_student_private``(默认 False) / ``preview_session_ids`` + 出参：``{ok,total,by_type,coverage,groups,watermark,preview_filter,truncated}`` + """ + tid = _require_tenant() + cols = await table_columns('pbl_evidence') + where = {'tenant_id': tid} + for key in ('blueprint_id', 'session_id', 'artifact_id'): + if not _blank(kw.get(key)): + where[key] = kw[key] + learner = kw.get('learner_id') or kw.get('student_id') + if not _blank(learner): + where['learner_id'] = str(learner) + + limit = min(MAX_SCAN_ROWS, max(1, _as_int(kw.get('scan_limit'), MAX_SCAN_ROWS))) + rows, truncated = await scan_evidence_rows(where, kw, limit=limit, cols=cols) + kept, stats = apply_default_filters(rows, kw) + agg = aggregate_rows(kept, kw.get('group_by') or 'learner') + agg.update({ + 'ok': True, 'status': 'OK', 'truncated': bool(truncated), + 'preview_filter': stats, 'tenant_id': tid, 'generated_at': now_str(), + }) + return agg + + +# ══════════════════════════════════════════════════════════════════════ +# M5b 契约 ③：证据列表查询（分页 + 总数 + 水位，US-17 含 payload） +# ══════════════════════════════════════════════════════════════════════ +async def pbl_evidence_query(**kw): + """证据查询契约（分页列表 + 总数 + 类型分布 + 水位；预览默认过滤）。 + + 出参：``{ok,status,total,page,rows_per_page,rows,by_type,coverage, + watermark,preview_filter,truncated}`` + """ + tid = _require_tenant() + cols = await table_columns('pbl_evidence') + where = {'tenant_id': tid} + for key in ('blueprint_id', 'session_id', 'artifact_id', 'evidence_type'): + if not _blank(kw.get(key)): + where[key] = kw[key] + learner = kw.get('learner_id') or kw.get('student_id') + if not _blank(learner): + where['learner_id'] = str(learner) + if not _blank(kw.get('source_event_id')): + where['source_event_id'] = str(kw['source_event_id']).strip()[:32] + + limit = min(MAX_SCAN_ROWS, max(1, _as_int(kw.get('scan_limit'), MAX_SCAN_ROWS))) + rows, truncated = await scan_evidence_rows(where, kw, limit=limit, cols=cols) + kept, stats = apply_default_filters(rows, kw) + + page = max(1, _as_int(kw.get('page'), 1)) + rows_per_page = min(MAX_PAGE_ROWS, + max(1, _as_int(kw.get('rows') or kw.get('page_size'), + DEFAULT_PAGE_ROWS))) + start = (page - 1) * rows_per_page + window = kept[start:start + rows_per_page] + for row in window: + row['payload'] = json_loads(row.get('payload_json'), default={}) + + agg = aggregate_rows(kept, 'type') + return { + 'ok': True, 'status': 'OK', 'total': len(kept), 'page': page, + 'rows_per_page': rows_per_page, 'rows': window, 'data': window, + 'by_type': agg['by_type'], 'coverage': agg['coverage'], + 'watermark': agg['watermark'], 'preview_filter': stats, + 'truncated': bool(truncated), 'tenant_id': tid, + } + + +# ══════════════════════════════════════════════════════════════════════ +# M5b 契约 ④：主体证据档案（list_evidence_by_student / by_team 等价） +# ══════════════════════════════════════════════════════════════════════ +async def pbl_evidence_subject_summary(**kw): + """学生 / 团队维度证据档案：四类分组明细 + 产出物计数 + 水位。 + + 入参：``subject_type``(student|learner|team) + ``subject_id`` + （或 ``student_id`` / ``team_id``）/ ``blueprint_id``(可选) + / ``include_preview``(默认 False) + 出参：``{ok,subject_type,subject_id,evidence_total,by_type,evidence, + artifact_count,watermark,preview_filter}`` + """ + tid = _require_tenant() + stype = str(kw.get('subject_type') or '').strip().lower() + sid = kw.get('subject_id') or kw.get('student_id') or kw.get('team_id') + if not stype: + stype = 'team' if not _blank(kw.get('team_id')) else 'student' + if _blank(sid): + raise PblError('PBL_E_PARAM', '缺少 subject_id（student_id / team_id）') + if stype not in ('student', 'learner', 'team'): + raise PblError('PBL_E_PARAM', 'subject_type 须为 student / team') + + cols = await table_columns('pbl_evidence') + where = {'tenant_id': tid} + if stype in ('student', 'learner'): + where['learner_id'] = str(sid) + else: + where['session_id'] = sid + if not _blank(kw.get('blueprint_id')): + where['blueprint_id'] = kw['blueprint_id'] + + rows, truncated = await scan_evidence_rows(where, kw, cols=cols) + kept, stats = apply_default_filters(rows, kw) + + grouped = {t: [] for t in EVIDENCE_TYPES} + for row in kept: + etype = str(row.get('evidence_type') or '') + if etype not in grouped: + etype = 'observation' + grouped[etype].append({ + 'id': row.get('id'), + 'source_event_id': row.get('source_event_id'), + 'occurred_at': row.get('occurred_at'), + 'session_id': row.get('session_id'), + 'blueprint_id': row.get('blueprint_id'), + 'artifact_id': row.get('artifact_id'), + 'dedup_key': row.get('dedup_key'), + 'payload': json_loads(row.get('payload_json'), default={}), + }) + + agg = aggregate_rows(kept, 'blueprint') + artifact_count = None + try: + acols = await table_columns('pbl_artifact') + if acols and 'tenant_id' in acols: + aw = {'tenant_id': tid} + if stype in ('student', 'learner') and 'creator_id' in acols: + aw['creator_id'] = str(sid) + if not _blank(kw.get('blueprint_id')) and 'blueprint_id' in acols: + aw['blueprint_id'] = kw['blueprint_id'] + asql, aargs = build_where(aw, acols) + cnt = await q_one('SELECT COUNT(*) AS c FROM `pbl_artifact` WHERE %s' % asql, aargs) + artifact_count = _as_int((cnt or {}).get('c')) + except Exception as exc: # 产出物表不可用不影响证据档案主口径 + artifact_count = None + stats['artifact_count_error'] = str(exc)[:200] + + return { + 'ok': True, 'status': 'OK', 'subject_type': stype, 'subject_id': str(sid), + 'tenant_id': tid, 'evidence_total': len(kept), + 'by_type': {t: len(grouped.get(t) or []) for t in EVIDENCE_TYPES}, + 'evidence': grouped, 'watermark': agg['watermark'], + 'artifact_count': artifact_count, 'preview_filter': stats, + 'truncated': bool(truncated), 'generated_at': now_str(), + } diff --git a/pbl_evidence/init.py b/pbl_evidence/init.py index 293df63..a3e680a 100644 --- a/pbl_evidence/init.py +++ b/pbl_evidence/init.py @@ -25,6 +25,12 @@ from pbl_evidence.api import ( pbl_evidence_watermark, ) from pbl_evidence.collector import collect_evidence_from_events, resolve_event_table +from pbl_evidence.aggregation import ( + pbl_evidence_collect_idempotent, + pbl_evidence_aggregate, + pbl_evidence_query, + pbl_evidence_subject_summary, +) from pbl_evidence.crud_api import ( pbl_artifact_options, pbl_evidence_delete, @@ -52,6 +58,12 @@ def load_pbl_evidence(): env.pbl_artifact_options = pbl_artifact_options env.pbl_evidence_update = pbl_evidence_update env.pbl_evidence_delete = pbl_evidence_delete + # ── M5b 证据聚合与查询契约（幂等采集包装 / 聚合 / 分页查询 / 主体档案） + # 漏这 4 行 → wwwroot/api/pbl_evidence_aggregate.dspy 等调用必 NameError（三处同步之 ③） + env.pbl_evidence_collect_idempotent = pbl_evidence_collect_idempotent + env.pbl_evidence_aggregate = pbl_evidence_aggregate + env.pbl_evidence_query = pbl_evidence_query + env.pbl_evidence_subject_summary = pbl_evidence_subject_summary # ── 底层能力（供 M5b 回放 / M6 评估直接复用）────────────────── env.pbl_collect_evidence_from_events = collect_evidence_from_events env.pbl_resolve_event_table = resolve_event_table diff --git a/scripts/load_path.py b/scripts/load_path.py index 5fa8778..d9a50d7 100644 --- a/scripts/load_path.py +++ b/scripts/load_path.py @@ -59,6 +59,12 @@ PATHS = [ # 证据查询 / 统计 ('/pbl_evidence/api/pbl_evidence_list.dspy', 'logined'), ('/pbl_evidence/api/pbl_evidence_stats.dspy', 'logined'), + # ── M5b 证据聚合与查询契约（幂等采集包装 / 聚合 / 分页查询 / 主体档案） + # 新增 .dspy 必须同步登记，漏登记 = 部署后 403（load_path.py --check 会报 MISSING） + ('/pbl_evidence/api/pbl_evidence_collect_idempotent.dspy', 'logined'), + ('/pbl_evidence/api/pbl_evidence_aggregate.dspy', 'logined'), + ('/pbl_evidence/api/pbl_evidence_query.dspy', 'logined'), + ('/pbl_evidence/api/pbl_evidence_subject_summary.dspy', 'logined'), # CRUD 框架适配端点（json/pbl_evidence.json editable 三 URL + artifact_id 下拉） ('/pbl_evidence/api/pbl_artifact_options.dspy', 'logined'), ('/pbl_evidence/api/pbl_evidence_update.dspy', 'logined'), diff --git a/scripts/selftest_m5b.py b/scripts/selftest_m5b.py new file mode 100644 index 0000000..dc7a36e --- /dev/null +++ b/scripts/selftest_m5b.py @@ -0,0 +1,408 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""M5b 自测（pbl_evidence 证据聚合与查询契约）—— 纯 stdlib，无需 DB / 无需 pytest。 + +覆盖设计 docs/01-design/modules/pbl_evidence.md §3.2 的 M5b 需求点： + +1. **四类证据幂等采集**：``pbl_evidence_collect_idempotent`` 首次 ``created=True``， + 重复（同 idempotency_key / 同唯一索引三元组）``created=False`` 且不产生第二行； +2. **idempotency_key 归一**：显式键原样透传；缺省键由 ``tenant|source_event_id| + evidence_type`` 派生且稳定（重放同键）；非四类 evidence_type 报 ``PBL_E_VALIDATION``； +3. **聚合契约**：``pbl_evidence_aggregate`` 四类计数 + 覆盖度 + 分组 + 时间水位； +4. **列表查询契约**：``pbl_evidence_query`` 分页 total/rows + payload 解析 + 水位； +5. **主体档案契约**：``pbl_evidence_subject_summary`` 学生 / 团队四类分组明细； +6. **Q-OPEN-10**：预览 / Playtest 数据（payload 标记、显式预览会话、STUDENT 私有可见性） + 默认剔除，``include_preview=True`` 才纳入，且 ``preview_filter`` 回显剔除口径； +7. **租户 fail-closed**：无 tenant_id 直接 ``PBL_E_TENANT_MISSING``，不落到 SQL。 + +内存版 DB 会**校验拼出的 SQL**（列清单、LIMIT 占位、无 ``None`` 泄漏）并按 params 过滤， +因此不只是「函数不报错」，而是验证了查询参数与 SQL 的对应关系。 + +运行： + cd modules/pbl_evidence && python3 scripts/selftest_m5b.py + # 或： python3 -m pytest scripts/selftest_m5b.py -q （test_* 函数可直接被收集） +""" + +import asyncio +import json +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +for sib in ('pbl_common',): + p = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname( + os.path.abspath(__file__)))), sib) + if os.path.isdir(p) and p not in sys.path: + sys.path.insert(0, p) + +from pbl_common.api import PblError # noqa: E402 +import pbl_evidence.aggregation as A # noqa: E402 + +TENANT = '1' +EVIDENCE_COLS = ['id', 'tenant_id', 'artifact_id', 'evidence_type', 'source_event_id', + 'session_id', 'learner_id', 'blueprint_id', 'payload_json', + 'occurred_at', 'dedup_key', 'collect_batch_id', 'created_at', 'updated_at'] +ARTIFACT_COLS = ['id', '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'] + +_CAPTURED = {'sql': [], 'params': []} + + +def _row(idx, etype, learner='S01', session='101', blueprint='9', artifact=0, + occurred='2026-09-20 10:00:0%d' % (idx % 9), payload=None, source=None): + return { + 'id': idx, 'tenant_id': TENANT, 'artifact_id': artifact, 'evidence_type': etype, + 'source_event_id': source or ('EV%03d' % idx), 'session_id': session, + 'learner_id': learner, 'blueprint_id': blueprint, + 'payload_json': json.dumps(payload or {'event_type': 'mission.completed'}, + ensure_ascii=False), + 'occurred_at': occurred, 'dedup_key': 'K%03d' % idx, + } + + +def seed_rows(): + """正式环境证据 5 条（四类齐）+ 预览/Playtest 证据 4 条（应被默认过滤）。""" + return [ + _row(1, 'decision'), + _row(2, 'action'), + _row(3, 'observation', learner='S02'), + _row(4, 'artifact_version', artifact=77, learner='S02'), + _row(5, 'decision', learner='S03', occurred='2026-09-21 08:00:00'), + # ── 预览 / Playtest 族（Q-OPEN-10 默认剔除）──────────────── + _row(6, 'action', payload={'pbl_preview': True, 'event_type': 'mission.completed'}), + _row(7, 'observation', payload={'env_mode': 'playtest'}), + _row(8, 'decision', session='999', + payload={'event_type': 'team.decision'}), # 显式预览会话 999 + _row(9, 'artifact_version', artifact=88, + payload={'visibility': 'student', 'content_ref': 'draft'}), # STUDENT 私有 + ] + + +class FakeDB: + """内存版证据库：按 params 过滤，同时校验 SQL 文本合法性。""" + + def __init__(self, rows): + self.rows = rows + + async def table_columns(self, table): + if table == 'pbl_evidence': + return list(EVIDENCE_COLS) + if table == 'pbl_artifact': + return list(ARTIFACT_COLS) + return [] + + @staticmethod + def _match(row, params): + for key, val in (params or {}).items(): + if key in ('scan_limit', 'time_from', 'time_to') or key.startswith('etype'): + continue + if key in row and str(row[key]) != str(val): + return False + if 'time_from' in (params or {}) and row.get('occurred_at'): + if str(row['occurred_at']) < str(params['time_from']): + return False + if 'time_to' in (params or {}) and row.get('occurred_at'): + if str(row['occurred_at']) > str(params['time_to']): + return False + types = [v for k, v in (params or {}).items() if k.startswith('etype')] + if types and str(row.get('evidence_type')) not in [str(t) for t in types]: + return False + return True + + async def q_all(self, sql, params=None): + _CAPTURED['sql'].append(sql) + _CAPTURED['params'].append(dict(params or {})) + low = sql.lower() + assert 'none' not in low.replace('is null', ''), 'SQL 泄漏 None：%s' % sql + if 'from `pbl_evidence`' in low: + assert 'limit ${scan_limit}$' in low, '缺少扫描上限（防全表拖库）：%s' % sql + assert 'tenant_id' in low, '证据查询必须带租户条件：%s' % sql + limit = int((params or {}).get('scan_limit') or A.MAX_SCAN_ROWS) + rows = [r for r in self.rows if self._match(r, params)] + rows.sort(key=lambda r: (str(r.get('occurred_at') or ''), int(r.get('id') or 0)), + reverse=True) + return rows[:limit] + if 'count(*)' in low and 'pbl_artifact' in low: + return [{'c': 2}] + return [] + + async def q_one(self, sql, params=None): + rows = await self.q_all(sql, params) + return rows[0] if rows else None + + +def install_fake(rows, tenant=TENANT): + """把 aggregation 的 IO 依赖换成内存实现（不碰真实 DB）。""" + db = FakeDB(rows) + A.table_columns = db.table_columns + A.q_all = db.q_all + A.q_one = db.q_one + A.tenant_id = lambda *a, **k: tenant + return db + + +def run(coro): + return asyncio.get_event_loop().run_until_complete(coro) + + +# ══════════════════════════════════════════════════════════════════════ +# 1. 幂等采集（含 created 布尔 + 重复不产生第二行） +# ══════════════════════════════════════════════════════════════════════ +def test_idempotent_collect_created_flag(): + rows = seed_rows() + install_fake(rows) + store = {} + + async def fake_collect(**kw): + key = (str(kw.get('tenant_id') or TENANT), str(kw.get('source_event_id')), + str(kw.get('evidence_type'))) + if key in store: + return {'ok': True, 'deduped': True, 'id': store[key], + 'unique_index': 'uk_ev_dedup', 'evidence_type': key[2]} + nid = 1000 + len(store) + store[key] = nid + rows.append(_row(nid, key[2], source=key[1])) + return {'ok': True, 'deduped': False, 'id': nid, 'unique_index': 'uk_ev_dedup', + 'evidence_type': key[2]} + + import pbl_evidence.api as api + origin = api.pbl_evidence_collect + api.pbl_evidence_collect = fake_collect + try: + first = run(A.pbl_evidence_collect_idempotent( + source_event_id='EV900', evidence_type='decision', learner_id='S01', + blueprint_id='9', session_id='101', payload={'choice': 'A'})) + assert first['created'] is True, first + assert first['deduped'] is False, first + assert first['idempotency_key'].startswith('IK') and len(first['idempotency_key']) == 26 + assert first['idempotency_key_derived'] is True, first + second = run(A.pbl_evidence_collect_idempotent( + source_event_id='EV900', evidence_type='decision', learner_id='S01', + blueprint_id='9', session_id='101', payload={'choice': 'A'})) + assert second['created'] is False, second + assert second['deduped'] is True, second + assert second['id'] == first['id'], (first, second) + assert second['idempotency_key'] == first['idempotency_key'], '派生键必须稳定' + assert len(store) == 1, '重复采集不得产生第二行：%s' % store + # 显式 idempotency_key 原样透传 + third = run(A.pbl_evidence_collect_idempotent( + source_event_id='EV901', evidence_type='action', idempotency_key='IK-EXTERNAL-01', + learner_id='S01')) + assert third['idempotency_key'] == 'IK-EXTERNAL-01', third + assert third['idempotency_key_derived'] is False, third + finally: + api.pbl_evidence_collect = origin + + +def test_collect_rejects_non_fourth_type(): + install_fake(seed_rows()) + try: + run(A.pbl_evidence_collect_idempotent(source_event_id='EV902', + evidence_type='research_upload')) + raise AssertionError('非四类 evidence_type 必须被拒') + except PblError as exc: + assert exc.code == 'PBL_E_VALIDATION', exc.code + + +def test_tenant_fail_closed(): + install_fake(seed_rows(), tenant=None) + for fn, kw in ((A.pbl_evidence_aggregate, {}), (A.pbl_evidence_query, {}), + (A.pbl_evidence_subject_summary, {'subject_id': 'S01'}), + (A.pbl_evidence_collect_idempotent, {'source_event_id': 'EV1'})): + try: + run(fn(**kw)) + raise AssertionError('%s 缺租户时必须 fail-closed' % fn.__name__) + except PblError as exc: + assert exc.code == 'PBL_E_TENANT_MISSING', (fn.__name__, exc.code) + + +# ══════════════════════════════════════════════════════════════════════ +# 2. 聚合契约 +# ══════════════════════════════════════════════════════════════════════ +def test_aggregate_four_types_and_coverage(): + install_fake(seed_rows()) + res = run(A.pbl_evidence_aggregate(group_by='learner')) + assert res['ok'] is True and res['status'] == 'OK', res + assert res['total'] == 5, res['total'] # 9 行中 4 行预览族被默认剔除 + assert res['by_type'] == {'decision': 2, 'action': 1, 'observation': 1, + 'artifact_version': 1}, res['by_type'] + assert res['coverage']['covered_types'] == 4, res['coverage'] + assert res['coverage']['missing_types'] == [], res['coverage'] + keys = {g['group_key'] for g in res['groups']} + assert keys == {'S01', 'S02', 'S03'}, keys + assert res['watermark']['max_occurred_at'] == '2026-09-21 08:00:00', res['watermark'] + assert res['preview_filter']['excluded_preview'] == 3, res['preview_filter'] + assert res['preview_filter']['excluded_student'] == 1, res['preview_filter'] + assert res['preview_filter']['applied'] is True, res['preview_filter'] + assert res['truncated'] is False, res + + +def test_aggregate_group_by_type_and_time_range(): + install_fake(seed_rows()) + res = run(A.pbl_evidence_aggregate(group_by='type', + time_from='2026-09-21 00:00:00', + time_to='2026-09-21 23:59:59')) + assert res['total'] == 1, res + assert res['groups'][0]['group_key'] == 'decision', res['groups'] + res2 = run(A.pbl_evidence_aggregate(group_by='session', + time_range={'from': '2026-09-20 00:00:00', + 'to': '2026-09-20 23:59:59'})) + assert res2['group_by'] == 'session_id', res2 + assert res2['total'] == 3, res2 + + +def test_aggregate_evidence_types_filter(): + install_fake(seed_rows()) + res = run(A.pbl_evidence_aggregate(evidence_types=['decision', 'action'])) + assert set(res['by_type']) == {'decision', 'action', 'observation', 'artifact_version'} + assert res['total'] == 2, res['total'] + assert res['coverage']['missing_types'] == ['observation', 'artifact_version'], res['coverage'] + try: + run(A.pbl_evidence_aggregate(evidence_types='decision,bogus')) + raise AssertionError('非法 evidence_types 必须被拒') + except PblError as exc: + assert exc.code == 'PBL_E_VALIDATION', exc.code + + +# ══════════════════════════════════════════════════════════════════════ +# 3. 列表查询契约（分页 + payload + 水位） +# ══════════════════════════════════════════════════════════════════════ +def test_query_pagination_and_payload(): + install_fake(seed_rows()) + res = run(A.pbl_evidence_query(page=1, rows=2)) + assert res['total'] == 5, res['total'] + assert len(res['rows']) == 2, len(res['rows']) + assert res['rows'][0]['payload'].get('event_type'), res['rows'][0] + page2 = run(A.pbl_evidence_query(page=2, rows=2)) + ids = {r['id'] for r in res['rows']} ^ {r['id'] for r in page2['rows']} + assert len(ids) == 4, ids # 翻页不重叠 + tail = run(A.pbl_evidence_query(page=3, rows=2)) + assert len(tail['rows']) == 1, tail['rows'] + assert run(A.pbl_evidence_query(page=9, rows=20))['rows'] == [] + + +def test_query_filters_by_learner_and_type(): + install_fake(seed_rows()) + res = run(A.pbl_evidence_query(learner_id='S02')) + assert res['total'] == 2, res['total'] + assert {r['learner_id'] for r in res['rows']} == {'S02'}, res['rows'] + res2 = run(A.pbl_evidence_query(evidence_type='decision')) + assert res2['total'] == 2, res2['total'] + + +# ══════════════════════════════════════════════════════════════════════ +# 4. 主体档案契约（学生 / 团队） +# ══════════════════════════════════════════════════════════════════════ +def test_subject_summary_student_and_team(): + install_fake(seed_rows()) + stu = run(A.pbl_evidence_subject_summary(subject_type='student', subject_id='S02')) + assert stu['ok'] is True, stu + assert stu['evidence_total'] == 2, stu['evidence_total'] + assert stu['by_type']['action'] == 1 and stu['by_type']['artifact_version'] == 1, stu['by_type'] + assert set(stu['evidence']) == {'decision', 'action', 'observation', 'artifact_version'} + assert stu['artifact_count'] == 2, stu['artifact_count'] + tea = run(A.pbl_evidence_subject_summary(subject_type='team', team_id='101', + blueprint_id='9')) + assert tea['subject_id'] == '101', tea + assert tea['evidence_total'] == 2, tea['evidence_total'] # 999 预览会话已剔除 + try: + run(A.pbl_evidence_subject_summary(subject_type='school', subject_id='X')) + raise AssertionError('非法 subject_type 必须被拒') + except PblError as exc: + assert exc.code == 'PBL_E_PARAM', exc.code + try: + run(A.pbl_evidence_subject_summary(subject_type='student')) + raise AssertionError('缺 subject_id 必须被拒') + except PblError as exc: + assert exc.code == 'PBL_E_PARAM', exc.code + + +# ══════════════════════════════════════════════════════════════════════ +# 5. Q-OPEN-10 预览 / Playtest 默认过滤 +# ══════════════════════════════════════════════════════════════════════ +def test_qopen10_preview_excluded_by_default(): + install_fake(seed_rows()) + default = run(A.pbl_evidence_query()) + withprev = run(A.pbl_evidence_query(include_preview=True, + include_student_private=True, + preview_session_ids=['999'])) + assert default['total'] == 5, default['total'] + assert withprev['total'] == 9, withprev['total'] + assert withprev['preview_filter']['applied'] is False, withprev['preview_filter'] + assert withprev['by_type']['observation'] == 2, withprev['by_type'] + + +def test_qopen10_rules_pure(): + prev, rule = A.is_preview_row({'payload_json': json.dumps({'pbl_preview': True}), + 'session_id': '1'}) + assert prev and rule == 'payload:pbl_preview', rule + prev, rule = A.is_preview_row({'payload_json': json.dumps({'env_mode': 'playtest'}), + 'session_id': '1'}) + assert prev and rule == 'payload:env_mode', rule + prev, rule = A.is_preview_row({'payload_json': json.dumps({'mode': 'demo'}), + 'session_id': '1'}) + assert prev and rule == 'payload:mode', rule + prev, rule = A.is_preview_row({'payload_json': json.dumps({'event_type': 'x'}), + 'session_id': '999'}, ['999']) + assert prev and rule == 'preview_session:999', rule + prev, _ = A.is_preview_row({'payload_json': '{bad json', 'session_id': '1'}) + assert prev is False, '坏 JSON 不得误判为预览' + priv, rule = A.is_student_private({'payload_json': json.dumps({'visibility': 'STUDENT'})}) + assert priv and rule.startswith('payload:visibility'), rule + priv, _ = A.is_student_private({'payload_json': json.dumps({'visibility': 'teacher'})}) + assert priv is False + kept, stats = A.apply_default_filters( + [{'id': 1, 'payload_json': '{}', 'session_id': '1', 'learner_id': 'S01'}], + {'subject_type': 'student', 'subject_id': 'S02'}) + assert kept == [] and stats['excluded_subject'] == 1, stats + key, derived = A.resolve_idempotency_key({'tenant_id': '1', 'source_event_id': 'EV1', + 'evidence_type': 'action'}) + key2, _ = A.resolve_idempotency_key({'tenant_id': '1', 'source_event_id': 'EV1', + 'evidence_type': 'action'}) + assert key == key2 and derived is True, (key, key2) + assert A.resolve_idempotency_key({'idempotency_key': 'abc'})[0] == 'abc' + + +def test_sql_shape_and_column_pruning(): + install_fake(seed_rows()) + _CAPTURED['sql'].clear() + run(A.pbl_evidence_aggregate(group_by='learner')) + sql = [s for s in _CAPTURED['sql'] if 'from `pbl_evidence`' in s.lower()][0] + assert 'select `id`, `evidence_type`' in sql.lower(), sql + assert '`source_event_id`' in sql and '`dedup_key`' in sql and '`payload_json`' in sql, sql + assert 'order by `occurred_at` desc' in sql.lower(), sql + assert 'limit ${scan_limit}$' in sql.lower(), sql + + +def test_watermark_contract(): + install_fake(seed_rows()) + agg = run(A.pbl_evidence_aggregate(group_by='blueprint')) + wm = agg['watermark'] + assert wm['count'] == 5 and wm['max_id'] == 5, wm + assert wm['min_occurred_at'] == '2026-09-20 10:00:01', wm + q = run(A.pbl_evidence_query(learner_id='S03')) + assert q['watermark']['max_occurred_at'] == '2026-09-21 08:00:00', q['watermark'] + + +ALL_TESTS = sorted([v for k, v in list(globals().items()) + if k.startswith('test_') and callable(v)]) + + +def main(): + passed, failed = 0, 0 + for fn in ALL_TESTS: + try: + fn() + except Exception as exc: + failed += 1 + print('FAIL %s: %s' % (fn.__name__, exc)) + else: + passed += 1 + print('PASS %s' % fn.__name__) + print('\n== M5b selftest: %d passed, %d failed, %d total ==' % (passed, failed, len(ALL_TESTS))) + return 1 if failed else 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/wwwroot/api/pbl_evidence_aggregate.dspy b/wwwroot/api/pbl_evidence_aggregate.dspy new file mode 100644 index 0000000..0012b36 --- /dev/null +++ b/wwwroot/api/pbl_evidence_aggregate.dspy @@ -0,0 +1,20 @@ +# pbl_evidence/api/pbl_evidence_aggregate.dspy +# M5b 契约②：证据聚合（四类计数 + 覆盖度 + 分组 + 时间水位）。 +# Q-OPEN-10：预览 / Playtest 数据默认剔除，需显式 include_preview=true 才纳入。 +# dspy 铁律：无 import、无 ServerEnv()、显式 return。 +debug(f'pbl_evidence/api/pbl_evidence_aggregate.dspy: START params_kw={dict(params_kw)}') + +try: + kw = dict(params_kw) + for flag in ('include_preview', 'include_student_private'): + if isinstance(kw.get(flag), str): + kw[flag] = kw[flag].strip().lower() in ('1', 'true', 'yes', 'y') + res = await pbl_evidence_aggregate(**kw) + status = 'OK' if res.get('ok') else 'ERROR' + debug(f'pbl_evidence_aggregate.dspy: DONE total={res.get("total")} ' + f'groups={len(res.get("groups") or [])}') + return {'status': status, 'data': res, 'message': res.get('message') or '', + 'total': res.get('total')} +except Exception as e: + error(f'pbl_evidence_aggregate.dspy: FAIL {format_exc()}') + return {'status': 'ERROR', 'data': None, 'message': str(e)} diff --git a/wwwroot/api/pbl_evidence_collect_idempotent.dspy b/wwwroot/api/pbl_evidence_collect_idempotent.dspy new file mode 100644 index 0000000..43b559f --- /dev/null +++ b/wwwroot/api/pbl_evidence_collect_idempotent.dspy @@ -0,0 +1,15 @@ +# pbl_evidence/api/pbl_evidence_collect_idempotent.dspy +# M5b 契约①：四类证据（decision/action/observation/artifact_version）幂等采集。 +# 重复采集（同 idempotency_key / 同唯一索引三元组）不报错，返回 created=false。 +# dspy 铁律：无 import、无 ServerEnv()、显式 return；函数由 load_pbl_evidence() 挂 ServerEnv。 +debug(f'pbl_evidence/api/pbl_evidence_collect_idempotent.dspy: START params_kw={dict(params_kw)}') + +try: + res = await pbl_evidence_collect_idempotent(**params_kw) + status = 'OK' if res.get('ok') else 'ERROR' + debug(f'pbl_evidence_collect_idempotent.dspy: DONE created={res.get("created")} ' + f'deduped={res.get("deduped")} id={res.get("id")}') + return {'status': status, 'data': res, 'message': res.get('message') or ''} +except Exception as e: + error(f'pbl_evidence_collect_idempotent.dspy: FAIL {format_exc()}') + return {'status': 'ERROR', 'data': None, 'message': str(e)} diff --git a/wwwroot/api/pbl_evidence_query.dspy b/wwwroot/api/pbl_evidence_query.dspy new file mode 100644 index 0000000..ed874d2 --- /dev/null +++ b/wwwroot/api/pbl_evidence_query.dspy @@ -0,0 +1,21 @@ +# pbl_evidence/api/pbl_evidence_query.dspy +# M5b 契约③：证据分页查询（含 timestamp + payload，US-17「证据不只是分数」）。 +# 支持 blueprint_id / learner_id / session_id / artifact_id / evidence_type / +# evidence_types / time_from,time_to / page / rows；预览数据默认过滤（Q-OPEN-10）。 +# dspy 铁律：无 import、无 ServerEnv()、显式 return。 +debug(f'pbl_evidence/api/pbl_evidence_query.dspy: START params_kw={dict(params_kw)}') + +try: + kw = dict(params_kw) + for flag in ('include_preview', 'include_student_private'): + if isinstance(kw.get(flag), str): + kw[flag] = kw[flag].strip().lower() in ('1', 'true', 'yes', 'y') + res = await pbl_evidence_query(**kw) + status = 'OK' if res.get('ok') else 'ERROR' + debug(f'pbl_evidence_query.dspy: DONE total={res.get("total")} ' + f'page={res.get("page")} rows={len(res.get("rows") or [])}') + return {'status': status, 'data': res, 'message': res.get('message') or '', + 'total': res.get('total')} +except Exception as e: + error(f'pbl_evidence_query.dspy: FAIL {format_exc()}') + return {'status': 'ERROR', 'data': None, 'message': str(e)} diff --git a/wwwroot/api/pbl_evidence_subject_summary.dspy b/wwwroot/api/pbl_evidence_subject_summary.dspy new file mode 100644 index 0000000..39b2347 --- /dev/null +++ b/wwwroot/api/pbl_evidence_subject_summary.dspy @@ -0,0 +1,22 @@ +# pbl_evidence/api/pbl_evidence_subject_summary.dspy +# M5b 契约④：学生 / 团队证据档案（设计 §3.2 list_evidence_by_student / by_team 等价实现）。 +# 入参：subject_type=student|team + subject_id（或 student_id / team_id）+ blueprint_id(可选) +# 出参：四类分组明细 + 产出物计数 + 水位；预览/Playtest 数据默认过滤（Q-OPEN-10）。 +# dspy 铁律：无 import、无 ServerEnv()、显式 return。 +debug(f'pbl_evidence/api/pbl_evidence_subject_summary.dspy: START params_kw={dict(params_kw)}') + +try: + kw = dict(params_kw) + if isinstance(kw.get('include_preview'), str): + kw['include_preview'] = kw['include_preview'].strip().lower() in ('1', 'true', 'yes', 'y') + if isinstance(kw.get('include_student_private'), str): + kw['include_student_private'] = kw['include_student_private'].strip().lower() in ('1', 'true', 'yes', 'y') + res = await pbl_evidence_subject_summary(**kw) + status = 'OK' if res.get('ok') else 'ERROR' + debug(f'pbl_evidence_subject_summary.dspy: DONE subject={res.get("subject_type")}:' + f'{res.get("subject_id")} total={res.get("evidence_total")}') + return {'status': status, 'data': res, 'message': res.get('message') or '', + 'total': res.get('evidence_total')} +except Exception as e: + error(f'pbl_evidence_subject_summary.dspy: FAIL {format_exc()}') + return {'status': 'ERROR', 'data': None, 'message': str(e)} diff --git a/wwwroot/index.ui b/wwwroot/index.ui index 0ea4cc3..1d92405 100644 --- a/wwwroot/index.ui +++ b/wwwroot/index.ui @@ -271,6 +271,118 @@ } } ] + }, + { + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_evidence_content", + "options": { + "url": "{{entire_url('/pbl_evidence/api/pbl_evidence_collect_idempotent.dspy')}}" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "[M5b] 证据幂等采集 collect_idempotent" + } + } + ] + }, + { + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_evidence_content", + "options": { + "url": "{{entire_url('/pbl_evidence/api/pbl_evidence_aggregate.dspy')}}" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "[M5b] 证据聚合 aggregate" + } + } + ] + }, + { + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_evidence_content", + "options": { + "url": "{{entire_url('/pbl_evidence/api/pbl_evidence_query.dspy')}}" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "[M5b] 证据查询 query" + } + } + ] + }, + { + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_evidence_content", + "options": { + "url": "{{entire_url('/pbl_evidence/api/pbl_evidence_subject_summary.dspy')}}" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "[M5b] 学生/团队证据档案" + } + } + ] } ] }, @@ -284,4 +396,4 @@ } } ] -} +} \ No newline at end of file