66 lines
2.0 KiB
Python
66 lines
2.0 KiB
Python
"""英国物流价格数据访问类"""
|
|
import pandas as pd
|
|
from typing import Dict, List, Optional
|
|
from dataaccess.base_dao import BaseDAO
|
|
|
|
|
|
class UKPriceDAO(BaseDAO):
|
|
"""英国物流价格数据访问"""
|
|
|
|
def __init__(self):
|
|
super().__init__("uk_logistics_price")
|
|
|
|
def get_all_companies(self) -> List[str]:
|
|
"""获取所有英国物流公司"""
|
|
df = self.get_all()
|
|
if df.empty:
|
|
return []
|
|
return df["company"].unique().tolist()
|
|
|
|
def get_company_price(self, company: str) -> pd.DataFrame:
|
|
"""获取指定物流公司的价格数据"""
|
|
return self.get_by_condition({"company": company})
|
|
|
|
def get_company_config(self, company: str) -> Dict:
|
|
"""获取物流公司配置信息"""
|
|
df = self.get_by_condition({"company": company})
|
|
if df.empty:
|
|
return {}
|
|
|
|
config = {}
|
|
for _, row in df.iterrows():
|
|
key = row.get("config_key")
|
|
value = row.get("config_value")
|
|
if key:
|
|
config[key] = value
|
|
|
|
return config
|
|
|
|
|
|
class UKPostcodeDAO(BaseDAO):
|
|
"""英国邮编分区数据访问"""
|
|
|
|
def __init__(self):
|
|
super().__init__("uk_postcode_zone")
|
|
|
|
def get_zone(self, postcode: str) -> Optional[str]:
|
|
"""根据邮编获取分区"""
|
|
postcode_prefix = postcode.split()[0].upper()
|
|
df = self.get_by_condition({"postcode_prefix": postcode_prefix})
|
|
|
|
if df.empty:
|
|
return None
|
|
|
|
return df.iloc[0].get("zone")
|
|
|
|
def is_remote(self, postcode: str) -> bool:
|
|
"""判断是否偏远"""
|
|
postcode_prefix = postcode.split()[0].upper()
|
|
remote_prefixes = ["BT", "IM", "JE", "ZE", "GY", "HS", "PO", "IV", "KA", "KW", "PH", "PA"]
|
|
|
|
if not any(postcode_prefix.startswith(p) for p in remote_prefixes):
|
|
return False
|
|
|
|
df = self.get_by_condition({"postcode_prefix": postcode_prefix, "is_remote": "1"})
|
|
return not df.empty
|