From e234bfbe7a8d9f7850f83a222544e61c74a2eacc Mon Sep 17 00:00:00 2001 From: BXY0414 <1079823339@qq.com> Date: Wed, 22 Jul 2026 14:28:23 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AEfeat:=20=E6=B8=A0=E9=81=93=E5=BA=93?= =?UTF-8?q?=E5=AD=98=E5=90=8C=E6=AD=A5=E6=94=AF=E6=8C=81=E5=AD=97=E6=AE=B5?= =?UTF-8?q?=E4=BB=BB=E6=84=8F=E6=8E=92=E5=BA=8F=E5=B9=B6=E5=A2=9E=E5=8A=A0?= =?UTF-8?q?=E5=AD=97=E6=AE=B5=E5=BC=82=E5=B8=B8=E5=91=8A=E8=AD=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 按字段名校验并写入渠道库存，取消字段顺序校验及表头重建 - 严格校验必填字段名称、类型和数字格式，失败时不修改表头或记录 - 字段校验失败时发送飞书群通知 - 添加字段校验与通知流程测试 - 忽略 Python 字节码、__pycache__ 和 .DS_Store --- .DS_Store | Bin 8196 -> 0 bytes .gitignore | 10 +- .vscode/settings.json | 4 + README.md | 1 + inventory_sync/feishu.py | 123 ++++++++++--------------- inventory_sync/service.py | 27 +++++- tests/test_channel_inventory_schema.py | 100 ++++++++++++++++++++ 7 files changed, 189 insertions(+), 76 deletions(-) delete mode 100644 .DS_Store create mode 100644 .vscode/settings.json create mode 100644 tests/test_channel_inventory_schema.py diff --git a/.DS_Store b/.DS_Store deleted file mode 100644 index 199d5c142361b5a625a39a33efb298d233a87899..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 8196 zcmeHM&2G~`5T0!Vbrk_Q6e&VNvV?>{szU!jR9r&Zf`Sl5O@RX-V<({vxV9BL4W)|6 z8Qy_c;L4M5;Ym1^Z+5pOb{)7OM0TXz*`N9L=Ns=j;}DUk^;(NWWg@bWSeJ^(a)kRi zosuctQi2x16E(?2+NCx*r>sqbVZbn87%&VN1`GrL0t0wwvsh!^`>NOEh5^ICL^8nl zgN?+xs`XH5d37L@Apmp%%~DY2aDXwk)>W;CN}~rN6ck!Pg$^-f zL^MYpqWwh1`wn#|pnb&s0a`-pVl0PRvfoXQBQ9sgxk@_{*9MBm<8KEC^bRHGk27>( z9m)9;s_+gnbA8Ipbv}2UG3E=XnZf?ZU@wsmbS&5P8uM~_mwI}Yv0Qh89eA*7qc3MY zbT>0*0k$FbAE)P@+R4}zv9~TP-(ch>R$$XLTBPe~6nzKk_K1hZpuBz@X}^B;BEJ(S z3k%=G)O6v@*%@obDq1fayJ5ePxJfUrx!vbn+YUqDJ1)E4(@wkZl;+mLAaOfEyDI^m zRvRWSo^^s&*sq1XpcPAOLv>iQ)~r)1A095=T$s0)mRAS!_TlOh$}7va27_7a;*}eB zww$;AUJ!o7P9Q>G_H<6wPjdF_nBDC+x`7`C%pjvL None: self.config = config @@ -133,6 +136,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 +180,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 +222,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("未找到主字段") - 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("删除重复字段后未找到主字段") + raise FeishuFieldValidationError("未找到渠道库存表主字段") + validation_errors: list[str] = [] if primary_field.get("field_name") != PRIMARY_FIELD_NAME: - payload = {"field_name": PRIMARY_FIELD_NAME, "type": primary_field.get("type")} - if primary_field.get("property") is not None: - payload["property"] = primary_field.get("property") - self.update_field(access_token, app_token, table_id, primary_field["field_id"], payload) - time.sleep(0.2) - - 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) + validation_errors.append( + f"主字段名称错误：期望“{PRIMARY_FIELD_NAME}”，实际“{primary_field.get('field_name') or '未命名'}”" + ) + if primary_field.get("type") != 1: + validation_errors.append( + f"主字段类型错误：字段“{PRIMARY_FIELD_NAME}”应为文本，实际类型 {primary_field.get('type')}" + ) + 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} - self.create_field(access_token, app_token, table_id, payload) - time.sleep(0.2) + actual_formatter = (field.get("property") or {}).get("formatter") + if actual_formatter != spec.formatter: + 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")] diff --git a/inventory_sync/service.py b/inventory_sync/service.py index 121c416..1d007db 100644 --- a/inventory_sync/service.py +++ b/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) - available_fields = client.ensure_target_fields(access_token, app_token, table_id, field_specs) + 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( diff --git a/tests/test_channel_inventory_schema.py b/tests/test_channel_inventory_schema.py new file mode 100644 index 0000000..0d68b3d --- /dev/null +++ b/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)