update

Dockerfile

@@ -13,4 +13,4 @@ COPY --chown=user ./requirements.txt requirements.txt
 RUN pip install --no-cache-dir --upgrade -r requirements.txt
 
 COPY --chown=user . /app
-CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

app.py

@@ -1,277 +0,0 @@
-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:
-    # return "t-g1045baIWTQUCO2WMLFF4LZXMB4MFPGZ6JL4ZNN" # 测试，使用固定值
-    """
-    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("\n\n\n**************************\n\nmessage in...")
-            # 此处后续优化，先用str.decode()变成json字符串、再用json.loads()转换为json对象
-            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:
-                    reply_string = generate_reply_string(data_dict['msg_id'], tenant_access_token, data_dict['content']['body'], data_dict['content']['type'])
-                    # 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")
-                        reply_content = reply_body.get("content")
-                        reply_content_string = json.dumps(reply_content)
-                        reply_body={
-                            "content":reply_content_string,
-                            "msg_type":reply_body.get("msg_type")
-                        }
-                        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)}

feishu_service.py

@@ -0,0 +1,76 @@
+import os
+import json
+import httpx
+from dotenv import load_dotenv
+
+load_dotenv()
+
+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
+
+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
+
+async def send_feishu_reply(msg_id: str, tenant_access_token: str, content: dict, msg_type: str):
+    """
+    Sends a reply message to Feishu API.
+
+    Args:
+        msg_id: The message ID to reply to.
+        tenant_access_token: The tenant access token.
+        content: The content of the reply message (dictionary).
+        msg_type: The type of the message (e.g., 'text').
+    """
+    url = f"https://open.feishu.cn/open-apis/im/v1/messages/{msg_id}/reply"
+    headers = {
+        "Authorization": f"Bearer {tenant_access_token}",
+        "Content-Type": "application/json"
+    }
+    # The content needs to be a JSON string within the 'content' field of the body
+    body = {
+        "content": json.dumps(content),
+        "msg_type": msg_type
+    }
+
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.post(url, headers=headers, json=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 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}")

main.py

@@ -0,0 +1,114 @@
+from fastapi import FastAPI, Request
+import time
+import asyncio
+import json
+from dotenv import load_dotenv
+
+# Import functions from the new service modules
+from redis_service import connect_redis, redis_subscriber, r as redis_client, publish_message as publish_to_redis
+from feishu_service import get_valid_tenant_access_token, send_feishu_reply
+
+load_dotenv()
+
+# Replace 'my_feishu_channel' with the actual channel name you want to subscribe to
+REDIS_CHANNEL_NAME = 'my_feishu_channel'
+
+app = FastAPI()
+
+# Startup event to connect to Redis and start the subscriber task
+@app.on_event("startup")
+async def startup_event():
+    # Connect to Redis
+    await connect_redis()
+    # Start the Redis subscriber as a background task
+    asyncio.create_task(redis_subscriber(redis_client, REDIS_CHANNEL_NAME))
+    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
+        print("Received URL verification request.")
+        return {"challenge": data.get('challenge')}
+
+    # Process other event types
+    # The full event data is under data['event']
+    event = data.get('event', {})
+    message = event.get('message', {})
+    chat_type = message.get('chat_type')
+
+    message_type = 'unknown'
+    if chat_type == 'p2p':
+        message_type = 'private'
+    elif chat_type == 'group':
+        if message.get('mentions'):
+            message_type = 'group_mention'
+        else:
+            message_type = 'group_all'
+
+    # Extract content safely
+    content_str = message.get('content')
+    content_dict = {}
+    if content_str:
+        try:
+            content_dict = json.loads(content_str)
+        except json.JSONDecodeError:
+            print(f"Warning: Could not parse message content as JSON: {content_str}")
+            # Handle cases where content is not valid JSON, maybe store as plain text
+            content_dict = {"text": content_str} # Store raw content as text
+
+    ret_data = {
+        'bot_id': data.get('header', {}).get('app_id'),
+        'message_type': message_type, # This is chat type (private/group)
+        'original_msg_type': message.get('message_type'), # Add original message type (text/image etc.)
+        'msg_id': message.get('message_id'),
+        "content": content_dict, # Use the parsed content dictionary
+        "sender": {
+            "id": event.get('sender', {}).get('sender_id', {}).get('open_id'),
+            "sender_type": event.get('sender', {}).get('sender_type'),
+            "name": "", # Name is not directly available in this payload structure
+            "is_human": None # Not directly available
+        },
+        "timestamp": time.time(),
+        "platform_specific": {
+            "app_id": data.get('header', {}).get('app_id')
+        }
+    }
+
+    # Publish the processed message data to Redis
+    await publish_to_redis(REDIS_CHANNEL_NAME, json.dumps(ret_data))
+
+    print('\n\n\nret_data: ', json.dumps(ret_data, indent=2)) # Use indent for readability
+    return {'status': 'OK'}
+
+# The /publish endpoint is now handled by the publish_message function in redis_service.py
+# We can expose it here if needed, or remove it if it's only for internal use.
+# For now, let's keep it as an example of calling the service function from a route.
+@app.post("/publish")
+async def publish_message_route(request: Request):
+    """
+    Publishes a message to a Redis channel via the redis_service.
+    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"}
+
+        # Call the publish function from redis_service
+        result = await publish_to_redis(channel, message)
+        return result
+
+    except Exception as e:
+        return {"status": "error", "message": str(e)}

redis_service.py

@@ -0,0 +1,105 @@
+import os
+import asyncio
+import redis.asyncio as redis
+from dotenv import load_dotenv
+from redis.exceptions import ConnectionError
+import json
+from feishu_service import get_valid_tenant_access_token, send_feishu_reply # Import Feishu service functions
+
+load_dotenv()
+
+redis_host = os.getenv('REDIS_HOST')
+redis_password = os.getenv('REDIS_PASSWORD')
+redis_port = os.getenv('REDIS_PORT', 6379)
+
+# Use async Redis client
+r = redis.Redis(
+  host=redis_host,
+  port=int(redis_port),
+  password=redis_password,
+  ssl=True
+)
+
+async def connect_redis():
+    """
+    Connects to Redis and performs a ping test.
+    """
+    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
+        raise # Re-raise the exception to indicate connection failure
+
+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:
+        message = await pubsub.get_message(ignore_subscribe_messages=True)
+        if message:
+            print("\n\n\n**************************\n\nmessage 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 using the Feishu service function
+                tenant_access_token = await get_valid_tenant_access_token()
+                if tenant_access_token:
+                    # Extract necessary info from data_dict for sending reply
+                    msg_id = data_dict.get('msg_id')
+                    content = data_dict.get('content') # This should be the dictionary like {"text": "..."}
+                    # Correctly extract msg_type from the original_msg_type field
+                    msg_type = data_dict.get('original_msg_type')
+
+                    if msg_id and content and msg_type:
+                         # Send the reply using the Feishu service function
+                        await send_feishu_reply(msg_id, tenant_access_token, content, msg_type)
+                    else:
+                        print("Missing msg_id, content, or msg_type in received data, cannot send reply.")
+
+                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
+            except Exception as e:
+                print(f"An unexpected error occurred while processing message: {e}")
+
+
+        await asyncio.sleep(0.01) # Prevent blocking the event loop
+
+async def publish_message(channel: str, message: str):
+    """
+    Publishes a message to a Redis channel.
+    """
+    try:
+        # Publish the message to the specified channel (publish is asynchronous)
+        await r.publish(channel, message)
+        print(f"Published message to channel '{channel}'")
+        return {"status": "success", "message": f"Published message to channel '{channel}'"}
+
+    except Exception as e:
+        print(f"Error publishing message: {e}")
+        return {"status": "error", "message": str(e)}