shanxia2.0/backend/app/services/pl_mobile_stop_meta.py

"""Optional independent stop metadata. No old-table columns or default-path queries."""
from decimal import Decimal
from fastapi import HTTPException

META_FIELDS = ('alias_name', 'brand_mode', 'brand_id', 'brand_name')
IDENTITY_FIELDS = ('contract_id', 'contract_product_id', 'product_id', 'package_spec', 'model', 'unit')
SOURCE_FIELDS = ('product_name', 'brand_mode', 'brand_id', 'brand_name')


def _value(row, field):
    return str(getattr(row, field, None) or '').strip()


def _saved_meta_values(row):
    values = {field: _value(row, field) for field in META_FIELDS}
    # NULL asks the mobile reader to infer the current Web name; empty means explicitly cleared.
    if getattr(row, 'alias_name', None) is None:
        values['alias_name'] = None
    return values


def assert_mobile_meta_enabled(enabled, requested):
    """Kept for existing callers. Alias and brand now follow the Web plan-product fields."""
    return None


def fold_mobile_selection_into_web_fields(payload):
    """Write the mobile alias into product_name and the load-line brand onto the plan product."""
    if not getattr(payload, 'mobile_product_meta', False):
        return
    by_key, by_id = {}, {}
    for product in payload.products or []:
        key = _value(product, 'client_key')
        if key:
            by_key[key] = product
        row_id = _value(product, 'id')
        if row_id:
            by_id[row_id] = product
    load_rows = list(payload.load.products) if getattr(payload, 'load', None) else []
    for stop in load_rows:
        _apply_mobile_alias(stop)
        linked = by_key.get(_value(stop, 'plan_product_client_key')) or by_id.get(_value(stop, 'plan_product_id'))
        if linked is None:
            continue
        meta = getattr(stop, 'mobile_product_meta', None)
        alias = getattr(meta, 'alias_name', None) if meta is not None else None
        if isinstance(alias, str) and alias.strip():
            linked.product_name = alias.strip()
        _apply_mobile_brand(linked, meta)
    for delivery in getattr(payload, 'deliveries', None) or []:
        for stop in delivery.products or []:
            _apply_mobile_alias(stop)


def _apply_mobile_alias(stop):
    meta = getattr(stop, 'mobile_product_meta', None)
    alias = getattr(meta, 'alias_name', None) if meta is not None else None
    if isinstance(alias, str) and alias.strip():
        stop.product_name = alias.strip()


def _apply_mobile_brand(plan_product, meta):
    if meta is None:
        return
    mode = _value(meta, 'brand_mode')
    if mode == 'required':
        plan_product.brand_mode = 'required'
        plan_product.brand_id = _value(meta, 'brand_id') or None
        plan_product.brand_name = _value(meta, 'brand_name') or None
    elif mode == 'any':
        plan_product.brand_mode = 'any'
        plan_product.brand_id = None
        plan_product.brand_name = None


def validate_mobile_meta_request(payload, enabled):
    requested = bool(getattr(payload, 'mobile_product_meta', False))
    stops = ([payload.load] if payload.load else []) + list(payload.deliveries or [])
    rows = [p for stop in stops for p in stop.products]
    if not requested and any(getattr(p, 'mobile_product_meta', None) is not None or getattr(p, 'unlinked_draft', False) for p in rows):
        raise HTTPException(422, '段产品快照必须显式启用移动保存')

    if requested:
        if any(not _value(p, 'client_key') for p in payload.products or []) or any(not _value(p, 'plan_product_client_key') and not (getattr(payload, 'save_mode', '') == 'draft' and getattr(p, 'unlinked_draft', False)) for p in rows):
            raise HTTPException(422, '移动产品关联键不完整，请重新加载')
        values = [getattr(p, 'plan_qty', None) for p in rows] + [getattr(p, 'quantity', None) for p in payload.products or []]
        for value in values:
            if value is not None:
                qty = Decimal(str(value))
                if not qty.is_finite() or qty < 0:
                    raise HTTPException(422, '移动产品数量须为非负有限数')


def resolve_stop_meta(snapshots, used, tenant_no, plan_id, stop_type, stop_id, product, plan_product):
    explicit = getattr(product, 'mobile_product_meta', None)
    if explicit is not None:
        return {field: _value(explicit, field) for field in META_FIELDS}
    matches = [row for row in snapshots if row.id not in used
               and row.tenant_no == tenant_no and row.plan_id == plan_id
               and row.stop_type == stop_type and row.stop_id == stop_id
               and all(_value(row, field) == _value(product, field) for field in IDENTITY_FIELDS)]
    source_id = _value(product, 'id')
    if source_id:
        matches = [row for row in matches if row.stop_product_id == source_id]
    if len(matches) > 1:
        raise HTTPException(422, '该段有无法唯一关联的产品快照，请在移动端重新加载后保存')
    if not matches:
        return None
    row = matches[0]
    used.add(row.id)
    result = _saved_meta_values(row)
    # Existing Web legal fields take precedence only when actually changed.
    if (_value(product, 'product_name') != _value(row, 'product_name')
            or _value(plan_product, 'product_name') != _value(row, 'source_product_name')):
        result['alias_name'] = None
    if any(_value(plan_product, f) != _value(row, 'source_' + f) for f in SOURCE_FIELDS[1:]):
        result.update({f: _value(plan_product, f) for f in SOURCE_FIELDS[1:]})
    return result


def load_stop_meta(db, tenant_no, plan_id, enabled):
    if not enabled:
        return []
    from app.models.pl_mobile_stop_meta import PlPlanStopProductMeta
    from sqlalchemy.exc import OperationalError, ProgrammingError
    try:
        return db.query(PlPlanStopProductMeta).filter(
            PlPlanStopProductMeta.tenant_no == tenant_no,
            PlPlanStopProductMeta.plan_id == plan_id,
            PlPlanStopProductMeta.del_flag == 0,
        ).all()
    except (OperationalError, ProgrammingError) as exc:
        raise HTTPException(503, '移动运输产品快照不可用，请联系管理员核查迁移后重试；本次未保存') from exc


def write_stop_meta(db, snapshots, used, tenant_no, plan_id, stop_type, stop_id,
                    stop_product_id, product, plan_product, user_id, dept_id, now):
    values = resolve_stop_meta(snapshots, used, tenant_no, plan_id, stop_type, stop_id, product, plan_product)
    if values is None:
        return
    from app.models.pl_mobile_stop_meta import PlPlanStopProductMeta
    from app.core.utils import generate_uuid
    db.add(PlPlanStopProductMeta(
        id=generate_uuid(), tenant_no=tenant_no, plan_id=plan_id, stop_type=stop_type,
        stop_id=stop_id, stop_product_id=stop_product_id,
        **{f: _value(product, f) for f in IDENTITY_FIELDS},
        product_name=_value(product, 'product_name'),
        **{'source_' + f: _value(plan_product, f) for f in SOURCE_FIELDS},
        **values, order_num=0, create_by=user_id, create_dept_id=dept_id or '',
        create_time=now, update_by=user_id, update_time=now, del_flag=0,
    ))


def attach_stop_meta(result, snapshots):
    from app.schemas.pl.plan import PlanStopProductMeta
    by_id = {(row.stop_type, row.stop_product_id): row for row in snapshots}
    stops = [('load', result.load)] + [('delivery', row) for row in result.deliveries or []]
    for stop_type, stop in stops:
        if stop is None:
            continue
        for product in stop.products:
            row = by_id.get((stop_type, product.id))
            if row:
                product.mobile_product_meta = PlanStopProductMeta(**_saved_meta_values(row))


def attach_web_selection_as_mobile_meta(result, standard_by_id, alias_options_by_id):
    """Expose the Web plan-product alias and brand in the mobile read shape. No side table."""
    from app.schemas.pl.plan import PlanStopProductMeta
    from app.services.pl_product_display import resolve_alias_from_saved_name
    products = {row.id: row for row in result.products or []}
    stops = [result.load] + list(result.deliveries or [])
    for stop in stops:
        if stop is None:
            continue
        for product in stop.products:
            linked = products.get(getattr(product, 'plan_product_id', None))
            if linked is None:
                continue
            product_id = _value(linked, 'product_id')
            alias = resolve_alias_from_saved_name(
                getattr(linked, 'product_name', None),
                standard_by_id.get(product_id),
                alias_options_by_id.get(product_id) or [],
            )
            product.mobile_product_meta = PlanStopProductMeta(
                alias_name=alias or '',
                brand_mode=_value(linked, 'brand_mode'),
                brand_id=_value(linked, 'brand_id'),
                brand_name=_value(linked, 'brand_name'),
            )