2025-06-17 13:40:20 +08:00
|
|
|
|
"""物流模块基类,不做具体实现"""
|
|
|
|
|
|
from enum import Enum
|
|
|
|
|
|
|
|
|
|
|
|
class LogisticsType(Enum):
|
|
|
|
|
|
EXPRESS = '快递'
|
2025-06-23 16:08:03 +08:00
|
|
|
|
LTL = '卡派'
|
2025-06-17 13:40:20 +08:00
|
|
|
|
OCEAN = '海运'
|
|
|
|
|
|
AIR = '空运'
|
|
|
|
|
|
class PortType(Enum):
|
|
|
|
|
|
DEFAULT = '默认'
|
|
|
|
|
|
WEST = 'west'
|
|
|
|
|
|
CENTRAL = 'central'
|
|
|
|
|
|
EAST = 'east'
|
|
|
|
|
|
SOUTH = 'south'
|
|
|
|
|
|
NORTH = 'north'
|
|
|
|
|
|
|
|
|
|
|
|
# 基础物流费用类
|
|
|
|
|
|
class BaseLogistics():
|
|
|
|
|
|
"""基础物流类,强制要求子类定义特定属性"""
|
|
|
|
|
|
company: str
|
|
|
|
|
|
currency:str = 'USD' # 货币单位,默认美元
|
|
|
|
|
|
port: PortType = PortType.DEFAULT
|
|
|
|
|
|
logistics_type:LogisticsType
|
|
|
|
|
|
active: bool = True # 是否启用该物流,默认未启用
|
|
|
|
|
|
def __init__(self):
|
|
|
|
|
|
# 检查子类是否定义了必要的类变量
|
|
|
|
|
|
for attr in ['company', 'country_code', 'country']:
|
|
|
|
|
|
if not getattr(self.__class__, attr, None):
|
|
|
|
|
|
raise AttributeError(f"Subclass {self.__class__.__name__} is missing required attribute '{attr}'.")
|
|
|
|
|
|
|
|
|
|
|
|
def __repr__(self):
|
|
|
|
|
|
return f"类名:{self.__class__.__name__}, 物流类型:{self.logistics_type.value}"
|
|
|
|
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
|
|
def get_open_subclasses(cls):
|
|
|
|
|
|
"""递归获取所有启用状态为 True 的子类"""
|
|
|
|
|
|
subclasses = set(cls.__subclasses__())
|
|
|
|
|
|
for subclass in cls.__subclasses__():
|
|
|
|
|
|
subclasses.update(subclass.get_open_subclasses())
|
|
|
|
|
|
subclasses = {subclass for subclass in subclasses if subclass.active ==True}
|
|
|
|
|
|
return subclasses
|
|
|
|
|
|
@classmethod
|
|
|
|
|
|
def get_close_subclasses(cls):
|
|
|
|
|
|
"""获取关闭状态的子类,状态为 False 的子类"""
|
|
|
|
|
|
subclasses = set(cls.__subclasses__())
|
|
|
|
|
|
for subclass in cls.__subclasses__():
|
|
|
|
|
|
subclasses.update(subclass.get_close_subclasses())
|
|
|
|
|
|
return {subclass for subclass in subclasses if subclass.active == False}
|
|
|
|
|
|
@classmethod
|
|
|
|
|
|
def open_logistics(cls):
|
|
|
|
|
|
"""打开该类"""
|
|
|
|
|
|
cls.active = True
|
|
|
|
|
|
@classmethod
|
|
|
|
|
|
def close_logistics(cls):
|
|
|
|
|
|
"""关闭该类"""
|
|
|
|
|
|
cls.active = False
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class TailLogistics(BaseLogistics):
|
|
|
|
|
|
"""尾端物流类,包含快递和卡派"""
|
|
|
|
|
|
# 费用结果保存在实例对象中,不需要cal函数返回
|
|
|
|
|
|
country_code: str
|
|
|
|
|
|
country: str
|
|
|
|
|
|
company: str
|
|
|
|
|
|
currency:str = 'USD' # 货币单位,默认美元
|
|
|
|
|
|
port: PortType = PortType.DEFAULT
|
|
|
|
|
|
logistics_type = LogisticsType.EXPRESS # 默认快递
|
|
|
|
|
|
def __init__(self):
|
|
|
|
|
|
super().__init__()
|
|
|
|
|
|
def calculate_fee(self):
|
|
|
|
|
|
"""计算费用"""
|
|
|
|
|
|
raise NotImplementedError("Subclasses must implement express fee calculation.")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class HeadLogistics(BaseLogistics):
|
|
|
|
|
|
"""头程物流类,包含海运和空运"""
|
|
|
|
|
|
country_code: list[str]
|
|
|
|
|
|
country: list[str]
|
|
|
|
|
|
logistics_type = LogisticsType.OCEAN # 默认海运
|
|
|
|
|
|
currency = 'CNY'
|
|
|
|
|
|
def __init__(self):
|
|
|
|
|
|
super().__init__()
|
|
|
|
|
|
self.head_ratio:float
|
|
|
|
|
|
|
|
|
|
|
|
def calculate_fee(self, packages):
|
|
|
|
|
|
"""计算费用"""
|
|
|
|
|
|
detail_amount = {
|
|
|
|
|
|
"volume_weight":0.00,
|
|
|
|
|
|
"head_per":0.00,
|
|
|
|
|
|
"head_amount":0.00
|
|
|
|
|
|
}
|
|
|
|
|
|
total_weight = sum([package.get_volume_weight(6000) for package in packages])
|
|
|
|
|
|
detail_amount['volume_weight'] = round(total_weight,2)
|
|
|
|
|
|
detail_amount['head_per'] = self.head_ratio
|
|
|
|
|
|
detail_amount["head_amount"] = total_weight * self.head_ratio
|
|
|
|
|
|
return detail_amount
|