功能实现说明

1. 准备好config.json文件，用于存放配置文件信息

2. 准备好config.py用于读取配置文件信息

3. feishu.py用于实现获取临时token以及推送消息至指定机器人并推到指定飞书用户

config.json

{"auth":{"app_id": "cli_xxxxxc","app_secret": "xxxxxxx"},"user_ids":{"replace_your_feishu_user_name":"replace_your_feishu_user_name's_user_id"}
}

config.py

"""
This module provides a Config class for managing configuration settings for the Feishu API integration.It reads configuration data from a JSON file and provides methods to access various settings.
"""import jsonclass Config:"""A class to manage configuration settings for the Feishu API integration.This class reads configuration data from a JSON file and provides methodsto access app credentials and user IDs."""def __init__(self, config_file='config.json'):"""Initialize the Config instance.Args:config_file (str): Path to the configuration JSON file. Defaults to 'config.json'.Raises:FileNotFoundError: If the specified config file is not found.json.JSONDecodeError: If the config file is not valid JSON.KeyError: If expected keys are missing in the config file."""with open(config_file, 'r') as f:config = json.load(f)# Extract configuration valuesself.app_id = config['auth']['app_id']self.app_secret = config['auth']['app_secret']self.user_ids = config['user_ids']def get_app_id(self):"""Get the Feishu app ID.Returns:str: The app ID."""return self.app_iddef get_app_secret(self):"""Get the Feishu app secret.Returns:str: The app secret."""return self.app_secretdef get_user_id(self, name):"""Get a user ID by name.Args:name (str): The name of the user.Returns:str: The user ID if found, None otherwise."""return self.user_ids.get(name)if __name__=='__main__':# Example usage of the Config classconfig = Config()print(config.get_app_id())print(config.get_app_secret())print(config.get_user_id('test'))

feishu.py

飞书机器人消息推送功能实现

import urllib3
import json
import ssl
import warnings
from config import Config# Suppress only the single InsecureRequestWarning from urllib3 needed.
warnings.filterwarnings('ignore', category=urllib3.exceptions.InsecureRequestWarning)class Feishu:"""A class to interact with the Feishu API.This class provides methods to authenticate and communicate with the Feishu API,including obtaining a tenant access token for further API operations."""def __init__(self):"""Initialize the Feishu instance.Sets up the configuration, base URL, and creates a custom SSL contextto allow unverified HTTPS connections (for development purposes only)."""self.config = Config()self.base_url = "https://open.feishu.cn/open-apis"self.tenant_access_token = None# Create a custom SSL contextssl_context = ssl.create_default_context()ssl_context.check_hostname = Falsessl_context.verify_mode = ssl.CERT_NONE# Use the custom SSL context in PoolManagerself.http = urllib3.PoolManager(ssl_context=ssl_context)def get_tenant_access_token(self):"""Obtain a tenant access token from the Feishu API.This method sends a POST request to the Feishu API to get a tenant access token.It uses the app_id and app_secret from the config to authenticate the request.Returns:str: The tenant access token if successful, None otherwise.Raises:Exception: If an error occurs during the API request."""url = f"{self.base_url}/auth/v3/tenant_access_token/internal"headers = {"Content-Type": "application/json"}data = {"app_id": self.config.get_app_id(),"app_secret": self.config.get_app_secret()}try:response = self.http.request("POST", url, headers=headers, body=json.dumps(data))response_data = json.loads(response.data.decode('utf-8'))if response.status == 200:self.tenant_access_token = response_data.get("tenant_access_token")if self.tenant_access_token:return self.tenant_access_tokenelse:print("Error: tenant_access_token not found in the response")else:print(f"Error: HTTP {response.status}")print("Response data:", response_data)except Exception as e:print(f"An error occurred: {str(e)}")return Noneif __name__ == "__main__":feishu = Feishu()token = feishu.get_tenant_access_token()if token:print(f"Successfully obtained tenant access token: {token}")else:print("Failed to obtain tenant access token")