feat: integrate Feishu multidimensional table API (#96)

feishu_api.py

@@ -0,0 +1,99 @@
+"""
+Feishu (Lark) API integration for Open Source Bazaar.
+Issue: https://github.com/Open-Source-Bazaar/Open-Source-Bazaar.github.io/issues/96
+"""
+
+import requests
+import json
+from typing import Dict, List, Optional
+
+
+class FeishuAPI:
+    """Client for Feishu multidimensional table API."""
+
+    BASE_URL = "https://open.feishu.cn/open-apis"
+
+    def __init__(self, app_id: str, app_secret: str):
+        self.app_id = app_id
+        self.app_secret = app_secret
+        self.tenant_token = None
+
+    def get_tenant_token(self) -> str:
+        """Authenticate and get tenant access token."""
+        url = f"{self.BASE_URL}/auth/v3/tenant_access_token/internal"
+        headers = {"Content-Type": "application/json"}
+        data = {
+            "app_id": self.app_id,
+            "app_secret": self.app_secret
+        }
+
+        response = requests.post(url, headers=headers, json=data)
+        response.raise_for_status()
+
+        result = response.json()
+        self.tenant_token = result.get("tenant_access_token")
+        return self.tenant_token
+
+    def get_headers(self) -> Dict[str, str]:
+        """Get authenticated headers for API requests."""
+        if not self.tenant_token:
+            self.get_tenant_token()
+
+        return {
+            "Authorization": f"Bearer {self.tenant_token}",
+            "Content-Type": "application/json"
+        }
+
+    def get_table_records(self, app_token: str, table_id: str) -> List[Dict]:
+        """Fetch records from a multidimensional table."""
+        url = f"{self.BASE_URL}/bitable/v1/apps/{app_token}/tables/{table_id}/records"
+        headers = self.get_headers()
+
+        response = requests.get(url, headers=headers)
+        response.raise_for_status()
+
+        return response.json().get("data", {}).get("items", [])
+
+    def create_record(self, app_token: str, table_id: str, fields: Dict) -> Dict:
+        """Create a new record in the table."""
+        url = f"{self.BASE_URL}/bitable/v1/apps/{app_token}/tables/{table_id}/records"
+        headers = self.get_headers()
+
+        data = {"fields": fields}
+        response = requests.post(url, headers=headers, json=data)
+        response.raise_for_status()
+
+        return response.json()
+
+    def update_record(self, app_token: str, table_id: str, record_id: str, fields: Dict) -> Dict:
+        """Update an existing record."""
+        url = f"{self.BASE_URL}/bitable/v1/apps/{app_token}/tables/{table_id}/records/{record_id}"
+        headers = self.get_headers()
+
+        data = {"fields": fields}
+        response = requests.put(url, headers=headers, json=data)
+        response.raise_for_status()
+
+        return response.json()
+
+
+def main():
+    """Example usage."""
+    # Configuration (should be loaded from environment variables)
+    import os
+    app_id = os.getenv("FEISHU_APP_ID", "")
+    app_secret = os.getenv("FEISHU_APP_SECRET", "")
+
+    if not app_id or not app_secret:
+        print("Please set FEISHU_APP_ID and FEISHU_APP_SECRET environment variables")
+        return
+
+    client = FeishuAPI(app_id, app_secret)
+
+    # Example: Get records from a table
+    # records = client.get_table_records("app_token_here", "table_id_here")
+    # print(json.dumps(records, indent=2, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1 @@
+requests>=2.28.0

test_feishu_api.py

@@ -0,0 +1,79 @@
+"""
+Tests for Feishu API integration.
+"""
+
+import unittest
+from unittest.mock import Mock, patch, MagicMock
+import json
+from feishu_api import FeishuAPI
+
+
+class TestFeishuAPI(unittest.TestCase):
+    """Test cases for FeishuAPI client."""
+
+    def setUp(self):
+        self.client = FeishuAPI("test_app_id", "test_app_secret")
+
+    @patch('requests.post')
+    def test_get_tenant_token(self, mock_post):
+        """Test authentication and token retrieval."""
+        mock_response = Mock()
+        mock_response.json.return_value = {
+            "code": 0,
+            "tenant_access_token": "test_token_123",
+            "expire": 7200
+        }
+        mock_response.raise_for_status = Mock()
+        mock_post.return_value = mock_response
+
+        token = self.client.get_tenant_token()
+
+        self.assertEqual(token, "test_token_123")
+        self.assertEqual(self.client.tenant_token, "test_token_123")
+        mock_post.assert_called_once()
+
+    @patch('requests.get')
+    @patch.object(FeishuAPI, 'get_tenant_token')
+    def test_get_table_records(self, mock_auth, mock_get):
+        """Test fetching table records."""
+        mock_auth.return_value = "test_token"
+
+        mock_response = Mock()
+        mock_response.json.return_value = {
+            "code": 0,
+            "data": {
+                "items": [
+                    {"record_id": "rec1", "fields": {"Name": "Book 1"}},
+                    {"record_id": "rec2", "fields": {"Name": "Book 2"}}
+                ]
+            }
+        }
+        mock_response.raise_for_status = Mock()
+        mock_get.return_value = mock_response
+
+        records = self.client.get_table_records("app123", "tbl456")
+
+        self.assertEqual(len(records), 2)
+        self.assertEqual(records[0]["fields"]["Name"], "Book 1")
+
+    @patch('requests.post')
+    @patch.object(FeishuAPI, 'get_tenant_token')
+    def test_create_record(self, mock_auth, mock_post):
+        """Test creating a new record."""
+        mock_auth.return_value = "test_token"
+
+        mock_response = Mock()
+        mock_response.json.return_value = {
+            "code": 0,
+            "data": {"record": {"record_id": "new_rec_123"}}
+        }
+        mock_response.raise_for_status = Mock()
+        mock_post.return_value = mock_response
+
+        result = self.client.create_record("app123", "tbl456", {"Name": "New Book"})
+
+        self.assertEqual(result["data"]["record"]["record_id"], "new_rec_123")
+
+
+if __name__ == "__main__":
+    unittest.main()