inventory_analytics/inventory_sync/feishu.py

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")]