approve: 复测已修复 Bug(pbls-初始迭代)

This commit is contained in:
agent.develop 2026-10-13 14:48:36 +08:00
parent 8446faafea
commit a9a938a260

183
pbl_assessment/seed.py Normal file
View File

@ -0,0 +1,183 @@
# -*- coding: utf-8 -*-
"""pbl_assessment 种子数据 —— Rubric 默认量规 rubric_default_5level(幂等注入)。
对应 env/test.json seed.items[3]:Rubric 默认量规 rubric_default_5level(5 级准则)。
种子数据使用 __system__ 租户,由 build.sh 第 12 步种子段调用。
幂等策略:按 rubric.code 存在即跳过(不更新、不覆盖),二次执行零差异。
"""
import json
import time
import hashlib
MODULE = 'pbl_assessment'
TBL_RUBRIC = 'pbl_rubric'
TBL_RUBRIC_CRITERION = 'pbl_rubric_criterion'
SYSTEM_TENANT = '__system__'
# ---- 默认量规定义(5 级准则,权重合计 100%)----
DEFAULT_RUBRIC = {
"tenant_id": SYSTEM_TENANT,
"id": "rubric_default_5level",
"code": "RUBRIC_DEFAULT_5LEVEL",
"title": "PBL 通用量规(5 级)",
"blueprint_id": "",
"criteria_json": json.dumps([
{
"code": "C01_PROBLEM_DEFINITION",
"name": "问题定义与理解",
"learning_goal_id": "LG_PROBLEM",
"weight": 20.0,
"max_points": 5,
"descriptors": {
"5": "独立识别复杂问题,提出清晰且可验证的问题陈述",
"4": "识别主要问题,提出可验证的问题陈述",
"3": "在引导下识别问题,问题陈述基本清晰",
"2": "问题识别不完整,问题陈述模糊",
"1": "无法识别问题或问题陈述缺失",
},
},
{
"code": "C02_RESEARCH_INQUIRY",
"name": "探究与研究",
"learning_goal_id": "LG_RESEARCH",
"weight": 20.0,
"max_points": 5,
"descriptors": {
"5": "多源系统性探究,数据收集方法科学,分析深入",
"4": "使用多种可靠来源,分析有逻辑支撑",
"3": "使用2-3个来源,分析基本合理",
"2": "来源单一或不可靠,分析缺乏深度",
"1": "未进行有效探究或数据来源不明",
},
},
{
"code": "C03_SOLUTION_DESIGN",
"name": "方案设计与创新",
"learning_goal_id": "LG_DESIGN",
"weight": 25.0,
"max_points": 5,
"descriptors": {
"5": "方案具有原创性、可行性高,考虑了约束与取舍",
"4": "方案可行且有一定创新元素",
"3": "方案可用但创新性不足",
"2": "方案存在明显缺陷或不完整",
"1": "方案不可行或与问题无关",
},
},
{
"code": "C04_IMPLEMENTATION",
"name": "实施与执行",
"learning_goal_id": "LG_IMPLEMENT",
"weight": 20.0,
"max_points": 5,
"descriptors": {
"5": "按计划高质量完成全部交付,超出预期",
"4": "按计划完成交付,质量良好",
"3": "基本完成任务,质量可接受",
"2": "部分任务完成,存在质量缺陷",
"1": "未完成核心任务或质量不合格",
},
},
{
"code": "C05_REFLECTION_COMMUNICATION",
"name": "反思与沟通",
"learning_goal_id": "LG_REFLECT",
"weight": 15.0,
"max_points": 5,
"descriptors": {
"5": "深刻反思学习过程,沟通清晰有说服力,回应自如",
"4": "有反思内容,表达清晰",
"3": "有基本反思与表达",
"2": "反思表面化或表达不清",
"1": "缺少反思或沟通无效",
},
},
], ensure_ascii=False),
"total_weight": 100.00,
"version_no": 1,
"status": "active",
"approval_id": "",
}
# 量规准则表(与 pbl_rubric 一对多,每条准则一行)
SEED_CRITERIA = []
for c in json.loads(DEFAULT_RUBRIC['criteria_json']):
SEED_CRITERIA.append({
"tenant_id": SYSTEM_TENANT,
"id": "criterion_%s" % c['code'].lower(),
"rubric_code": "RUBRIC_DEFAULT_5LEVEL",
"code": c['code'],
"name": c['name'],
"learning_goal_id": c.get('learning_goal_id', ''),
"weight": c['weight'],
"max_points": c['max_points'],
"descriptors_json": json.dumps(c.get('descriptors', {}), ensure_ascii=False),
"sort_no": list(c['code'] for c in json.loads(DEFAULT_RUBRIC['criteria_json'])).index(c['code']) * 10 + 10,
})
async def seed_rubrics(dbname=None, tenant_id=None):
"""幂等写入默认量规种子。
:param dbname: 可选,数据库名(缺省自动解析)
:param tenant_id: 可选,租户 ID(缺省使用 __system__)
:return: dict {ok, rubrics_inserted, criteria_inserted, errors}
"""
from pbl_common.api import sql_exe, sql_rows
import time as _time
tid = tenant_id or SYSTEM_TENANT
now = str(int(_time.time() * 1000))
result = {"ok": True, "rubrics_inserted": 0, "criteria_inserted": 0, "errors": []}
# ---- 量规表 ----
exist = await sql_rows(
'SELECT id FROM `%s` WHERE `tenant_id` = ${t}$ AND `code` = ${c}$ LIMIT 1'
% TBL_RUBRIC, {'t': tid, 'c': DEFAULT_RUBRIC['code']})
if not exist:
try:
row = dict(DEFAULT_RUBRIC)
row['tenant_id'] = tid
row.setdefault('create_time', now)
row.setdefault('update_time', now)
cols = ','.join('`%s`' % k for k in row)
vals = ','.join('${%s}$' % k for k in row)
await sql_exe('INSERT INTO `%s` (%s) VALUES (%s)' % (TBL_RUBRIC, cols, vals), row)
result['rubrics_inserted'] = 1
except Exception as exc:
err = '默认量规写入失败: %s' % exc
result['errors'].append(err)
print('[pbl_assessment.seed][WARN] %s' % err)
# ---- 量规准则表 ----
for crit in SEED_CRITERIA:
try:
exist_c = await sql_rows(
'SELECT id FROM `%s` WHERE `tenant_id` = ${t}$ AND `code` = ${c}$ LIMIT 1'
% TBL_RUBRIC_CRITERION, {'t': tid, 'c': crit['code']})
if exist_c:
continue
row = dict(crit)
row['tenant_id'] = tid
row.setdefault('create_time', now)
row.setdefault('update_time', now)
cols = ','.join('`%s`' % k for k in row)
vals = ','.join('${%s}$' % k for k in row)
await sql_exe('INSERT INTO `%s` (%s) VALUES (%s)' % (TBL_RUBRIC_CRITERION, cols, vals), row)
result['criteria_inserted'] += 1
except Exception as exc:
err = '准则 %s 写入失败: %s' % (crit.get('code'), exc)
result['errors'].append(err)
print('[pbl_assessment.seed][WARN] %s' % err)
result['ok'] = len(result['errors']) == 0
return result
def seed_all_sync(dbname=None, tenant_id=None):
"""同步包装(供 build.sh heredoc 裸调)。"""
import asyncio
return asyncio.run(seed_rubrics(dbname=dbname, tenant_id=tenant_id))