Add environment configuration, enhance Feishu session handling, and implement tests for TLS 1.2 and retry logic

This commit is contained in:
BXY
2026-09-28 11:20:14 +08:00
parent a71dd66504
commit 152cb5a65b
5 changed files with 143 additions and 22 deletions
+32
View File
@@ -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
+2
View File
@@ -1,6 +1,8 @@
__pycache__/* __pycache__/*
*.pyc *.pyc
.vscode/ .vscode/
# Local credentials and machine-specific configuration
.env .env
created/ created/
创建编组.zip 创建编组.zip
+2 -2
View File
@@ -7,7 +7,7 @@ pull: []
deployments: deployments:
- name: gmv-daily-data-sync-8am - name: gmv-daily-data-sync-8am
version: "1.0.0" version: "1.0.1"
description: "每天上午 8 点执行 GMV 数据同步并写入飞书多维表格。" description: "每天上午 8 点执行 GMV 数据同步并写入飞书多维表格。"
entrypoint: prefect_gmv_scheduler.py:gmv_daily_data_sync_flow entrypoint: prefect_gmv_scheduler.py:gmv_daily_data_sync_flow
parameters: parameters:
@@ -29,7 +29,7 @@ deployments:
name: gmv-analytics-pool name: gmv-analytics-pool
- name: gmv-daily-notify-9am - name: gmv-daily-notify-9am
version: "1.0.0" version: "1.0.1"
description: "每天上午 9 点推送 GMV 数据同步结果到飞书群。" description: "每天上午 9 点推送 GMV 数据同步结果到飞书群。"
entrypoint: prefect_gmv_scheduler.py:gmv_daily_notify_flow entrypoint: prefect_gmv_scheduler.py:gmv_daily_notify_flow
parameters: parameters:
+77 -19
View File
@@ -1,6 +1,7 @@
import json import json
import os import os
import re import re
import ssl
from collections import defaultdict from collections import defaultdict
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone from datetime import date, datetime, timedelta, timezone
@@ -12,6 +13,8 @@ from urllib.parse import urlparse
import psycopg2 import psycopg2
import requests import requests
from openpyxl import load_workbook from openpyxl import load_workbook
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from logging_config import get_logger, setup_logging from logging_config import get_logger, setup_logging
@@ -23,6 +26,51 @@ DEFAULT_START_DATE = "2026-04-01"
DEFAULT_STORE_FIELD_NAME = "店铺名称" DEFAULT_STORE_FIELD_NAME = "店铺名称"
STORE_FIELD_CANDIDATES = ["业务店铺名称", "店铺名称", "门店名称", "店铺", "名称"] STORE_FIELD_CANDIDATES = ["业务店铺名称", "店铺名称", "门店名称", "店铺", "名称"]
SUPPORTED_PROJECTS = ["大眼睛", "羽心堂", "MMMJ", "EM", "春回盏", "BROKEN ZOO"] 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) @dataclass(frozen=True)
@@ -679,12 +727,13 @@ def get_feishu_token() -> str:
app_secret = os.getenv("FEISHU_APP_SECRET") app_secret = os.getenv("FEISHU_APP_SECRET")
if not app_id or not app_secret: if not app_id or not app_secret:
raise ValueError("缺少飞书环境变量 FEISHU_APP_ID / FEISHU_APP_SECRET") raise ValueError("缺少飞书环境变量 FEISHU_APP_ID / FEISHU_APP_SECRET")
log_step("开始获取飞书 tenant_access_token") log_step(f"开始获取飞书 tenant_access_token(TLS 1.2,最多重试 {FEISHU_HTTP_RETRY_COUNT} 次)")
resp = requests.post( with create_feishu_session(retry_post=True) as session:
"https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal", resp = session.post(
json={"app_id": app_id, "app_secret": app_secret}, "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
timeout=30, json={"app_id": app_id, "app_secret": app_secret},
) timeout=30,
)
resp.raise_for_status() resp.raise_for_status()
payload = resp.json() payload = resp.json()
if payload.get("code") != 0: if payload.get("code") != 0:
@@ -702,7 +751,7 @@ class FeishuBitableClient:
self._fields_cache: dict[str, list[dict[str, Any]]] = {} self._fields_cache: dict[str, list[dict[str, Any]]] = {}
self._records_cache: dict[str, list[dict[str, Any]]] = {} self._records_cache: dict[str, list[dict[str, Any]]] = {}
self._tables_cache: list[TableInfo] | None = None self._tables_cache: list[TableInfo] | None = None
self.session = requests.Session() self.session = create_feishu_session()
self.session.headers.update( self.session.headers.update(
{ {
"Authorization": f"Bearer {self.token}", "Authorization": f"Bearer {self.token}",
@@ -979,7 +1028,8 @@ def send_feishu_sync_card(
}, },
} }
log_step("开始推送飞书群卡片消息") 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() resp.raise_for_status()
payload = resp.json() payload = resp.json()
if payload.get("code") not in (0, None): 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}") raise FileNotFoundError(f"未找到预览图文件: {path}")
token = token or get_feishu_token() token = token or get_feishu_token()
headers = {"Authorization": f"Bearer {token}"} headers = {"Authorization": f"Bearer {token}"}
with path.open("rb") as image_file: with create_feishu_session() as session:
files = {"image": (path.name, image_file, "application/octet-stream")} with path.open("rb") as image_file:
data = {"image_type": "message"} files = {"image": (path.name, image_file, "application/octet-stream")}
log_step(f"开始上传飞书消息图片: {path}") data = {"image_type": "message"}
resp = requests.post("https://open.feishu.cn/open-apis/im/v1/images", headers=headers, data=data, files=files, timeout=60) 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() resp.raise_for_status()
payload = resp.json() payload = resp.json()
if payload.get("code") != 0: if payload.get("code") != 0:
@@ -1071,12 +1128,13 @@ def send_feishu_app_card(
"Content-Type": "application/json; charset=utf-8", "Content-Type": "application/json; charset=utf-8",
} }
log_step(f"开始通过应用机器人发送群卡片: chat_id={chat_id}") log_step(f"开始通过应用机器人发送群卡片: chat_id={chat_id}")
resp = requests.post( with create_feishu_session() as session:
"https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id", resp = session.post(
headers=headers, "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id",
json=message, headers=headers,
timeout=60, json=message,
) timeout=60,
)
try: try:
payload = resp.json() payload = resp.json()
except ValueError: except ValueError:
+29
View File
@@ -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()