diff --git a/feishu_api.py b/feishu_api.py new file mode 100644 index 0000000..4fd64fb --- /dev/null +++ b/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() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..39aae79 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +requests>=2.28.0 \ No newline at end of file diff --git a/test_feishu_api.py b/test_feishu_api.py new file mode 100644 index 0000000..4612eda --- /dev/null +++ b/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() \ No newline at end of file