logistics/logistics_service.py

"""物流费用计算统一入口服务"""
import re
from typing import Dict, List, Any, Optional

from utils.Package import Package, Package_group
from utils.countryOperator import OperateCountry
from logisticsClass.logisticsBaseClass import PortType


class LogisticsService:
    """物流费用计算统一服务"""

    @staticmethod
    def _detect_country(postcode: str) -> str:
        """根据邮编格式自动识别国家"""
        postcode = postcode.strip().upper()

        # 英国邮编格式：字母+数字(+字母)+空格+数字+字母
        # 支持格式: SW1A 1AA, M1 1AA, BT1 1AA, AA11 1AA 等
        if re.match(r'^[A-Z]{1,2}[0-9][A-Z0-9]?\s?[0-9][A-Z]{2}$', postcode):
            return "UK"

        # 美国邮编格式：5位数字或5位-4位
        if re.match(r'^\d{5}(-\d{4})?$', postcode):
            return "US"

        # 澳洲邮编格式：4位数字
        if re.match(r'^\d{4}$', postcode):
            return "AU"

        # 欧洲格式（德国、法国等）
        if re.match(r'^\d{5}$', postcode):
            return "DE"  # 默认德国

        raise ValueError(f"无法识别的邮编格式: {postcode}")

    @staticmethod
    def _parse_packages(packages_data: List[Dict]) -> Package_group:
        """解析包裹数据"""
        packages = []
        for i, pkg in enumerate(packages_data):
            name = pkg.get("name", f"包裹{i+1}")
            length = pkg.get("length", 0)
            width = pkg.get("width", 0)
            height = pkg.get("height", 0)
            weight = pkg.get("weight", 0)

            package = Package(name, length, width, height, weight)
            packages.append(package)

        return Package_group(packages)

    @staticmethod
    def calculate(postcode: str, packages_data: List[Dict],
                  port: PortType = PortType.DEFAULT) -> Dict[str, Any]:
        """
        计算物流费用并返回最优渠道

        Args:
            postcode: 收件人邮编
            packages_data: 包裹数据列表，格式为：
                [
                    {"length": 63, "width": 59, "height": 48, "weight": 8000},  # 单位：cm, g
                    ...
                ]
            port: 港口类型，默认DEFAULT

        Returns:
            包含最优渠道和所有渠道费用的字典
        """
        # 1. 识别国家
        country = LogisticsService._detect_country(postcode)

        # 2. 解析包裹
        packages = LogisticsService._parse_packages(packages_data)

        # 3. 创建国家操作对象
        op_country = OperateCountry(country, port, packages, postcode)

        # 4. 获取所有渠道费用
        all_fees = op_country.get_all_tail_info()

        # 5. 找出最优渠道
        valid_fees = {k: v for k, v in all_fees.items() if v < 99999}
        if not valid_fees:
            return {
                "country": country,
                "postcode": postcode,
                "optimal_channel": None,
                "optimal_fee": None,
                "currency": None,
                "all_channels": all_fees,
                "error": "所有渠道均不可用"
            }

        optimal_channel = min(valid_fees, key=valid_fees.get)
        optimal_fee = valid_fees[optimal_channel]
        currency = op_country.get_tail_currency(optimal_channel)

        # 6. 构建结果
        result = {
            "country": country,
            "postcode": postcode,
            "optimal_channel": optimal_channel,
            "optimal_fee": optimal_fee,
            "currency": currency,
            "all_channels": {},
            "package_count": len(packages_data),
            "total_weight": sum(p.weight for p in packages) / 1000,  # kg
        }

        # 7. 添加所有渠道详情
        for company, fee in all_fees.items():
            company_type = op_country.get_logistic_type(company)
            company_currency = op_country.get_tail_currency(company)
            result["all_channels"][company] = {
                "fee": fee if fee < 99999 else None,
                "currency": company_currency,
                "type": company_type,
                "available": fee < 99999
            }

        return result

    @staticmethod
    def calculate_us(postcode: str, packages_data: List[Dict],
                     port: PortType = PortType.DEFAULT) -> Dict[str, Any]:
        """计算美国物流费用"""
        return LogisticsService.calculate(postcode, packages_data, port)

    @staticmethod
    def calculate_uk(postcode: str, packages_data: List[Dict],
                    port: PortType = PortType.DEFAULT) -> Dict[str, Any]:
        """计算英国物流费用"""
        return LogisticsService.calculate(postcode, packages_data, port)

    @staticmethod
    def calculate_au(postcode: str, packages_data: List[Dict],
                    port: PortType = PortType.DEFAULT) -> Dict[str, Any]:
        """计算澳洲物流费用"""
        return LogisticsService.calculate(postcode, packages_data, port)

    @staticmethod
    def calculate_eur(postcode: str, packages_data: List[Dict],
                      port: PortType = PortType.DEFAULT) -> Dict[str, Any]:
        """计算欧洲物流费用"""
        return LogisticsService.calculate(postcode, packages_data, port)

    @staticmethod
    def get_company_detail(postcode: str, packages_data: List[Dict],
                          company_name: str) -> Dict[str, Any]:
        """获取指定物流公司的费用明细"""
        country = LogisticsService._detect_country(postcode)
        packages = LogisticsService._parse_packages(packages_data)
        op_country = OperateCountry(country, PortType.DEFAULT, packages, postcode)

        detail = op_country.get_detail_amount(company_name, packages, postcode)
        currency = op_country.get_tail_currency(company_name)

        return {
            "company": company_name,
            "currency": currency,
            "detail": detail,
            "total": detail.get("tail_amount", 0)
        }