"""公称压力(PN)处理服务
对标 VBA function.vb 确认压力() 函数，提供：
1. 型号→压力对照查询（从 SQLite model_param_mappings 表）
2. 英制(LB)/公制(MPa)自动判断
3. 单位剥离和标准化
4. 不同模板类型的显示文本生成
5. 自学习机制（新型号自动记录到数据库）
v2.0 新增：
- 支持三种模板类型（全中文/中英文/俄英文）的 PN 显示格式
- 全中文模板：纯数值（底图已印 MPa），英制带 Lb
- 中英文/俄英文模板：数值+单位（底图可能无单位印刷）
"""
from __future__ import annotations
import re
from typing import Any
from sqlalchemy.orm import Session
from certflow.utils.logger import logger
# ============================================================
# 英制压力规则
# ============================================================
# 对标 VBA 确认压力() 第312-315行
LB_TO_MPA_MAP: dict[int, float] = {
150: 2.0,
300: 5.0,
600: 10.0,
800: 13.0,
900: 15.0,
1500: 25.0,
2500: 42.0,
}
# 英制型号关键词（VBA print.vb 第634行）
IMPERIAL_KEYWORDS = ["LB", "2500"]
# 单位后缀（需要剥离的）
UNIT_SUFFIXES = ["MPA", "MPa", "Mpa", "LB", "Lb", "lb"]
[文档]
class PNService:
"""公称压力处理服务
负责从产品型号解析/查询公称压力，提供统一的压力值表示。
对标 VBA function.vb 确认压力() 函数（第295-399行）。
Attributes:
session: SQLAlchemy 数据库会话
"""
def __init__(self, session: Session | None = None) -> None:
self.session = session
# ============================================================
# 英制判断
# ============================================================
[文档]
@staticmethod
def is_imperial(model: str) -> bool:
"""判断型号是否为英制压力
对标 VBA print.vb 第634行：
.. code-block:: text
If (UCase(.Cells(iR, 14+OfSet)) Like "*LB*") Or
(UCase(.Cells(iR, 14+OfSet)) Like "*2500*") Then
Args:
model: 产品型号字符串，如 "Z41H-150LB"
Returns:
bool: 是否英制
"""
if not model:
return False
upper_model = model.upper()
return any(kw.upper() in upper_model for kw in IMPERIAL_KEYWORDS)
[文档]
@staticmethod
def lb_to_mpa(lb_level: int) -> float:
"""英制 LB 等级转换为公制 MPa 压力值
对标 VBA 确认压力() 第312-315行：
If 压力返回值 = "150Lb" Then 压力返回值 = 2#
If 压力返回值 = "300Lb" Then 压力返回值 = 5#
If 压力返回值 = "600Lb" Then 压力返回值 = 10#
If 压力返回值 = "800Lb" Then 压力返回值 = 13#
#30 P0 DN/PN 字典 DB 化：换算表 yaml 化（config.yaml 的
sales_plan.pn_convert.lb_to_mpa），配置优先；内置 ``LB_TO_MPA_MAP``
作为回退（配置缺失时仍可用，不阻断打印）。
Args:
lb_level: LB 等级值
Returns:
float: 对应 MPa 压力值
"""
try:
from certflow.config.settings import PN_LB_TO_MPA
if PN_LB_TO_MPA:
return PN_LB_TO_MPA.get(lb_level, LB_TO_MPA_MAP.get(lb_level, 0.0))
except Exception:
pass
return LB_TO_MPA_MAP.get(lb_level, 0.0)
# ============================================================
# 单位剥离
# ============================================================
[文档]
@staticmethod
def strip_unit(pressure_str: str) -> tuple[str, str]:
"""剥离压力字符串中的单位后缀
对标 VBA 确认压力() 第316行：
.. code-block:: text
If 压力返回值 Like "*MPa" Then
压力返回值 = VBA.Replace(UCase(压力返回值), UCase("MPa"), "")
Args:
pressure_str: 带可能的单位后缀的压力字符串，如 "2.5MPa", "150Lb", "1.0"
Returns:
(数值部分, 单位) 元组，如 ("2.5", "MPa"), ("150", "Lb"), ("1.0", "")
"""
if not pressure_str:
return "", ""
trimmed = pressure_str.strip()
for suffix in UNIT_SUFFIXES:
if trimmed.upper().endswith(suffix.upper()):
value = trimmed[: -len(suffix)].strip()
unit = suffix.upper()
if unit in ("MPA", "MPA"):
unit = "MPa"
if unit in ("LB", "LB"):
unit = "Lb"
return value, unit
# 没有匹配到已知后缀，检查是否为纯数值
if re.match(r"^-?\d+\.?\d*$", trimmed):
return trimmed, ""
return trimmed, ""
[文档]
@staticmethod
def normalize_pn_value(pressure_str: str) -> str:
"""标准化压力数值（去除单位，返回纯数值字符串）
Args:
pressure_str: 压力字符串
Returns:
str: 纯数值字符串，如 "2.5", "150"
"""
value, _ = PNService.strip_unit(pressure_str)
return value
[文档]
@staticmethod
def derive_pn_fields(pn_raw: str, family: str) -> tuple[str, str, str]:
"""以 ``pn_display``（权威输入串，任意格式）反向拆出存储三件套。
单一真相来源（docs/PATCH_LOG 补丁 29/30/32）：合格证压力只用 ``pn_display``
落库，``pn_value`` / ``pn_unit`` 由其反向拆出，渲染再按模板语言族决定带不带单位
（见 ``TemplateFieldFormatter.format_pn`` / ``output_guard.validate_pn_text``）。
规则：
- ``pn_display`` 始终保存**完整串**（含单位），如 ``"25.0MPa"`` / ``"150Lb"``；
- 中文数值型（无单位）→ 默认补 ``MPa``（底板预印 MPa），即 ``"25.0"`` →
``("25.0MPa", "25.0", "MPa")``；
- 非中文数值型被门限拦截，不会传入此处；带单位（如 ``"150Lb"``）正常拆；
- 非 MPa 单位（如 ``Lb``/``KPa``）原样保留，不做换算。
Args:
pn_raw: 用户输入/字典原文（如 ``"25.0"`` / ``"32.0MPa"`` / ``"150Lb"``）。
family: 模板语言族（全中文/中英文/全英文/俄英文）。
Returns:
(pn_display, pn_value, pn_unit) 三元组。
"""
raw = (pn_raw or "").strip()
if not raw:
return "", "", ""
val, unit = PNService.strip_unit(raw)
if not val:
return "", "", ""
# 中文数值型默认 MPa；非中文数值型由门限拦截，此处不会走到缺单位分支
eff_unit = unit or ("MPa" if family == "全中文" else "")
pn_display = f"{val}{eff_unit}" if eff_unit else val
return pn_display, val, eff_unit
# ============================================================
# 数据库查询
# ============================================================
[文档]
def lookup_pn_from_db(self, model: str) -> dict[str, Any] | None:
"""从 model_param_mappings 表查询型号对应的压力信息
注意：此方法依赖 model_param_mappings 表已存在。
如果表尚未创建，返回 None。
Args:
model: 产品型号
Returns:
dict: {pressure_value, test_standard, working_temp, working_medium} 或 None
"""
if not self.session or not model:
return None
try:
from certflow.models.model_param_mapping import ModelParamMapping
result = (
self.session.query(ModelParamMapping)
.filter(ModelParamMapping.product_model == model.strip())
.first()
)
if result:
return {
"pressure_value": result.pressure_value,
"test_standard": result.test_standard,
"working_temp": result.working_temp,
"working_medium": result.working_medium,
}
except ImportError:
# model_param_mappings 表尚未创建
pass
except Exception as e:
logger.debug(f"从数据库查询型号压力失败: {model} — {e}")
return None
# ============================================================
# 核心方法：从型号获取完整 PN 信息
# ============================================================
def _resolve_pn_via_model_spec_parser(self, model: str) -> dict[str, str] | None:
"""用 ModelSpecParser 稳健解析型号内嵌压力（白名单，优先于脆弱后缀提取）
仅对标准公称压力代号（GB/T 阀门口径压力系列）返回结果，不会伪造；
解析器为纯函数，异常时回退 None，由调用方继续既有兜底逻辑。
Args:
model: 产品型号
Returns:
dict: ``{"pn_value", "pn_unit", "pn_display", "source"}``；
未命中标准代号或异常时返回 ``None``。
"""
try:
from certflow.services.model_spec_parser import ModelSpecParser
mspec = ModelSpecParser.parse_pressure(model)
if mspec and mspec.get("pn_value"):
return mspec
except Exception: # pragma: no cover - 纯函数解析器，理论上不抛异常
logger.debug("ModelSpecParser.parse_pressure 解析失败，回退既有逻辑", exc_info=True)
return None
[文档]
def resolve_pn(self, model: str, template_pn: str | None = None) -> dict[str, Any]:
"""从型号解析公称压力，返回完整的 PN 信息字典
处理流程：
1. 判断是否英制型号（LB）→ 提取磅级并转 MPa
2. 否则优先用 ModelSpecParser 稳健解析型号内嵌压力（白名单，识别
PNxx/CLASSxx/标准压力代号，避免末尾温度/材质数字误判）
3. 否则回退既有逻辑：数据库 model_param_mappings → 模板 PN →
型号后缀脆弱兜底提取
4. 剥离单位后缀、返回标准化结果
Args:
model: 产品型号，如 "Z41H-25C", "Z41H-150LB"
template_pn: 模板中已有的 PN 值（可选），如 "2.5MPa"
Returns:
dict: 包含以下键的字典
- **pn_value** (str): 纯数值，如 "2.5", "150"
- **pn_display** (str): 带单位显示，如 "2.5MPa", "150Lb"
- **pn_unit** (str): 单位 "MPa" / "Lb" / ""
- **is_imperial** (bool): 是否英制
- **lb_level** (int | None): LB 等级（仅英制）
- **mpa_equiv** (float | None): 等效 MPa（仅英制）
- **source** (str): 来源 "model" / "template" / "db" / "unknown"
"""
result = {
"pn_value": "",
"pn_display": "",
"pn_unit": "",
"is_imperial": False,
"lb_level": None,
"mpa_equiv": None,
"source": "unknown",
"test_standard": "",
}
if not model:
return result
model_upper = model.upper().strip()
# 统一预查型号压力转换表，填充标准号（各分支共享，避免重复查询）
_db = self.lookup_pn_from_db(model)
if _db:
result["test_standard"] = _db.get("test_standard", "") or ""
# === 第1步：判断英制 ===
if self.is_imperial(model):
result["is_imperial"] = True
lb_level = self.extract_lb_level(model)
if lb_level:
result["lb_level"] = lb_level
result["mpa_equiv"] = self.lb_to_mpa(lb_level)
result["pn_value"] = str(lb_level)
result["pn_unit"] = "Lb"
result["pn_display"] = f"{lb_level}Lb"
result["source"] = "model"
return result
# === 第2步：用 ModelSpecParser 稳健解析型号内嵌压力（白名单，优先于脆弱后缀提取）===
# 可识别显式 PNxx / CLASSxx，以及型号中段/后缀的标准公称压力代号
# （如 H61H-250 的 250 → 25MPa）；白名单校验能避免把型号末尾的温度/材质数字
# （如 "材质：20g" 的 20、"425℃"）误判为公称压力。该解析器仅对标准代号
# (GB/T 阀门口径压力系列) 返回结果，不会伪造，因此优先于脆弱的后缀提取。
mspec = self._resolve_pn_via_model_spec_parser(model)
if mspec:
result["pn_value"] = mspec["pn_value"]
result["pn_unit"] = mspec["pn_unit"]
result["pn_display"] = mspec["pn_display"]
result["source"] = "model"
return result
# === 第3步：从数据库查询（如果有 model_param_mappings 表）===
db_result = self.lookup_pn_from_db(model)
if db_result and db_result.get("pressure_value"):
raw_pn = db_result["pressure_value"]
value, unit = self.strip_unit(raw_pn)
result["pn_value"] = value
result["pn_unit"] = unit
result["pn_display"] = f"{value}{unit}" if unit else value
result["source"] = "db"
return result
# === 第4步：从模板 PN 解析 ===
if template_pn:
value, unit = self.strip_unit(template_pn)
if value:
result["pn_value"] = value
result["pn_unit"] = unit
result["pn_display"] = f"{value}{unit}" if unit else value
result["source"] = "template"
return result
# === 第5步：从型号后缀兜底提取（脆弱，仅当上述均未命中）===
# 常见格式: Z41H-25C → 25 表示 2.5MPa
# Z41H-16C → 16 表示 1.6MPa
# D942X-10Q → 10 表示 1.0MPa
# 注意：仅作最后兜底，因无法区分型号末尾非压力数字（如材质代号）。
pressure_from_model = self._extract_pn_from_model_suffix(model_upper)
if pressure_from_model:
result["pn_value"] = pressure_from_model
result["pn_unit"] = "MPa"
result["pn_display"] = f"{pressure_from_model}MPa"
result["source"] = "model"
return result
return result
@staticmethod
def _extract_pn_from_model_suffix(model_upper: str) -> str:
"""从型号后缀提取公称压力值
常见格式：
- Z41H-25C → 型号后缀 "25" → 2.5MPa
- Z41H-16C → 型号后缀 "16" → 1.6MPa
- D942X-10Q → 型号后缀 "10" → 1.0MPa
Args:
model_upper: 大写型号字符串
Returns:
str: 压力值如 "2.5", "1.6"，未识别返回空字符串
"""
# 模式：型号末尾的 -XX 部分（XX 为数字+可选字母）
match = re.search(r"-?(\d+)\s*([A-Z]*)$", model_upper)
if not match:
return ""
num_str = match.group(1)
_ = match.group(2) # suffix, reserved for future use
if not num_str:
return ""
num = int(num_str)
# 排除明显不是压力的数字（如口径值）
# 压力后缀通常在 6-64 之间
if num < 6 or num > 64:
return ""
# 常见阀门型号编码：数字 / 10 = MPa
return f"{num / 10:.1f}".rstrip("0").rstrip(".")
# ============================================================
# 显示文本生成（根据模板类型）
# ============================================================
[文档]
@staticmethod
def get_display_text(
pn_value: str,
pn_unit: str,
template_type: str = "全中文",
) -> str:
"""根据模板类型生成适合打印的 PN 显示文本
对标 VBA print.vb 中三种模板的不同处理：
- 全中文模板：纯数值（底图已印 MPa），英制带 Lb
- 中英文模板：数值+单位（底图可能无单位印刷）
- 俄英文模板：数值+单位（底图可能无单位印刷）
各模板的 VBA 实现：
.. code-block:: text
' 全中文模板
.Cells(6,2) = Format(确认压力(.Cells(2,14+OfSet)), "#0.0")
' 英制: .Cells(6,2) = Space(14-Len(Trim(str_ylA))) & Trim(str_ylA)
' 中英文/俄英文模板
.Cells(6,2) = str_ylB
Args:
pn_value: 纯数值，如 "2.5", "150"
pn_unit: 单位，如 "MPa", "Lb", ""
template_type: 模板类型 "全中文" / "中英文" / "全英文" / "俄英文"
Returns:
str: 适合打印的显示文本
"""
if not pn_value:
return ""
if template_type == "全中文":
# 全中文底图已印刷 MPa，只需数值
# 但英制压力需要带 Lb 后缀
if pn_unit == "Lb":
return f"{pn_value}Lb"
# 特殊值保持原样
if pn_value in ("2.5Kpa", "-----", "/"):
return pn_value
# 格式化数值
try:
return f"{float(pn_value):.1f}".rstrip("0").rstrip(".")
except ValueError:
return pn_value
else:
# 中英文/全英文/俄英文：数值+单位
if pn_unit:
return f"{pn_value}{pn_unit}"
# 没有单位信息，默认 MPa
return f"{pn_value}MPa"
# ============================================================
# 自学习
# ============================================================
[文档]
def auto_learn(
self,
model: str,
pn_value: str,
pn_unit: str,
test_standard: str = "",
certificate_no: str = "",
) -> bool:
"""自动记录新型号到 model_param_mappings 表
对标 VBA CanShuSheet 的自动追加行为。
Args:
model: 产品型号
pn_value: 压力数值
pn_unit: 压力单位
test_standard: 试压标准号
certificate_no: 关联合格证编号
Returns:
bool: 是否成功
"""
if not self.session or not model or not pn_value:
return False
try:
from certflow.models.model_param_mapping import ModelParamMapping
# 检查是否已存在
existing = (
self.session.query(ModelParamMapping)
.filter(ModelParamMapping.product_model == model.strip())
.first()
)
if existing:
# 更新已有记录
if pn_value and not existing.pressure_value:
existing.pressure_value = f"{pn_value}{pn_unit}" if pn_unit else pn_value
if test_standard and not existing.test_standard:
existing.test_standard = test_standard
existing.usage_count = (existing.usage_count or 0) + 1
else:
# 新增记录
new_mapping = ModelParamMapping(
product_model=model.strip(),
pressure_value=f"{pn_value}{pn_unit}" if pn_unit else pn_value,
test_standard=test_standard or None,
source="auto-learned",
usage_count=1,
)
self.session.add(new_mapping)
# 记录自学习日志
try:
from certflow.models.param_auto_learn_log import ParamAutoLearnLog
log_entry = ParamAutoLearnLog(
table_name="model_param_mappings",
field_name="pressure_value",
lookup_key=model.strip(),
user_value=f"{pn_value}{pn_unit}" if pn_unit else pn_value,
certificate_no=certificate_no or None,
)
self.session.add(log_entry)
except ImportError:
pass
self.session.commit()
logger.info(f"自学习记录: 型号={model}, 压力={pn_value}{pn_unit}")
return True
except ImportError:
logger.debug("model_param_mappings 表尚未创建，跳过自学习")
return False
except Exception as e:
logger.error(f"自学习失败: {model} — {e}")
if self.session:
self.session.rollback()
return False