from __future__ import annotations import json import ssl 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" 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, url: str, *, 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"} if headers: request_headers.update(headers) 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 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, ssl.SSLError) as 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}") 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")]