cjw/geMoldInsight
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

D17 批 1：Human-in-Loop 老师傅经验反馈（数据 + 权限 + 写入 API）

Browse Source 
新增 experience_feedback 表（32 表迁移，alembic head b7d1f4a92c3e），
老师傅对系统推荐方案给出"采纳 / 调整 / 拒绝"反馈，按"产品指纹 +
工艺参数"为键跨任务匹配，下次同指纹产品分析自动消费。

变更内容：
- src/moldinsight/models/experience_feedback.py（new）ORM：Base 单点来源、
  跨模块裸 FK（user_id / processing_task_id / stp_file_id）、不建 ORM
  relationship；fingerprint JSON 列存跨任务匹配键
- src/moldinsight/models/__init__.py 导出 ExperienceFeedback
- migrations/versions/b7d1f4a92c3e_add_experience_feedback.py（new）32 表
  迁移；fingerprint 列 PG 下加 GIN 索引（jsonb_path_query 支持）
- src/shared/database/init_db.py 加 3 个权限码（view_experience_feedback /
  feedback_experience_hint / manage_experience_feedback）+ 新角色
  process_engineer；admin 角色 permissions 同步补齐；init_permissions /
  init_roles 改为按 code 比对（新增保留已有 id，避免 FK 引用失效）——
  修复既有 DB 启动期漏掉新权限的幂等 bug
- src/moldinsight/services/experience_feedback_service.py（new）service：
  compute_fingerprint 分桶（bbox_aspect / volume_bucket / face_bucket /
  undercut_class / material_family / is_foam）/ record_feedback（D9 边界：
  service.flush + 路由 commit；D17 衰减：同 stp_file_id 整体续期 90 天 TTL，
  无 celery beat 依赖）/ list_hints_for_task / resolve_for_process_params
- src/moldinsight/api/experience_feedback_router.py（new）路由：Pydantic
  模型写在路由文件内（项目硬规则）；POST /api/tasks/{task_id}/experience-feedback
  + GET /api/tasks/{task_id}/experience-hints；归属 TaskQueryService.ensure_task_access
  + User.has_permission 全仓首次调用点
- src/moldinsight/api/__init__.py ROUTE_MODULES 注册新路由
- tests/test_model_ownership.py EXPECTED_TABLES 加 experience_feedback
  （31→32）
- tests/test_experience_feedback_fingerprint.py（new）分桶参数化覆盖
  bbox / volume / face / undercut / material / is_foam 各边界值
- tests/test_experience_feedback_router.py（new）API 契约 9 例
  （401/403/422/200 路径 + 衰减续期 + 任务归属校验 + ORM 注册收口）
- docs/STATUS.md 顶部加 2026-09-23 批 1 日志条目
- docs/TECH_DEBT.md D17 加批 1 已完成描述 + 剩余工作清单
- docs/API_CONTRACT.md §3.2 加 D17 端点表格

测试基线：185 passed, 9 skipped（净增 59 测试）。

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
chenjw28
2026-09-23 16:11:17 +08:00
parent 79441a8a87
commit 2c9ba9d6b3
13 changed files with 1338 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/moldinsight/api/__init__.py
+1
View File
@@ -27,6 +27,7 @@ ROUTE_MODULES = [
("加工", "moldinsight.api.machining_router", False),
("导出", "moldinsight.api.export_router", False),
("铝价", "moldinsight.api.aluminum_price_routes", False),
("老师傅经验反馈", "moldinsight.api.experience_feedback_router", False), # D17 Human-in-Loop
# 调试端点会 dump 全量任务数据，仅 DEBUG 模式注册（双重防线：还需登录）
("调试", "moldinsight.api.debug_router", True),
]
src/moldinsight/api/experience_feedback_router.py
+199
View File
@@ -0,0 +1,199 @@
"""老师傅经验反馈 API：D17 Human-in-Loop 闭环。
端点：
- POST /api/tasks/{task_id}/experience-feedback 提交方案级反馈
- GET /api/tasks/{task_id}/experience-hints 拉取同指纹历史 hints 摘要
权限：
- 写入：Depends(get_current_active_user) + ensure_task_access + 行内 has_permission
- 读取：Depends(get_current_active_user) + ensure_task_access（所有登录用户可看）
Pydantic 模型写在路由文件内（项目硬规则，shared/models/schemas.py 不扩张）。
"""
from datetime import datetime
from typing import Dict, Any, List, Literal, Optional
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from shared.database.database import get_db_session
from shared.models.identity import User
from shared.services.auth_service import get_current_active_user
from shared.utils.logger import get_logger
from moldinsight.services.experience_feedback_service import (
ExperienceFeedbackService,
compute_fingerprint,
)
from moldinsight.services.task_query_service import TaskQueryService
from moldinsight.models import ExperienceFeedback, ProcessingTask, GeometryData
logger = get_logger(__name__)
router = APIRouter()
# ── Pydantic 请求 / 响应模型（写在路由文件内，硬规则）──
class ExperienceFeedbackCreate(BaseModel):
"""老师傅方案级反馈请求体。"""
scheme_id: str = Field(..., min_length=1, max_length=64)
feedback_status: Literal["adopted", "adjust", "rejected"]
feedback_reason: Optional[str] = Field(None, max_length=2000)
adjust_suggestion: Optional[str] = Field(None, max_length=2000)
confidence_at_submit: Optional[float] = Field(None, ge=0.0, le=1.0)
score_at_submit: Optional[float] = Field(None, ge=0.0, le=100.0)
class ExperienceFeedbackResponse(BaseModel):
"""反馈写入响应。"""
id: int
scheme_id: str
scheme_axis: str
feedback_status: str
created_at: datetime
class ExperienceHintItem(BaseModel):
"""同指纹历史 hints 摘要（按 scheme_axis 聚合）。"""
scheme_axis: str
adopted_count: int
rejected_count: int
adjust_count: int
confidence: float
weight: float
sample_count: int
class ExperienceHintsResponse(BaseModel):
"""GET /experience-hints 响应。"""
task_id: str
stp_file_id: int
material_name: str
is_foam: bool
fingerprint: Dict[str, str]
hints: List[ExperienceHintItem]
# ── 端点 ──
@router.post(
"/tasks/{task_id}/experience-feedback",
response_model=ExperienceFeedbackResponse,
)
async def submit_feedback(
task_id: str,
body: ExperienceFeedbackCreate,
db_session: AsyncSession = Depends(get_db_session),
current_user: User = Depends(get_current_active_user),
):
"""提交方案级反馈。
权限：登录用户 + 任务归属 + feedback_experience_hint。
写入后由路由 commit（D9 边界）+ invalidate_task_view（task_view 60s TTL 失效）。
"""
# 1. 任务归属校验（与 task_router / design_router 同一约定）
await TaskQueryService.ensure_task_access(db_session, task_id, current_user.id)
# 2. 权限校验：行内 has_permission（identity.py:38 全仓首次调用）
if not current_user.has_permission("feedback_experience_hint"):
raise HTTPException(403, "需要工艺工程师或管理员权限")
# 3. 写反馈（仅 flush，D9 边界由本路由 commit）
feedback = await ExperienceFeedbackService().record_feedback(
session=db_session,
task_id=task_id,
scheme_id=body.scheme_id,
feedback_status=body.feedback_status,
feedback_reason=body.feedback_reason,
adjust_suggestion=body.adjust_suggestion,
user=current_user,
confidence_at_submit=body.confidence_at_submit,
score_at_submit=body.score_at_submit,
process_params_snapshot=None, # 路由不接管 process_params，由算法层填
)
try:
await db_session.commit()
except Exception as exc:
await db_session.rollback()
logger.error(f"反馈提交失败: {exc}")
raise HTTPException(500, "反馈提交失败")
# 4. 失效任务视图缓存（写反馈后 next view 立即反映 hints）
TaskQueryService.invalidate_task_view(task_id)
return ExperienceFeedbackResponse(
id=feedback.id,
scheme_id=feedback.scheme_id,
scheme_axis=feedback.scheme_axis,
feedback_status=feedback.feedback_status,
created_at=feedback.created_at or datetime.utcnow(),
)
@router.get(
"/tasks/{task_id}/experience-hints",
response_model=ExperienceHintsResponse,
)
async def get_experience_hints(
task_id: str,
db_session: AsyncSession = Depends(get_db_session),
current_user: User = Depends(get_current_active_user),
):
"""拉取该任务的同指纹历史 hints 摘要。
权限：登录用户 + 任务归属。组织知识对所有人可见（不要求工艺工程师权限）。
"""
# 1. 任务归属校验
row = await TaskQueryService.ensure_task_access(db_session, task_id, current_user.id)
_, stp_file = row
# 2. 取 material / is_foam
pt_row = await db_session.execute(
select(ProcessingTask).where(ProcessingTask.task_id == task_id)
)
processing_task = pt_row.scalar_one_or_none()
params = (processing_task.parameters if processing_task else None) or {}
material_name = str(params.get("material") or "ABS")
is_foam = bool(params.get("is_foam_material", False))
# 3. 计算 fingerprint（用于回显 + 与 record_feedback 用同一函数）
geo_row = await db_session.execute(
select(GeometryData).where(GeometryData.stp_file_id == stp_file.id)
)
geo = geo_row.scalar_one_or_none()
geometry_summary: Dict[str, Any] = {}
if geo is not None:
geometry_summary = {
"volume": geo.volume,
"bounding_box": {
"min": geo.bounding_box_min,
"max": geo.bounding_box_max,
},
"topology_faces": geo.topology_faces,
}
fingerprint = compute_fingerprint(geometry_summary, material_name, is_foam)
# 4. 拉 hints 聚合
hints = await ExperienceFeedbackService().list_hints_for_task(
session=db_session,
stp_file_id=stp_file.id,
material_name=material_name,
is_foam=is_foam,
)
return ExperienceHintsResponse(
task_id=task_id,
stp_file_id=stp_file.id,
material_name=material_name,
is_foam=is_foam,
fingerprint=fingerprint,
hints=[ExperienceHintItem(**h) for h in hints],
)
src/moldinsight/models/__init__.py
+2
View File
@@ -14,6 +14,7 @@ from moldinsight.models.stp_analysis import (
DesignRecommendation,
AnalysisMetrics,
)
from moldinsight.models.experience_feedback import ExperienceFeedback
__all__ = [
"STPFile",
@@ -25,4 +26,5 @@ __all__ = [
"FeatureDetection",
"DesignRecommendation",
"AnalysisMetrics",
"ExperienceFeedback",
]
src/moldinsight/models/experience_feedback.py
+91
View File
@@ -0,0 +1,91 @@
"""moldinsight 域模型：老师傅经验反馈表。
Human-in-Loop 闭环（D17，2026-09）：老师傅对系统推荐方案给出"采纳 / 调整 / 拒绝"
反馈，按"产品指纹 + 工艺参数"为索引跨任务匹配；下次同指纹产品分析自动消费
（OCC worker payload 透传 → MultiSchemeMoldPlanner → PartingSchemeScorer 加成）。
跨模块桥接只保留裸 FK，不建 ORM relationship（base.py 约定）：
- processing_task_id -> processing_tasks.id
- stp_file_id -> stp_files.id
- user_id -> users.id
写入 feedback 时由 service 层填充 fingerprint JSON（bbox_aspect / volume_bucket /
face_bucket / undercut_class / material_family / is_foam）。fingerprint 是跨任务
匹配的索引列（PG 下有 GIN 索引支持 jsonb_path_query 类查询）。
"""
from sqlalchemy import Column, Integer, String, Text, DateTime, JSON, Float, ForeignKey, Index
from sqlalchemy.sql import func
from shared.models.base import Base
class ExperienceFeedback(Base):
"""老师傅经验反馈：方案级整体反馈 + 跨任务指纹匹配。
关键字段：
- feedback_status: 'adopted' / 'adjust' / 'rejected'
- fingerprint: 跨任务匹配键，结构见 moldinsight.services.experience_feedback_service
- expires_at: 90 天 TTL；写新反馈时同 stp_file_id 整体续期（D17 衰减机制）
"""
__tablename__ = "experience_feedback"
id = Column(Integer, primary_key=True, index=True)
# ── 任务归属（跨模块裸 FK，CASCADE 随任务 / 文件清理）──
processing_task_id = Column(
Integer,
ForeignKey("processing_tasks.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
stp_file_id = Column(
Integer,
ForeignKey("stp_files.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
# ── 方案标识 ──
scheme_id = Column(String(64), nullable=False, index=True)
scheme_axis = Column(String(1), nullable=False)
scheme_method = Column(String(50), nullable=True)
# ── 反馈主体 ──
feedback_status = Column(String(20), nullable=False, index=True)
feedback_reason = Column(Text, nullable=True)
adjust_suggestion = Column(Text, nullable=True)
# ── 上下文快照（用于回放）──
process_params_snapshot = Column(JSON, nullable=True)
fingerprint = Column(JSON, nullable=False)
confidence_at_submit = Column(Float, nullable=True)
score_at_submit = Column(Float, nullable=True)
# ── 审计（user_id RESTRICT：禁止级联删，保留审计归因）──
user_id = Column(
Integer,
ForeignKey("users.id", ondelete="RESTRICT"),
nullable=False,
index=True,
)
role_code = Column(String(50), nullable=False)
# ── 时间戳与衰减 ──
created_at = Column(DateTime, default=func.now(), index=True)
expires_at = Column(DateTime, nullable=True, index=True)
__table_args__ = (
Index(
"ix_experience_feedback_stp_axis_status",
"stp_file_id",
"scheme_axis",
"feedback_status",
),
)
def __repr__(self) -> str:
return (
f"<ExperienceFeedback(id={self.id}, stp_file_id={self.stp_file_id}, "
f"scheme_id='{self.scheme_id}', axis='{self.scheme_axis}', "
f"status='{self.feedback_status}')>"
)
src/moldinsight/services/experience_feedback_service.py
+394
View File
@@ -0,0 +1,394 @@
"""老师傅经验反馈服务：D17 Human-in-Loop 闭环。
三层使用：
1. `record_feedback` — 路由层 POST 调用；写入 ExperienceFeedback；同 stp_file_id
整体续期（D17 衰减机制）；只 flush，由路由 commit（D9 边界）。
2. `list_hints_for_task` — 路由层 GET 调用；按 stp_file_id + material_family +
is_foam 锚定，聚合返回前端 ResultView 用 hints 摘要。
3. `resolve_for_process_params` — ProcessingService 调用；返回 OCC worker payload
用的 hints dict，OCC 子进程透传给 MultiSchemeMoldPlanner。
"""
from datetime import datetime, timedelta, timezone
from typing import Dict, Any, List, Optional
from sqlalchemy import select, update, or_
from sqlalchemy.ext.asyncio import AsyncSession
from shared.models.identity import User, Role, UserRole
from shared.utils.logger import get_logger
from moldinsight.models import (
ExperienceFeedback,
ProcessingTask,
STPFile,
GeometryData,
MoldCavityData,
)
logger = get_logger(__name__)
FEEDBACK_TTL_DAYS = 90
def compute_fingerprint(
geometry_data: Optional[dict],
material_name: str,
is_foam: bool,
) -> Dict[str, str]:
"""计算产品指纹（跨任务匹配键）。
分桶策略见 plan §5.1：
- bbox_aspect: cube/compact/slab/elongated/long_bar
- volume_bucket: xs/s/m/l/xl（mm³ → cm³）
- face_bucket: simple/normal/complex/dense
- undercut_class: none/mild/moderate/heavy
- material_family: foam/abs/pp/pa/other（粗粒度，避免材料名变化导致匹配失效）
- is_foam: "true"/"false"
"""
geo = geometry_data or {}
dims = geo.get("bounding_box", {}).get("dimensions") or [0, 0, 0]
sorted_dims = sorted(dims or [0, 0, 0])
if sorted_dims[0] <= 0:
ratio = 1.0
else:
ratio = sorted_dims[1] / sorted_dims[0]
if ratio < 1.0:
bbox_aspect = "cube"
elif ratio < 1.5:
bbox_aspect = "compact"
elif ratio < 3.0:
bbox_aspect = "slab"
elif ratio < 6.0:
bbox_aspect = "elongated"
else:
bbox_aspect = "long_bar"
volume_cm3 = (geo.get("volume", 0) or 0) / 1000.0
if volume_cm3 < 10:
volume_bucket = "xs"
elif volume_cm3 < 100:
volume_bucket = "s"
elif volume_cm3 < 500:
volume_bucket = "m"
elif volume_cm3 < 2000:
volume_bucket = "l"
else:
volume_bucket = "xl"
face_count = (
geo.get("topology_faces", 0)
or geo.get("topology", {}).get("faces", 0)
or 0
)
if face_count < 100:
face_bucket = "simple"
elif face_count < 500:
face_bucket = "normal"
elif face_count < 2000:
face_bucket = "complex"
else:
face_bucket = "dense"
undercut_count = geo.get("undercut_count", 0) or 0
if undercut_count == 0:
undercut_class = "none"
elif undercut_count < 4:
undercut_class = "mild"
elif undercut_count < 9:
undercut_class = "moderate"
else:
undercut_class = "heavy"
mat_lower = (material_name or "").lower()
if "al" in mat_lower and "si" in mat_lower:
material_family = "foam"
elif "abs" in mat_lower:
material_family = "abs"
elif "pp" in mat_lower:
material_family = "pp"
elif "pa" in mat_lower:
material_family = "pa"
else:
material_family = "other"
return {
"bbox_aspect": bbox_aspect,
"volume_bucket": volume_bucket,
"face_bucket": face_bucket,
"undercut_class": undercut_class,
"material_family": material_family,
"is_foam": "true" if is_foam else "false",
}
class ExperienceFeedbackService:
"""老师傅经验反馈：写入 / 同指纹 hints 摘要 / OCC worker 用 hints 解析。"""
def __init__(self) -> None:
# 显式无依赖：与 processing_service / task_storage_service 范式一致
pass
async def record_feedback(
self,
session: AsyncSession,
*,
task_id: str,
scheme_id: str,
feedback_status: str,
feedback_reason: Optional[str],
adjust_suggestion: Optional[str],
user: User,
confidence_at_submit: Optional[float] = None,
score_at_submit: Optional[float] = None,
process_params_snapshot: Optional[Dict[str, Any]] = None,
) -> ExperienceFeedback:
"""写入方案级反馈。仅 flush，由路由 commit（D9 边界）。
同时同 stp_file_id 整体续期（expires_at = now() + 90d），
"写入即消费"的语义保证新反馈立刻进入有效集。
"""
if feedback_status not in ("adopted", "adjust", "rejected"):
raise ValueError(f"feedback_status 非法: {feedback_status}")
# 1. 找 task + stp_file
row = await session.execute(
select(ProcessingTask, STPFile)
.join(STPFile, ProcessingTask.stp_file_id == STPFile.id)
.where(ProcessingTask.task_id == task_id)
)
row = row.first()
if not row:
raise ValueError(f"任务不存在: {task_id}")
processing_task, stp_file = row
# 2. 解析 material_name + is_foam_material
params = processing_task.parameters or {}
material_name = str(params.get("material") or "ABS")
is_foam_material = bool(params.get("is_foam_material", False))
# 3. 取 geometry_data / cavity_data 计算 fingerprint
geometry_summary: Dict[str, Any] = {}
geo_row = await session.execute(
select(GeometryData).where(GeometryData.stp_file_id == stp_file.id)
)
geo = geo_row.scalar_one_or_none()
if geo is not None:
geometry_summary = {
"volume": geo.volume,
"bounding_box": {
"min": geo.bounding_box_min,
"max": geo.bounding_box_max,
},
"topology_faces": geo.topology_faces,
}
cavity_row = await session.execute(
select(MoldCavityData).where(MoldCavityData.stp_file_id == stp_file.id)
)
cavity = cavity_row.scalar_one_or_none()
if cavity is not None:
ki = cavity.cavity_key_info or {}
side_actions = (
ki.get("quality_considerations") if isinstance(ki, dict) else None
) or {}
if isinstance(side_actions, dict):
geometry_summary["undercut_count"] = side_actions.get("undercut_count", 0) or 0
fingerprint = compute_fingerprint(geometry_summary, material_name, is_foam_material)
# 4. 找 scheme 的 axis + method（从 cavity_key_info.candidate_schemes）
scheme_axis = "Z"
scheme_method: Optional[str] = None
if cavity is not None:
ki = cavity.cavity_key_info or {}
candidate_schemes = ki.get("candidate_schemes") if isinstance(ki, dict) else None
if isinstance(candidate_schemes, list):
for cs in candidate_schemes:
if isinstance(cs, dict) and cs.get("scheme_id") == scheme_id:
scheme_axis = (
cs.get("axis")
or cs.get("parting_axis")
or "Z"
)
scheme_method = cs.get("method") or cs.get("scheme_method")
break
# 5. 取用户角色（显式 JOIN 查询，避免 user.roles 在跨 session 下 lazy load 失败）
role_codes = await self._fetch_user_role_codes(session, user.id)
if getattr(user, "is_superuser", False):
role_code = "admin"
elif role_codes:
role_code = role_codes[0]
else:
role_code = "user"
# 6. 写 ExperienceFeedback
now = datetime.now(timezone.utc)
new_ttl = now + timedelta(days=FEEDBACK_TTL_DAYS)
feedback = ExperienceFeedback(
processing_task_id=processing_task.id,
stp_file_id=stp_file.id,
scheme_id=scheme_id,
scheme_axis=str(scheme_axis)[:1],
scheme_method=scheme_method,
feedback_status=feedback_status,
feedback_reason=feedback_reason,
adjust_suggestion=adjust_suggestion,
process_params_snapshot=process_params_snapshot,
fingerprint=fingerprint,
confidence_at_submit=confidence_at_submit,
score_at_submit=score_at_submit,
user_id=user.id,
role_code=role_code,
expires_at=new_ttl,
)
session.add(feedback)
await session.flush()
# 7. 同 stp_file_id 整体续期（D17 衰减：仅刷新过期 / NULL 行）
await session.execute(
update(ExperienceFeedback)
.where(
ExperienceFeedback.stp_file_id == stp_file.id,
or_(
ExperienceFeedback.expires_at.is_(None),
ExperienceFeedback.expires_at < now,
),
)
.values(expires_at=new_ttl)
)
await session.flush()
return feedback
async def list_hints_for_task(
self,
session: AsyncSession,
*,
stp_file_id: int,
material_name: str,
is_foam: bool,
limit: int = 10,
) -> List[Dict[str, Any]]:
"""同指纹历史采纳摘要，给前端 ResultView 用。
排除 expires_at < now() 的过期反馈；按 material_family + is_foam 锚定；
按 scheme_axis 聚合（adopted/rejected/adjust 计数 + 加权 confidence）。
"""
now = datetime.now(timezone.utc)
is_foam_str = "true" if is_foam else "false"
mat_lower = (material_name or "").lower()
if "al" in mat_lower and "si" in mat_lower:
material_family = "foam"
elif "abs" in mat_lower:
material_family = "abs"
elif "pp" in mat_lower:
material_family = "pp"
elif "pa" in mat_lower:
material_family = "pa"
else:
material_family = "other"
rows = await session.execute(
select(ExperienceFeedback)
.where(
ExperienceFeedback.stp_file_id == stp_file_id,
or_(
ExperienceFeedback.expires_at.is_(None),
ExperienceFeedback.expires_at > now,
),
)
.order_by(ExperienceFeedback.created_at.desc())
.limit(limit * 4)
)
feedbacks = rows.scalars().all()
axis_summary: Dict[str, Dict[str, Any]] = {}
for fb in feedbacks:
fp = fb.fingerprint or {}
# 锚定：material_family + is_foam 必须一致
if fp.get("material_family") != material_family:
continue
if fp.get("is_foam") != is_foam_str:
continue
axis = fb.scheme_axis or "Z"
summary = axis_summary.setdefault(axis, {
"scheme_axis": axis,
"adopted_count": 0,
"rejected_count": 0,
"adjust_count": 0,
"sample_count": 0,
})
summary["sample_count"] += 1
if fb.feedback_status == "adopted":
summary["adopted_count"] += 1
elif fb.feedback_status == "rejected":
summary["rejected_count"] += 1
elif fb.feedback_status == "adjust":
summary["adjust_count"] += 1
result: List[Dict[str, Any]] = []
for axis, s in axis_summary.items():
total = s["adopted_count"] + s["rejected_count"] + s["adjust_count"]
if total == 0:
continue
confidence = (s["adopted_count"] - s["rejected_count"]) / max(total, 1)
confidence = max(-1.0, min(1.0, confidence))
weight = max(0.0, confidence) # weight 仅正向上有效（不"扣分"老算法）
s["confidence"] = round(confidence, 3)
s["weight"] = round(weight, 3)
result.append(s)
result.sort(key=lambda x: (-x["weight"], -x["sample_count"]))
return result[:limit]
@staticmethod
async def _fetch_user_role_codes(session: AsyncSession, user_id: int) -> List[str]:
"""显式 JOIN 拿用户角色 codes，避免 user.roles 在跨 session 下 detached lazy load 失败。
测试场景下 user 是从一个 session 取出传到另一个 session，访问 user.roles 会触发
DetachedInstanceError；生产场景下也以显式查询更稳（不依赖 ORM relationship 配置）。
"""
rows = await session.execute(
select(Role.code)
.join(UserRole, UserRole.role_id == Role.id)
.where(UserRole.user_id == user_id)
)
return [row[0] for row in rows.all()]
async def resolve_for_process_params(
self,
session: AsyncSession,
*,
task_id: str,
process_params: Dict[str, Any],
bucket_hint: Optional[Dict[str, str]] = None,
) -> List[Dict[str, Any]]:
"""返回 OCC worker payload 用的 hints。
按 fingerprint bucket 找同指纹最近 N 条采纳，每条
{scheme_axis, weight, sample_count, summary}，传给 PartingSchemeScorer
加成和 PartingCandidateGenerator 优先级加成。
"""
row = await session.execute(
select(ProcessingTask, STPFile)
.join(STPFile, ProcessingTask.stp_file_id == STPFile.id)
.where(ProcessingTask.task_id == task_id)
)
row = row.first()
if not row:
return []
processing_task, stp_file = row
params = processing_task.parameters or {}
material_name = str(params.get("material") or "ABS")
is_foam = bool(params.get("is_foam_material", False))
return await self.list_hints_for_task(
session,
stp_file_id=stp_file.id,
material_name=material_name,
is_foam=is_foam,
)
# 模块级单例（与 processing_service / task_storage_service 范式一致）
experience_feedback_service = ExperienceFeedbackService()
src/shared/database/init_db.py
+54 -23
View File
@@ -65,55 +65,86 @@ DEFAULT_PERMISSIONS = [
{"code": "view_users", "name": "查看用户", "module": "admin"},
{"code": "manage_users", "name": "管理用户", "module": "admin"},
{"code": "manage_roles", "name": "管理角色", "module": "admin"},
# D17 Human-in-Loop：老师傅经验反馈
{"code": "view_experience_feedback", "name": "查看老师傅反馈", "module": "moldinsight"},
{"code": "feedback_experience_hint", "name": "提交方案级反馈", "module": "moldinsight"},
{"code": "manage_experience_feedback", "name": "管理老师傅反馈", "module": "moldinsight"},
]
DEFAULT_ROLES = [
{"code": "admin", "name": "管理员", "description": "系统管理员，拥有所有权限", "is_system": True, "permissions": ["view_dashboard", "view_moldinsight", "upload_file", "view_history", "view_inventory", "manage_inventory", "view_products", "manage_products", "view_suppliers", "manage_suppliers", "view_customers", "manage_customers", "view_finance", "manage_receipts", "manage_payments", "void_finance_transaction", "view_users", "manage_users", "manage_roles"]},
{"code": "admin", "name": "管理员", "description": "系统管理员，拥有所有权限", "is_system": True, "permissions": ["view_dashboard", "view_moldinsight", "upload_file", "view_history", "view_inventory", "manage_inventory", "view_products", "manage_products", "view_suppliers", "manage_suppliers", "view_customers", "manage_customers", "view_finance", "manage_receipts", "manage_payments", "void_finance_transaction", "view_users", "manage_users", "manage_roles", "view_experience_feedback", "feedback_experience_hint", "manage_experience_feedback"]},
{"code": "user", "name": "普通用户", "description": "普通用户，可使用模具分析和查看库存", "is_system": False, "permissions": ["view_dashboard", "view_moldinsight", "upload_file", "view_history", "view_inventory", "view_products", "view_suppliers", "view_customers", "view_finance", "manage_receipts", "manage_payments"]},
{"code": "viewer", "name": "只读用户", "description": "只读用户，只能查看数据", "is_system": False, "permissions": ["view_dashboard", "view_moldinsight", "view_history", "view_inventory", "view_products", "view_suppliers", "view_customers", "view_finance"]},
# D17 Human-in-Loop：工艺工程师角色——可查看 + 提交老师傅反馈
{"code": "process_engineer", "name": "工艺工程师", "description": "工艺工程师，可查看 + 提交老师傅经验反馈", "is_system": False, "permissions": ["view_dashboard", "view_moldinsight", "upload_file", "view_history", "view_experience_feedback", "feedback_experience_hint"]},
]
async def init_permissions(session):
"""初始化权限"""
"""初始化权限（按 code 补登：已存在跳过，缺失新增）
设计要点（D17 修复）：
- 旧实现 `if existing_perms: return` 会让既存 DB 启动期漏掉新增权限码
- 改为按 code 比对：已存在的 permission 保留 id（避免 FK 引用失效），
缺失的新增；这样后续 DEFAULT_PERMISSIONS 追加的项也能在升级时落到既存 DB
"""
result = await session.execute(select(Permission))
existing_perms = result.scalars().all()
if existing_perms:
logger.info("权限已初始化")
return
perm_map = {}
existing_perms = {p.code: p for p in result.scalars().all()}
perm_map = {p.code: p.id for p in existing_perms.values()}
new_count = 0
for perm_data in DEFAULT_PERMISSIONS:
if perm_data["code"] in existing_perms:
continue
perm = Permission(**perm_data)
session.add(perm)
await session.flush()
perm_map[perm.code] = perm.id
logger.info(f"创建了 {len(DEFAULT_PERMISSIONS)} 个权限")
new_count += 1
if not existing_perms:
logger.info(f"创建了 {len(DEFAULT_PERMISSIONS)} 个权限")
elif new_count:
logger.info(f"补登了 {new_count} 个权限（既有 DB 升级）")
else:
logger.info("权限已初始化（无新增）")
return perm_map
async def init_roles(session, perm_map):
"""初始化角色"""
"""初始化角色（按 code 补登：已存在跳过，缺失新建 + 完整绑定 permissions）
设计要点（D17 修复）：
- 旧实现 `if existing_roles: return` 会让既存 DB 启动期漏掉新角色
- 改为按 code 比对：已存在的角色不重置其 RolePermission 绑定
（避免重建关联破坏 user / role 关系），缺失的角色按 DEFAULT_ROLES 完整创建
"""
result = await session.execute(select(Role))
existing_roles = result.scalars().all()
if existing_roles:
logger.info("角色已初始化")
return
existing_roles = {r.code: r for r in result.scalars().all()}
new_count = 0
for role_data in DEFAULT_ROLES:
perm_ids = [perm_map[code] for code in role_data.pop("permissions")]
code = role_data["code"]
if code in existing_roles:
continue
perm_codes = role_data.pop("permissions")
role = Role(**role_data)
session.add(role)
await session.flush()
for perm_id in perm_ids:
for perm_code in perm_codes:
perm_id = perm_map.get(perm_code)
if perm_id is None:
continue
rp = RolePermission(role_id=role.id, permission_id=perm_id)
session.add(rp)
logger.info(f"创建了 {len(DEFAULT_ROLES)} 个角色")
new_count += 1
if not existing_roles:
logger.info(f"创建了 {len(DEFAULT_ROLES)} 个角色")
elif new_count:
logger.info(f"补登了 {new_count} 个角色（既有 DB 升级）")
else:
logger.info("角色已初始化（无新增）")
async def create_admin_user(session):