45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
"""Resolve supplier aliases to canonical master-data identities."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import ProgrammingError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.entities import SupplierMaster
|
|
from app.services.supplier_cleaning import supplier_match_key
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SupplierIdentity:
|
|
id: int
|
|
canonical_name: str
|
|
short_name: str | None
|
|
aliases: tuple[str, ...]
|
|
|
|
@property
|
|
def match_names(self) -> tuple[str, ...]:
|
|
return tuple(dict.fromkeys(filter(None, (self.short_name, *self.aliases, self.canonical_name))))
|
|
|
|
|
|
async def resolve_supplier(session: AsyncSession, value: str | None) -> SupplierIdentity | None:
|
|
if not value or not isinstance(session, AsyncSession):
|
|
return None
|
|
key = supplier_match_key(value)
|
|
try:
|
|
async with session.begin_nested():
|
|
rows = (await session.execute(select(SupplierMaster).where(SupplierMaster.status == "启用"))).scalars().all()
|
|
except ProgrammingError:
|
|
return None
|
|
matches = []
|
|
for row in rows:
|
|
names = (row.canonical_name, row.short_name, *(row.aliases or []))
|
|
if any(name and supplier_match_key(name) == key for name in names):
|
|
matches.append(row)
|
|
if len(matches) != 1:
|
|
return None
|
|
row = matches[0]
|
|
return SupplierIdentity(row.id, row.canonical_name, row.short_name, tuple(row.aliases or []))
|