一、业务背景
在日本直邮业务中，运费计算是一个复杂的问题。不同物流线路（EMS、航空小包、SAL、包税专线）的计费规则完全不同：
EMS：按重量分段计价，500g以内一个价，之后每500g递增
航空小包：按重量+体积计费，取较大值
包税专线：按重量计费 + 固定报关费
大货专线：首重+续重模式
而且规则经常变化——物流公司调价、汇率波动、燃油附加费调整。如果把这些规则硬编码在代码里，每次调价都要改代码、发版、重启服务，运维成本极高。
二、策略模式：每个线路一个独立策略
python
from abc import ABC, abstractmethodclass ShippingStrategy(ABC): @abstractmethod def calculate(self, weight_kg, volume_cm3, declared_value): pass @abstractmethod def get_estimated_days(self): passclass EMSStrategy(ShippingStrategy): def init(self): # 从配置中心加载费率 self.rate_table = self._load_rate_table() def calculate(self, weight_kg, volume_cm3, declared_value): # EMS按重量分段计价 weight_g = weight_kg 1000 if weight_g <= 500: return self.rate_table['first_500'] else: extra = (weight_g - 500 + 499) // 500 # 向上取整 return self.rate_table['first_500'] + extra self.rate_table['extra_500'] def get_estimated_days(self): return "3-5 days"class TaxFreeStrategy(ShippingStrategy): def calculate(self, weight_kg, volume_cm3, declared_value): # 包税线路：重量费 + 报关费 weight_fee = weight_kg self.rate_table['per_kg'] declaration_fee = 50 # 固定报关费 # 体积费（如果体积重量大于实际重量） volume_weight = volume_cm3 / 6000 # 体积重计算公式 volume_fee = max(0, (volume_weight - weight_kg)) self.rate_table['per_kg'] return weight_fee + declaration_fee + volume_fee
三、规则引擎：让计费规则可配置
虽然策略模式把每条线路的逻辑封装了，但费率本身还是硬编码在代码里。我引入了一个规则引擎，让非技术人员也能修改费率：
python
import jsonimport redisclass RuleEngine: def init(self): self.redis = redis.Redis(decode_responses=True) self.cache_key = "shipping:rules" def get_rule(self, strategy_name): """从Redis获取最新的规则配置""" rules = self.redis.hget(self.cache_key, strategy_name) if rules: return json.loads(rules) return self._get_default_rule(strategy_name) def evaluate(self, strategy_name, context): """执行规则引擎""" rule = self.get_rule(strategy_name) # 检查条件 for condition in rule.get('conditions', []): if not self._check_condition(condition, context): return None # 计算费用 return self._calculate(rule['formula'], context)
规则配置示例（JSON格式，存储在Redis中）：
json
{ "ems": { "conditions": [ {"field": "weight_kg", "operator": "<=", "value": 30} ], "formula": { "type": "piecewise", "segments": [ {"max": 0.5, "price": 1400}, {"max": 1.0, "price": 1800}, {"max": 1.5, "price": 2100} ], "extra_per_500g": 300 } }}
四、合箱运费的分摊算法
合箱是日本代购用户常用的省钱方式。合箱后，运费需要按比例分摊到每个商品上：
python
def split_shipping_cost(items, total_weight, total_volume, total_cost): """按重量+体积的加权平均分摊运费""" total_score = 0 scores = [] for item in items: # 重量分 + 体积分（各占50%） weight_score = item['weight'] / total_weight 0.5 volume_score = item['volume'] / total_volume 0.5 score = weight_score + volume_score scores.append(score) total_score += score # 按比例分配 results = [] for i, item in enumerate(items): share = total_cost * (scores[i] / total_score) results.append({ 'item_id': item['id'], 'shipping_cost': round(share, 2) }) return results
五、实时汇率集成
运费以日元计价，但用户看到的是人民币。需要集成实时汇率：
python
import aiohttpimport asyncioclass ExchangeRateService: def init(self): self.cached_rate = None self.cache_time = 0 async def get_rate(self): """获取实时汇率（带缓存）""" now = time.time() if self.cached_rate and (now - self.cache_time) < 300: return self.cached_rate async with aiohttp.ClientSession() as session: async with session.get( "https://api.exchangerate.host/latest", params={"base": "JPY", "symbols": "CNY"} ) as resp: data = await resp.json() self.cached_rate = data['rates']['CNY'] self.cache_time = now return self.cached_rate
六、总结
运费计算引擎的核心设计思路是：策略模式封装逻辑，规则引擎管理配置，实时汇率动态更新。这套架构让运费计算模块的维护成本降低了80%，每次调价只需要在Redis中更新配置，不需要发版。
