diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..90e8e61 --- /dev/null +++ b/.env.example @@ -0,0 +1,32 @@ +# PostgreSQL analytics database (required) +ANALYTICS_DB_HOST=postgres.example.internal +ANALYTICS_DB_PORT=5432 +ANALYTICS_DB_USER=your_database_user +ANALYTICS_DB_PASSWORD=your_database_password +ANALYTICS_DB_NAME=your_database_name + +# Feishu application and Bitable (required) +FEISHU_APP_ID=cli_your_app_id +FEISHU_APP_SECRET=your_feishu_app_secret +FEISHU_BASE_URL=https://example.feishu.cn/base/your_app_token + +# Sync settings +PROJECT_MAPPING_FILE=店铺映射表.xlsx +FEISHU_STORE_FIELD_NAME=店铺名称 +SYNC_START_DATE=2026-04-01 + +# Feishu notification (optional; configure one or both as needed) +FEISHU_BOT_WEBHOOK= +FEISHU_CHAT_ID= +FEISHU_REPORT_IMAGE_PATH= + +# Prefect (required only for Prefect deployment/worker mode) +PREFECT_API_URL=http://host.docker.internal:4200/api +PREFECT_SYNC_STATE_FILE= + +# Logging (optional) +LOG_DIR= +LOG_FILE_NAME=gmv_sync.log +LOG_LEVEL=INFO +LOG_MAX_BYTES=10485760 +LOG_BACKUP_COUNT=7 diff --git a/.gitignore b/.gitignore index c7edfd2..2d48adc 100644 --- a/.gitignore +++ b/.gitignore @@ -1,8 +1,10 @@ __pycache__/* *.pyc .vscode/ + +# Local credentials and machine-specific configuration .env created/ 创建编组.zip .DS_Store -logs/ \ No newline at end of file +logs/ diff --git a/prefect.yaml b/prefect.yaml index 30ee8fe..19e217c 100644 --- a/prefect.yaml +++ b/prefect.yaml @@ -7,7 +7,7 @@ pull: [] deployments: - name: gmv-daily-data-sync-8am - version: "1.0.0" + version: "1.0.1" description: "每天上午 8 点执行 GMV 数据同步并写入飞书多维表格。" entrypoint: prefect_gmv_scheduler.py:gmv_daily_data_sync_flow parameters: @@ -29,7 +29,7 @@ deployments: name: gmv-analytics-pool - name: gmv-daily-notify-9am - version: "1.0.0" + version: "1.0.1" description: "每天上午 9 点推送 GMV 数据同步结果到飞书群。" entrypoint: prefect_gmv_scheduler.py:gmv_daily_notify_flow parameters: diff --git a/project_bitable_common.py b/project_bitable_common.py index fd6d619..7160587 100644 --- a/project_bitable_common.py +++ b/project_bitable_common.py @@ -1,6 +1,7 @@ import json import os import re +import ssl from collections import defaultdict from dataclasses import dataclass, field from datetime import date, datetime, timedelta, timezone @@ -12,6 +13,8 @@ from urllib.parse import urlparse import psycopg2 import requests from openpyxl import load_workbook +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry from logging_config import get_logger, setup_logging @@ -23,6 +26,51 @@ DEFAULT_START_DATE = "2026-04-01" DEFAULT_STORE_FIELD_NAME = "店铺名称" STORE_FIELD_CANDIDATES = ["业务店铺名称", "店铺名称", "门店名称", "店铺", "名称"] SUPPORTED_PROJECTS = ["大眼睛", "羽心堂", "MMMJ", "EM", "春回盏", "BROKEN ZOO"] +FEISHU_HTTP_RETRY_COUNT = 3 + + +class TLS12HTTPAdapter(HTTPAdapter): + """Requests adapter that keeps the TLS policy consistent for Feishu calls.""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + self._ssl_context = ssl.create_default_context() + # TLS 1.3 handshakes are reset by the current server-side network path. + # Keep certificate verification enabled while using the compatible TLS 1.2. + self._ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2 + self._ssl_context.maximum_version = ssl.TLSVersion.TLSv1_2 + super().__init__(*args, **kwargs) + + def init_poolmanager(self, connections: int, maxsize: int, block: bool = False, **pool_kwargs: Any) -> None: + pool_kwargs["ssl_context"] = self._ssl_context + super().init_poolmanager(connections, maxsize, block=block, **pool_kwargs) + + def proxy_manager_for(self, proxy: str, **proxy_kwargs: Any) -> Any: + proxy_kwargs["ssl_context"] = self._ssl_context + return super().proxy_manager_for(proxy, **proxy_kwargs) + + +def create_feishu_session(*, retry_post: bool = False) -> requests.Session: + """Create a verified TLS 1.2 session for Feishu APIs with bounded retries. + + Only the token endpoint opts into POST retries because obtaining a token is + idempotent. Data-writing POST requests retain their original at-most-once + behaviour, while all calls still use TLS 1.2. + """ + + retry = Retry( + total=FEISHU_HTTP_RETRY_COUNT, + connect=FEISHU_HTTP_RETRY_COUNT, + read=FEISHU_HTTP_RETRY_COUNT, + status=0, + other=0, + backoff_factor=1, + allowed_methods=None if retry_post else frozenset({"GET", "HEAD", "OPTIONS"}), + raise_on_status=False, + ) + session = requests.Session() + adapter = TLS12HTTPAdapter(max_retries=retry) + session.mount("https://", adapter) + return session @dataclass(frozen=True) @@ -679,12 +727,13 @@ def get_feishu_token() -> str: app_secret = os.getenv("FEISHU_APP_SECRET") if not app_id or not app_secret: raise ValueError("缺少飞书环境变量 FEISHU_APP_ID / FEISHU_APP_SECRET") - log_step("开始获取飞书 tenant_access_token") - resp = requests.post( - "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal", - json={"app_id": app_id, "app_secret": app_secret}, - timeout=30, - ) + log_step(f"开始获取飞书 tenant_access_token（TLS 1.2，最多重试 {FEISHU_HTTP_RETRY_COUNT} 次）") + with create_feishu_session(retry_post=True) as session: + resp = session.post( + "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal", + json={"app_id": app_id, "app_secret": app_secret}, + timeout=30, + ) resp.raise_for_status() payload = resp.json() if payload.get("code") != 0: @@ -702,7 +751,7 @@ class FeishuBitableClient: self._fields_cache: dict[str, list[dict[str, Any]]] = {} self._records_cache: dict[str, list[dict[str, Any]]] = {} self._tables_cache: list[TableInfo] | None = None - self.session = requests.Session() + self.session = create_feishu_session() self.session.headers.update( { "Authorization": f"Bearer {self.token}", @@ -979,7 +1028,8 @@ def send_feishu_sync_card( }, } log_step("开始推送飞书群卡片消息") - resp = requests.post(webhook_url, json=card, timeout=30) + with create_feishu_session() as session: + resp = session.post(webhook_url, json=card, timeout=30) resp.raise_for_status() payload = resp.json() if payload.get("code") not in (0, None): @@ -995,11 +1045,18 @@ def upload_feishu_image(image_path: str, token: str | None = None) -> str: raise FileNotFoundError(f"未找到预览图文件: {path}") token = token or get_feishu_token() headers = {"Authorization": f"Bearer {token}"} - with path.open("rb") as image_file: - files = {"image": (path.name, image_file, "application/octet-stream")} - data = {"image_type": "message"} - log_step(f"开始上传飞书消息图片: {path}") - resp = requests.post("https://open.feishu.cn/open-apis/im/v1/images", headers=headers, data=data, files=files, timeout=60) + with create_feishu_session() as session: + with path.open("rb") as image_file: + files = {"image": (path.name, image_file, "application/octet-stream")} + data = {"image_type": "message"} + log_step(f"开始上传飞书消息图片: {path}") + resp = session.post( + "https://open.feishu.cn/open-apis/im/v1/images", + headers=headers, + data=data, + files=files, + timeout=60, + ) resp.raise_for_status() payload = resp.json() if payload.get("code") != 0: @@ -1071,12 +1128,13 @@ def send_feishu_app_card( "Content-Type": "application/json; charset=utf-8", } log_step(f"开始通过应用机器人发送群卡片: chat_id={chat_id}") - resp = requests.post( - "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id", - headers=headers, - json=message, - timeout=60, - ) + with create_feishu_session() as session: + resp = session.post( + "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id", + headers=headers, + json=message, + timeout=60, + ) try: payload = resp.json() except ValueError: diff --git a/tests/test_feishu_http.py b/tests/test_feishu_http.py new file mode 100644 index 0000000..5c6a11e --- /dev/null +++ b/tests/test_feishu_http.py @@ -0,0 +1,29 @@ +import ssl + +from project_bitable_common import FEISHU_HTTP_RETRY_COUNT, TLS12HTTPAdapter, create_feishu_session + + +def test_feishu_session_forces_tls_12() -> None: + session = create_feishu_session() + try: + adapter = session.get_adapter("https://open.feishu.cn") + assert isinstance(adapter, TLS12HTTPAdapter) + context = adapter._ssl_context + assert context.minimum_version == ssl.TLSVersion.TLSv1_2 + assert context.maximum_version == ssl.TLSVersion.TLSv1_2 + finally: + session.close() + + +def test_token_session_can_retry_post_but_regular_session_cannot() -> None: + regular_session = create_feishu_session() + token_session = create_feishu_session(retry_post=True) + try: + regular_retry = regular_session.get_adapter("https://open.feishu.cn").max_retries + token_retry = token_session.get_adapter("https://open.feishu.cn").max_retries + assert regular_retry.total == FEISHU_HTTP_RETRY_COUNT + assert regular_retry.allowed_methods == frozenset({"GET", "HEAD", "OPTIONS"}) + assert token_retry.allowed_methods is None + finally: + regular_session.close() + token_session.close()