Upload app.py

app.py

@@ -5,6 +5,8 @@ import time
 import os
 import tempfile
 import re
+import threading
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from collections import OrderedDict
 from datetime import datetime
 from zoneinfo import ZoneInfo
@@ -36,6 +38,7 @@ NOTION_API_BASE = "https://api.notion.com/v1"
 NOTION_PAGE_SIZE = 100
 REQUEST_TIMEOUT = 30
 API_RATE_LIMIT_DELAY = 0.3
+MAX_CONCURRENT_WORKERS = 3
 
 HEADERS = {
     "Authorization": f"Bearer {NOTION_TOKEN}",
@@ -217,13 +220,14 @@ def extract_page_properties(page):
     return result
 
 
-def query_all_pages(extract_func):
+def query_all_pages(extract_func, filter_body=None):
     """
-    通用 Notion 分页查询
+    通用 Notion 分页查询，支持可选服务端过滤
 
     参数:
         extract_func: 从每个 page 提取数据的函数，返回 (key, value) 元组
                      如果 key 为 None，则跳过该记录
+        filter_body:  可选，Notion filter 字典，传入后由服务端过滤，减少传输量
     返回:
         dict: 提取的数据字典
     """
@@ -235,6 +239,8 @@ def query_all_pages(extract_func):
         body = {"page_size": NOTION_PAGE_SIZE}
         if start_cursor:
             body["start_cursor"] = start_cursor
+        if filter_body:
+            body["filter"] = filter_body
 
         r = notion_post(f"databases/{DATABASE_ID}/query", body)
         data = r.json()
@@ -847,15 +853,12 @@ def update_cwt(file, logger=None):
     existing_pages = query_all_pages(extract_mawb_with_id_ar_ap)
     logger.log(f"Total {len(existing_pages)} MAWBs in Notion")
 
-    # -------- 更新数据 --------
-    updated_count = 0
+    # -------- Phase 1：预处理，构建更新任务列表 --------
     skipped = []
     skipped_cwt_ar_ap = []
-    total = len(mawb_list)
-
-    for idx, (mawb, weight, ata) in enumerate(zip(mawb_list, weight_list, ata_list), 1):
-        logger.log(f"Progress: {idx}/{total} | Processing MAWB: {mawb}")
+    tasks = []  # (mawb, page_id, properties, log_suffix)
 
+    for mawb, weight, ata in zip(mawb_list, weight_list, ata_list):
         if mawb not in existing_pages:
             logger.log(f"MAWB {mawb} not found in Notion, skipped")
             skipped.append(mawb)
@@ -867,6 +870,7 @@ def update_cwt(file, logger=None):
         ap_done = page_info["ap_checked"]
 
         properties = {}
+        log_parts = []
 
         # CWT：如果 AR 或 AP 已完成则跳过
         if ar_done or ap_done:
@@ -875,38 +879,49 @@ def update_cwt(file, logger=None):
                 reasons.append("AR")
             if ap_done:
                 reasons.append("AP")
-            logger.log(f"CWT skipped ({' & '.join(reasons)} already checked)")
+            logger.log(f"MAWB {mawb}: CWT skipped ({' & '.join(reasons)} already checked)")
             skipped_cwt_ar_ap.append(f"{mawb} - {' & '.join(reasons)} already checked")
         else:
             properties["CWT"] = {"number": weight}
+            log_parts.append(f"CWT={weight:.2f}")
 
         # ATA 仅在 Excel 中有值时才写入（不受 AR/AP 影响）
         ata_str = None
         if pd.notna(ata):
             ata_str = ata.strftime("%Y/%m/%d %H:%M")
-            properties["ATA"] = {
-                "rich_text": [{"text": {"content": ata_str}}]
-            }
+            properties["ATA"] = {"rich_text": [{"text": {"content": ata_str}}]}
+            log_parts.append(f"ATA={ata_str}")
 
-        # 如果没有任何属性需要更新，跳过 API 调用
+        # 没有任何属性需要更新，跳过
         if not properties:
-            logger.log(f"MAWB {mawb} nothing to update, skipped")
+            logger.log(f"MAWB {mawb}: nothing to update, skipped")
             continue
 
-        r = notion_patch(page_id, properties)
+        tasks.append((mawb, page_id, properties, " | ".join(log_parts)))
 
-        if r.status_code == 200:
-            parts = []
-            if "CWT" in properties:
-                parts.append(f"CWT={weight:.2f}")
-            if "ATA" in properties:
-                parts.append(f"ATA={ata_str}")
-            logger.log(f"MAWB {mawb} updated | {' | '.join(parts)}")
-            updated_count += 1
-        else:
-            logger.log(f"MAWB {mawb} update failed: {r.text}")
+    logger.log(f"\nPrepared {len(tasks)} update tasks, starting concurrent PATCH (workers={MAX_CONCURRENT_WORKERS})...")
+
+    # -------- Phase 2：并发 PATCH --------
+    updated_count = 0
+    lock = threading.Lock()
+
+    def _patch_task(mawb, page_id, properties, log_suffix):
+        r = notion_patch(page_id, properties)
+        return mawb, r.status_code, r.text, log_suffix
 
-        time.sleep(API_RATE_LIMIT_DELAY) 
+    with ThreadPoolExecutor(max_workers=MAX_CONCURRENT_WORKERS) as executor:
+        futures = {
+            executor.submit(_patch_task, mawb, page_id, props, log_sfx): mawb
+            for mawb, page_id, props, log_sfx in tasks
+        }
+        for future in as_completed(futures):
+            mawb, status, text, log_sfx = future.result()
+            with lock:
+                if status == 200:
+                    logger.log(f"MAWB {mawb} updated | {log_sfx}")
+                    updated_count += 1
+                else:
+                    logger.log(f"MAWB {mawb} update failed: {text}")
 
     logger.log("\n================ Process Completed ================")
     logger.log(f"Updated MAWB: {updated_count}")
@@ -1068,62 +1083,77 @@ def update_whout_date(file, logger=None):
     df = df.dropna(subset=[mawb_col])
     logger.log(f"Excel read completed, total {len(df)} records")
 
-    # 查询 Notion
-    logger.log("Querying Notion database...")
-    existing_pages = query_all_pages(extract_mawb_with_id_whout)
-    logger.log(f"Total {len(existing_pages)} MAWBs in Notion")
+    # -------- 方案二：服务端过滤，只拉取 WH-OUT Date 为空的记录 --------
+    logger.log("Querying Notion (filter: WH-OUT Date is empty)...")
+    whout_empty_filter = {"property": "WH-OUT Date", "date": {"is_empty": True}}
+    pages_to_update = query_all_pages(extract_mawb_with_id_whout, filter_body=whout_empty_filter)
+    logger.log(f"Records without WH-OUT Date: {len(pages_to_update)}")
 
-    updated = 0
+    # 第二次轻量查询：获取全部 MAWB，用于区分「未找到」与「已有日期」
+    logger.log("Querying all MAWBs for existence check...")
+    all_mawbs = set(query_all_pages(extract_mawb_only).keys())
+    logger.log(f"Total MAWBs in Notion: {len(all_mawbs)}")
+
+    # -------- Phase 1：预处理，构建更新任务列表 --------
     skipped_no_date = []
     skipped_already = []
     not_found = []
-    total = len(df)
+    tasks = []  # (mawb, page_id, date_value)
 
-    for current, (_, row) in enumerate(df.iterrows(), 1):
+    for _, row in df.iterrows():
         mawb = str(row[mawb_col]).strip()
-        logger.log(f"Progress: {current}/{total} | Processing MAWB: {mawb}")
 
         if not mawb or mawb.lower() == 'nan':
             logger.log("Empty MAWB, skipped")
             continue
 
-        if mawb not in existing_pages:
-            logger.log("Not found in Notion")
-            not_found.append(mawb)
-            continue
+        if mawb in pages_to_update:
+            # 日期清洗
+            raw_date = row.get(whout_col)
+            if pd.isna(raw_date):
+                logger.log(f"MAWB {mawb}: WH-OUT Date is empty in Excel, skipped")
+                skipped_no_date.append(mawb)
+                continue
 
-        page_info = existing_pages[mawb]
+            try:
+                date_value = pd.to_datetime(raw_date).strftime("%Y-%m-%d")
+            except Exception:
+                logger.log(f"MAWB {mawb}: Date parse failed '{raw_date}', skipped")
+                skipped_no_date.append(f"{mawb} - parse failed: {raw_date}")
+                continue
+
+            tasks.append((mawb, pages_to_update[mawb]["id"], date_value))
 
-        # 已有日期则跳过，不做覆盖
-        if page_info["has_whout"]:
-            logger.log("WH-OUT Date already exists in Notion, skipped")
+        elif mawb in all_mawbs:
+            logger.log(f"MAWB {mawb}: WH-OUT Date already exists in Notion, skipped")
             skipped_already.append(mawb)
-            continue
+        else:
+            logger.log(f"MAWB {mawb}: Not found in Notion")
+            not_found.append(mawb)
 
-        # 日期清洗
-        raw_date = row.get(whout_col)
-        if pd.isna(raw_date):
-            logger.log("WH-OUT Date is empty in Excel, skipped")
-            skipped_no_date.append(mawb)
-            continue
+    logger.log(f"\nPrepared {len(tasks)} update tasks, starting concurrent PATCH (workers={MAX_CONCURRENT_WORKERS})...")
 
-        try:
-            date_value = pd.to_datetime(raw_date).strftime("%Y-%m-%d")
-        except Exception:
-            logger.log(f"Date parse failed: '{raw_date}', skipped")
-            skipped_no_date.append(f"{mawb} - parse failed: {raw_date}")
-            continue
+    # -------- Phase 2：并发 PATCH --------
+    updated = 0
+    lock = threading.Lock()
 
-        page_id = page_info["id"]
+    def _patch_task(mawb, page_id, date_value):
         r = notion_patch(page_id, {"WH-OUT Date": {"date": {"start": date_value}}})
+        return mawb, r.status_code, r.text, date_value
 
-        if r.status_code == 200:
-            logger.log(f"WH-OUT Date updated to {date_value}")
-            updated += 1
-        else:
-            logger.log(f"Update failed: {r.text}")
-
-        time.sleep(API_RATE_LIMIT_DELAY)
+    with ThreadPoolExecutor(max_workers=MAX_CONCURRENT_WORKERS) as executor:
+        futures = {
+            executor.submit(_patch_task, mawb, page_id, date_val): mawb
+            for mawb, page_id, date_val in tasks
+        }
+        for future in as_completed(futures):
+            mawb, status, text, date_val = future.result()
+            with lock:
+                if status == 200:
+                    logger.log(f"MAWB {mawb}: WH-OUT Date updated to {date_val}")
+                    updated += 1
+                else:
+                    logger.log(f"MAWB {mawb}: Update failed: {text}")
 
     logger.log("\n================ Process Completed ================")
     logger.log(f"Updated: {updated}")