import os import pandas as pd from utils.Package import Package,Package_group from utils.gtools import MySQLconnect from utils.countryOperator import OperateCountry from utils.logisticsBill import BillFactory, Billing import requests ods = MySQLconnect("ods") engine = ods.engine() cursor = ods.connect().cursor() def get_package_real_vol_by_api(packages_id): packages = Package_group() packages_str = "" index = 1 for package_id in packages_id: # 计算往年利润率 # sql = f"SELECT length,width,hight,weight FROM package_vol_info WHERE package = %s" # cursor.execute(sql, (package_id,)) # resp = cursor.fetchall() # if len(resp) == 0: # continue # else: # weight = resp[0][3] # package_length = resp[0][0] # package_width = resp[0][1] # package_hight = resp[0][2] # 拦截 url = f'https://cp.maso.hk/index.php?main=biphp&act=package_fund&key=W6BOYJ7BH27YCGRFCA0LWBVKMU1KRU5Q&package={package_id}' resp = requests.get(url).json() if resp['code'] == "0": weight = int(float(resp['data'][0]['weight'])*1000) package_length = resp['data'][0]['l'].replace(",","") if len(resp['data'][0]['l'])>0 else "0" package_width = resp['data'][0]['w'].replace(",","") if len(resp['data'][0]['w'])>0 else "0" package_hight = resp['data'][0]['h'].replace(",","") if len(resp['data'][0]['h'])>0 else "0" package_str = f"{weight/1000}|{package_length}*{package_width}*{package_hight}" packages_str += package_str + "," index += 1 package = Package(str(package_id),float(package_length),float(package_width),float(package_hight),weight) packages.add_package(package) return packages,packages_str def get_order_bill(opCountry,order_id,packages_id,postcode,convey,amount): print(order_id) beizhu = amount conveys = 1 if convey == "海运" else 0 try: packages,packages_str = get_package_real_vol_by_api(packages_id=packages_id) bill = Billing(str(order_id),opCountry,packages,postcode,company_name=None,head_type=conveys,beizhu=beizhu) print(bill) bill_data = bill.bill_dict() # ✅ 只调用一次 result = bill_data.get("预测尾端", 0) _type = bill_data.get("尾端渠道", "") head_amount = bill_data.get("预测头程CNY", 0) total_amount = bill_data.get("总金额USD", 0) volume_weight = bill_data.get("体积重", 0) per_head = bill_data.get("头程单价", 0) return volume_weight,_type,per_head,head_amount,result,total_amount,packages_str except ZeroDivisionError as e: print(e) return "没测量" ,0,0,0,0,0,"" excel_path = r'D:\test\logistics\拦截数据\订单数据.xlsx' country_list = ['United Kingdom','United States','Australia','Germany','Spain','France'] # country_list = ['United States'] for country in country_list: order_id = f""" SELECT # ol.order_date, CONCAT("[",GROUP_CONCAT(pr.包裹号),"]") AS package_group , ol.order_id , ol.postcode, ol.delivery_country, ol.convey, ol.order_price_dollar FROM parcel pr LEFT JOIN dwd.order_list ol ON ol.order_id = pr.订单号 WHERE # ol.order_date between "2024-10-01" and "2025-03-01" pr.生成时间 >= DATE_SUB(NOW(), INTERVAL 20 DAY) AND fund_status NOT REGEXP "等待" AND site_name REGEXP "litfad|kwoking|lakiq" # AND convey = "海运" # AND delivery_country = "United States" AND pr.订单号 = ol.order_id AND delivery_country regexp '{country}' AND NOT EXISTS ( SELECT 1 FROM `order_express` oe WHERE oe.包裹号 = pr.包裹号 AND oe.包裹状态 = "已作废") GROUP BY ol.order_id """ order_id_df = pd.read_sql(order_id,engine) if order_id_df.empty: print(f"{country}无订单") continue order_id_df.sort_values(by=['delivery_country'],inplace=True) countries = order_id_df['delivery_country'].unique() opCountry = OperateCountry(country) order_id_df_cal = order_id_df.copy() # 删除order_id_df_cal中已经存在的订单 if os.path.exists(excel_path): existing_df = pd.read_excel(excel_path) order_id_df_cal = order_id_df_cal[~(order_id_df_cal['order_id'].isin(existing_df['order_id']) & order_id_df_cal['package_group'].isin(existing_df['package_group']))] if order_id_df_cal.empty: continue order_id_df_cal[["体积重","_type","头程单价","头程(CNY)","尾端","总金额(USD)","实际体积"]] = order_id_df_cal.apply(lambda x: get_order_bill(opCountry, x['order_id'], eval(x['package_group']), x['postcode'], x['convey'], x['order_price_dollar']),axis=1,result_type='expand') order_id_df_cal = order_id_df_cal[~(order_id_df_cal['体积重']== "没测量")] order_id_df_cal['总金额(USD)'] = pd.to_numeric(order_id_df_cal['总金额(USD)'], errors='coerce') order_id_df_cal = order_id_df_cal.dropna(subset=['总金额(USD)']) order_id_df_cal = order_id_df_cal[order_id_df_cal['总金额(USD)'] <9999] order_id_df_cal['生成日期'] = pd.Timestamp.now().strftime('%Y/%m/%d') order_id_df_cal['order_id'] = order_id_df_cal['order_id'].astype(int) if order_id_df_cal.empty: print(f"{country}没有需要测量的订单。") continue if not os.path.exists(excel_path): order_id_df_cal.to_excel(excel_path, index=False, sheet_name='Sheet1') else: exist_df = pd.read_excel(excel_path) new_rows = order_id_df_cal[~(order_id_df_cal['order_id'].isin(exist_df['order_id']) & order_id_df_cal['package_group'].isin(exist_df['package_group']))] updated_df = pd.concat([exist_df, new_rows], ignore_index=True) # order_id去重 updated_df = updated_df.drop_duplicates(subset=['order_id'], keep='last') if not new_rows.empty: # 将更新后的 DataFrame 写回 Excel 文件 updated_df.to_excel(excel_path, index=False, sheet_name='Sheet1') print(f"已写入 {len(new_rows)} 条新数据。") order_id_df_cal = order_id_df_cal.drop(columns=['生成日期']) order_id_list = order_id_df_cal["order_id"].tolist() param = ",".join(f"'{order_id}'" for order_id in order_id_list) purchase_order_sql = f""" with t1 AS (SELECT LEFT ( ol.out_detials_outlink_id, 15 ) AS order_id, SUM( out_detials_qty * price )/ 7 AS instock_cost, NULL AS buy_cost FROM ods.outstock_list ol JOIN ods.instock_list il ON ol.store_in_id = il.id WHERE LEFT ( ol.out_detials_outlink_id, 15 ) IN ({param}) GROUP BY LEFT ( ol.out_detials_outlink_id, 15 ) UNION ALL SELECT LEFT ( order_product_id, 15 ) as order_id, NULL as instock_cost, SUM(buy_num * actual_price)/7 AS buy_cost FROM `warehouse_purchasing` WHERE LEFT ( order_product_id, 15 ) IN ({param}) AND buy_audit = "采购完成" group by LEFT ( order_product_id, 15 ) ) SELECT order_id, SUM(CASE WHEN instock_cost is null THEN buy_cost ELSE instock_cost END) AS pur_cost FROM t1 GROUP BY order_id """ purchase_order_df = pd.read_sql(purchase_order_sql, con=engine) purchase_order_df["order_id"] = purchase_order_df["order_id"].astype(str) order_id_df_cal["order_id"] = order_id_df_cal["order_id"].astype(str) order_id_df_cal = pd.merge(order_id_df_cal, purchase_order_df, on='order_id', how='left') def profit_cal(amount,cost): transaction_fees_rate = 0.031 cur_tran_rate = 0 store_cost = 2 profit = amount * (1 - transaction_fees_rate - cur_tran_rate) - store_cost - cost return round(profit, 2) order_id_df_cal["profit"] = order_id_df_cal.apply(lambda x: profit_cal(x["order_price_dollar"] , x["pur_cost"] + x ["总金额(USD)"]), axis=1) order_id_df_cal["profit_rate"] = round(order_id_df_cal["profit"] / order_id_df_cal["order_price_dollar"],2) order_id_df_cal["rejust"] = order_id_df_cal.apply(lambda x: 1 if x["profit_rate"] < 0 else 0, axis=1) order_id_df_cal[order_id_df_cal['rejust'] == 1] order_id_df_cal['拦截处理日期'] = pd.to_datetime(pd.Timestamp.now().date()) # order_id_df_cal = order_id_df_cal[order_id_df_cal['rejust'] == 1] # import pendulum # dt = pendulum.now() # # 将日期时间对象转换为字符串 # dt_str = dt.to_format('YYYY-MM-DD') # finename = dt_str + "拦截.xlsx" def update_excel_with_new_data(excel_path, new_data_df): try: new_data_df['order_id'] = new_data_df['order_id'].astype(str) if not os.path.exists(excel_path): # 如果文件不存在，创建一个新的 Excel 文件 new_data_df = new_data_df[new_data_df['pur_cost'].notna()] new_data_df.to_excel(excel_path, index=False, sheet_name='Sheet1') print(f"文件不存在，已创建新文件并写入 {len(new_data_df)} 条数据。") df['拦截处理日期'] = pd.to_datetime(df['拦截处理日期']).dt.strftime('%Y/%m/%d') else: existing_df = pd.read_excel(excel_path) existing_df['order_id'] = existing_df['order_id'].astype(str) new_rows = new_data_df[~(new_data_df['order_id'].isin(existing_df['order_id']) & new_data_df['package_group'].isin(existing_df['package_group']))] new_rows = new_rows[new_rows['pur_cost'].notna()] updated_df = pd.concat([existing_df, new_rows], ignore_index=True) # order_id去重 updated_df = updated_df.drop_duplicates(subset=['order_id'], keep='last') updated_df.to_excel(excel_path, index=False, sheet_name='Sheet1') df = pd.read_excel(excel_path) df = df.drop_duplicates(subset=['order_id'], keep='last') df['拦截处理日期'] = pd.to_datetime(df['拦截处理日期']) df.to_excel(excel_path, index=False, sheet_name='Sheet1') if not new_rows.empty: # 将更新后的 DataFrame 写回 Excel 文件 print(f"已写入 {len(new_rows)} 条新数据。") else: print("没有新数据需要添加。") except Exception as e: print(f"发生错误: {e}") update_excel_with_new_data(r'D:\test\logistics\拦截数据\拦截总表.xlsx',order_id_df_cal)