78 lines
3.4 KiB
Python
78 lines
3.4 KiB
Python
"""Portable, read-only material-code/name lookup for delivery replies."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
MATERIAL_NAME_CACHE_FILE = (
|
|
Path(__file__).resolve().parents[2]
|
|
/ "capabilities/am516-delivery-prediction/data/material_names.json"
|
|
)
|
|
|
|
|
|
class MaterialNameCacheError(ValueError):
|
|
"""A cache error that contains no credentials or raw ERP records."""
|
|
|
|
|
|
def material_names_hash(materials: dict[str, str]) -> str:
|
|
content = json.dumps(materials, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
return hashlib.sha256(content.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def material_names_from_records(records: list[dict[str, Any]]) -> dict[str, str]:
|
|
if not records:
|
|
raise MaterialNameCacheError("物料查询结果为空，保留原缓存。")
|
|
materials: dict[str, str] = {}
|
|
for row in records:
|
|
code, name = row.get("FNumber"), row.get("FName")
|
|
if not isinstance(code, str) or not code.strip() or not isinstance(name, str) or not name.strip():
|
|
raise MaterialNameCacheError("物料记录缺少有效编码或名称，保留原缓存。")
|
|
code, name = code.strip(), " ".join(name.split())
|
|
if code in materials and materials[code] != name:
|
|
raise MaterialNameCacheError("同一物料编码存在不同名称，需核对后再刷新缓存。")
|
|
materials[code] = name
|
|
return dict(sorted(materials.items()))
|
|
|
|
|
|
def load_material_names(path: Path = MATERIAL_NAME_CACHE_FILE) -> dict[str, str]:
|
|
"""Reuse memory until the local file changes; never access ERP."""
|
|
try:
|
|
stat = path.stat()
|
|
except FileNotFoundError as exc:
|
|
raise MaterialNameCacheError("本地物料名称缓存文件缺失") from exc
|
|
except OSError as exc:
|
|
raise MaterialNameCacheError("本地物料名称缓存文件不可读取") from exc
|
|
return _load_material_names(path, (stat.st_mtime_ns, stat.st_size, stat.st_ino))
|
|
|
|
|
|
@lru_cache(maxsize=4)
|
|
def _load_material_names(path: Path, version: tuple[int, int, int]) -> dict[str, str]:
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except FileNotFoundError as exc:
|
|
raise MaterialNameCacheError("本地物料名称缓存文件缺失") from exc
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
|
raise MaterialNameCacheError("本地物料名称缓存文件不可读取或格式错误") from exc
|
|
if not isinstance(payload, dict) or payload.get("schema_version") != 1:
|
|
raise MaterialNameCacheError("本地物料名称缓存版本或结构无效")
|
|
materials = payload.get("materials")
|
|
if not isinstance(materials, dict) or not materials:
|
|
raise MaterialNameCacheError("本地物料名称缓存为空或结构无效")
|
|
if any(
|
|
not isinstance(code, str) or not code.strip()
|
|
or not isinstance(name, str) or not name.strip()
|
|
for code, name in materials.items()
|
|
):
|
|
raise MaterialNameCacheError("本地物料名称缓存包含无效编码或名称")
|
|
if payload.get("material_count") != len(materials) or payload.get("content_sha256") != material_names_hash(materials):
|
|
raise MaterialNameCacheError("本地物料名称缓存计数或完整性校验失败")
|
|
return materials
|
|
|
|
|
|
load_material_names.cache_clear = _load_material_names.cache_clear
|