feishu_bitable.py

from __future__ import annotations

import time
from typing import Any, Dict, Iterable, List, Optional, Set, Tuple

import requests

from .logger import get_logger

logger = get_logger(__name__)


class FeishuError(RuntimeError):
    pass


class FeishuBitableClient:
    def __init__(
        self,
        app_id: str,
        app_secret: str,
        app_token: str,
        timeout: int = 20,
        dry_run: bool = False,
    ) -> None:
        self.app_id = app_id
        self.app_secret = app_secret
        self.app_token = app_token
        self.timeout = timeout
        self.dry_run = dry_run
        self.base_url = "https://open.feishu.cn/open-apis"
        self._tenant_access_token: Optional[str] = None
        self._token_expire_at = 0.0

    def _headers(self) -> Dict[str, str]:
        token = self.get_tenant_access_token()
        return {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json; charset=utf-8",
        }

    def get_tenant_access_token(self) -> str:
        if self.dry_run:
            return "dry_run_token"

        now = time.time()
        if self._tenant_access_token and now < self._token_expire_at - 60:
            return self._tenant_access_token

        if not self.app_id or not self.app_secret:
            raise FeishuError("Missing FEISHU_APP_ID or FEISHU_APP_SECRET")

        url = f"{self.base_url}/auth/v3/tenant_access_token/internal"
        resp = requests.post(
            url,
            json={"app_id": self.app_id, "app_secret": self.app_secret},
            timeout=self.timeout,
        )
        resp.raise_for_status()
        data = resp.json()
        if data.get("code") != 0:
            raise FeishuError(f"Failed to get tenant token: {data}")

        self._tenant_access_token = data["tenant_access_token"]
        self._token_expire_at = now + int(data.get("expire", 7200))
        return self._tenant_access_token

    def list_records(
        self,
        table_id: str,
        page_size: int = 500,
        field_names: Optional[List[str]] = None,
    ) -> List[Dict[str, Any]]:
        if self.dry_run:
            logger.info("[DRY_RUN] list_records table=%s", table_id)
            return []

        if not self.app_token or not table_id:
            raise FeishuError("Missing FEISHU_BITABLE_APP_TOKEN or table_id")

        records: List[Dict[str, Any]] = []
        page_token = None
        while True:
            params: Dict[str, Any] = {"page_size": page_size}
            if page_token:
                params["page_token"] = page_token
            if field_names:
                # Feishu supports field_names as repeated query params in some SDKs.
                # For raw HTTP, leaving this out is more compatible. We filter locally instead.
                pass

            url = f"{self.base_url}/bitable/v1/apps/{self.app_token}/tables/{table_id}/records"
            resp = requests.get(url, headers=self._headers(), params=params, timeout=self.timeout)
            resp.raise_for_status()
            data = resp.json()
            if data.get("code") != 0:
                raise FeishuError(f"Failed to list records: {data}")

            items = data.get("data", {}).get("items", [])
            records.extend(items)
            if not data.get("data", {}).get("has_more"):
                break
            page_token = data.get("data", {}).get("page_token")
            if not page_token:
                break

        return records

    def get_existing_hashes(self, raw_table_id: str) -> Tuple[Set[str], Set[str]]:
        records = self.list_records(raw_table_id)
        url_hashes: Set[str] = set()
        title_hashes: Set[str] = set()
        for record in records:
            fields = record.get("fields", {}) or {}
            url_hash = fields.get("url_hash")
            title_hash = fields.get("title_hash")
            if url_hash:
                url_hashes.add(str(url_hash))
            if title_hash:
                title_hashes.add(str(title_hash))
        logger.info("Loaded existing hashes: url=%s title=%s", len(url_hashes), len(title_hashes))
        return url_hashes, title_hashes

    def batch_create_records(self, table_id: str, rows: Iterable[Dict[str, Any]], batch_size: int = 100) -> int:
        rows_list = list(rows)
        if not rows_list:
            return 0

        if self.dry_run:
            logger.info("[DRY_RUN] batch_create_records table=%s count=%s", table_id, len(rows_list))
            for row in rows_list[:3]:
                logger.info("[DRY_RUN] sample row: %s", row)
            return len(rows_list)

        if not self.app_token or not table_id:
            raise FeishuError("Missing FEISHU_BITABLE_APP_TOKEN or table_id")

        created = 0
        for i in range(0, len(rows_list), batch_size):
            chunk = rows_list[i : i + batch_size]
            url = f"{self.base_url}/bitable/v1/apps/{self.app_token}/tables/{table_id}/records/batch_create"
            payload = {"records": [{"fields": row} for row in chunk]}
            resp = requests.post(url, headers=self._headers(), json=payload, timeout=self.timeout)
            resp.raise_for_status()
            data = resp.json()
            if data.get("code") != 0:
                raise FeishuError(f"Failed to create records: {data}")
            created += len(chunk)
        return created

    def read_watch_configs(self, config_table_id: str) -> List[Dict[str, Any]]:
        records = self.list_records(config_table_id)
        configs: List[Dict[str, Any]] = []
        for record in records:
            fields = record.get("fields", {}) or {}
            if fields:
                configs.append(fields)
        return configs


def send_feishu_webhook(
    webhook_url: str,
    text: str,
    card: dict | None = None,
    timeout: int = 20,
    dry_run: bool = False,
) -> bool:
    """Send a Feishu webhook message.

    If *card* is provided (a dict with msg_type='interactive'), send as
    Interactive Card.  Otherwise fall back to plain text.
    """
    if not webhook_url:
        logger.info("No FEISHU_WEBHOOK_URL configured, skip webhook push.")
        return False

    if dry_run:
        import json as _json
        if card:
            logger.info("[DRY_RUN] send CARD webhook:\n%s", _json.dumps(card, ensure_ascii=False, indent=2)[:2000])
        else:
            logger.info("[DRY_RUN] send TEXT webhook:\n%s", text[:1000])
        return True

    if card:
        payload = card  # already {msg_type: "interactive", card: {...}}
    else:
        payload = {
            "msg_type": "text",
            "content": {
                "text": text,
            },
        }

    resp = requests.post(webhook_url, json=payload, timeout=timeout)
    resp.raise_for_status()
    data = resp.json()
    if data.get("StatusCode", 0) not in (0, None) or data.get("code", 0) not in (0, None):
        logger.warning("Feishu webhook returned non-zero response: %s", data)
    return True