douyin/api/webhook.py

import os
import sys
sys.path.append(os.path.dirname(os.path.dirname(__file__)))
import asyncio
import json
import hashlib  
import hmac  
from typing import Optional,Any
from fastapi import FastAPI,APIRouter,  File, HTTPException, Depends, Request,Header, UploadFile  
from fastapi.responses import JSONResponse 
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials  
from config import logger,CLIENT_KEY,CLIENT_SECRET
from grpc_m.send_data_to_vector import langchain_chat
from api.comment import item_comment_reply

webhook_route = APIRouter()  

def verify_signature(body: bytes, signature: bytes, secret: bytes) -> bool:  
    combined = secret + body  # 直接拼接字节
    computed_signature = hashlib.sha1(combined).hexdigest()  # 计算签名并转为十六进制字符串
    logger.debug(f"computed_signature {computed_signature}")
    return hmac.compare_digest(computed_signature.encode('utf-8'), signature)
  
async def get_verified_data(request: Request) -> Any:  
    data = await request.json()  
    signature = request.headers.get('X-Douyin-Signature')  
      
    if not signature:  
        raise HTTPException(status_code=400, detail="Signature header is missing")  
  
    if isinstance(signature, str):  
        signature = signature.encode('utf-8')  
  
    body_bytes = await request.body()  
  
    logger.debug(f"body_bytes {type(body_bytes)} {body_bytes}")
    logger.debug(f"signature {type(signature)} {signature}")
    if not verify_signature(body_bytes, signature, CLIENT_SECRET.encode('utf-8')):  
        raise HTTPException(status_code=403, detail="Invalid signature")  
  
    return data  

@webhook_route.post("/webhook")  
async def webhook(verified_data: Any = Depends(get_verified_data)): 
    event = verified_data.get("event")  
  
    if event == "verify_webhook":  
        challenge = verified_data.get("content").get("challenge")  
        return JSONResponse(content={"challenge": challenge})  
    await handle_event(verified_data)
    return JSONResponse(status_code=200, content={"message": "success"})  

# https://developer.open-douyin.com/docs/resource/zh-CN/dop/develop/openapi/interaction-management/comment-management-user/accept-comment-reply-event
async def handle_event(data:dict):
    # 处理其他类型的消息  
    event = data.get("event")  
    if event == "item_comment_reply":  
        asyncio.create_task(item_comment_reply(data))
  

def main():
    body = b'{"event":"item_comment_reply","client_key":"aw6aipmfdtplwtyq","from_user_id":"_000QadFMhmU1jNCI3JdPnyVDL6XavC70dFy","to_user_id":"_000LiV_o0FGKMwaZgqGMCvcVSf-UAnNU_kk","content":"{\\"at_user_id\\":\\"\\",\\"avatar\\":\\"https://p3.douyinpic.com/aweme/720x720/aweme-avatar/tos-cn-i-0813_66c4e34ae8834399bbf967c3d3c919db.jpeg?from=3782654143\\",\\"comment_id\\":\\"@9VxS1/qCUc80K2etd8wkUc791mbgP/iCPZZwrQmjLFUVafP660zdRmYqig357zEBbPDOginb1DAizzy/rp2MrA==\\",\\"comment_user_id\\":\\"_000QadFMhmU1jNCI3JdPnyVDL6XavC70dFy\\",\\"content\\":\\"\xe5\x95\x8a\xe5\x95\x8a\\",\\"create_time\\":1708802048,\\"digg_count\\":0,\\"level\\":1,\\"nick_name\\":\\"\xe7\xa8\x8b\xe5\xba\x8f\xe5\x91\x98\xe9\xa9\xac\xe5\xb7\xa5\\",\\"parent_id\\":\\"7332749358502333711\\",\\"reply_comment_total\\":0,\\"reply_to_comment_id\\":\\"0\\",\\"reply_to_item_id\\":\\"@9VxS1/qCUc80K2etd8wkUc791mbrOvyLOJB5rAqkLFIQafD960zdRmYqig357zEBrOff1pAnFpeMgJjQVMqang==\\"}","log_id":"021708802052557fdbdfdbdfdbdfdbd00000000000000ab9c49ee"}'
    body_json_dict  ={'event': 'item_comment_reply', 'client_key': 'aw6aipmfdtplwtyq', 'from_user_id': '_000QadFMhmU1jNCI3JdPnyVDL6XavC70dFy', 'to_user_id': '_000LiV_o0FGKMwaZgqGMCvcVSf-UAnNU_kk', 'content': '{"at_user_id":"","avatar":"https://p3.douyinpic.com/aweme/720x720/aweme-avatar/tos-cn-i-0813_66c4e34ae8834399bbf967c3d3c919db.jpeg?from=3782654143","comment_id":"@9VxS1/qCUc80K2etd8wkUc791mbgOfqKOZZ2rwiiL1QTavn+60zdRmYqig357zEBuXn2PviubsArHSFhW7SksQ==","comment_user_id":"_000QadFMhmU1jNCI3JdPnyVDL6XavC70dFy","content":"123","create_time":1708849799,"digg_count":0,"level":1,"nick_name":"程序员马工","parent_id":"7332749358502333711","reply_comment_total":0,"reply_to_comment_id":"0","reply_to_item_id":"@9VxS1/qCUc80K2etd8wkUc791mbrOvyLOJB5rAqkLFIQafD960zdRmYqig357zEBrOff1pAnFpeMgJjQVMqang=="}', 'log_id': '021708849808460fdbdfdbdfdbdfdbd00000000000000ab5553a6'}
    body_json_str = json.dumps(body_json_dict)  
    body_json_bytes = body_json_str.encode('utf-8')  
    body_str = body.decode('utf-8').rstrip('\n')  
    body_bytes = body_str.encode('utf-8')
    signature = b'b3513d6b1e5c80db0f82d88892e68826063f3b18'
    
    v = verify_signature(body_json_str, signature, CLIENT_SECRET.encode())
    logger.info(f"{v}")

if __name__ == "__main__":
    main()