"""材质牌号服务
封装 material_grades 表的查询与写入逻辑，供两处复用：
1. 打印参数补充对话框：get_grades_by_category() 按阀门零件类别分组返回牌号列表；
2. 材质牌号管理界面：list_all / create / update / delete 提供完整 CRUD。
数据模型说明（与领域约束对齐）：
- grade + standard_code + element 唯一确定一条「牌号在某标准下的某元素性能范围」记录；
- category_large / category_medium 表示材料形态维度（铸件类/锻件类/板材类/棒材类），
同一牌号挂不同形态/标准时范围可能不同；
- category 表示阀门零件维度（body/stem/disc/common），仅用于打印对话框分组，
与材料形态维度正交。
"""
from __future__ import annotations
import random
from typing import Any
from sqlalchemy import distinct, func
from sqlalchemy.orm import Session
from certflow.data.common_grades import COMMON_GRADES
from certflow.models.material_grade import MaterialGrade
from certflow.services.bom_material_service import canonical_grade
from certflow.utils.logger import logger
# 阀门零件类别合法值域
VALID_CATEGORIES = ("body", "stem", "disc", "common")
# 数据类型合法值域
VALID_KINDS = ("chemical", "mechanical")
# 力学性能单位映射：伸长率/收缩率为百分数，冲击功为焦耳，其余默认 MPa
_MECH_UNIT: dict[str, str] = {
"伸长率": "%",
"收缩率": "%",
"冲击功": "J",
}
# 力学性能元素名 → 报告模板列键别名（质保书用 ReL/Rm/A/Z，材质证明书用 σs/σb/δ/ψ）。
# 一个物理量映射到两套键，使同一 MaterialPart 同时满足两类模板，引擎按各自 spec 取键即可。
_MECH_KEY_ALIASES: dict[str, tuple[str, ...]] = {
"屈服强度": ("ReL", "σs"),
"抗拉强度": ("Rm", "σb"),
"伸长率": ("A", "δ"),
"收缩率": ("Z", "ψ"),
}
[文档]
class MaterialGradeService:
"""材质牌号服务
封装 material_grades 表的读写，所有写操作由调用方（控制器）负责提交事务。
"""
# C3：种子数据集由独立数据模块维护（覆盖常见阀门材料，含完整 standard_min/max 与
# is_active=True）；此处作为类属性暴露，供 seed_common_grades 引用。
_SEED_GRADES = COMMON_GRADES
def __init__(self, session: Session):
"""初始化服务
Args:
session: SQLAlchemy 数据库会话
"""
self.session = session
# ============================================================
# 查询
# ============================================================
[文档]
def list_all(self, active_only: bool = False) -> list[MaterialGrade]:
"""返回全部材质牌号记录（按牌号、标准号、元素排序）
Args:
active_only: 仅返回启用记录
Returns:
MaterialGrade 记录列表
"""
query = self.session.query(MaterialGrade)
if active_only:
query = query.filter(MaterialGrade.is_active.is_(True))
return query.order_by(
MaterialGrade.grade,
MaterialGrade.standard_code,
MaterialGrade.element,
).all()
[文档]
def get_by_id(self, grade_id: int) -> MaterialGrade | None:
"""按主键获取单条记录。
Args:
grade_id: 主键 ID。
Returns:
MaterialGrade | None: 对应记录，不存在时返回 None。
"""
return self.session.get(MaterialGrade, grade_id)
[文档]
def get_by_grade(self, grade: str) -> list[MaterialGrade]:
"""按牌号查询全部材质记录（化学成分 + 力学性能），大小写不敏感。
C0/C3：供 ``ReportBomEnricher`` 在报告生成时按命中牌号取化学/力学性能范围，
比 ``list_all``（全量、调用方自行过滤）更高效、语义更清晰；与
``generate_composition_values`` 同口径经 ``canonical_grade`` 归并。
Args:
grade: 材质牌号（如 "WCB" / "20" / "304"）。
Returns:
list[MaterialGrade]：该牌号的全部记录（按 标准号、元素 排序）；
牌号空 / 归一化后为空 → 返回 ``[]``；不存在该牌号 → 返回 ``[]``。
"""
grade = canonical_grade(grade)
if not grade:
return []
return (
self.session.query(MaterialGrade)
.filter(func.lower(MaterialGrade.grade) == grade)
.order_by(MaterialGrade.standard_code, MaterialGrade.element)
.all()
)
[文档]
def get_distinct_grades(self, active_only: bool = True) -> list[str]:
"""返回去重后的牌号列表（用于下拉框等）。
Args:
active_only: 仅返回启用记录，默认 True。
Returns:
list[str]: 去重并排序后的牌号列表。
"""
query = self.session.query(distinct(MaterialGrade.grade))
if active_only:
query = query.filter(MaterialGrade.is_active.is_(True))
return sorted(row[0] for row in query.all() if row[0])
[文档]
def get_grades_by_category(self) -> dict[str, list[str]]:
"""按阀门零件类别分组返回牌号列表（仅启用记录）
分组规则：
- category 字段非空 → 归入对应类别（body/stem/disc）；
- category 为空 → 归入 "common"，对话框会将 common 合并进各零件下拉框。
Returns:
{类别: [牌号列表]}，如 {"body": ["WCB", "CF8"], "common": ["304"]}
空表时返回 {}。
"""
rows = (
self.session.query(MaterialGrade.grade, MaterialGrade.category)
.filter(MaterialGrade.is_active.is_(True))
.all()
)
if not rows:
return {}
grouped: dict[str, list[str]] = {}
for grade, category in rows:
if not grade:
continue
key = category if category in VALID_CATEGORIES else "common"
grouped.setdefault(key, [])
if grade not in grouped[key]:
grouped[key].append(grade)
# 每类排序，保证 UI 稳定
return {k: sorted(v) for k, v in grouped.items()}
# ============================================================
# 写入
# ============================================================
[文档]
def create(self, data: dict[str, Any]) -> MaterialGrade:
"""新增一条材质牌号记录
Args:
data: 字段字典，至少包含 grade / standard_code / element；
可选 category_large / category_medium / category /
standard_min / standard_max / decimal_places /
actual_min / actual_max / is_active
Returns:
新建的 MaterialGrade 实例
Raises:
ValueError: 缺少必填字段，或 (grade, standard_code, element) 已存在
"""
grade = (data.get("grade") or "").strip()
standard_code = (data.get("standard_code") or "").strip()
element = (data.get("element") or "").strip()
if not grade or not standard_code or not element:
raise ValueError("牌号(grade)、标准号(standard_code)、元素(element) 均为必填")
# 唯一性校验
self._assert_no_clash(grade, standard_code, element, exclude_id=-1)
mg = MaterialGrade(
grade=grade,
standard_code=standard_code,
category_large=(data.get("category_large") or None),
category_medium=(data.get("category_medium") or None),
category=self._normalize_category(data.get("category")),
kind=self._normalize_kind(data.get("kind")),
element=element,
standard_min=data.get("standard_min"),
standard_max=data.get("standard_max"),
decimal_places=data.get("decimal_places", 2),
actual_min=data.get("actual_min"),
actual_max=data.get("actual_max"),
is_active=data.get("is_active", True),
)
self.session.add(mg)
self.session.flush()
logger.info(f"新增材质牌号记录: id={mg.id}, grade={grade}, element={element}")
return mg
[文档]
def update(self, grade_id: int, data: dict[str, Any]) -> MaterialGrade | None:
"""更新指定记录
Args:
grade_id: 主键
data: 待更新字段字典（空值字段忽略）
Returns:
更新后的记录；id 不存在返回 None
"""
mg = self.get_by_id(grade_id)
if mg is None:
logger.warning(f"更新失败：材质牌号 id={grade_id} 不存在")
return None
# 牌号/标准号/元素若变更，需重新做唯一性校验
new_grade = (data.get("grade") or mg.grade).strip()
new_std = (data.get("standard_code") or mg.standard_code).strip()
new_elem = (data.get("element") or mg.element).strip()
if new_grade != mg.grade or new_std != mg.standard_code or new_elem != mg.element:
self._assert_no_clash(new_grade, new_std, new_elem, exclude_id=grade_id)
mg.grade = new_grade
mg.standard_code = new_std
mg.element = new_elem
mg.category_large = data.get("category_large", mg.category_large)
mg.category_medium = data.get("category_medium", mg.category_medium)
if "category" in data:
mg.category = self._normalize_category(data["category"])
if "kind" in data:
mg.kind = self._normalize_kind(data["kind"])
# 可选标量字段：仅在传入时覆盖
scalar_fields = (
"standard_min",
"standard_max",
"decimal_places",
"actual_min",
"actual_max",
"is_active",
)
for f in scalar_fields:
if f in data:
setattr(mg, f, data[f])
self.session.flush()
logger.info(f"更新材质牌号记录: id={grade_id}, grade={new_grade}, element={new_elem}")
return mg
def _assert_no_clash(
self, grade: str, standard_code: str, element: str, exclude_id: int
) -> None:
"""校验 (grade, standard_code, element) 在排除自身外是否已存在
Raises:
ValueError: 存在冲突记录
"""
clash = (
self.session.query(MaterialGrade)
.filter(
MaterialGrade.grade == grade,
MaterialGrade.standard_code == standard_code,
MaterialGrade.element == element,
MaterialGrade.id != exclude_id,
)
.first()
)
if clash:
raise ValueError(f"记录已存在: 牌号={grade}, 标准号={standard_code}, 元素={element}")
[文档]
def delete(self, grade_id: int) -> bool:
"""删除指定记录
Args:
grade_id: 主键
Returns:
是否删除成功
"""
mg = self.get_by_id(grade_id)
if mg is None:
logger.warning(f"删除失败：材质牌号 id={grade_id} 不存在")
return False
self.session.delete(mg)
self.session.flush()
logger.info(f"删除材质牌号记录: id={grade_id}, grade={mg.grade}")
return True
# ============================================================
# 内部工具
# ============================================================
@staticmethod
def _normalize_category(value: Any) -> str | None:
"""规范化阀门零件类别字段
空值 / 非法值 → None（归入 common）；合法值原样返回。
"""
if not value:
return None
value = str(value).strip().lower()
return value if value in VALID_CATEGORIES else None
@staticmethod
def _normalize_kind(value: Any) -> str | None:
"""规范化数据类型字段
空值 / 非法值 → "chemical"（默认化学成分）；合法值原样返回。
"""
if not value:
return "chemical"
value = str(value).strip().lower()
return value if value in VALID_KINDS else "chemical"
# ============================================================
# 质保书模块：成分生成 + 种子
# ============================================================
[文档]
def generate_composition(self, grade: str, category_large: str | None = None) -> dict[str, str]:
"""按牌号生成质保书所需的化学成分 / 力学性能字符串
从 material_grades 表中取出该牌号（可按材料形态过滤）的全部启用的
化学成分(chemical)与力学性能(mechanical)记录，对每条记录用 RandBetween
在 [standard_min, standard_max] 范围内按 decimal_places 取随机实测值，
分别拼成 "C:0.20%, Si:0.40%, ..." 与 "屈服强度:245MPa, ..." 形式的字符串。
Args:
grade: 材质牌号，如 "WCB" / "304"
category_large: 可选材料形态过滤（铸件类/锻件类/板材类/棒材类）
Returns:
{"chemical_composition": str, "mechanical_property": str}
牌号无数据（或库为空）时两者均返回空串，由调用方决定是否回退占位值。
Examples:
>>> service = MaterialGradeService(session)
>>> result = service.generate_composition("WCB", category_large="铸件类")
>>> print(result["chemical_composition"])
"""
# 别名归并：BOM 中 "20"/"q335b" 等同写归一化到库内规范牌号；
# 大小写不敏感比较，兼容库内混合大小写键（WCB/20#/16Mn）。
grade = canonical_grade(grade)
query = self.session.query(MaterialGrade).filter(
func.lower(MaterialGrade.grade) == grade,
MaterialGrade.is_active.is_(True),
)
if category_large:
query = query.filter(MaterialGrade.category_large == category_large)
rows = query.order_by(MaterialGrade.kind, MaterialGrade.element).all()
chem_parts: list[str] = []
mech_parts: list[str] = []
for r in rows:
decimals = r.decimal_places if r.decimal_places is not None else 2
value = self._randbetween_value(r.standard_min, r.standard_max, decimals)
if value is None:
continue
text = f"{value:.{decimals}f}"
kind = r.kind or "chemical"
if kind == "mechanical":
# 力学性能单位：伸长率/收缩率为百分数，冲击功为焦耳，其余为 MPa
unit = _MECH_UNIT.get(r.element, "MPa")
mech_parts.append(f"{r.element}:{text}{unit}")
else:
chem_parts.append(f"{r.element}:{text}%")
return {
"chemical_composition": ", ".join(chem_parts),
"mechanical_property": ", ".join(mech_parts),
}
@staticmethod
def _randbetween_value(min_v: float | None, max_v: float | None, decimals: int) -> float | None:
"""在 [min, max] 内取一个物理自洽的随机实测值，按 decimals 四舍五入。
C2：不再用 ``random.uniform`` 均匀采样（易在区间两端聚集、出现「所有元素同时贴边」
的非物理组合），改为**中值 + 小扰动**——以三角分布取样，众数落在区间中点，实测值
自然向标称中值靠拢、越靠边界概率越低，更贴近真实炉批。
- min / max 均为 None → 无范围，返回 None（跳过该元素）；
- 仅有一端 → 取该端定值（如 P≤0.04 取 0.04）。
"""
if min_v is None and max_v is None:
return None
if min_v is None:
lo = hi = max_v
elif max_v is None:
lo = hi = min_v
else:
lo, hi = min_v, max_v
if lo == hi:
return round(lo, decimals)
# 三角分布：mode=中点 → 中值+小扰动，边界概率低，物理自洽
return round(random.triangular(lo, hi, (lo + hi) / 2), decimals)
[文档]
def generate_composition_values(
self, grade: str, category_large: str | None = None
) -> dict[str, dict[str, float]]:
"""按牌号生成质保书材质块所需的**数值字典**（供填充引擎逐列写入）。
与 :meth:`generate_composition`（返回展示字符串）同源取数，但返回 float 数值并把
列键规范化到报告模板列：化学成分直接用元素符号（C/Mn/Si/...，两类模板一致），
力学性能经 :data:`_MECH_KEY_ALIASES` 映射到 ReL/Rm/A/Z 与 σs/σb/δ/ψ 两套别名，
使返回值同时适配质保书与材质证明书模板。
Args:
grade: 材质牌号，如 "WCB" / "304"。
category_large: 可选材料形态过滤（铸件类/锻件类/板材类/棒材类）。
Returns:
{"化学成分": {元素: 值}, "力学性能": {列键: 值}}；牌号无数据时两者均为空 dict。
"""
# 别名归并 + 大小写不敏感比较（同 generate_composition）
grade = canonical_grade(grade)
query = self.session.query(MaterialGrade).filter(
func.lower(MaterialGrade.grade) == grade,
MaterialGrade.is_active.is_(True),
)
if category_large:
query = query.filter(MaterialGrade.category_large == category_large)
rows = query.order_by(MaterialGrade.kind, MaterialGrade.element).all()
chem: dict[str, float] = {}
mech: dict[str, float] = {}
for r in rows:
decimals = r.decimal_places if r.decimal_places is not None else 2
value = self._randbetween_value(r.standard_min, r.standard_max, decimals)
if value is None:
continue
if (r.kind or "chemical") == "mechanical":
for key in _MECH_KEY_ALIASES.get(r.element, (r.element,)):
mech[key] = value
else:
chem[r.element] = value
return {"化学成分": chem, "力学性能": mech}
# 常见牌号的化学成分 + 力学性能种子数据已迁至 ``certflow.data.common_grades``
# （COMMON_GRADES，C3 扩展覆盖 + 保证 is_active 与 standard_min/max 完整）。
# 本类保留 ``_SEED_GRADES`` 别名指向同一对象，供潜在外部引用兼容。
[文档]
def seed_common_grades(self) -> int:
"""幂等灌入常见牌号的化学成分 + 力学性能示例数据
仅当库中不存在任何 material_grades 记录时才写入，重复调用安全。
供管理界面「载入示例」按钮、测试初始化使用。
C3 保证：种子每条元素记录均带 ``standard_min`` / ``standard_max``
（单侧上限元素取 min=0.00）且 ``is_active=True``，从源头杜绝
「范围缺失 → 报告该列恒空」。若种子数据自身出现两端皆空的元素，
则跳过该元素（不写入），并记录告警，保证落库数据必然完整。
Returns:
实际写入的记录条数（0 表示已存在，未写入）
"""
if self.session.query(MaterialGrade).first() is not None:
logger.info("material_grades 已有数据，跳过种子灌入")
return 0
count = 0
for spec in self._SEED_GRADES:
for elem, mn, mx, dec, kind in spec["elements"]:
# C3：两端皆空的元素不落库（缺失范围会让报告该列恒空），跳过并告警
if mn is None and mx is None:
logger.warning(f"种子跳过两端皆空的元素: grade={spec['grade']}, element={elem}")
continue
# 单侧缺失按 0.00 兜底，保证 standard_min/max 完整可查
std_min = 0.0 if mn is None else mn
std_max = 0.0 if mx is None else mx
self.session.add(
MaterialGrade(
grade=spec["grade"],
standard_code=spec["standard_code"],
category_large=spec["category_large"],
category_medium=spec["category_medium"],
category=spec["category"],
kind=kind,
element=elem,
standard_min=std_min,
standard_max=std_max,
decimal_places=dec,
is_active=True,
)
)
count += 1
self.session.flush()
logger.info(f"种子灌入完成: {count} 条材质牌号记录")
return count
# ============================================================
# 数据质量诊断（C3：范围缺失 → 报告恒空的可见性）
# ============================================================
[文档]
def find_incomplete_records(self) -> list[MaterialGrade]:
"""返回「启用但范围缺失」的牌号记录——这些会让质保书/材质报告对应列恒空。
判定：``is_active == True`` 且 ``standard_min`` 或 ``standard_max`` 为 NULL。
这类记录可被 ``generate_composition_values`` 命中（is_active 通过），但其
某元素无范围可取 → ``_randbetween_value`` 返回 None → 该元素被跳过 → 报告列空。
Returns:
缺失范围的启用记录列表（按 牌号、元素 排序）；空表/无缺失返回 []。
"""
return (
self.session.query(MaterialGrade)
.filter(
MaterialGrade.is_active.is_(True),
MaterialGrade.standard_min.is_(None) | MaterialGrade.standard_max.is_(None),
)
.order_by(MaterialGrade.grade, MaterialGrade.element)
.all()
)
[文档]
def count_incomplete_records(self) -> int:
"""启用但范围缺失的记录条数（供管理界面提示数据质量）。"""
return len(self.find_incomplete_records())
[文档]
def get_data_quality_summary(self) -> dict[str, int]:
"""材质牌号表数据质量概览（供管理界面状态栏展示）。
Returns:
{total, active, inactive, incomplete} 四项计数。
"""
total = self.session.query(MaterialGrade).count()
active = self.session.query(MaterialGrade).filter(MaterialGrade.is_active.is_(True)).count()
return {
"total": total,
"active": active,
"inactive": total - active,
"incomplete": self.count_incomplete_records(),
}