lixu
786b7d3721
- api/: Go(chi) 只读 API 骨架, /healthz + 版本化路由(占位), Dockerfile, 单测 - ingestion/: Python 采集/ETL 包骨架, units 单位归一化(纯函数+测试), adapter 协议 - docker-compose.yml: postgres + redis + minio + api - .github/workflows/ci.yml: Go build/vet/test + Python ruff/pytest - docs/data-contract.md(两端共享契约) + docs/disclaimer.md(不提供购买声明) - migrations/ 占位(M1 起填充) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
98 lines
2.8 KiB
Python
98 lines
2.8 KiB
Python
"""Unit normalization for OpenGoods.
|
|
|
|
Product parameters arrive in many units (g/kg/ml/L, kcal/kJ, ...). To make
|
|
values comparable and searchable we store both the original value and a
|
|
normalized value expressed in a canonical unit per dimension.
|
|
|
|
This module is intentionally dependency-free and pure so it is easy to test.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from decimal import Decimal
|
|
|
|
# Conversion factor maps each unit to its canonical unit within a dimension.
|
|
# canonical_value = value * factor
|
|
_FACTORS: dict[str, tuple[str, str, Decimal]] = {
|
|
# mass -> g
|
|
"mg": ("mass", "g", Decimal("0.001")),
|
|
"g": ("mass", "g", Decimal("1")),
|
|
"kg": ("mass", "g", Decimal("1000")),
|
|
# volume -> ml
|
|
"ml": ("volume", "ml", Decimal("1")),
|
|
"cl": ("volume", "ml", Decimal("10")),
|
|
"l": ("volume", "ml", Decimal("1000")),
|
|
# energy -> kJ
|
|
"kj": ("energy", "kJ", Decimal("1")),
|
|
"kcal": ("energy", "kJ", Decimal("4.184")),
|
|
}
|
|
|
|
# Alias map normalizes common spellings/locales to a canonical unit code.
|
|
_ALIASES: dict[str, str] = {
|
|
"kgs": "kg",
|
|
"千克": "kg",
|
|
"公斤": "kg",
|
|
"克": "g",
|
|
"毫升": "ml",
|
|
"升": "l",
|
|
"L": "l",
|
|
"litre": "l",
|
|
"liter": "l",
|
|
"kj": "kj",
|
|
"kJ": "kj",
|
|
"千焦": "kj",
|
|
"千卡": "kcal",
|
|
"大卡": "kcal",
|
|
}
|
|
|
|
|
|
class UnitError(ValueError):
|
|
"""Raised when a unit cannot be recognized."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Normalized:
|
|
"""Result of normalizing a (value, unit) pair to its canonical unit."""
|
|
|
|
value: Decimal
|
|
unit: str
|
|
dimension: str
|
|
canonical_value: Decimal
|
|
canonical_unit: str
|
|
|
|
|
|
def canonical_unit_code(unit: str) -> str:
|
|
"""Resolve a raw unit string to a known canonical unit code."""
|
|
cleaned = unit.strip()
|
|
cleaned = _ALIASES.get(cleaned, cleaned).lower()
|
|
if cleaned not in _FACTORS:
|
|
raise UnitError(f"unknown unit: {unit!r}")
|
|
return cleaned
|
|
|
|
|
|
def normalize(value: Decimal | float | int | str, unit: str) -> Normalized:
|
|
"""Normalize a value+unit to its canonical unit within its dimension."""
|
|
code = canonical_unit_code(unit)
|
|
dimension, canonical, factor = _FACTORS[code]
|
|
dec = value if isinstance(value, Decimal) else Decimal(str(value))
|
|
return Normalized(
|
|
value=dec,
|
|
unit=code,
|
|
dimension=dimension,
|
|
canonical_value=dec * factor,
|
|
canonical_unit=canonical,
|
|
)
|
|
|
|
|
|
def kcal_to_kj(kcal: Decimal | float | int | str) -> Decimal:
|
|
"""Convert energy in kcal to kJ (1 kcal = 4.184 kJ)."""
|
|
dec = kcal if isinstance(kcal, Decimal) else Decimal(str(kcal))
|
|
return dec * Decimal("4.184")
|
|
|
|
|
|
def kj_to_kcal(kj: Decimal | float | int | str) -> Decimal:
|
|
"""Convert energy in kJ to kcal."""
|
|
dec = kj if isinstance(kj, Decimal) else Decimal(str(kj))
|
|
return dec / Decimal("4.184")
|