該段代碼主要實現從數據庫和 Excel 文件中讀取數據，并對兩者進行字段匹配，最終找出 Excel 中未匹配到的數據庫記錄。功能如下：

1. [sqlSelect()]：連接 MySQL 數據庫并查詢比價單及其商品信息。
2. [BiJiaDaoChu()]：調用外部 API 導出 Excel 文件（注釋中未被調用）。
3. [read_excel_to_dict()]：將 Excel 文件讀取為字典列表。
4. [normalize_value()]：統一不同格式的值（如時間、空值、數字轉字符串），便于后續比較。
5. [match_list_to_list()]：根據字段映射匹配兩個字典列表中的條目，允許時間差 2 秒。
6. [find_unmatched_in_list_b()]：找出在 Excel 中沒有匹配到的數據庫記錄。

最終輸出：打印出數據庫中在 Excel 中未找到匹配項的數據。

import jsonimport pandas as pd
import pymysql
import requests
import datetime
import numpy as np
from pymysql import Timestampdef sqlSelect():global result_skudb = pymysql.Connect(host='IP',port=3306,user='user',password='password',db='test',charset='utf8',cursorclass=pymysql.cursors.DictCursor)cur = db.cursor()sql = 'select * from aaa a join bbb b  on a.id=b.sheet_id where a.status in(1,2)'cur.execute(sql)result_sku = cur.fetchall()cur.close()db.close()# print(result_sku)return result_skudef BiJiaDaoChu():data = {"askSheetCode": None}headers = {"Authorization": "440d9854d7434d1f998081abc6785fab","Content-Type": "application/json"}url = 'http:test/export'response = requests.post(url=url, data=json.dumps(data), headers=headers)# 判斷響應類型content_type = response.headers.get('Content-Type', '')if 'application/json' in content_type:try:print(response.json())  # 嘗試解析 JSONexcept json.JSONDecodeError:print("無法解析 JSON 響應")elif 'application/octet-stream' in content_type or 'application/vnd.ms-excel' in content_type:with open("../data/exported_data_bijia_test.xls", "wb") as f:f.write(response.content)  # 保存 Excel 文件print("文件已保存為 exported_data_bijia_test.xls")else:print("未知響應類型:", content_type)print(response.text)def read_excel_to_dict(file_path):"""讀取 Excel 文件并將數據以字典的形式返回。:param file_path: Excel 文件的路徑:return: 包含數據的字典列表"""try:# 讀取 Excel 文件df = pd.read_excel(file_path)# 將 DataFrame 轉換為字典列表data = df.to_dict(orient='records')# print("Excel 數據讀取成功", data)return dataexcept Exception as e:print(f"讀取 Excel 文件時出錯: {e}")return []def normalize_value(value):# 處理空值if value is None or (isinstance(value, float) and np.isnan(value)) or value == '':return None# 統一時間格式為 datetime.datetimeif isinstance(value, pd.Timestamp):return value.to_pydatetime()elif isinstance(value, datetime.datetime):return valueelif isinstance(value, datetime.date):return datetime.datetime.combine(value, datetime.time())# 統一數字類型為字符串if isinstance(value, (int, float)):return str(int(value)) if isinstance(value, float) else str(value)# 統一字符串類型：去除前后空格if isinstance(value, str):return value.strip()return valuedef match_list_to_list(list_a, list_b, field_mapping):"""比較兩個字典列表，返回匹配成功的對。增加字段級調試打印 + 時間字段允許最多相差 2 秒。"""matched_pairs = []for a_item in list_a:for b_item in list_b:matched = Truefor key_a, key_b in field_mapping.items():val_a = normalize_value(a_item.get(key_a))val_b = normalize_value(b_item.get(key_b))# 如果都是時間類型，允許最多差 2 秒if isinstance(val_a, datetime.datetime) and isinstance(val_b, datetime.datetime):diff_seconds = abs((val_a - val_b).total_seconds())if diff_seconds <= 2:continue  # 允許匹配成功elif val_a != val_b:#print(f"[字段不匹配] {key_a}({val_a!r}) vs {key_b}({val_b!r})")matched = Falsebreakif matched:matched_pairs.append((a_item, b_item))#print("[匹配成功] 找到一對匹配項")break  # 可選：找到第一個就停止return matched_pairsdef find_unmatched_in_list_b(matched_pairs, list_b):# print("完整匹配對 matched_pairs：", matched_pairs)matched_keys = []for idx, (a_item, b_item) in enumerate(matched_pairs):# print(f"[{idx}] 提取 b.id: {b_item['b.id']}")  # 調試每條提取matched_keys.append(b_item['b.id'])# print("提取到的所有 b.id 列表:", matched_keys)unmatched_b_items = []for item in list_b:if item['b.id'] not in matched_keys:unmatched_b_items.append(item)return unmatched_b_itemsfile_path = "../data/exported_data_bijia_test.xls"
excel_data = read_excel_to_dict(file_path)
list_a = excel_data
list_b = sqlSelect()mapping = {'需求比價單號': 'sheet_sn', '業務分類': 'business_class', '報價開始時間': 'quotation_start_time','報價截止時間': 'quotation_end_time', '采購單位': 'purchase_unit', '聯系人': 'lixiren', '聯系方式': 'mobile','商品名稱': 'goods_name', '品牌': 'pinpai', '計量單位': 'unit', '采購數量': 'number', '規格描述': 'specifications',
'技術標準': 'standard_code', '備注': 'remark', '其他信息': 'other_info'}matches = match_list_to_list(list_a, list_b, mapping)
# print("匹配結果：=========", matches)
# for a, b in matches:
#     print("匹配成功：")
#     print("  list_a 項:", a)
#     print("  list_b 項:", b)
# 查找未匹配的 list_b 數據
unmatched_b = find_unmatched_in_list_b(matches, list_a, list_b)# 打印出來
print("=== list_b 中未在 list_a 匹配到的數據 ===")
for item in unmatched_b:print(item)