logistics/utils/logisticsBill.py

import sys
sys.path.append("D:\workspace\logistics")
from dataclasses import dataclass
from typing import List, Literal
import pandas as pd
from utils.countryOperator import OperateCountry
from utils.Package import Package, Package_group

@dataclass
class BillItem:
    """账单项"""
    amount: float   # 金额
    amount_usd: float   # 金额（美元）
    item_type: Literal['头程','尾程']  # 项目类型 头程或尾程
    item_detail: str  # 项目明细（如提单费，尾端基础运费，尾端附加费等）
    currency: str  # 货币单位（如 USD, EUR 等）
    


# 应该有个模式，计算实际账单金额，或者得出最小费用金额,这个由工厂来控制，写两个账单类
class Billing:
    """账单"""
    def __init__(self, bill_name="wxx", operator: OperateCountry = None, packages: Package_group=None,postcode=None, company_name=None,head_type = 1,beizhu = None):
        self.operator = operator
        self.name = bill_name
        self.packages = packages
        self.postcode = postcode
        self.country = operator.country_name
        self.company_name = company_name
        self.head_type = head_type
        self.beizhu = beizhu
        self.items: List[BillItem] = []  # 存储账单项
        self.volume_weight = 0
        self.head_per = 0
        self.logistic_type = None
        self.add_items_from_operator()
        
        
    def add_item(self, item: BillItem):
        """添加账单项"""
        self.items.append(item)

    def add_items_from_operator(self):
        """从Operator获取费用并添加账单项"""
        if self.packages is not None or self.postcode is not None:
            self.operator.set_packages_and_postcode(self.packages, self.postcode)  # 设置包裹信息
        company_name = self.company_name if self.company_name is not None else self.operator.get_min_company() # 取快递类型的最小公司
        logistic_type = self.logistic_type if self.logistic_type is not None else self.operator.get_logistic_type(company_name)
        self.company_name = company_name
        self.logistic_type = logistic_type

        # 获取头程费用
        head_detail = self.operator.get_ocean_fee() if self.head_type == 1 else self.operator.get_air_fee()
        head_fee = head_detail['head_amount']
        self.head_per = head_detail['head_per']
        self.volume_weight = head_detail['volume_weight']
        
        # 获取尾程费用
        detail_amount = self.operator.get_detail_amount(company_name)

        # 获取货币单位
        tail_currency = self.operator.get_tail_currency(company_name)
        # 添加账单项
        self.add_item(BillItem(amount = head_fee,amount_usd = self.operator.convert_to_usd(head_fee, 'CNY'), item_type = "头程", item_detail = "head_amount",currency = 'CNY'))
        for item_detail,amount in detail_amount.items():
            if amount == 0 and item_detail != "tail_amount":
                continue
            self.add_item(BillItem(amount = round(amount,2),amount_usd = self.operator.convert_to_usd(round(amount,2), tail_currency), item_type = "尾程",item_detail=item_detail, currency=tail_currency))

    def get_logistic_type(self):
        """获取渠道类型:快递/卡派"""
        return self.logistic_type
    
    def get_items(self):
        return self.items
    @property
    def tail_amount(self):
        for item in self.items:
            if item.item_type == "尾程" and item.item_detail == "tail_amount":
                tailfee = item.amount
                tailcurrency = item.currency
                break
        return tailfee,tailcurrency
    @property 
    def head_amount(self):
        for item in self.items:
            if item.item_type == "头程" and item.item_detail == "head_amount":
                headfee = item.amount
                headcurrency = item.currency
                break
        return headfee,headcurrency
    @property
    def total_amount_usd(self):
        for item in self.items:
            if item.item_type == "头程" and item.item_detail == "head_amount":
                headfee = item.amount_usd
            if item.item_type == "尾程" and item.item_detail == "tail_amount":
                tailfee = item.amount_usd
        return headfee + tailfee

    def bill_dict(self):
        """返回账单字典"""
        result = {}
        result['ID'] = self.name
        result['体积重'] = self.volume_weight
        result['头程单价'] = self.head_per
        result['预测头程CNY'] = self.head_amount[0]
        result['预测尾端'] = self.tail_amount[0]

        for item in self.items:
            if item.item_type == "头程" or item.item_detail == 'tail_amount':
                continue
            result[item.item_detail] = item.amount
        result['总金额USD'] = self.total_amount_usd
        result['货币单位'] = self.tail_amount[1]
        result['尾端渠道'] = self.company_name
        result['备注'] = self.beizhu
        return result
    # 附加费明细
    def get_other_fee(self):
        """返回账单附加费明细"""
        detail_items = {}
        for item in self.items:
            if item.item_type != "头程" and item.item_detail != "tail_amount" and item.item_detail != "base":
                detail_items[item.item_detail] = item.amount
        return detail_items
    
    def __repr__(self):
        result = "账单名称: " + str(self.name) + "\n"
        result += "账单项:"
        result += "   ".join(f"{item.item_detail}: {item.amount} {item.currency}" for item in self.items)
        result += "   总金额: " + str(self.total_amount_usd) + " USD"
        return result


class BillFactory:
    """账单工厂"""
    def __init__(self):
        self.bills = []
    def create_bill(self, bill_name,operator: OperateCountry,packages = None,postcode = None,company_name = None,head_type=1,beizhu = None):
        bill = Billing(bill_name,operator,packages,postcode,company_name,head_type,beizhu)
        self.add_bill(bill)
        return bill
    
    def add_bill(self, bill):
        self.bills.append(bill)
    def get_bills(self):
        """获取全部账单列表"""
        return self.bills
    def bills_to_df(self):
        """将账单列表导存为df"""
        # 先定义数据结构
        bills_dict = []
        for bill in self.bills:
            bills_dict.append(bill.bill_dict())
        df = pd.DataFrame(bills_dict)
        # 重排列顺序
        columns_to_move = ['预测尾端', '总金额USD', '货币单位', '尾端渠道', '备注']
        reordered_columns = [col for col in df.columns if col not in columns_to_move] + columns_to_move
        df = df[reordered_columns]
        return df
            
    def __repr__(self):
        result = "\n".join(str(bill) for bill in self.bills)
        return f"账单列表:\n{result}"
    
if __name__ == '__main__':
    # 测试
    
    package = Package("wxx",63,59,48,8000)
    packages= Package_group([package])
    postcode = "PA2 9BF"
    billFactory = BillFactory()
    opCountry= OperateCountry('UK')
    company_name = "智谷"
    bill = billFactory.create_bill("wxx",opCountry,packages,postcode,company_name,1,"无")
    print(billFactory)