fix: use API workdays and final delivery date in bot replies

shared/runtime/feishu_bot_bridge.py

@@ -12,7 +12,7 @@ import threading
 import time
 import unicodedata
 from dataclasses import dataclass
-from datetime import date, datetime, timedelta
+from datetime import date, datetime
 from pathlib import Path
 from typing import Any
 
@@ -82,40 +82,6 @@ CANDIDATE_CONTINUATION_TTL_SECONDS = 6 * 60 * 60
 RECEIVED_REACTION_TYPE = "OnIt"
 
 
-def _inclusive_date_range(start: date, end: date) -> frozenset[date]:
-    return frozenset(
-        start + timedelta(days=offset)
-        for offset in range((end - start).days + 1)
-    )
-
-
-# 国办发明电〔2025〕7号：国务院办公厅关于2026年部分节假日安排的通知。
-CHINA_WORKDAY_CALENDAR_SOURCE = (
-    "https://www.gov.cn/zhengce/zhengceku/202511/content_7047091.htm"
-)
-CHINA_OFFICIAL_HOLIDAYS = {
-    2026: frozenset().union(
-        _inclusive_date_range(date(2026, 1, 1), date(2026, 1, 3)),
-        _inclusive_date_range(date(2026, 2, 15), date(2026, 2, 23)),
-        _inclusive_date_range(date(2026, 4, 4), date(2026, 4, 6)),
-        _inclusive_date_range(date(2026, 5, 1), date(2026, 5, 5)),
-        _inclusive_date_range(date(2026, 6, 19), date(2026, 6, 21)),
-        _inclusive_date_range(date(2026, 9, 25), date(2026, 9, 27)),
-        _inclusive_date_range(date(2026, 10, 1), date(2026, 10, 7)),
-    )
-}
-CHINA_OFFICIAL_MAKEUP_WORKDAYS = {
-    2026: frozenset(
-        {
-            date(2026, 1, 4),
-            date(2026, 2, 14),
-            date(2026, 2, 28),
-            date(2026, 5, 9),
-            date(2026, 9, 20),
-            date(2026, 10, 10),
-        }
-    )
-}
 CANDIDATE_CONTINUATION_SHORT_PHRASES = {
     "还有吗",
     "还有呢",
@@ -209,12 +175,6 @@ class CandidateUnavailableError(RuntimeError):
     pass
 
 
-class WorkdayCalendarUnavailableError(ValueError):
-    def __init__(self, year: int) -> None:
-        self.year = year
-        super().__init__(f"{year}年国家法定工作日历未配置")
-
-
 _candidate_continuations: dict[str, CandidateContinuation] = {}
 _candidate_continuation_lock = threading.Lock()
 
@@ -536,52 +496,7 @@ def run_controlled_query(request: RequestInput) -> dict[str, Any]:
     return payload
 
 
-def natural_days_to_workdays(natural_days: int) -> int:
-    """Convert natural days with AR51's fixed quotient-and-remainder rule."""
-    if isinstance(natural_days, bool) or not isinstance(natural_days, int):
-        raise TypeError("自然日交期必须是整数")
-    if natural_days < 0:
-        raise ValueError("自然日交期不能小于0")
-
-    full_weeks, remainder = divmod(natural_days, 7)
-    if remainder == 0:
-        remainder_workdays = 0
-    elif remainder <= 4:
-        remainder_workdays = max(remainder - 1, 0)
-    else:
-        remainder_workdays = remainder - 2
-    return full_weeks * 5 + remainder_workdays
-
-
-def is_china_official_workday(day: date) -> bool:
-    holidays = CHINA_OFFICIAL_HOLIDAYS.get(day.year)
-    makeup_workdays = CHINA_OFFICIAL_MAKEUP_WORKDAYS.get(day.year)
-    if holidays is None or makeup_workdays is None:
-        raise WorkdayCalendarUnavailableError(day.year)
-    if day in makeup_workdays:
-        return True
-    if day in holidays:
-        return False
-    return day.weekday() < 5
-
-
-def add_china_official_workdays(start: date, workdays: int) -> date:
-    """Count official workdays after start; the start date itself is excluded."""
-    if isinstance(workdays, bool) or not isinstance(workdays, int):
-        raise TypeError("工作日交期必须是整数")
-    if workdays < 0:
-        raise ValueError("工作日交期不能小于0")
-
-    current = start
-    counted = 0
-    while counted < workdays:
-        current += timedelta(days=1)
-        if is_china_official_workday(current):
-            counted += 1
-    return current
-
-
-def whole_natural_day_count(value: Any) -> int | None:
+def whole_day_count(value: Any) -> int | None:
     if isinstance(value, bool) or not isinstance(value, (int, float)):
         return None
     if isinstance(value, float) and not value.is_integer():
@@ -593,9 +508,7 @@ def whole_natural_day_count(value: Any) -> int | None:
 def lead_time_details(
     payload: dict[str, Any],
     requested_quantity: int,
-    calculation_date: date | None = None,
 ) -> dict[str, Any]:
-    calculation_date = calculation_date or datetime.now().astimezone().date()
     data = payload.get("data")
     if not isinstance(data, dict):
         return {
@@ -635,31 +548,23 @@ def lead_time_details(
         if isinstance(main_factor, (str, int, float))
         else "接口未返回，需人工复核"
     )
-    predicted_date = summary.get("AM516预测交期")
-    total_days = whole_natural_day_count(summary.get("AM516总交期"))
+    total_days = whole_day_count(summary.get("AM516总交期"))
     if total_days is None:
-        total_days = whole_natural_day_count(summary.get("总交期"))
-    workdays: int | None = None
-    delivery_date: date | None = None
-    if total_days is not None:
-        workdays = natural_days_to_workdays(total_days)
+        total_days = whole_day_count(summary.get("总交期"))
+    workdays = whole_day_count(summary.get("AM516工作日交期"))
+    delivery_date = summary.get("AM516最终交货日期")
+    if isinstance(delivery_date, str) and re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", delivery_date):
         try:
-            delivery_date = add_china_official_workdays(calculation_date, workdays)
-        except WorkdayCalendarUnavailableError as exc:
-            prediction_value = (
-                f"【{exc}】（原始{total_days}个自然日，折算{workdays}个工作日）"
-            )
-        else:
-            prediction_value = (
-                f"{delivery_date.isoformat()}"
-                f"（原始{total_days}个自然日，折算{workdays}个工作日）"
-            )
-        prediction = f"AM516预测交期：{prediction_value}"
-    elif isinstance(predicted_date, (str, int, float)):
-        prediction_value = f"{predicted_date}（接口未返回自然日交期，未换算）"
+            date.fromisoformat(delivery_date)
+        except ValueError:
+            delivery_date = None
+    else:
+        delivery_date = None
+    if workdays is not None and delivery_date is not None:
+        prediction_value = f"{workdays}个工作日，最终交货日期：{delivery_date}"
         prediction = f"AM516预测交期：{prediction_value}"
     else:
-        prediction = "接口未返回可用交期汇总，需人工复核。"
+        prediction = "接口未返回有效工作日交期和最终交货日期，需人工复核。"
         prediction_value = None
 
     inventory_info = data.get("库存信息")
@@ -738,7 +643,7 @@ def lead_time_details(
         "prediction_value": prediction_value,
         "total_days": total_days,
         "workdays": workdays,
-        "delivery_date": delivery_date.isoformat() if delivery_date else None,
+        "delivery_date": delivery_date,
         "main_factor": main_factor_text,
         "inventory": inventory_text,
         "available_inventory": (
@@ -936,7 +841,6 @@ def build_report(request: RequestInput, candidate_offset: int = 0) -> str:
     if candidate_offset < 0:
         raise InputError("候选批次位置无效，请重新发起型号和数量查询。")
 
-    calculation_date = datetime.now().astimezone().date()
     all_candidate_specs = generate_candidate_specs(request.model)
     candidate_specs = all_candidate_specs[
         candidate_offset : candidate_offset + CANDIDATE_BATCH_SIZE
@@ -952,10 +856,9 @@ def build_report(request: RequestInput, candidate_offset: int = 0) -> str:
         baseline = lead_time_details(
             baseline_payload,
             request.quantity,
-            calculation_date=calculation_date,
         )
         if not baseline["has_lead_time"]:
-            raise RuntimeError("原型号接口未返回可用交期汇总，未生成候选结果。")
+            raise RuntimeError("原型号接口未返回有效工作日交期和最终交货日期，未生成候选结果。")
         candidates.append(
             {
                 "index": 1,
@@ -979,7 +882,6 @@ def build_report(request: RequestInput, candidate_offset: int = 0) -> str:
         details = lead_time_details(
             payload,
             request.quantity,
-            calculation_date=calculation_date,
         )
         return spec, details if details["has_lead_time"] else None