Files
inventory_analytics/inventory_sync/feishu.py
T
BXY0414 e234bfbe7a 修feat: 渠道库存同步支持字段任意排序并增加字段异常告警
- 按字段名校验并写入渠道库存,取消字段顺序校验及表头重建
- 严格校验必填字段名称、类型和数字格式,失败时不修改表头或记录
- 字段校验失败时发送飞书群通知
- 添加字段校验与通知流程测试
- 忽略 Python 字节码、__pycache__ 和 .DS_Store
2026-07-22 14:28:23 +08:00

262 lines
11 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from __future__ import annotations
import json
import time
import uuid
from collections.abc import Sequence
from typing import Any
from urllib import error, parse, request
from .settings import AppConfig
from .specs import FIELD_SPECS, PRIMARY_FIELD_NAME
FEISHU_API_BASE = "https://open.feishu.cn/open-apis"
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
def request_json(
self,
url: str,
*,
method: str = "GET",
headers: dict[str, str] | None = None,
payload: dict[str, Any] | None = None,
) -> dict[str, Any]:
body = None
request_headers = {"Content-Type": "application/json; charset=utf-8"}
if headers:
request_headers.update(headers)
if payload is not None:
body = json.dumps(payload).encode("utf-8")
req = request.Request(url, data=body, headers=request_headers, method=method)
try:
with request.urlopen(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:
raise RuntimeError(f"无法连接飞书接口: {exc.reason}") from exc
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},
)
if data.get("code") != 0:
raise FeishuApiError(f"获取 tenant_access_token 失败: {data}")
token = data.get("tenant_access_token")
if not token:
raise FeishuApiError("接口未返回 tenant_access_token")
return token
def build_base_url(self) -> str:
base_url = self.config.feishu_base_url
app_token = self.config.feishu_app_token
if app_token and "/base/" not in base_url and "/wiki/" not in base_url:
return f"{base_url}/{app_token}"
return base_url
def get_wiki_node_info(self, access_token: str, node_token: str) -> dict[str, Any]:
data = self.request_json(
f"{FEISHU_API_BASE}/wiki/v2/spaces/get_node?token={parse.quote(node_token)}",
headers={"Authorization": f"Bearer {access_token}"},
)
if data.get("code") != 0:
raise FeishuApiError(f"获取知识空间节点信息失败: {data}")
node_info = data.get("data", {}).get("node")
if not node_info:
raise FeishuApiError("未获取到知识空间节点信息")
return node_info
def parse_base_info(self, base_url: str, access_token: str) -> dict[str, str | None]:
parsed_url = parse.urlparse(base_url)
pathname = parsed_url.path
app_token = pathname.rstrip("/").split("/")[-1]
if "/wiki/" in pathname:
node_info = self.get_wiki_node_info(access_token, app_token)
app_token = node_info.get("obj_token", app_token)
query_params = parse.parse_qs(parsed_url.query)
return {
"base_url": base_url,
"app_token": app_token,
"selected_table_id": query_params.get("table", [None])[0],
"selected_view_id": query_params.get("view", [None])[0],
}
def enable_advanced_permission(self, access_token: str, app_token: str) -> bool:
try:
data = self.request_json(
f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}",
method="PUT",
headers={"Authorization": f"Bearer {access_token}"},
payload={"is_advanced": True},
)
except Exception:
return False
return data.get("code") == 0
def _paginate(self, access_token: str, url_prefix: str, page_size: int) -> list[dict[str, Any]]:
items: list[dict[str, Any]] = []
page_token = ""
while True:
query = parse.urlencode({"page_size": page_size, **({"page_token": page_token} if page_token else {})})
data = self.request_json(
f"{url_prefix}?{query}",
headers={"Authorization": f"Bearer {access_token}"},
)
if data.get("code") != 0:
raise FeishuApiError(f"分页接口失败: {data}")
items.extend(data.get("data", {}).get("items") or [])
page_token = data.get("data", {}).get("page_token") or ""
if not data.get("data", {}).get("has_more", False):
break
return items
def list_tables(self, access_token: str, app_token: str) -> list[dict[str, Any]]:
return self._paginate(access_token, f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables", 100)
def list_fields(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}/fields", 100)
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}",
method="DELETE",
headers={"Authorization": f"Bearer {access_token}"},
)
if data.get("code") != 0:
raise FeishuApiError(f"删除字段失败({field_id}): {data}")
def update_field(self, access_token: str, app_token: str, table_id: str, field_id: str, payload: dict[str, Any]) -> None:
data = self.request_json(
f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables/{table_id}/fields/{field_id}",
method="PUT",
headers={"Authorization": f"Bearer {access_token}"},
payload=payload,
)
if data.get("code") != 0:
raise FeishuApiError(f"更新字段失败({payload.get('field_name')}): {data}")
def create_field(self, access_token: str, app_token: str, table_id: str, payload: dict[str, Any]) -> None:
data = self.request_json(
f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables/{table_id}/fields?client_token={uuid.uuid4()}",
method="POST",
headers={"Authorization": f"Bearer {access_token}"},
payload=payload,
)
if data.get("code") != 0:
raise FeishuApiError(f"新增字段失败({payload.get('field_name')}): {data}")
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):
chunk = record_ids[start : start + 500]
data = self.request_json(
f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_delete",
method="POST",
headers={"Authorization": f"Bearer {access_token}"},
payload={"records": chunk},
)
if data.get("code") != 0:
raise FeishuApiError(f"删除记录失败: {data}")
deleted += len(chunk)
time.sleep(0.2)
return deleted
def batch_create_records(self, access_token: str, app_token: str, table_id: str, records: list[dict[str, Any]]) -> int:
created = 0
for start in range(0, len(records), 500):
chunk = records[start : start + 500]
data = self.request_json(
f"{FEISHU_API_BASE}/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_create?client_token={uuid.uuid4()}",
method="POST",
headers={"Authorization": f"Bearer {access_token}"},
payload={"records": chunk},
)
if data.get("code") != 0:
raise FeishuApiError(f"新增记录失败: {data}")
created += len(chunk)
time.sleep(0.2)
return created
def ensure_target_fields(
self,
access_token: str,
app_token: str,
table_id: str,
field_specs: Sequence[Any] = FIELD_SPECS,
) -> list[str]:
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 FeishuFieldValidationError("未找到渠道库存表主字段")
validation_errors: list[str] = []
if primary_field.get("field_name") != PRIMARY_FIELD_NAME:
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:
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:
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 '未设置'}"
)
if validation_errors:
raise FeishuFieldValidationError(";".join(validation_errors))
return [field.get("field_name") for field in fields if field.get("field_name")]