"""公称通径(DN)处理服务
对标 VBA function.vb 确认口径() 函数，提供：
1. 口径字符串标准化（去除 DN 前缀、英制转换）
2. 口径映射查询（从 SQLite caliber_mappings 表）
3. 自学习机制（新口径自动记录到数据库）
4. 多口径处理（如 DN100*80 → 100*80）
v2.0 新增：
- 对标 VBA 确认口径() 全部逻辑
- 英制口径识别（2\", 2-1/2\" 等格式）
- 哨兵值处理（VBA 的 CInt(-20) 对应 None）
"""
from __future__ import annotations
import re
from typing import Any
from sqlalchemy.orm import Session
from certflow.utils.logger import logger
# ============================================================
# 英制口径模式
# ============================================================
# 常见英制口径格式：2", 2-1/2", 3/4", 1/2"
# 扩展：可携带 LB 磅级（VBA 英制口径 + 磅级），如 2"150LB、2-1/2"600LB
IMPERIAL_CALIBER_PATTERN = re.compile(
r'^(\d+(?:-\d+/\d+)?)\s*"\s*(?:(\d+)\s*LB)?$'
r'|^(\d+/\d+)\s*"\s*(?:(\d+)\s*LB)?$'
)
# DN 前缀模式（需要剥离的）
DN_PREFIX_PATTERN = re.compile(r"^DN\s*", re.IGNORECASE)
# LB 磅级提取（如 2"150LB / DN50 300LB 中的 150/300）
LB_CLASS_PATTERN = re.compile(r"(\d+)\s*LB", re.IGNORECASE)
# VBA 哨兵值：CInt(-20) → None，表示口径无效、需人工补充
SENTINEL_VALUES = {"-20", "-20.0"}
# 英制型号关键字（VBA 确认压力()：UCase Like "*LB*" Or "*2500*"）
IMPERIAL_MODEL_KEYWORDS = ("LB", "2500")
# DN 口径尾部需清理的公制单位后缀（VBA：去 "MM"）
IMPERIAL_SUFFIX_MM = "MM"
[文档]
class DNService:
"""公称通径处理服务
负责从口径字符串解析/查询标准化口径值。
对标 VBA function.vb 确认口径() 函数（第156-293行）。
Attributes:
session: SQLAlchemy 数据库会话
"""
def __init__(self, session: Session | None = None) -> None:
self.session = session
# ============================================================
# 口径标准化
# ============================================================
[文档]
@staticmethod
def normalize_caliber(raw_text: str) -> str:
"""标准化口径字符串
对标 VBA print.vb 第681行：
.Cells(5, 2) = Replace(str_kjA, "DN", "")
Args:
raw_text: 原始口径字符串，如 "DN100", "DN100*80", "2\""
Returns:
str: 标准化口径值，如 "100", "100*80", "50"
"""
if not raw_text:
return ""
text = raw_text.strip()
# 去掉 DN 前缀
text = DN_PREFIX_PATTERN.sub("", text).strip()
# VBA print.vb：口径尾部 "MM" 清理（如 "50MM" → "50"）
if text.upper().endswith(IMPERIAL_SUFFIX_MM):
text = text[: -len(IMPERIAL_SUFFIX_MM)].strip()
return text
[文档]
@staticmethod
def is_imperial_caliber(raw_text: str) -> bool:
"""判断是否为英制口径
Args:
raw_text: 口径字符串
Returns:
bool: 是否英制
"""
if not raw_text:
return False
return bool(IMPERIAL_CALIBER_PATTERN.match(raw_text.strip()))
[文档]
@staticmethod
def is_imperial_model(model: str) -> bool:
"""判断产品型号是否为英制（含 LB 磅级或 2500 等级）
对标 VBA 确认压力() 第312行：
If (UCase(... Like "*LB*") Or (UCase(... Like "*2500*")) Then ...
Args:
model: 产品型号字符串
Returns:
bool: 是否英制型号
"""
if not model:
return False
upper = model.upper()
return any(kw in upper for kw in IMPERIAL_MODEL_KEYWORDS)
[文档]
@staticmethod
def imperial_to_mm(inch_str: str) -> str | None:
"""英制口径转换为 mm 值
对标 VBA 确认口径() 中隐含的英制转换逻辑。
常见英制口径对照：
- 1/2\" → 15mm
- 3/4\" → 20mm
- 1\" → 25mm
- 1-1/4\" → 32mm
- 1-1/2\" → 40mm
- 2\" → 50mm
- 2-1/2\" → 65mm
- 3\" → 80mm
- 4\" → 100mm
- 5\" → 125mm
- 6\" → 150mm
- 8\" → 200mm
- 10\" → 250mm
- 12\" → 300mm
Args:
inch_str: 英制口径字符串，如 "2\"", "2-1/2\""
Returns:
str: mm 值字符串，如 "50", "65"，未匹配返回 None
"""
if not inch_str:
return None
# 去除引号和空格
clean = inch_str.strip().rstrip('"').strip()
# 解析分数格式：2-1/2 → 2.5
match = re.match(r"^(\d+)-(\d+)/(\d+)$", clean)
if match:
whole = int(match.group(1))
num = int(match.group(2))
den = int(match.group(3))
inches = whole + num / den
elif re.match(r"^(\d+)/(\d+)$", clean):
match = re.match(r"^(\d+)/(\d+)$", clean)
inches = int(match.group(1)) / int(match.group(2))
else:
try:
inches = float(clean)
except ValueError:
return None
# 英寸 → mm 对照表
inch_to_mm: dict[float, str] = {
0.5: "15",
0.75: "20",
1.0: "25",
1.25: "32",
1.5: "40",
2.0: "50",
2.5: "65",
3.0: "80",
4.0: "100",
5.0: "125",
6.0: "150",
7.0: "175",
8.0: "200",
9.0: "225",
10.0: "250",
11.0: "275",
12.0: "300",
13.0: "325",
14.0: "350",
15.0: "375",
16.0: "400",
17.0: "425",
18.0: "450",
19.0: "475",
20.0: "500",
21.0: "525",
22.0: "550",
23.0: "575",
24.0: "600",
}
return inch_to_mm.get(inches)
# ============================================================
# 数据库查询
# ============================================================
[文档]
def lookup_caliber_from_db(self, raw_text: str) -> dict[str, Any] | None:
"""从 caliber_mappings 表查询口径映射
对标 VBA 确认口径() 第164-166行：
Arr_KouJingDuiYing = CanShuSheet.Range("口径替换数据表")
Args:
raw_text: 原始口径字符串
Returns:
dict: {caliber_value, is_imperial} 或 None
"""
if not self.session or not raw_text:
return None
try:
from certflow.models.caliber_mapping import CaliberMapping
# 精确匹配（去除 DN 前缀后匹配）
normalized = self.normalize_caliber(raw_text)
result = (
self.session.query(CaliberMapping)
.filter(CaliberMapping.raw_text == normalized)
.first()
)
if not result:
# 用原始文本再试
result = (
self.session.query(CaliberMapping)
.filter(CaliberMapping.raw_text == raw_text.strip())
.first()
)
if result:
# 更新使用计数
result.usage_count = (result.usage_count or 0) + 1
self.session.commit()
return {
"caliber_value": result.caliber_value,
"is_imperial": result.is_imperial,
}
except ImportError:
pass
except Exception as e:
logger.debug(f"从数据库查询口径映射失败: {raw_text} — {e}")
return None
# ============================================================
# 核心方法：从口径字符串获取标准化口径
# ============================================================
@staticmethod
def _is_valid_caliber_token(normalized: str) -> bool:
"""归一后是否为合法口径标识
合法口径应为纯数字或 ``数字*数字``（如 "100"、"100*80"），
不含字母/汉字；含字母者视为未映射口径，应交由回退或人工补充。
Args:
normalized: 归一后的口径字符串
Returns:
bool: 是否可作为合法口径值
"""
if not normalized:
return False
return not re.search(r"[A-Za-z\u4e00-\u9fff]", normalized)
def _resolve_from_fallback(
self, fallback_raw: str, result: dict[str, Any]
) -> dict[str, Any] | None:
"""C5/C6 备用口径回退（B5 为空则用 C5）
Args:
fallback_raw: 备用口径原始串
result: 当前结果字典（用于回退失败时回退）
Returns:
dict | None: 回退解析得到的口径字典（已标记 source="fallback"），无则 None
"""
if not fallback_raw or not fallback_raw.strip():
return None
fb = self.resolve_caliber(fallback_raw.strip())
if fb["caliber_value"]:
fb["source"] = "fallback"
return fb
return None
[文档]
def resolve_caliber(
self, caliber_str: str, model: str = "", fallback_raw: str = ""
) -> dict[str, Any]:
"""解析口径字符串，返回标准化口径信息
处理流程对标 VBA 确认口径() 函数（含 B5/C5 回退与英制 LB 判断）：
1. 英制口径判断和转换（支持携带 LB 磅级，如 2"150LB）
2. 数据库口径映射表查询（剥离 LB 后缀后匹配）
3. 直接标准化（去除 DN 前缀、清理 "MM" 后缀）
4. C5/C6 备用口径回退（主口径为空或解析无效时启用）
5. 无法解析，标记 needs_manual
Args:
caliber_str: 原始口径字符串，如 "DN100", "2\"", "DN100*80", "2\"150LB"
model: 产品型号（用于自学习关联及英制型号判断）
fallback_raw: 备用口径原始串（C5/C6 备用列），主口径为空或解析失败时回退
Returns:
dict: 包含以下键的字典
- **caliber_value** (str): 标准化口径值，如 "100", "50"
- **is_imperial** (bool): 是否英制
- **raw_text** (str): 原始文本
- **source** (str): 来源 "db" / "imperial" / "normalize" / "fallback" / "unknown"
- **needs_manual** (bool): 是否需要手动补充（对标 VBA CInt(-20) 哨兵）
- **lb_class** (str | None): 提取到的 LB 磅级（如 "150"），无则 None
"""
result: dict[str, Any] = {
"caliber_value": "",
"is_imperial": False,
"raw_text": caliber_str or "",
"source": "unknown",
"needs_manual": False,
"lb_class": None,
}
# 主口径为空 → 直接用备用口径（B5 空则用 C5）
if not caliber_str or not caliber_str.strip():
fb = self._resolve_from_fallback(fallback_raw, result)
return fb if fb is not None else result
text = caliber_str.strip()
# === 哨兵值：VBA CInt(-20) → None，表示口径无效需人工补充 ===
if self.normalize_caliber(text) in SENTINEL_VALUES:
result["needs_manual"] = True
return result
# 先提取 LB 磅级（可能与 inch/DN 同串出现）
lb_match = LB_CLASS_PATTERN.search(text)
lb_class = lb_match.group(1) if lb_match else None
cleaned = LB_CLASS_PATTERN.sub("", text).strip()
# === 第1步：英制判断（inch 形式，可带 LB） ===
m = IMPERIAL_CALIBER_PATTERN.match(text)
if m:
inch_part = (m.group(1) or m.group(3) or "") + '"'
mm_value = self.imperial_to_mm(inch_part)
if mm_value:
result.update(
caliber_value=mm_value,
is_imperial=True,
source="imperial",
lb_class=lb_class,
)
return result
# === 第2步：数据库查询（剥离 LB 后缀后匹配） ===
db_result = self.lookup_caliber_from_db(cleaned)
if db_result and db_result.get("caliber_value"):
result.update(
caliber_value=db_result["caliber_value"],
is_imperial=db_result.get("is_imperial", False),
source="db",
lb_class=lb_class,
)
return result
# === 第3步：直接标准化（去 DN 前缀、清 "MM" 后缀） ===
normalized = self.normalize_caliber(cleaned)
if self._is_valid_caliber_token(normalized):
result.update(
caliber_value=normalized,
source="normalize",
lb_class=lb_class,
)
return result
# === 第4步：C5/C6 备用口径回退（主口径解析无效时启用） ===
fb = self._resolve_from_fallback(fallback_raw, result)
if fb is not None:
return fb
# === 第5步：无法解析，需要手动补充 ===
result["needs_manual"] = True
return result
# ============================================================
# 自学习
# ============================================================
[文档]
def auto_learn(
self,
raw_text: str,
caliber_value: str,
is_imperial: bool = False,
certificate_no: str = "",
) -> bool:
"""自动记录新口径到 caliber_mappings 表
对标 VBA 确认口径() 第263-283行的自动追加行为：
Arr_KouJingDuiYing(UBound(...), 1) = 口径字符串
Arr_KouJingDuiYing(UBound(...), 2) = 确认口径
Args:
raw_text: 原始口径字符串
caliber_value: 标准化口径值
is_imperial: 是否英制
certificate_no: 关联合格证编号
Returns:
bool: 是否成功
"""
if not self.session or not raw_text or not caliber_value:
return False
try:
from certflow.models.caliber_mapping import CaliberMapping
normalized = self.normalize_caliber(raw_text)
# 检查是否已存在
existing = (
self.session.query(CaliberMapping)
.filter(CaliberMapping.raw_text == normalized)
.first()
)
if existing:
# 更新已有记录
existing.usage_count = (existing.usage_count or 0) + 1
else:
# 新增记录
new_mapping = CaliberMapping(
raw_text=normalized,
caliber_value=caliber_value,
is_imperial=is_imperial,
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="caliber_mappings",
field_name="caliber_value",
lookup_key=normalized,
user_value=caliber_value,
certificate_no=certificate_no or None,
)
self.session.add(log_entry)
except ImportError:
pass
self.session.commit()
logger.info(
f"自学习口径: {normalized} → {caliber_value}{' (英制)' if is_imperial else ''}"
)
return True
except ImportError:
logger.debug("caliber_mappings 表尚未创建，跳过自学习")
return False
except Exception as e:
logger.error(f"口径自学习失败: {raw_text} — {e}")
if self.session:
self.session.rollback()
return False