"""口径映射模型
将各种格式的口径原始值（如 "DN15"、"1/2\""、"G1/2"、"3/8\"" 等）映射为标准口径。
支持公制和英制两种表示。
"""
from __future__ import annotations
from sqlalchemy import Boolean, Column, DateTime, Integer, String
from sqlalchemy.orm import Session
from sqlalchemy.sql import func
from certflow.models.base import Base
[文档]
class CaliberMapping(Base):
"""口径字符串映射表
对标 VBA function.vb 确认口径() 第164-166行：
.. code-block:: text
Arr_KouJingDuiYing = CanShuSheet.Range("口径替换数据表")
For i = LBound(Arr_KouJingDuiYing, 1) To UBound(Arr_KouJingDuiYing, 1)
If VBA.Trim(Arr_KouJingDuiYing(i, 1)) = VBA.Trim(口径字符串) Then ...
Attributes:
id: 主键ID
raw_text: 原始口径字符串（如 "DN100", "2\"", "2-1/2\"", "DN100*80"）
caliber_value: 标准化口径值（如 "100", "50", "65", "100*80"）
is_imperial: 是否英制口径
source: 来源（manual/auto-learned/import）
usage_count: 使用次数
"""
__tablename__ = "caliber_mappings"
id = Column(Integer, primary_key=True, autoincrement=True)
raw_text = Column(String(50), nullable=False, unique=True, index=True, comment="原始口径文本")
caliber_value = Column(String(20), nullable=False, comment="口径值（mm或英寸表示）")
is_imperial = Column(Boolean, default=False, comment="是否英制（True=英寸，False=毫米）")
source = Column(String(20), default="manual", comment="来源：manual/auto_learn/import")
usage_count = Column(Integer, default=0, comment="使用次数")
created_at = Column(DateTime, server_default=func.now(), comment="创建时间")
updated_at = Column(
DateTime, server_default=func.now(), onupdate=func.now(), comment="更新时间"
)
# raw_text 已设为 unique=True，SQLite 自动生成唯一索引，无需重复显式索引（阶段1b 清理）
__table_args__ = ()
def __repr__(self) -> str:
return f"<CaliberMapping(raw='{self.raw_text}' → {self.caliber_value}{'英寸' if self.is_imperial else 'mm'})>"
[文档]
def to_mm(self) -> int | None:
"""转换为毫米值
Returns:
int | None: 标准化后的毫米整数值；无法解析时返回 None。
"""
if self.is_imperial:
try:
if "/" in self.caliber_value:
num, den = self.caliber_value.split("/")
inch = float(num) / float(den)
else:
inch = float(self.caliber_value)
return round(inch * 25.4)
except (ValueError, ZeroDivisionError):
return None
else:
try:
return int(float(self.caliber_value))
except ValueError:
return None
class CaliberMappingService:
"""口径映射服务"""
# 内置正则规则（作为数据库映射的补充/回退）
BUILTIN_PATTERNS = [
# DN格式：DN15, DN 15, DN15A
(r"DN\s*(\d+)", lambda m: ("mm", int(m.group(1)))),
# 数字+mm：15mm, 15 mm
(r"(\d+)\s*mm", lambda m: ("mm", int(m.group(1)))),
# 纯数字（毫米）
(r"^(\d+)$", lambda m: ("mm", int(m.group(1)))),
# 英寸分数：1/2", 1/2″
(r"(\d+)/(\d+)[\"″']?", lambda m: ("inch", f"{m.group(1)}/{m.group(2)}")),
# 英寸整数：2", 2″
(r"(\d+(?:\.\d+)?)[\"″']", lambda m: ("inch", m.group(1))),
# G螺纹：G1/2 → 1/2英寸
(r"G(\d+)/(\d+)", lambda m: ("inch", f"{m.group(1)}/{m.group(2)}")),
(r"G(\d+)", lambda m: ("inch", m.group(1)) if int(m.group(1)) <= 4 else None),
# 带PN后缀：DN15 PN315 → 15mm
(r"DN(\d+)\s*PN", lambda m: ("mm", int(m.group(1)))),
# 带抗硫后缀：DN80抗硫 → 80mm
(r"DN(\d+)抗硫", lambda m: ("mm", int(m.group(1)))),
# 范围格式：DN65/40 → 65mm
(r"DN(\d+)/\d+", lambda m: ("mm", int(m.group(1)))),
]
def __init__(self, session: Session) -> None:
"""初始化口径映射服务。
Args:
session: SQLAlchemy 会话，用于查询与持久化。
"""
self.session = session
def resolve_caliber(self, raw_text: str) -> tuple[str, str] | None:
"""解析口径，返回 (单位, 值)
Args:
raw_text: 原始口径文本
Returns:
(单位, 值) 如 ("mm", "15") 或 ("inch", "3/8")，无法解析返回 None
"""
if not raw_text or not isinstance(raw_text, str):
return None
raw_text = raw_text.strip()
if not raw_text:
return None
# 1. 精确匹配数据库
mapping = (
self.session.query(CaliberMapping).filter(CaliberMapping.raw_text == raw_text).first()
)
if mapping:
mapping.usage_count = (mapping.usage_count or 0) + 1
self.session.commit()
unit = "inch" if mapping.is_imperial else "mm"
return (unit, mapping.caliber_value)
# 2. 标准化后匹配（去空格、转大写）
normalized = self._normalize(raw_text)
mapping = (
self.session.query(CaliberMapping).filter(CaliberMapping.raw_text == normalized).first()
)
if mapping:
mapping.usage_count = (mapping.usage_count or 0) + 1
self.session.commit()
unit = "inch" if mapping.is_imperial else "mm"
return (unit, mapping.caliber_value)
# 3. 内置规则
import re
for pattern, extractor in self.BUILTIN_PATTERNS:
match = re.search(pattern, raw_text, re.IGNORECASE)
if match:
try:
result = extractor(match)
if result:
unit, value = result
self.auto_learn(raw_text, unit, str(value))
return (unit, str(value))
except Exception:
continue
return None
def resolve_to_mm(self, raw_text: str) -> int | None:
"""解析口径并返回毫米值
Args:
raw_text: 原始口径文本。
Returns:
int | None: 换算后的毫米整数值；无法解析时返回 None。
"""
result = self.resolve_caliber(raw_text)
if not result:
return None
unit, value = result
if unit == "mm":
try:
return int(float(value))
except ValueError:
return None
else:
try:
if "/" in value:
num, den = value.split("/")
inch = float(num) / float(den)
else:
inch = float(value)
return round(inch * 25.4)
except (ValueError, ZeroDivisionError):
return None
def _normalize(self, text: str) -> str:
"""标准化文本"""
if not text:
return ""
text = text.strip()
text = text.replace("″", '"').replace("′", "'")
return text.upper()
def add_mapping(
self, raw_text: str, unit: str, value: str, source: str = "manual"
) -> CaliberMapping | None:
"""添加口径映射
Args:
raw_text: 原始口径文本。
unit: 单位（mm/inch）。
value: 口径值。
source: 来源。
Returns:
CaliberMapping | None: 新建或更新后的映射对象；raw_text 或 value 为空时返回 None。
"""
if not raw_text or not value:
return None
raw_text = raw_text.strip()
is_imperial = unit == "inch"
existing = (
self.session.query(CaliberMapping).filter(CaliberMapping.raw_text == raw_text).first()
)
if existing:
existing.caliber_value = value
existing.is_imperial = is_imperial
existing.updated_at = func.now()
self.session.commit()
return existing
mapping = CaliberMapping(
raw_text=raw_text,
caliber_value=value,
is_imperial=is_imperial,
source=source,
usage_count=0,
)
self.session.add(mapping)
self.session.commit()
return mapping
def auto_learn(self, raw_text: str, unit: str, value: str) -> None:
"""自动学习新映射
Args:
raw_text: 原始口径文本。
unit: 单位（mm/inch）。
value: 口径值。
"""
if not raw_text:
return
raw_text = raw_text.strip()
existing = (
self.session.query(CaliberMapping).filter(CaliberMapping.raw_text == raw_text).first()
)
if existing:
return
mapping = CaliberMapping(
raw_text=raw_text,
caliber_value=value,
is_imperial=(unit == "inch"),
source="auto_learn",
usage_count=1,
)
self.session.add(mapping)
self.session.commit()
def get_all_mappings(self) -> list[dict]:
"""获取所有映射（按使用次数降序）
Returns:
list[dict]: 每条映射的字典列表，按 usage_count 降序、raw_text 升序排列。
"""
mappings = (
self.session.query(CaliberMapping)
.order_by(CaliberMapping.usage_count.desc(), CaliberMapping.raw_text)
.all()
)
return [
{
"id": m.id,
"raw_text": m.raw_text,
"caliber_value": m.caliber_value,
"is_imperial": m.is_imperial,
"display": f"{m.caliber_value}{'英寸' if m.is_imperial else 'mm'}",
"source": m.source,
"usage_count": m.usage_count,
}
for m in mappings
]