Spaces:
Running
Running
| import json | |
| import base64 | |
| import urllib.request | |
| from http.server import BaseHTTPRequestHandler, HTTPServer | |
| # --- 配置区 (建议在 HF Space 的 Settings -> Variables 中设置) --- | |
| FEISHU_APP_ID = "你的AppID" | |
| FEISHU_APP_SECRET = "你的AppSecret" | |
| HF_TOKEN = "你的HuggingFaceToken" | |
| HF_MODEL_URL = "https://api-inference.huggingface.co/models/Salesforce/blip-image-captioning-base" | |
| def get_feishu_token(): | |
| url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal" | |
| data = json.dumps({"app_id": FEISHU_APP_ID, "app_secret": FEISHU_APP_SECRET}).encode() | |
| req = urllib.request.Request(url, data=data, method="POST") | |
| with urllib.request.urlopen(req) as f: | |
| return json.loads(f.read()).get("tenant_access_token") | |
| class FeishuHandler(BaseHTTPRequestHandler): | |
| def do_POST(self): | |
| content_length = int(self.headers['Content-Length']) | |
| post_data = self.rfile.read(content_length) | |
| data = json.loads(post_data.decode()) | |
| # 1. 处理飞书 URL 验证 | |
| if "challenge" in data: | |
| self.send_response(200) | |
| self.end_headers() | |
| self.wfile.write(json.dumps({"challenge": data["challenge"]}).encode()) | |
| return | |
| # 2. 处理图片消息 | |
| event = data.get("event", {}) | |
| msg = event.get("message", {}) | |
| if msg.get("msg_type") == "image": | |
| msg_id = msg["message_id"] | |
| img_key = json.loads(msg["content"])["image_key"] | |
| open_id = event["sender"]["sender_id"]["open_id"] | |
| # 开启后台处理，防止 Webhook 超时 | |
| self.process_task(msg_id, img_key, open_id) | |
| self.send_response(200) | |
| self.end_headers() | |
| def process_task(self, msg_id, img_key, open_id): | |
| token = get_feishu_token() | |
| # A. 内存中获取图片 | |
| img_url = f"https://open.feishu.cn/open-apis/im/v1/messages/{msg_id}/resources/{img_key}?type=image" | |
| img_req = urllib.request.Request(img_url, headers={"Authorization": f"Bearer {token}"}) | |
| with urllib.request.urlopen(img_req) as f: | |
| img_bin = f.read() # 图片数据直接在内存变量里 | |
| # B. 调用 HF 模型进行识别 | |
| hf_req = urllib.request.Request(HF_MODEL_URL, data=img_bin, headers={"Authorization": f"Bearer {HF_TOKEN}"}) | |
| with urllib.request.urlopen(hf_req) as f: | |
| desc = json.loads(f.read())[0].get("generated_text", "识别失败") | |
| # C. 回复飞书 | |
| reply_url = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=open_id" | |
| reply_data = json.dumps({ | |
| "receive_id": open_id, | |
| "msg_type": "text", | |
| "content": json.dumps({"text": f"🤖 AI 识别结果：\n{desc}"}) | |
| }).encode() | |
| urllib.request.urlopen(urllib.request.Request(reply_url, data=reply_data, | |
| headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"})) | |
| # 启动服务 | |
| if __name__ == "__main__": | |
| # Hugging Face 默认监听 7860 端口 | |
| server = HTTPServer(('0.0.0.0', 7860), FeishuHandler) | |
| print("Serving on port 7860...") | |
| server.serve_forever() |