BXY0414/inventory_analytics
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

调整 字段顺序、名称

Browse Source
This commit is contained in:
BXY0414
2026-04-22 17:12:37 +08:00
parent 86f120bc18
commit 7b45d04a6b
6 changed files with 56 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
inventory_sync/feishu.py
+22 -32
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import json
import re
import time
import uuid
from typing import Any
@@ -160,6 +161,14 @@ 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):
@@ -217,40 +226,21 @@ class FeishuClient:
self.update_field(access_token, app_token, table_id, primary_field["field_id"], payload)
time.sleep(0.2)
for old_name, new_name in LEGACY_NAME_MAP.items():
fields = self.list_fields(access_token, app_token, table_id)
field = next((item for item in fields if item.get("field_name") == old_name), None)
if field and old_name != new_name:
payload = {"field_name": new_name, "type": field.get("type")}
if field.get("property") is not None:
payload["property"] = field.get("property")
self.update_field(access_token, app_token, table_id, field["field_id"], payload)
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 self._is_managed_field(field_name):
self.delete_field(access_token, app_token, table_id, field_id)
time.sleep(0.2)
fields = self.list_fields(access_token, app_token, table_id)
fields_by_name = {field.get("field_name"): field for field in fields if field.get("field_name")}
for spec in FIELD_SPECS:
field = fields_by_name.get(spec.feishu_name)
if not field:
payload = {"field_name": spec.feishu_name, "type": spec.field_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)
fields = self.list_fields(access_token, app_token, table_id)
fields_by_name = {item.get("field_name"): item for item in fields if item.get("field_name")}
field = fields_by_name.get(spec.feishu_name)
if not field:
continue
current_formatter = (field.get("property") or {}).get("formatter")
if field.get("type") != spec.field_type or current_formatter != spec.formatter:
payload = {"field_name": spec.feishu_name, "type": spec.field_type}
if spec.formatter is not None:
payload["property"] = {"formatter": spec.formatter}
elif field.get("property") is not None:
payload["property"] = field.get("property")
self.update_field(access_token, app_token, table_id, field["field_id"], payload)
time.sleep(0.2)
payload = {"field_name": spec.feishu_name, "type": spec.field_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)
return [field.get("field_name") for field in self.list_fields(access_token, app_token, table_id) if field.get("field_name")]