update

.gitignore

@@ -0,0 +1,2 @@
+__pycache__/
+.env

app.py

@@ -1,7 +1,277 @@
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
+import os, time
+import asyncio
+import redis.asyncio as redis # Import async Redis client
+from dotenv import load_dotenv
+from redis.exceptions import ConnectionError # Import ConnectionError
+import json # Import the json module
+import httpx # Import httpx for async HTTP requests
+
+load_dotenv()
+
+redis_host = os.getenv('REDIS_HOST')
+redis_password = os.getenv('REDIS_PASSWORD')
+redis_port = os.getenv('REDIS_PORT', 6379) # Default to 6379 if not set
+feishu_app_id = os.getenv('FEISHU_APP_ID', 'YOUR_FEISHU_APP_ID') # Placeholder
+feishu_app_secret = os.getenv('FEISHU_APP_SECRET', 'YOUR_FEISHU_APP_SECRET') # Placeholder
+
+# Use async Redis client
+r = redis.Redis(
+  host=redis_host,
+  port=int(redis_port), # Ensure port is an integer
+  password=redis_password,
+  ssl=True
+)
+
+reply_template = {
+    "url":"https://open.feishu.cn/open-apis/im/v1/messages/{{msg_id}}/reply",
+    "Authorization": f"Bearer {{tenant_access_token}}",
+    "body": {
+        "content": "{{body}}",
+        "msg_type": "{{ type }}"
+    }
+}
+
+def generate_reply_string(msg_id: str, tenant_access_token: str, body: str, type: str) -> str:
+    """
+    Generates a reply string based on the reply_template.
+
+    Args:
+        msg_id: The message ID.
+        tenant_access_token: The tenant access token.
+        body: The content body of the reply (should be a dictionary).
+        type: The message type.
+
+    Returns:
+        A formatted string based on the reply_template.
+    """
+    # Use json.dumps to convert the body dictionary to a JSON string
+    # and then escape the quotes for inclusion in the f-string
+    body_json_string = json.dumps(body)
+
+    # Construct the final string using an f-string
+    # Note: The original template had {{body}} directly, which implies the body
+    # should be inserted as a JSON object, not a string.
+    # The f-string needs to correctly format the JSON structure.
+    # The url and Authorization header use f-strings within the template itself,
+    # so we need to evaluate those first, then insert the variables.
+    # Let's reconstruct the template string directly with f-strings.
+    reply_string = f"""{{
+    "url":"https://open.feishu.cn/open-apis/im/v1/messages/{msg_id}/reply",
+    "Authorization": "Bearer {tenant_access_token}",
+    "body": {{
+        "content": {{
+            "text": "{body}"
+        }},
+        "msg_type": "{type}"
+    }}
+}}"""
+    return reply_string
+
+async def get_valid_tenant_access_token() -> str:
+    """
+    Retrieves a valid tenant access token from Feishu API.
+    """
+    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
+    headers = {
+        "Content-Type": "application/json"
+    }
+    payload = {
+        "app_id": feishu_app_id,
+        "app_secret": feishu_app_secret
+    }
+
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.post(url, headers=headers, json=payload)
+            response.raise_for_status() # Raise an exception for bad status codes
+            data = response.json()
+            if data.get("code") == 0:
+                return data.get("tenant_access_token")
+            else:
+                print(f"Error getting tenant access token: {data.get('msg')}")
+                return None
+        except httpx.HTTPStatusError as e:
+            print(f"HTTP error occurred: {e}")
+            return None
+        except httpx.RequestError as e:
+            print(f"An error occurred while requesting {e.request.url!r}: {e}")
+            return None
+        except Exception as e:
+            print(f"An unexpected error occurred: {e}")
+            return None
+
+
+# def get_tenant_access_token():
+#     return "tenant_access_token"
 
 app = FastAPI()
 
+# Optional: Test connection (async ping)
+@app.on_event("startup")
+async def connect_redis():
+    try:
+        await r.ping()
+        print("Connected to Redis!")
+
+        # Perform SET and GET test after successful connection
+        try:
+            await r.set("mykey", "myvalue")
+            print("Redis SET test successful: mykey = myvalue")
+            value = await r.get("mykey")
+            print(f"Redis GET test successful: mykey = {value.decode()}")
+        except Exception as e:
+            print(f"Redis SET/GET test failed: {e}")
+
+    except ConnectionError as e:
+        print(f"Could not connect to Redis: {e}")
+        # If connection fails, we might not be able to proceed further
+        return # Exit the startup event if connection fails
+
+
+# Define the Redis subscriber function
+async def redis_subscriber(redis_client, channel_name):
+    """
+    Asynchronous function to subscribe to a Redis channel and process messages.
+    """
+    print(f"Subscribing to channel: {channel_name}")
+    pubsub = redis_client.pubsub()
+    await asyncio.sleep(1) # Add a small delay before subscribing
+    await pubsub.subscribe(channel_name) # Use await for async subscribe
+
+    # Listen for messages
+    while True:
+        # print("Waiting for message...")
+        message = await pubsub.get_message(ignore_subscribe_messages=True)
+        # print(f"Message received: {message}")
+        if message:
+            print("message in...")
+            channel = message['channel'].decode()
+            data_str = message['data'].decode()
+            print(f"\n\n\nReceived message on channel '{channel}': {data_str}")
+
+            # Convert the received JSON string back to a Python dictionary
+            try:
+                data_dict = json.loads(data_str)
+                # Get valid tenant access token
+                tenant_access_token = await get_valid_tenant_access_token()
+                if tenant_access_token:
+                    print(f"Parsed message data as dictionary: {data_dict}")
+                    print(f"msg_id: {data_dict['msg_id']}")
+                    print(f"tenant_access_token: {tenant_access_token}")
+                    print(f"body: {data_dict['content']['body']}")
+                    print(f"type: {data_dict['content']['type']}")
+                    reply_string = generate_reply_string(data_dict['msg_id'], tenant_access_token, data_dict['content']['body'], data_dict['content']['type'])
+                    print(reply_string)
+
+                    # Send the reply string as an HTTP POST request
+                    try:
+                        reply_data = json.loads(reply_string)
+                        reply_url = reply_data.get("url")
+                        reply_headers = {"Authorization": reply_data.get("Authorization")}
+                        reply_body = reply_data.get("body")
+
+                        async with httpx.AsyncClient() as client:
+                            response = await client.post(reply_url, headers=reply_headers, json=reply_body)
+                            response.raise_for_status() # Raise an exception for bad status codes
+                            print(f"Successfully sent reply: Status Code {response.status_code}")
+                            print(f"Reply Response: {response.text}")
+
+                    except json.JSONDecodeError as e:
+                        print(f"Could not parse reply_string as JSON: {e}")
+                    except httpx.HTTPStatusError as e:
+                        print(f"HTTP error sending reply: {e}")
+                    except httpx.RequestError as e:
+                        print(f"An error occurred while requesting {e.request.url!r} to send reply: {e}")
+                    except Exception as e:
+                        print(f"An unexpected error occurred while sending reply: {e}")
+
+                else:
+                    print("Failed to get tenant access token, cannot process message.")
+
+
+            except json.JSONDecodeError as e:
+                print(f"Could not parse message data as JSON: {e}")
+                # Handle cases where the message is not valid JSON
+
+        await asyncio.sleep(0.01) # Prevent blocking the event loop
+
+# Run the subscriber as a background task on startup
+@app.on_event("startup")
+async def startup_event():
+    # Replace 'my_feishu_channel' with the actual channel name you want to subscribe to
+    asyncio.create_task(redis_subscriber(r, 'my_feishu_channel'))
+    print("Redis subscriber task started.")
+
+
 @app.get("/")
 def greet_json():
     return {"Hello": "World!"}
+
+@app.post("/webhook/feishu")
+async def handle_webhook(request: Request):
+    # Read the incoming JSON body
+    data = await request.json()
+    # Check if 'type' exists in data and is equal to "url_verification"
+    if data.get('type') == "url_verification":
+        # If it's a URL verification request, return the challenge
+        return {"challenge": data.get('challenge')}
+
+    # Process other event types
+    # The full event data is under data['body']['event'] based on previous examples
+    event = data['event'] # Use .get() with default {} for safe access
+    message = event['message']
+    chat_type=message.get('chat_type')
+    if chat_type=='p2p':
+        message_type = 'private'
+    elif chat_type=='group':
+        if message.get('mentions'):
+            message_type = 'group_mention'
+        else:
+            message_type = 'group_all'
+    
+    ret_data = {
+        'bot_id': data['header']['app_id'],
+        'message_type': message_type,
+        'msg_id': message['message_id'],
+        "content":{
+            "type": message['message_type'],
+            "body": json.loads(message.get('content')).get('text')
+        },
+        "sender": {
+            "id": event['sender']['sender_id']['open_id'],
+            "sender_type": event['sender']['sender_type'],
+            "name": "",
+            "is_human": None
+        },
+        "timestamp":time.time(),
+        "platform_specific": {
+            "app_id": data['header']['app_id']
+        }
+    }
+    await r.publish('my_feishu_channel', json.dumps(ret_data))
+
+    print('\n\n\nret_data: ',json.dumps(ret_data))
+    return {'status': 'OK'} 
+
+@app.post("/publish")
+async def publish_message(request: Request):
+    """
+    Publishes a message to a Redis channel.
+    Expects JSON body with 'channel' and 'message'.
+    """
+    try:
+        data = await request.json()
+        channel = data.get("channel")
+        message = data.get("message")
+
+        if not channel or not message:
+            return {"status": "error", "message": "Missing channel or message"}
+
+        # Publish the message to the specified channel (publish is asynchronous)
+        await r.publish(channel, message)
+
+        return {"status": "success", "message": f"Published message to channel '{channel}'"}
+
+    except Exception as e:
+        return {"status": "error", "message": str(e)}

requirements.txt

@@ -1,3 +1,6 @@
 # clear && uvicorn app:app --host 0.0.0.0 --port 7860 --reload
 fastapi
 uvicorn[standard]
+redis
+python-dotenv
+httpx