first commit

- 按字段名校验并写入渠道库存，取消字段顺序校验及表头重建
- 严格校验必填字段名称、类型和数字格式，失败时不修改表头或记录
- 字段校验失败时发送飞书群通知
- 添加字段校验与通知流程测试
- 忽略 Python 字节码、__pycache__ 和 .DS_Store

.gitignore

@@ -1,7 +1,13 @@
 .env
 .feishu_oauth_tokens.json
-inventory_sync/__pycache__
+
+# Python bytecode (at any directory depth)
+__pycache__/
+*.pyc
+
+# macOS Finder metadata (at any directory depth)
 .DS_Store
+
 batch_inventory_data/*
 logs
 reports

.vscode/settings.json

@@ -0,0 +1,4 @@
+{
+    "python-envs.defaultEnvManager": "ms-python.python:conda",
+    "python-envs.defaultPackageManager": "ms-python.python:conda"
+}

PROJECT_GUIDE.md

@@ -10,7 +10,7 @@
 
 命名说明：
 
-- “渠道库存统计”是框架层命名，当前默认业务目标仍是羽心堂渠道库存-9：00和19：00。
+- “渠道库存统计”是框架层命名，当前默认业务目标仍是羽心堂渠道库存-9：30和20：00。
 - “批次库存统计”是框架层命名，当前默认业务目标仍是羽心堂批次库存。
 - 后续扩展到其他品牌时，优先复用这两条通用链路，而不是继续新增带品牌名的脚本和包名。
 
@@ -278,7 +278,7 @@ all
 说明：
 
 - 当前脚本名已经泛化为“渠道库存统计”，但内部动作参数 `sync-brand` 仍来自历史实现，后续可再统一。
-- 当前默认同步目标仍是羽心堂渠道库存-9：00和19：00。
+- 当前默认同步目标仍是羽心堂渠道库存-9：30和20：00。
 
 渠道库存统计目标和字段映射当前仍维护在：
 
@@ -379,7 +379,7 @@ prefect_inventory_scheduler.py
 
 ### 10.1 调度目标
 
-当前通过 Prefect 定义了 2 个 deployment，都会在每天 `09:15` 和 `19:15`（`Asia/Shanghai`）执行：
+当前通过 Prefect 定义了 2 个 deployment，每个 deployment 用一条 cron 覆盖每天 `09:15` 和 `19:15`（`Asia/Shanghai`）：
 
 - 渠道库存统计并写入飞书
 - 批次库存统计报表并写入飞书
@@ -417,6 +417,14 @@ docker compose exec -T prefect-inventory-worker prefect --no-prompt deploy --nam
 docker compose up -d --build
 ```
 
+如果调整了 Python 依赖或 Prefect 版本，需要重新构建镜像后再注册 deployment：
+
+```bash
+docker compose build --no-cache prefect-inventory-worker
+docker compose up -d prefect-inventory-worker
+sh scripts/register_prefect_deployments.sh
+```
+
 这会启动一个容器，容器内通过 `supervisord` 常驻运行：
 
 ```text

README.md

@@ -166,10 +166,11 @@ PREFECT_API_URL
 - `.env` 不应提交到版本库。
 - 日志默认不会输出数据库连接信息、飞书 app token、飞书 base URL、表 ID、密码或密钥。
 - `FEISHU_APP_TOKEN` 可选；当 `FEISHU_BASE_URL` 已包含 `/base/` 或 `/wiki/` 时，可以不单独配置。
+- 渠道库存同步允许飞书字段任意排序，但会严格校验必填字段名称、类型和数字格式。校验失败时不会修改表头或记录，并会向指定的飞书通知群发送错误消息。应用需开启机器人能力、具备发送群消息权限，并已加入该通知群。
 
 ## Prefect 调度
 
-项目定义了两个 Prefect deployment：
+项目定义了两个 Prefect deployment，每个 deployment 用一条 cron 覆盖 09:15 和 19:15：
 
 - `channel-inventory-sync-0915-1915`：每天 09:15 和 19:15 执行渠道库存统计并写入飞书。
 - `batch-inventory-report-0915-1915`：每天 09:15 和 19:15 执行批次库存统计并写入飞书。
@@ -193,6 +194,14 @@ docker compose exec -T prefect-inventory-worker prefect --no-prompt deploy --nam
 docker compose up -d --build
 ```
 
+如果调整了 Python 依赖或 Prefect 版本，需要重新构建镜像后再注册 deployment：
+
+```bash
+docker compose build --no-cache prefect-inventory-worker
+docker compose up -d prefect-inventory-worker
+sh scripts/register_prefect_deployments.sh
+```
+
 手动触发：
 
 ```bash

REFACTOR_PLAN.md

@@ -247,7 +247,7 @@ ANALYTICS_DB_NAME
 
 ```python
 BRAND_SYNC_TARGETS = [
-    BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂渠道库存-9：00和19：00", warehouse_keys=(2, 3, 7, 13)),
+    BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂渠道库存-9：30和20：00", warehouse_keys=(2, 3, 7, 13)),
 ]
 ```
 

inventory_analytics/batch_inventory_report/service.py

@@ -18,6 +18,7 @@ import psycopg2
 from ..config import DbConfig, build_app_config, load_analytics_db_config, load_raw_db_config
 from ..feishu.client import FeishuApiError, FeishuClient
 from ..logging_utils import configure_logging as configure_app_logging
+from ..logging_utils import configure_logging_with_extra_handler
 from ..logging_utils import exception_fields
 from ..logging_utils import log_event as emit_log_event
 from ..paths import LOG_DIR, REPORT_DIR
@@ -212,6 +213,16 @@ LOG_FIELD_LABELS = {
 }
 
 def configure_logging() -> None:
+    if os.getenv("INVENTORY_PREFECT_LOG_BRIDGE") == "1":
+        handler = PrefectLogHandler()
+        handler.setFormatter(
+            logging.Formatter(
+                fmt="%(asctime)s %(levelname)s %(name)s %(message)s",
+                datefmt="%Y-%m-%d %H:%M:%S",
+            )
+        )
+        configure_logging_with_extra_handler(LOG_FILE, handler)
+    else:
         configure_app_logging(LOG_FILE)
 
 
@@ -219,6 +230,23 @@ def log_event(level: int, event: str, **fields: Any) -> None:
     emit_log_event(logger, LOG_FIELD_LABELS, level, event, **fields)
 
 
+class PrefectLogHandler(logging.Handler):
+    def emit(self, record: logging.LogRecord) -> None:
+        if record.name.startswith("prefect"):
+            return
+        try:
+            from prefect.logging import get_run_logger
+
+            prefect_logger = get_run_logger()
+        except Exception:
+            return
+        message = record.getMessage()
+        log_method = getattr(prefect_logger, record.levelname.lower(), None)
+        if log_method is None:
+            log_method = prefect_logger.info
+        log_method(message)
+
+
 @dataclass(frozen=True)
 class FeishuFieldSpec:
     field_name: str      # 飞书字段名

inventory_analytics/channel_inventory_sync/entrypoint.py

@@ -8,6 +8,7 @@ import time
 
 from .cli import main as channel_inventory_sync_main
 from ..logging_utils import configure_logging, log_event
+from ..logging_utils import configure_logging_with_extra_handler
 from ..logging_utils import exception_fields
 from ..paths import LOG_DIR
 
@@ -28,7 +29,34 @@ LOG_FIELD_LABELS = {
 }
 
 
+class PrefectLogHandler(logging.Handler):
+    def emit(self, record: logging.LogRecord) -> None:
+        if record.name.startswith("prefect"):
+            return
+        try:
+            from prefect.logging import get_run_logger
+
+            prefect_logger = get_run_logger()
+        except Exception:
+            return
+        message = record.getMessage()
+        log_method = getattr(prefect_logger, record.levelname.lower(), None)
+        if log_method is None:
+            log_method = prefect_logger.info
+        log_method(message)
+
+
 def main() -> int:
+    if os.getenv("INVENTORY_PREFECT_LOG_BRIDGE") == "1":
+        handler = PrefectLogHandler()
+        handler.setFormatter(
+            logging.Formatter(
+                fmt="%(asctime)s %(levelname)s %(name)s %(message)s",
+                datefmt="%Y-%m-%d %H:%M:%S",
+            )
+        )
+        configure_logging_with_extra_handler(LOG_FILE, handler, stream=sys.stderr)
+    else:
         configure_logging(LOG_FILE, stream=sys.stderr)
     started_at = time.monotonic()
     start_fields = {"argv": sys.argv[1:], "stage": "start"}

inventory_analytics/logging_utils.py

@@ -84,6 +84,18 @@ def configure_logging(log_file: Path, *, stream = None, level: int = logging.INF
     root_logger.addHandler(file_handler)
 
 
+def configure_logging_with_extra_handler(
+    log_file: Path,
+    extra_handler: logging.Handler,
+    *,
+    stream = None,
+    level: int = logging.INFO,
+) -> None:
+    configure_logging(log_file, stream=stream, level=level)
+    extra_handler.setLevel(level)
+    logging.getLogger().addHandler(extra_handler)
+
+
 def log_event(logger: logging.Logger, labels: dict[str, str], level: int, event: str, **fields: Any) -> None:
     safe_event = sanitize_value(event)
     details = " ".join(

inventory_sync/feishu.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import json
-import re
+import ssl
 import time
 import uuid
 from collections.abc import Sequence
@@ -9,19 +9,35 @@ from typing import Any
 from urllib import error, parse, request
 
 from .settings import AppConfig
-from .specs import FIELD_SPECS, LEGACY_NAME_MAP, PRIMARY_FIELD_NAME
+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):
     """Raised when the Feishu API returns a non-zero code."""
 
 
+class FeishuFieldValidationError(FeishuApiError):
+    """Raised when a channel-inventory table does not match its required schema."""
+
+
 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,
@@ -30,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"}
@@ -38,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}")
@@ -133,6 +161,21 @@ class FeishuClient:
     def list_records(self, access_token: str, app_token: str, table_id: str) -> list[dict[str, Any]]:
         return self._paginate(access_token, f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables/{table_id}/records", 500)
 
+    def send_text_message(self, access_token: str, chat_id: str, text: str) -> None:
+        """Send a plain-text notification to a Feishu group chat."""
+        data = self.request_json(
+            f"{FEISHU_API_BASE}/im/v1/messages?receive_id_type=chat_id",
+            method="POST",
+            headers={"Authorization": f"Bearer {access_token}"},
+            payload={
+                "receive_id": chat_id,
+                "msg_type": "text",
+                "content": json.dumps({"text": text}, ensure_ascii=False),
+            },
+        )
+        if data.get("code") != 0:
+            raise FeishuApiError(f"发送飞书群消息失败: {data}")
+
     def delete_field(self, access_token: str, app_token: str, table_id: str, field_id: str) -> None:
         data = self.request_json(
             f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables/{table_id}/fields/{field_id}",
@@ -162,14 +205,6 @@ class FeishuClient:
         if data.get("code") != 0:
             raise FeishuApiError(f"新增字段失败({payload.get('field_name')}): {data}")
 
-    def _is_managed_field(self, field_name: str) -> bool:
-        managed_names = {spec.feishu_name for spec in FIELD_SPECS}
-        managed_names.update(LEGACY_NAME_MAP.keys())
-        managed_names.update(LEGACY_NAME_MAP.values())
-        if field_name in managed_names:
-            return True
-        return bool(re.fullmatch(r"\d{1,2}月销售数量", field_name) or re.fullmatch(r"\d{6}销售数量", field_name))
-
     def batch_delete_records(self, access_token: str, app_token: str, table_id: str, record_ids: list[str]) -> int:
         deleted = 0
         for start in range(0, len(record_ids), 500):
@@ -212,71 +247,40 @@ class FeishuClient:
         fields = self.list_fields(access_token, app_token, table_id)
         primary_field = next((field for field in fields if field.get("is_primary")), None)
         if not primary_field:
-            raise FeishuApiError("未找到主字段")
+            raise FeishuFieldValidationError("未找到渠道库存表主字段")
-        expected_field_names = {spec.feishu_name for spec in field_specs}
-        expected_field_order = [spec.feishu_name for spec in field_specs]
-        actual_managed_field_names = {
-            field.get("field_name")
-            for field in fields
-            if not field.get("is_primary") and field.get("field_name") in expected_field_names
-        }
-        actual_expected_field_order = [
-            field.get("field_name")
-            for field in fields
-            if not field.get("is_primary") and field.get("field_name") in expected_field_names
-        ]
-        obsolete_managed_field_names = {
-            field.get("field_name")
-            for field in fields
-            if (
-                not field.get("is_primary")
-                and field.get("field_name") not in expected_field_names
-                and self._is_managed_field(field.get("field_name") or "")
-            )
-        }
-        primary_matches = primary_field.get("field_name") == PRIMARY_FIELD_NAME
-        fields_match = (
-            actual_managed_field_names == expected_field_names
-            and actual_expected_field_order == expected_field_order
-            and not obsolete_managed_field_names
-        )
-        if primary_matches and fields_match:
-            return [field.get("field_name") for field in fields if field.get("field_name")]
-
-        duplicate_primary = next(
-            (field for field in fields if not field.get("is_primary") and field.get("field_name") == PRIMARY_FIELD_NAME),
-            None,
-        )
-        if duplicate_primary:
-            self.delete_field(access_token, app_token, table_id, duplicate_primary["field_id"])
-            time.sleep(0.2)
-            fields = self.list_fields(access_token, app_token, table_id)
-            primary_field = next((field for field in fields if field.get("is_primary")), None)
-            if not primary_field:
-                raise FeishuApiError("删除重复字段后未找到主字段")
 
+        validation_errors: list[str] = []
         if primary_field.get("field_name") != PRIMARY_FIELD_NAME:
-            payload = {"field_name": PRIMARY_FIELD_NAME, "type": primary_field.get("type")}
+            validation_errors.append(
-            if primary_field.get("property") is not None:
+                f"主字段名称错误：期望“{PRIMARY_FIELD_NAME}”，实际“{primary_field.get('field_name') or '未命名'}”"
-                payload["property"] = primary_field.get("property")
+            )
-            self.update_field(access_token, app_token, table_id, primary_field["field_id"], payload)
+        if primary_field.get("type") != 1:
-            time.sleep(0.2)
+            validation_errors.append(
-
+                f"主字段类型错误：字段“{PRIMARY_FIELD_NAME}”应为文本，实际类型 {primary_field.get('type')}"
-        fields = self.list_fields(access_token, app_token, table_id)
+            )
-        for field in fields:
-            field_name = field.get("field_name")
-            field_id = field.get("field_id")
-            if field.get("is_primary") or not field_name or not field_id:
-                continue
-            if field_name in expected_field_names or self._is_managed_field(field_name):
-                self.delete_field(access_token, app_token, table_id, field_id)
-                time.sleep(0.2)
 
+        fields_by_name = {
+            field.get("field_name"): field
+            for field in fields
+            if not field.get("is_primary") and field.get("field_name")
+        }
         for spec in field_specs:
-            payload = {"field_name": spec.feishu_name, "type": spec.field_type}
+            field = fields_by_name.get(spec.feishu_name)
+            if not field:
+                validation_errors.append(f"缺少必需字段或字段已改名：“{spec.feishu_name}”")
+                continue
+            if field.get("type") != spec.field_type:
+                validation_errors.append(
+                    f"字段类型错误：“{spec.feishu_name}”应为 {spec.field_type}，实际为 {field.get('type')}"
+                )
             if spec.formatter is not None:
-                payload["property"] = {"formatter": spec.formatter}
+                actual_formatter = (field.get("property") or {}).get("formatter")
-            self.create_field(access_token, app_token, table_id, payload)
+                if actual_formatter != spec.formatter:
-            time.sleep(0.2)
+                    validation_errors.append(
+                        f"字段格式错误：“{spec.feishu_name}”应为 {spec.formatter}，实际为 {actual_formatter or '未设置'}"
+                    )
 
-        return [field.get("field_name") for field in self.list_fields(access_token, app_token, table_id) if field.get("field_name")]
+        if validation_errors:
+            raise FeishuFieldValidationError("；".join(validation_errors))
+
+        return [field.get("field_name") for field in fields if field.get("field_name")]

inventory_sync/postgres.py

@@ -2,21 +2,41 @@ from __future__ import annotations
 
 import csv
 import io
-import os
+import logging
-import subprocess
+import time
 from typing import Any
 
+import psycopg2
+
+from inventory_analytics.logging_utils import log_event as emit_log_event
+
 from .settings import AppConfig
 from .specs import BrandSyncTarget
 from .sql import DEFAULT_SYNC_TARGET, POSTGRES_QUERY, build_postgres_query
 
 
+logger = logging.getLogger("inventory_sync.postgres")
+
+LOG_FIELD_LABELS = {
+    "elapsed_ms": "耗时毫秒",
+    "row_count": "行数",
+    "stage": "阶段",
+    "statement_timeout_ms": "查询超时毫秒",
+}
+
+
+def log_event(level: int, event: str, **fields: Any) -> None:
+    emit_log_event(logger, LOG_FIELD_LABELS, level, event, **fields)
+
+
 def query_postgres(
     config: AppConfig,
     brand_name: str | None = None,
     warehouse_keys: tuple[int | str, ...] | None = None,
     target: BrandSyncTarget | None = None,
 ) -> dict[str, Any]:
+    started_at = time.monotonic()
+    statement_timeout_ms = 300_000
     active_brand_name = brand_name or (target.brand_name if target else DEFAULT_SYNC_TARGET.brand_name)
     active_warehouse_keys = warehouse_keys or (target.warehouse_keys if target else DEFAULT_SYNC_TARGET.warehouse_keys)
     query = (
@@ -27,40 +47,51 @@ def query_postgres(
             warehouse_keys=active_warehouse_keys,
         )
     )
-    env = os.environ.copy()
+    output = io.StringIO()
-    env["PGPASSWORD"] = config.analytics_db_password
-    command = [
-        "/usr/local/bin/psql",
-        "-h",
-        config.analytics_db_host,
-        "-p",
-        config.analytics_db_port,
-        "-U",
-        config.analytics_db_user,
-        "-d",
-        config.analytics_db_name,
-        "-v",
-        "ON_ERROR_STOP=1",
-        "-c",
-        "SET TIME ZONE 'Asia/Shanghai';",
-        "-c",
-        f"COPY ({query}) TO STDOUT WITH CSV HEADER",
-    ]
-
     try:
-        completed = subprocess.run(
+        log_event(logging.INFO, "开始连接PostgreSQL", stage="database_connect")
-            command,
+        with psycopg2.connect(
-            check=True,
+            host=config.analytics_db_host,
-            capture_output=True,
+            port=config.analytics_db_port,
-            text=True,
+            user=config.analytics_db_user,
-            env=env,
+            password=config.analytics_db_password,
+            dbname=config.analytics_db_name,
+            connect_timeout=10,
+        ) as conn:
+            log_event(
+                logging.INFO,
+                "PostgreSQL连接完成",
+                stage="database_connect",
+                elapsed_ms=int((time.monotonic() - started_at) * 1000),
             )
-    except subprocess.CalledProcessError as exc:
+            with conn.cursor() as cur:
-        detail = exc.stderr.strip() or exc.stdout.strip()
+                cur.execute("SET TIME ZONE 'Asia/Shanghai';")
-        raise RuntimeError(f"PostgreSQL 查询失败: {detail}") from exc
+                cur.execute("SET statement_timeout = %s;", (statement_timeout_ms,))
+                copy_started_at = time.monotonic()
+                log_event(
+                    logging.INFO,
+                    "开始执行渠道库存SQL",
+                    stage="database_query",
+                    statement_timeout_ms=statement_timeout_ms,
+                )
+                cur.copy_expert(f"COPY ({query}) TO STDOUT WITH CSV HEADER", output)
+                log_event(
+                    logging.INFO,
+                    "渠道库存SQL执行完成",
+                    stage="database_query",
+                    elapsed_ms=int((time.monotonic() - copy_started_at) * 1000),
+                )
+    except psycopg2.Error as exc:
+        raise RuntimeError(f"PostgreSQL 查询失败: {exc}") from exc
 
-    cleaned_output = "\n".join(
+    parse_started_at = time.monotonic()
-        line for line in completed.stdout.splitlines() if line.strip() and line.strip() != "SET"
+    output.seek(0)
+    rows = list(csv.DictReader(output))
+    log_event(
+        logging.INFO,
+        "渠道库存查询结果解析完成",
+        stage="database_parse",
+        row_count=len(rows),
+        elapsed_ms=int((time.monotonic() - parse_started_at) * 1000),
     )
-    rows = list(csv.DictReader(io.StringIO(cleaned_output)))
     return {"row_count": len(rows), "rows": rows}

inventory_sync/service.py

@@ -7,7 +7,7 @@ from typing import Any
 
 from inventory_analytics.logging_utils import log_event as emit_log_event
 
-from .feishu import FeishuApiError, FeishuClient
+from .feishu import FeishuApiError, FeishuClient, FeishuFieldValidationError
 from .postgres import query_postgres
 from .settings import AppConfig
 from .specs import BRAND_SYNC_TARGETS, FIELD_SPECS, PRIMARY_FIELD_NAME, PRIMARY_FIELD_SOURCE, BrandSyncTarget, FieldSpec
@@ -15,6 +15,8 @@ from .specs import BRAND_SYNC_TARGETS, FIELD_SPECS, PRIMARY_FIELD_NAME, PRIMARY_
 
 logger = logging.getLogger("inventory_sync.service")
 
+CHANNEL_INVENTORY_SCHEMA_ALERT_CHAT_ID = "oc_8a4b6891bade81d116914aa55054a49d"
+
 LOG_FIELD_LABELS = {
     "advanced_permission_enabled": "高级权限已开启",
     "brand_name": "品牌",
@@ -159,7 +161,28 @@ def sync_brand_target(
     source_rows = postgres_rows or []
     target_rows, shop_names = pivot_shop_stock_rows(source_rows)
     field_specs = build_channel_inventory_field_specs(shop_names)
+    try:
         available_fields = client.ensure_target_fields(access_token, app_token, table_id, field_specs)
+    except FeishuFieldValidationError as exc:
+        notification = (
+            "【渠道库存同步未执行】\n"
+            f"品牌：{target.brand_name}\n"
+            f"数据表：{target.table_name}\n"
+            "原因：飞书表头字段校验失败。本次不会删除或重建字段，也不会删除或写入记录。\n"
+            f"详情：{exc}"
+        )
+        try:
+            client.send_text_message(access_token, CHANNEL_INVENTORY_SCHEMA_ALERT_CHAT_ID, notification)
+        except Exception as notification_exc:  # noqa: BLE001
+            log_event(
+                logging.ERROR,
+                "渠道库存字段校验失败通知发送失败",
+                stage="schema_validation",
+                brand_name=target.brand_name,
+                table_name=target.table_name,
+                error=str(notification_exc),
+            )
+        raise
     existing_records = client.list_records(access_token, app_token, table_id)
     existing_record_ids = [record.get("record_id") for record in existing_records if record.get("record_id")]
     log_event(

inventory_sync/specs.py

@@ -25,7 +25,7 @@ class FieldSpec:
 
 
 BRAND_SYNC_TARGETS = [
-    BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂渠道库存-9：00和19：00", warehouse_keys=(2, 3, 7, 13)),
+    BrandSyncTarget(brand_name="羽心堂", table_name="羽心堂渠道库存-9：30和20：00", warehouse_keys=(2, 3, 7, 13)),
 ]
 
 

prefect.yaml

@@ -16,10 +16,7 @@ deployments:
       - feishu
       - channel-inventory-sync
     schedules:
-      - cron: "15 9 * * *"
+      - cron: "15 9,19 * * *"
-        timezone: "Asia/Shanghai"
-        active: true
-      - cron: "15 19 * * *"
         timezone: "Asia/Shanghai"
         active: true
     work_pool:
@@ -39,10 +36,7 @@ deployments:
       - feishu
       - batch-inventory-report
     schedules:
-      - cron: "15 9 * * *"
+      - cron: "15 9,19 * * *"
-        timezone: "Asia/Shanghai"
-        active: true
-      - cron: "15 19 * * *"
         timezone: "Asia/Shanghai"
         active: true
     work_pool:

prefect_inventory_scheduler.py

@@ -28,7 +28,19 @@ from prefect import flow, get_run_logger, task
 
 from inventory_analytics.logging_utils import exception_fields
 from inventory_analytics.channel_inventory_sync.entrypoint import main as channel_inventory_sync_main
+from inventory_analytics.channel_inventory_sync.entrypoint import LOG_FILE as CHANNEL_INVENTORY_SYNC_LOG_FILE
 from inventory_analytics.batch_inventory_report.cli import main as batch_inventory_report_main
+from inventory_analytics.batch_inventory_report.service import LOG_FILE as BATCH_INVENTORY_REPORT_LOG_FILE
+
+
+def read_log_tail(log_file: Path, line_count: int = 80) -> str:
+    try:
+        lines = log_file.read_text(encoding="utf-8", errors="replace").splitlines()
+    except FileNotFoundError:
+        return f"日志文件不存在: {log_file.name}"
+    except Exception as exc:  # noqa: BLE001
+        return f"读取日志文件失败: {exc.__class__.__name__}: {exc}"
+    return "\n".join(lines[-line_count:]) if lines else f"日志文件为空: {log_file.name}"
 
 
 @task(name="run-channel-inventory-sync")
@@ -38,11 +50,13 @@ def run_channel_inventory_sync_task() -> None:
     logger.info("渠道库存统计开始，action=all，安全日志模式=enabled")
     original_argv = sys.argv[:]
     original_safe_logs = os.environ.get("INVENTORY_SAFE_LOGS")
+    original_prefect_log_bridge = os.environ.get("INVENTORY_PREFECT_LOG_BRIDGE")
     root_logger = logging.getLogger()
     original_handlers = root_logger.handlers[:]
     original_level = root_logger.level
     try:
         os.environ["INVENTORY_SAFE_LOGS"] = "1"
+        os.environ["INVENTORY_PREFECT_LOG_BRIDGE"] = "1"
         sys.argv = ["sync_channel_inventory.py", "--action", "all"]
         exit_code = channel_inventory_sync_main()
     except Exception as exc:  # noqa: BLE001
@@ -54,10 +68,15 @@ def run_channel_inventory_sync_task() -> None:
             os.environ.pop("INVENTORY_SAFE_LOGS", None)
         else:
             os.environ["INVENTORY_SAFE_LOGS"] = original_safe_logs
+        if original_prefect_log_bridge is None:
+            os.environ.pop("INVENTORY_PREFECT_LOG_BRIDGE", None)
+        else:
+            os.environ["INVENTORY_PREFECT_LOG_BRIDGE"] = original_prefect_log_bridge
         root_logger.handlers = original_handlers
         root_logger.setLevel(original_level)
     if exit_code != 0:
         logger.error("渠道库存统计失败，退出码=%s", exit_code)
+        logger.error("渠道库存统计日志尾部:\n%s", read_log_tail(CHANNEL_INVENTORY_SYNC_LOG_FILE))
         raise RuntimeError(f"渠道库存统计执行失败，退出码: {exit_code}")
     logger.info("渠道库存统计完成，耗时毫秒=%s", int((time.monotonic() - started_at) * 1000))
 
@@ -78,11 +97,13 @@ def run_batch_inventory_report_task(
     )
     original_argv = sys.argv[:]
     original_safe_logs = os.environ.get("INVENTORY_SAFE_LOGS")
+    original_prefect_log_bridge = os.environ.get("INVENTORY_PREFECT_LOG_BRIDGE")
     root_logger = logging.getLogger()
     original_handlers = root_logger.handlers[:]
     original_level = root_logger.level
     try:
         os.environ["INVENTORY_SAFE_LOGS"] = "1"
+        os.environ["INVENTORY_PREFECT_LOG_BRIDGE"] = "1"
         sys.argv = ["generate_batch_inventory_report.py"]
         if output:
             sys.argv.extend(["--output", output])
@@ -103,10 +124,15 @@ def run_batch_inventory_report_task(
             os.environ.pop("INVENTORY_SAFE_LOGS", None)
         else:
             os.environ["INVENTORY_SAFE_LOGS"] = original_safe_logs
+        if original_prefect_log_bridge is None:
+            os.environ.pop("INVENTORY_PREFECT_LOG_BRIDGE", None)
+        else:
+            os.environ["INVENTORY_PREFECT_LOG_BRIDGE"] = original_prefect_log_bridge
         root_logger.handlers = original_handlers
         root_logger.setLevel(original_level)
     if exit_code != 0:
         logger.error("批次库存统计失败，退出码=%s", exit_code)
+        logger.error("批次库存统计日志尾部:\n%s", read_log_tail(BATCH_INVENTORY_REPORT_LOG_FILE))
         raise RuntimeError(f"批次库存统计报表执行失败，退出码: {exit_code}")
     logger.info("批次库存统计完成，耗时毫秒=%s", int((time.monotonic() - started_at) * 1000))
 
@@ -155,14 +181,14 @@ def main() -> None:
     if args.serve_channel_inventory_sync:
         inventory_channel_inventory_sync_flow.serve(
             name="channel-inventory-sync-0915-1915",
-            cron=("15 9 * * *", "15 19 * * *"),
+            cron="15 9,19 * * *",
             tags=["inventory", "feishu", "channel-inventory-sync"],
         )
         return
     if args.serve_batch_inventory_report:
         inventory_batch_inventory_report_flow.serve(
             name="batch-inventory-report-0915-1915",
-            cron=("15 9 * * *", "15 19 * * *"),
+            cron="15 9,19 * * *",
             parameters=flow_kwargs,
             tags=["inventory", "feishu", "batch-inventory-report"],
         )

pyproject.toml

@@ -8,9 +8,10 @@ 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.0.0",
+  "prefect==3.6.8",
   "psycopg2-binary",
   "python-dotenv",
   "xlrd>=2.0.1",

tests/test_channel_inventory_schema.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import json
+import unittest
+from unittest.mock import Mock
+
+from inventory_sync.feishu import FeishuClient, FeishuFieldValidationError
+from inventory_sync.service import CHANNEL_INVENTORY_SCHEMA_ALERT_CHAT_ID, sync_brand_target
+from inventory_sync.specs import BRAND_SYNC_TARGETS, FIELD_SPECS, PRIMARY_FIELD_NAME
+
+
+def valid_fields() -> list[dict[str, object]]:
+    fields: list[dict[str, object]] = [
+        {"field_name": PRIMARY_FIELD_NAME, "is_primary": True, "type": 1},
+    ]
+    for spec in FIELD_SPECS:
+        field: dict[str, object] = {
+            "field_name": spec.feishu_name,
+            "is_primary": False,
+            "type": spec.field_type,
+        }
+        if spec.formatter is not None:
+            field["property"] = {"formatter": spec.formatter}
+        fields.append(field)
+    return fields
+
+
+class ChannelInventorySchemaTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.client = FeishuClient(None)  # type: ignore[arg-type]
+        self.client.list_fields = Mock()  # type: ignore[method-assign]
+
+    def test_allows_fields_in_any_order(self) -> None:
+        fields = valid_fields()
+        self.client.list_fields.return_value = [fields[0], *reversed(fields[1:])]
+
+        available_fields = self.client.ensure_target_fields("token", "app", "table")
+
+        self.assertEqual(available_fields, [field["field_name"] for field in [fields[0], *reversed(fields[1:])]])
+
+    def test_rejects_missing_or_renamed_required_field(self) -> None:
+        fields = valid_fields()
+        fields = [field for field in fields if field["field_name"] != "品牌"]
+        self.client.list_fields.return_value = fields
+
+        with self.assertRaisesRegex(FeishuFieldValidationError, "缺少必需字段或字段已改名：“品牌”"):
+            self.client.ensure_target_fields("token", "app", "table")
+
+    def test_rejects_wrong_type_and_formatter(self) -> None:
+        fields = valid_fields()
+        price_field = next(field for field in fields if field["field_name"] == "零售价")
+        price_field["type"] = 1
+        price_field["property"] = {"formatter": "0"}
+        self.client.list_fields.return_value = fields
+
+        with self.assertRaisesRegex(FeishuFieldValidationError, "字段类型错误：“零售价”") as context:
+            self.client.ensure_target_fields("token", "app", "table")
+
+        self.assertIn("字段格式错误：“零售价”", str(context.exception))
+
+    def test_sends_group_text_message(self) -> None:
+        request_json = Mock(return_value={"code": 0})
+        self.client.request_json = request_json  # type: ignore[method-assign]
+
+        self.client.send_text_message("token", "oc_example", "字段校验失败")
+
+        url = request_json.call_args.args[0]
+        payload = request_json.call_args.kwargs["payload"]
+        self.assertEqual(url, "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id")
+        self.assertEqual(payload["receive_id"], "oc_example")
+        self.assertEqual(json.loads(payload["content"]), {"text": "字段校验失败"})
+
+    def test_schema_failure_notifies_group_before_records_are_changed(self) -> None:
+        class SchemaFailureClient:
+            def __init__(self) -> None:
+                self.notifications: list[tuple[str, str, str]] = []
+
+            def ensure_target_fields(self, *_args: object) -> list[str]:
+                raise FeishuFieldValidationError("缺少必需字段或字段已改名：“品牌”")
+
+            def send_text_message(self, access_token: str, chat_id: str, text: str) -> None:
+                self.notifications.append((access_token, chat_id, text))
+
+        client = SchemaFailureClient()
+        target = BRAND_SYNC_TARGETS[0]
+
+        with self.assertRaises(FeishuFieldValidationError):
+            sync_brand_target(
+                client,  # type: ignore[arg-type]
+                "token",
+                "app",
+                [{"name": target.table_name, "table_id": "table"}],
+                target,
+                [],
+            )
+
+        self.assertEqual(len(client.notifications), 1)
+        _, chat_id, text = client.notifications[0]
+        self.assertEqual(chat_id, CHANNEL_INVENTORY_SCHEMA_ALERT_CHAT_ID)
+        self.assertIn("不会删除或重建字段，也不会删除或写入记录", text)