Upload app.py

app.py

@@ -67,8 +67,7 @@ MANUAL_FEE_TYPES = [
 
 # 归档时需要迁移的属性（按 Notion 字段类型分组）
 ARCHIVE_RICH_TEXT_FIELDS = [
-    "Customer", "POD", "ETA", "ATA", "Service Level",
-    "LM Carrier", "AR Completed Date", "AP Completed Date"
+    "Customer", "POD", "Service Level", "LM Carrier"
 ]
 ARCHIVE_NUMBER_FIELDS = [
     "Parcels", "HTS Count", "Entry Count", "CWT",
@@ -78,7 +77,15 @@ ARCHIVE_NUMBER_FIELDS = [
     "WT Fee", "CES Transfer Fee"
 ]
 ARCHIVE_CHECKBOX_FIELDS = ["AR", "AP"]
-ARCHIVE_DATE_FIELDS = ["WH-OUT Date"]
+ARCHIVE_DATE_FIELDS = ["WH-OUT Date", "ETA", "ATA", "AR Completed Date", "AP Completed Date"]
+
+# POD → IANA timezone mapping (for localizing Excel dates when writing ETA/ATA)
+POD_TIMEZONE_MAP = {
+    "JFK": "America/New_York",
+    "ORD": "America/Chicago",
+    "LAX": "America/Los_Angeles",
+    "MIA": "America/New_York",
+}
 
 # ================== 会话与重试配置 ==================
 session = requests.Session()
@@ -306,14 +313,17 @@ def extract_mawb_with_id_whout(page):
 
 
 def extract_mawb_with_id_ar_ap(page):
-    """提取 MAWB 及页面 ID、AR/AP 状态"""
+    """提取 MAWB 及页面 ID、AR/AP 状态、POD"""
     title = page["properties"]["MAWB"]["title"]
     if title:
         mawb = title[0]["text"]["content"]
+        pod_rt = page["properties"].get("POD", {}).get("rich_text", [])
+        pod = pod_rt[0]["text"]["content"] if pod_rt else ""
         return mawb, {
             "id": page["id"],
             "ar_checked": page["properties"]["AR"]["checkbox"],
             "ap_checked": page["properties"]["AP"]["checkbox"],
+            "pod": pod,
         }
     return None, None
 
@@ -385,14 +395,17 @@ def create_mawb_eta(file, logger=None):
             skipped += 1
             continue
 
-        # -------- ETA 处理（文本写入）--------
-        eta_text = None
+        # -------- ETA 处理（日期写入，含目的地时区）--------
+        pod = str(row["Destination"]).strip()
+        tz_name = POD_TIMEZONE_MAP.get(pod.upper(), "America/Los_Angeles")
+        eta_date_prop = None
         if pd.notna(row["Estimated Time Arrival"]):
             try:
                 eta_dt = pd.to_datetime(row["Estimated Time Arrival"])
-                eta_text = eta_dt.strftime("%Y/%m/%d %H:%M")
+                eta_start = eta_dt.strftime("%Y-%m-%dT%H:%M:%S")
+                eta_date_prop = {"date": {"start": eta_start, "time_zone": tz_name}}
             except Exception:
-                eta_text = None
+                eta_date_prop = None
 
         service_level = str(row["Service Level"]).strip()
         cbm = row["CBM"]
@@ -407,7 +420,7 @@ def create_mawb_eta(file, logger=None):
                 "rich_text": [{"text": {"content": str(row["Customer"]).strip()}}]
             },
             "POD": {
-                "rich_text": [{"text": {"content": str(row["Destination"]).strip()}}]
+                "rich_text": [{"text": {"content": pod}}]
             },
             "Service Level": {
                 "rich_text": [{"text": {"content": service_level}}]
@@ -422,10 +435,8 @@ def create_mawb_eta(file, logger=None):
             "AP": {"checkbox": False}
         }
 
-        if eta_text:
-            properties["ETA"] = {
-                "rich_text": [{"text": {"content": eta_text}}]
-            }
+        if eta_date_prop:
+            properties["ETA"] = eta_date_prop
 
         # -------- HTS Count + Entry Count（仅 T01 服务等级）--------
         if service_level == "T01" and pd.notna(cbm):
@@ -486,77 +497,94 @@ def update_ar(file, logger=None):
     # 解析 Excel
     result = parse_ar_excel(file, logger)
     if isinstance(result, tuple) and len(result) == 2 and isinstance(result[1], str):
-        # 返回错误信息
         return result[1]
 
     df, (mawb_col, hts_fee_col, total_col) = result
 
-    # 查询 Notion
-    logger.log("Querying the existing data in Notion...")
-    existing_pages = query_all_pages(extract_mawb_with_ar)
-    logger.log(f"Total of {len(existing_pages)} MAWBs in Notion")
+    # -------- 服务端过滤：只查 AR=false 的记录 --------
+    logger.log("Querying Notion (filter: AR=false)...")
+    ar_false_filter = {"property": "AR", "checkbox": {"equals": False}}
+    pages_to_update = query_all_pages(extract_mawb_with_ar, filter_body=ar_false_filter)
+    logger.log(f"Records with AR unchecked: {len(pages_to_update)}")
+
+    # 轻量查询：全部 MAWB，用于区分「未找到」与「已勾选 AR」
+    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)}")
 
-    # 更新 AR
+    # 统一时间戳（洛杉矶时区，同批次一致）
+    la_tz = ZoneInfo("America/Los_Angeles")
+    _now_la = datetime.now(la_tz)
+    completed_time = _now_la.strftime("%Y/%m/%d %H:%M")         # 用于日志显示
+    completed_time_iso = _now_la.strftime("%Y-%m-%dT%H:%M:%S")  # 用于 Notion 日期字段
+
+    # -------- Phase 1：预处理，构建更新任务列表 --------
     updated, skipped, not_found = [], [], []
+    tasks = []  # (mawb, page_id, properties, log_suffix)
     total = len(df)
 
     for current, (_, row) in enumerate(df.iterrows(), 1):
         mawb = str(row[mawb_col]).strip()
-        logger.log(f"Progress: {current}/{total} | Processing MAWB: {mawb}")
+        logger.log(f"Preparing: {current}/{total} | 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:
+            properties = {
+                "AR": {"checkbox": True},
+                "AR Completed Date": {"date": {"start": completed_time_iso, "time_zone": "America/Los_Angeles"}}
+            }
+            log_parts = [f"AR checked, AR Completed Date: {completed_time}"]
+
+            if hts_fee_col and pd.notna(row.get(hts_fee_col)):
+                try:
+                    customs_fee = float(row[hts_fee_col])
+                    properties["Customs Clearance Fee"] = {"number": round(customs_fee, 2)}
+                    log_parts.append(f"Customs Clearance Fee: {customs_fee:.2f}")
+                except (ValueError, TypeError):
+                    logger.log("Warning: Invalid HTS Processing Fee value, skipped")
+
+            if total_col and pd.notna(row.get(total_col)):
+                try:
+                    ar_amount = float(row[total_col])
+                    properties["AR amount"] = {"number": round(ar_amount, 2)}
+                    log_parts.append(f"AR amount: {ar_amount:.2f}")
+                except (ValueError, TypeError):
+                    logger.log("Warning: Invalid Total value, skipped")
+
+            tasks.append((mawb, pages_to_update[mawb]["id"], properties, " | ".join(log_parts)))
 
-        page_info = existing_pages[mawb]
-        if page_info["ar_checked"]:
-            logger.log("AR already checked, skip")
+        elif mawb in all_mawbs:
+            logger.log("AR already checked, skipped")
             skipped.append(mawb)
-            continue
-
-        # 构建更新属性
-        properties = {}
-        properties["AR"] = {"checkbox": True}
-
-        # 获取当前时间（洛杉矶时区）
-        la_tz = ZoneInfo("America/Los_Angeles")
-        completed_time = datetime.now(la_tz).strftime("%Y/%m/%d %H:%M")
-        properties["AR Completed Date"] = {"rich_text": [{"text": {"content": completed_time}}]}
+        else:
+            logger.log("Not found in Notion")
+            not_found.append(mawb)
 
-        # 添加 Customs Clearance Fee（如果列存在）
-        if hts_fee_col and pd.notna(row.get(hts_fee_col)):
-            try:
-                customs_fee = float(row[hts_fee_col])
-                properties["Customs Clearance Fee"] = {"number": round(customs_fee, 2)}
-            except (ValueError, TypeError):
-                logger.log(f"Warning: Invalid HTS Processing Fee value, skipped")
+    logger.log(f"\nPrepared {len(tasks)} update tasks, starting concurrent PATCH (workers={MAX_CONCURRENT_WORKERS})...")
 
-        # 添加 AR amount（如果列存在）
-        if total_col and pd.notna(row.get(total_col)):
-            try:
-                ar_amount = float(row[total_col])
-                properties["AR amount"] = {"number": round(ar_amount, 2)}
-            except (ValueError, TypeError):
-                logger.log(f"Warning: Invalid Total value, skipped")
+    # -------- Phase 2：并发 PATCH --------
+    lock = threading.Lock()
 
-        r = notion_patch(page_info['id'], properties)
-        if r.status_code == 200:
-            log_parts = [f"AR checked successfully, AR Completed Date: {completed_time}"]
-            if "Customs Clearance Fee" in properties:
-                log_parts.append(f"Customs Clearance Fee: {properties['Customs Clearance Fee']['number']:.2f}")
-            if "AR amount" in properties:
-                log_parts.append(f"AR amount: {properties['AR amount']['number']:.2f}")
-            logger.log(" | ".join(log_parts))
-            updated.append(mawb)
-        else:
-            logger.log(f"Update failed: {r.text}")
+    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} | {log_sfx}")
+                    updated.append(mawb)
+                else:
+                    logger.log(f"MAWB {mawb} update failed: {text}")
 
     logger.log("\n================ Process Completed ================")
     logger.log(f"AR checked: {len(updated)}")
@@ -666,17 +694,25 @@ def aggregate_ap_fees(df, columns, logger):
     return mawb_dict, extra_fee_records, jfk_mawbs
 
 
-def update_ap_in_notion(mawb_dict, extra_fee_records, existing_pages, isc_summary, jfk_mawbs, logger):
+def update_ap_in_notion(mawb_dict, extra_fee_records, existing_pages, isc_summary, jfk_mawbs, all_mawbs, logger):
     """更新 AP 到 Notion"""
-    updated_count = 0
     skipped_ap_mawbs = []
     not_found_mawbs = []
     total = len(mawb_dict)
 
     extra_fee_mawbs = {x.split(" - ")[0] for x in extra_fee_records}
 
+    # 统一时间戳（洛杉矶时区，同批次一致）
+    la_tz = ZoneInfo("America/Los_Angeles")
+    _now_la = datetime.now(la_tz)
+    completed_time = _now_la.strftime("%Y/%m/%d %H:%M")         # 用于日志显示
+    completed_time_iso = _now_la.strftime("%Y-%m-%dT%H:%M:%S")  # 用于 Notion 日期字段
+
+    # -------- Phase 1：预处理，构建更新任务列表 --------
+    tasks = []  # (mawb, page_id, properties, ap_amount)
+
     for current, (mawb, fees) in enumerate(mawb_dict.items(), 1):
-        logger.log(f"Progress: {current}/{total} | Processing MAWB: {mawb}")
+        logger.log(f"Preparing: {current}/{total} | MAWB: {mawb}")
 
         # 跳过含有额外费用的 MAWB
         if mawb in extra_fee_mawbs:
@@ -684,20 +720,18 @@ def update_ap_in_notion(mawb_dict, extra_fee_records, existing_pages, isc_summar
             skipped_ap_mawbs.append(f"{mawb} - Extra fee present")
             continue
 
-        # MAWB 不在 Notion 中，跳过
+        # existing_pages 仅含 AP=false 记录；不在其中说明未找到或已勾选
         if mawb not in existing_pages:
-            logger.log("MAWB not found in Notion, skip")
-            not_found_mawbs.append(mawb)
+            if mawb in all_mawbs:
+                logger.log("AP already checked, skipped")
+                skipped_ap_mawbs.append(f"{mawb} - AP already checked")
+            else:
+                logger.log("MAWB not found in Notion, skip")
+                not_found_mawbs.append(mawb)
             continue
 
         page_info = existing_pages[mawb]
 
-        # 跳过已勾选 AP 的记录
-        if page_info["ap_checked"]:
-            logger.log("AP already checked, skip")
-            skipped_ap_mawbs.append(f"{mawb} - AP already checked")
-            continue
-
         # ===== CWT 前置强校验 =====
         cwt_value = page_info.get("CWT")
 
@@ -743,11 +777,9 @@ def update_ap_in_notion(mawb_dict, extra_fee_records, existing_pages, isc_summar
         for col_name in ["ISC Fee", "BreakDown Fee", "Bypass Fee", "CES Inspection Fee", "CES Transfer Fee"]:
             properties[col_name] = {"number": round(fees.get(col_name, 0.0), 2)}
 
-        # 写入 WT Fee
         wt_fee_value = round(fees.get("WT Fee", 0.0), 2)
         properties["WT Fee"] = {"number": wt_fee_value}
 
-        # 计算 AP amount（所有费用总和）
         ap_amount = (
             fees.get("ISC Fee", 0.0) +
             fees.get("BreakDown Fee", 0.0) +
@@ -757,21 +789,34 @@ def update_ap_in_notion(mawb_dict, extra_fee_records, existing_pages, isc_summar
             wt_fee_value
         )
         properties["AP amount"] = {"number": round(ap_amount, 2)}
-
         properties["AP"] = {"checkbox": True}
-        # 添加 AP Completed Date（洛杉矶时区）
-        la_tz = ZoneInfo("America/Los_Angeles")
-        completed_time = datetime.now(la_tz).strftime("%Y/%m/%d %H:%M")
-        properties["AP Completed Date"] = {"rich_text": [{"text": {"content": completed_time}}]}
+        properties["AP Completed Date"] = {"date": {"start": completed_time_iso, "time_zone": "America/Los_Angeles"}}
 
-        r = notion_patch(page_info['id'], properties)
-        if r.status_code == 200:
-            logger.log(f"Update successful, AP checked, AP amount: {ap_amount:.2f}, AP Completed Date: {completed_time}")
-            updated_count += 1
-        else:
-            logger.log(f"Update failed: {r.text}")
+        tasks.append((mawb, page_info["id"], properties, ap_amount))
 
-        time.sleep(API_RATE_LIMIT_DELAY)
+    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, ap_amount):
+        r = notion_patch(page_id, properties)
+        return mawb, r.status_code, r.text, ap_amount
+
+    with ThreadPoolExecutor(max_workers=MAX_CONCURRENT_WORKERS) as executor:
+        futures = {
+            executor.submit(_patch_task, mawb, page_id, props, ap_amt): mawb
+            for mawb, page_id, props, ap_amt in tasks
+        }
+        for future in as_completed(futures):
+            mawb, status, text, ap_amt = future.result()
+            with lock:
+                if status == 200:
+                    logger.log(f"MAWB {mawb} updated | AP checked, AP amount: {ap_amt:.2f}, AP Completed Date: {completed_time}")
+                    updated_count += 1
+                else:
+                    logger.log(f"MAWB {mawb} update failed: {text}")
 
     return updated_count, not_found_mawbs, skipped_ap_mawbs
 
@@ -798,14 +843,20 @@ def update_ap(file, isc_file, logger=None):
     else:
         logger.log("No JFK invoices found, ISC fee validation not required")
 
-    # 查询 Notion
-    logger.log("Querying Notion database...")
-    existing_pages = query_all_pages(extract_mawb_with_ap_cwt)
-    logger.log(f"Total {len(existing_pages)} MAWBs exist in Notion")
+    # -------- 服务端过滤：只查 AP=false 的记录 --------
+    logger.log("Querying Notion (filter: AP=false)...")
+    ap_false_filter = {"property": "AP", "checkbox": {"equals": False}}
+    existing_pages = query_all_pages(extract_mawb_with_ap_cwt, filter_body=ap_false_filter)
+    logger.log(f"Records with AP unchecked: {len(existing_pages)}")
+
+    # 轻量查询：全部 MAWB，用于区分「未找到」与「已勾选 AP」
+    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)}")
 
     # 更新 Notion
     updated_count, not_found_mawbs, skipped_ap_mawbs = update_ap_in_notion(
-        mawb_dict, extra_fee_records, existing_pages, isc_summary, jfk_mawbs, logger
+        mawb_dict, extra_fee_records, existing_pages, isc_summary, jfk_mawbs, all_mawbs, logger
     )
 
     logger.log("\n================ Process Completed ================")
@@ -886,11 +937,12 @@ def update_cwt(file, logger=None):
             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}}]}
-            log_parts.append(f"ATA={ata_str}")
+            pod = page_info.get("pod", "")
+            tz_name = POD_TIMEZONE_MAP.get(pod.upper(), "America/Los_Angeles")
+            ata_start = ata.strftime("%Y-%m-%dT%H:%M:%S")
+            properties["ATA"] = {"date": {"start": ata_start, "time_zone": tz_name}}
+            log_parts.append(f"ATA={ata_start} ({tz_name})")
 
         # 没有任何属性需要更新，跳过
         if not properties:
@@ -993,28 +1045,23 @@ def archive_old_records(cutoff_date_str=None, logger=None):
         logger.log(f"Progress: {idx}/{total} | Processing MAWB: {mawb}")
 
         # 解析 ETA，判断是否超过 3 个月
-        eta_rt = page["properties"].get("ETA", {}).get("rich_text", [])
-        if not eta_rt:
+        eta_date_val = page["properties"].get("ETA", {}).get("date")
+        if not eta_date_val or not eta_date_val.get("start"):
             logger.log("ETA is empty, skipped")
             skipped_eta.append(f"{mawb} - ETA empty")
             continue
 
-        eta_text = eta_rt[0]["text"]["content"]
-        eta_date = None
-        for fmt in ("%Y/%m/%d", "%Y-%m-%d"):
-            try:
-                eta_date = datetime.strptime(eta_text[:10], fmt)
-                break
-            except ValueError:
-                continue
-        if eta_date is None:
-            logger.log(f"ETA parse failed: '{eta_text}', skipped")
-            skipped_eta.append(f"{mawb} - ETA parse failed: {eta_text}")
+        eta_start = eta_date_val["start"]
+        try:
+            eta_date = datetime.strptime(eta_start[:10], "%Y-%m-%d")
+        except ValueError:
+            logger.log(f"ETA parse failed: '{eta_start}', skipped")
+            skipped_eta.append(f"{mawb} - ETA parse failed: {eta_start}")
             continue
 
         if eta_date > cutoff_date:
-            logger.log(f"ETA {eta_text} is after cutoff date, skipped")
-            skipped_eta.append(f"{mawb} - ETA {eta_text} after cutoff date")
+            logger.log(f"ETA {eta_start[:10]} is after cutoff date, skipped")
+            skipped_eta.append(f"{mawb} - ETA {eta_start[:10]} after cutoff date")
             continue
 
         # 检查 WH-OUT Date 是否为空
@@ -1036,7 +1083,7 @@ def archive_old_records(cutoff_date_str=None, logger=None):
 
         r_archive = notion_archive_page(page["id"])
         if r_archive.status_code == 200:
-            logger.log(f"Archived successfully (ETA: {eta_text})")
+            logger.log(f"Archived successfully (ETA: {eta_start[:10]})")
             archived_count += 1
         else:
             logger.log(f"Delete from main DB failed: {r_archive.text}")