181 lines
9.0 KiB
Python
181 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""pbl_assessment.api —— Rubric 加权评估与报告（第34章，M6）。
|
||
|
||
- criteria_json 内联准则数组（含 learning_goal_id/weight），**权重合计必须 = 100%**（Q-OPEN-3）；
|
||
- 准则变更需人工审批（14.2 modify_assessment_criteria），无 approval_id 的已存在 Rubric 变更直接拒绝；
|
||
- total_score 为派生值（Σ 得分×权重），不接收客户端直传。
|
||
"""
|
||
import hashlib
|
||
import json
|
||
|
||
from pbl_common.api import (PblError, actor_id, crud, json_dump, now_str, sql_exec,
|
||
sql_rows, sql_scalar, tenant_id)
|
||
|
||
BANDS = [(90, 'excellent'), (75, 'proficient'), (60, 'developing'), (0, 'beginning')]
|
||
|
||
|
||
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 {}
|
||
|
||
|
||
async def _tid():
|
||
return await tenant_id()
|
||
|
||
|
||
RUB = crud('pbl_rubric', 'pbl', ['code', 'title', 'blueprint_id', 'criteria_json',
|
||
'total_weight', 'version_no', 'status', 'approval_id'])
|
||
REC = crud('pbl_assessment_record', 'pbl', ['record_uid', 'learner_id', 'session_id',
|
||
'blueprint_id', 'artifact_id', 'evidence_id',
|
||
'rubric_id', 'scores_json', 'total_score',
|
||
'max_score', 'band', 'alignment_json',
|
||
'feedback_txt', 'assessor_type', 'assessor_id'])
|
||
|
||
|
||
async def pbl_rubric_save(**kw):
|
||
tid = await _tid()
|
||
criteria = _loads(kw.get('criteria_json'), [])
|
||
if not isinstance(criteria, list) or not criteria:
|
||
raise PblError('PBL_RUBRIC_EMPTY', 'criteria_json 须为非空数组')
|
||
total = sum(float(c.get('weight') or 0) for c in criteria)
|
||
if abs(total - 100.0) > 0.01:
|
||
raise PblError('PBL_RUBRIC_WEIGHT_INVALID', '权重合计须=100%%，实际 %.2f' % total)
|
||
for c in criteria:
|
||
if not (c.get('learning_goal_id') or c.get('learning_goal_uid')):
|
||
raise PblError('PBL_RUBRIC_ALIGNMENT_MISSING',
|
||
'准则 %s 缺 learning_goal_id（无法做对齐判定）' % c.get('code'))
|
||
payload = dict(kw)
|
||
payload['criteria_json'] = json_dump(criteria)
|
||
payload['total_weight'] = round(total, 2)
|
||
payload.setdefault('version_no', 1)
|
||
payload.setdefault('status', 'draft')
|
||
if kw.get('id'):
|
||
cur = await RUB['read'](id=kw['id'])
|
||
if cur['data'].get('version_no', 1) >= 1 and kw.get('status') in ('published', 'active') \
|
||
and not kw.get('approval_id'):
|
||
raise PblError('PBL_APPROVAL_REQUIRED',
|
||
'评估标准变更须人工审批（14.2），缺 approval_id')
|
||
return await RUB['update'](**payload)
|
||
payload.setdefault('code', 'RU%s' % hashlib.sha256(
|
||
(str(kw.get('title')) + now_str()).encode()).hexdigest()[:16])
|
||
r = await RUB['create'](**payload)
|
||
r['code'] = payload['code']
|
||
r['weight_total_ok'] = True
|
||
return r
|
||
|
||
|
||
async def pbl_rubric_get(**kw):
|
||
tid = await _tid()
|
||
rows = await sql_rows('SELECT * FROM `pbl_rubric` WHERE `tenant_id` = ${t}$'
|
||
' AND (`id` = ${id}$ OR `code` = ${c}$) LIMIT 1',
|
||
{'t': tid, 'id': kw.get('id') or 0, 'c': kw.get('code') or ''}, 'pbl')
|
||
if not rows:
|
||
raise PblError('PBL_NOT_FOUND', 'Rubric 不存在')
|
||
rows[0]['criteria_json'] = _loads(rows[0].get('criteria_json'), [])
|
||
return {'ok': True, 'data': rows[0]}
|
||
|
||
|
||
async def pbl_rubric_list(**kw):
|
||
return await RUB['list'](**kw)
|
||
|
||
|
||
async def pbl_assessment_score(**kw):
|
||
"""按 Rubric 逐准则打分 → 加权总分 + 等级 + 学习目标对齐结论。"""
|
||
tid = await _tid()
|
||
rubric_id = kw.get('rubric_id')
|
||
rr = await sql_rows('SELECT * FROM `pbl_rubric` WHERE `tenant_id` = ${t}$'
|
||
' AND `id` = ${id}$ LIMIT 1', {'t': tid, 'id': rubric_id}, 'pbl')
|
||
if not rr:
|
||
raise PblError('PBL_NOT_FOUND', 'Rubric 不存在')
|
||
rubric = rr[0]
|
||
criteria = _loads(rubric['criteria_json'], [])
|
||
scores = _loads(kw.get('scores_json'), {}) or {}
|
||
rows, weighted, max_w = [], 0.0, 0.0
|
||
for c in criteria:
|
||
w = float(c.get('weight') or 0)
|
||
raw = scores.get(c.get('code') or c.get('uid'))
|
||
pts = float(raw) if raw not in (None, '') else 0.0
|
||
cmax = float(c.get('max_points') or 5)
|
||
ratio = min(1.0, pts / cmax) if cmax else 0.0
|
||
weighted += ratio * 100 * w / 100.0
|
||
max_w += w
|
||
rows.append({'criterion': c.get('code'), 'learning_goal_id':
|
||
c.get('learning_goal_id') or c.get('learning_goal_uid'),
|
||
'weight': w, 'points': pts, 'max_points': cmax,
|
||
'contribution': round(ratio * w, 2)})
|
||
total = round(weighted, 2)
|
||
band = next(b[1] for b in BANDS if total >= b[0])
|
||
ev = await sql_rows('SELECT `evidence_type`,`id` FROM `pbl_evidence` WHERE `tenant_id` = ${t}$'
|
||
' AND `learner_id` = ${l}$ ORDER BY `occurred_at` DESC LIMIT ${n}$',
|
||
{'t': tid, 'l': kw.get('learner_id') or await actor_id(), 'n': 50}, 'pbl')
|
||
have = {r['evidence_type'] for r in ev}
|
||
spec_rows = await sql_rows('SELECT DISTINCT `evidence_type` FROM `pbl_blueprint_evidence_spec`'
|
||
' WHERE `tenant_id` = ${t}$ AND `blueprint_id` = ${b}$',
|
||
{'t': tid, 'b': rubric.get('blueprint_id') or 0}, 'pbl')
|
||
need = {s['evidence_type'] for s in spec_rows}
|
||
alignment = {'criteria_count': len(criteria), 'scored': len([r for r in rows if r['points']]),
|
||
'goals_covered': len({r['learning_goal_id'] for r in rows if r['points']}),
|
||
'evidence_present': sorted(have), 'evidence_required': sorted(need),
|
||
'evidence_missing': sorted(need - have)}
|
||
payload = {'record_uid': 'AS%s' % hashlib.sha256(
|
||
('%s%s%s' % (rubric_id, kw.get('learner_id'), now_str())).encode())
|
||
.hexdigest()[:20],
|
||
'learner_id': kw.get('learner_id') or await actor_id(),
|
||
'session_id': kw.get('session_id'), 'blueprint_id': rubric.get('blueprint_id'),
|
||
'artifact_id': kw.get('artifact_id'), 'evidence_id': kw.get('evidence_id'),
|
||
'rubric_id': rubric_id, 'scores_json': json_dump(rows),
|
||
'total_score': total, 'max_score': round(max_w * 100 / 100.0, 2), 'band': band,
|
||
'alignment_json': json_dump(alignment), 'feedback_txt': kw.get('feedback_txt'),
|
||
'assessor_type': kw.get('assessor_type') or 'teacher',
|
||
'assessor_id': await actor_id()}
|
||
r = await REC['create'](**payload)
|
||
r['record_uid'] = payload['record_uid']
|
||
r['total_score'], r['band'], r['alignment'] = total, band, alignment
|
||
r['derived'] = 'total_score 由 Σ(得分率×权重) 派生，客户端不可直传'
|
||
return r
|
||
|
||
|
||
async def pbl_assessment_record_get(**kw):
|
||
r = await REC['read'](**kw)
|
||
r['data']['scores_json'] = _loads(r['data'].get('scores_json'), [])
|
||
r['data']['alignment_json'] = _loads(r['data'].get('alignment_json'), {})
|
||
return r
|
||
|
||
|
||
async def pbl_assessment_report(**kw):
|
||
"""教师视图：按学习者聚合评估记录 + 证据覆盖 + 质量状态联动建议。"""
|
||
tid = await _tid()
|
||
where, params = ['`tenant_id` = ${t}$'], {'t': tid}
|
||
for col in ('learner_id', 'blueprint_id', 'session_id'):
|
||
if kw.get(col) not in (None, ''):
|
||
where.append('`%s` = ${%s}$' % (col, col))
|
||
params[col] = kw[col]
|
||
rows = await sql_rows('SELECT * FROM `pbl_assessment_record` WHERE %s'
|
||
' ORDER BY `total_score` DESC LIMIT ${n}$' % ' AND '.join(where),
|
||
dict(params, n=min(int(kw.get('limit') or 100), 500)), 'pbl')
|
||
learners = {}
|
||
for r in rows:
|
||
learners.setdefault(r['learner_id'], []).append(r)
|
||
summary = [{'learner_id': k, 'records': len(v),
|
||
'avg_score': round(sum(float(x['total_score'] or 0) for x in v) / len(v), 2),
|
||
'best_band': v[0]['band'],
|
||
'missing_evidence': sorted({m for x in v for m in
|
||
(_loads(x.get('alignment_json'), {})
|
||
.get('evidence_missing') or [])})}
|
||
for k, v in sorted(learners.items())]
|
||
bp_state = None
|
||
if kw.get('blueprint_id'):
|
||
bp_state = await sql_scalar('SELECT `quality_state` FROM `pbl_blueprint`'
|
||
' WHERE `tenant_id` = ${t}$ AND `id` = ${b}$',
|
||
{'t': tid, 'b': kw['blueprint_id']}, 'pbl')
|
||
return {'ok': True, 'learners': summary, 'record_count': len(rows),
|
||
'blueprint_quality_state': bp_state,
|
||
'playtest_ready_hint': ('已具备 playtest_ready 证据链' if rows and
|
||
all(not s['missing_evidence'] for s in summary)
|
||
else '存在证据缺口，建议保持 pbl_ready 并补证据')}
|