飞书机器人

import zmq
import json

import requests

Replace this with your actual Feishu bot webhook URL

WEBHOOK_URL = "your URL"

def send_feishu_message(text: str):
"""Send a text message to your Feishu chat via bot webhook."""
headers = {
"Content-Type": "application/json; charset=utf-8"
}
payload = {
"msg_type": "text",
"content": {
"text": text
}
}
response = requests.post(WEBHOOK_URL, headers=headers, data=json.dumps(payload))
if response.status_code == 200:
print("Message sent successfully!")
return response.json()
else:
print(f"Failed to send message. Status code: {response.status_code}, Response: {response.text}")
return None

def main():
# Initialize ZMQ context and socket
context = zmq.Context()
subscriber = context.socket(zmq.SUB)

# Connect to the publisher (replace with server A's IP address)
subscriber.connect("tcp://HOST:POST")  # Change to server A's IP# Subscribe to all messages (empty string means subscribe to everything)
subscriber.setsockopt_string(zmq.SUBSCRIBE, "")print("Subscriber started, waiting for messages...")try:while True:# Receive the JSON stringmessage = subscriber.recv_string()send_feishu_message(message)print(f"Received raw message: {message}")# Parse JSONtry:data = json.loads(message)# Access the deserialized dataprint("\nParsed SpecialMsg:")print(f"  name: {data['name']}")print(f"  serviceId: {data['serviceId']}")print(f"  stratId: {data['stratId']}")print(f"  content: {data['content']}")print("-" * 50)except json.JSONDecodeError as e:print(f"JSON parsing error: {e}")except KeyboardInterrupt:print("\nShutting down...")
finally:subscriber.close()context.term()

if name == "main":
main()