first commit

inventory_sync/feishu.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import ssl
 import time
 import uuid
 from collections.abc import Sequence
@@ -12,6 +13,17 @@ from .specs import FIELD_SPECS, PRIMARY_FIELD_NAME
 
 
 FEISHU_API_BASE = "https://open.feishu.cn/open-apis"
+FEISHU_HTTP_RETRY_COUNT = 3
+FEISHU_HTTP_RETRY_DELAY = 1.0
+
+
+def create_feishu_tls12_context() -> ssl.SSLContext:
+    """Create a verified TLS 1.2 context for the Feishu network path."""
+
+    context = ssl.create_default_context()
+    context.minimum_version = ssl.TLSVersion.TLSv1_2
+    context.maximum_version = ssl.TLSVersion.TLSv1_2
+    return context
 
 
 class FeishuApiError(RuntimeError):
@@ -25,6 +37,7 @@ class FeishuFieldValidationError(FeishuApiError):
 class FeishuClient:
     def __init__(self, config: AppConfig) -> None:
         self.config = config
+        self._opener = request.build_opener(request.HTTPSHandler(context=create_feishu_tls12_context()))
 
     def request_json(
         self,
@@ -33,6 +46,7 @@ class FeishuClient:
         method: str = "GET",
         headers: dict[str, str] | None = None,
         payload: dict[str, Any] | None = None,
+        retry_transport: bool = False,
     ) -> dict[str, Any]:
         body = None
         request_headers = {"Content-Type": "application/json; charset=utf-8"}
@@ -41,21 +55,32 @@ class FeishuClient:
         if payload is not None:
             body = json.dumps(payload).encode("utf-8")
 
+        # Reads and token acquisition are idempotent. Do not automatically retry
+        # record-writing POST requests because a connection can fail after send.
+        attempts = FEISHU_HTTP_RETRY_COUNT if retry_transport or method.upper() == "GET" else 1
+        for attempt in range(1, attempts + 1):
             req = request.Request(url, data=body, headers=request_headers, method=method)
             try:
-            with request.urlopen(req, timeout=30) as response:
+                with self._opener.open(req, timeout=30) as response:
                     return json.loads(response.read().decode("utf-8"))
             except error.HTTPError as exc:
                 detail = exc.read().decode("utf-8", errors="replace")
                 raise RuntimeError(f"请求失败: {exc.code} {detail}") from exc
-        except error.URLError as exc:
+            except (error.URLError, ssl.SSLError) as exc:
-            raise RuntimeError(f"无法连接飞书接口: {exc.reason}") from exc
+                if attempt == attempts:
+                    raise RuntimeError(
+                        f"无法连接飞书接口（TLS 1.2，已尝试 {attempts} 次）: {exc}"
+                    ) from exc
+                time.sleep(FEISHU_HTTP_RETRY_DELAY * attempt)
+
+        raise AssertionError("unreachable")
 
     def get_tenant_access_token(self) -> str:
         data = self.request_json(
             f"{FEISHU_API_BASE}/auth/v3/tenant_access_token/internal",
             method="POST",
             payload={"app_id": self.config.feishu_app_id, "app_secret": self.config.feishu_app_secret},
+            retry_transport=True,
         )
         if data.get("code") != 0:
             raise FeishuApiError(f"获取 tenant_access_token 失败: {data}")

pyproject.toml

@@ -8,6 +8,7 @@ version = "0.1.0"
 description = "Internal inventory analytics workflows for batch upload, channel inventory sync, and batch inventory reporting."
 requires-python = ">=3.11"
 dependencies = [
+  "importlib-metadata>=8.0.0",
   "openpyxl>=3.1.2",
   "pandas",
   "prefect==3.6.8",