from __future__ import annotations import json import re 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, LEGACY_NAME_MAP, 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 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 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 _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): 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 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("删除重复字段后未找到主字段") 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) for spec in field_specs: 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")]