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

"""运输计划现场链接读模型"""
from __future__ import annotations

import json

from sqlalchemy.orm import Session

from app.models.pl_models import PlTransportSiteLink
from app.schemas.pl.transport import TransportSiteLinkRead
from app.services.db_inventory_site_token import build_site_transport_url


def build_transport_site_link_read(
    db: Session,
    *,
    plan_product_id: str,
    product_name: str,
    site_link_id: str,
    plan_no: str,
    business_type: str,
    site_access_token: str | None = None,
) -> TransportSiteLinkRead:
    token = (site_access_token or "").strip()
    if not token and site_link_id:
        row = (
            db.query(PlTransportSiteLink)
            .filter(
                PlTransportSiteLink.id == site_link_id,
                PlTransportSiteLink.del_flag == 0,
            )
            .first()
        )
        if row:
            token = (row.site_access_token or "").strip()
    site_url = build_site_transport_url(token) if token else None
    return TransportSiteLinkRead(
        plan_product_id=plan_product_id,
        product_name=product_name,
        site_link_id=site_link_id,
        plan_no=plan_no,
        business_type=business_type,
        site_access_token=token or None,
        site_url=site_url,
    )


def build_transport_site_link_read_from_row(row: PlTransportSiteLink, product_name: str = "") -> TransportSiteLinkRead:
    token = (row.site_access_token or "").strip()
    return TransportSiteLinkRead(
        plan_product_id=row.plan_product_id,
        product_name=product_name or row.product or "",
        site_link_id=row.id,
        plan_no=row.plan_no or "",
        business_type=row.business_type,
        site_access_token=token or None,
        site_url=build_site_transport_url(token) if token else None,
        site_feedback_remark=row.site_feedback_remark,
        site_feedback_images=_parse_feedback_images(row.site_feedback_images),
        site_feedback_time=row.site_feedback_time,
    )


def _parse_feedback_images(raw: str | None) -> list[str]:
    if not raw:
        return []
    try:
        parsed = json.loads(raw)
        if isinstance(parsed, list):
            return [str(item) for item in parsed if item]
    except (TypeError, json.JSONDecodeError):
        pass
    return [part.strip() for part in raw.split(",") if part.strip()]