Spaces:
Running
Running
from fastapi import FastAPI, Request, BackgroundTasks, Header | |
from linebot import LineBotApi, WebhookHandler | |
from linebot.exceptions import InvalidSignatureError | |
from linebot.models import MessageEvent, TextMessage | |
from starlette.exceptions import HTTPException | |
import os | |
app = FastAPI() | |
# line_bot_api = LineBotApi(os.environ["LINE_CHANNEL_ACCESS_TOKEN"]) | |
# handler = WebhookHandler(os.environ["LINE_CHANNEL_SECRET"]) | |
line_bot_api = LineBotApi("ZwnEWwK46AUmRDOFXJOwUh1TqAELahP+kxHY0iBevF8JiEUe0CBQpkW2Gpo3C38LOaIX7hpW4IuTOSOjVof+DX9XgZg8MTeh5Iz+NDLlw6KKMtqNsyR6zt9CMnsso21ieifC/S8dcgXx3iKlhdS7UAdB04t89/1O/w1cDnyilFU=") | |
handler = WebhookHandler("7b9984bbc5c62445a79b95b15e34b727") | |
def root(): | |
return {"title": "Echo Bot"} | |
async def callback( | |
request: Request, | |
background_tasks: BackgroundTasks, | |
x_line_signature=Header(None), | |
): | |
body = await request.body() | |
try: | |
background_tasks.add_task( | |
handler.handle, body.decode("utf-8"), x_line_signature | |
) | |
except InvalidSignatureError: | |
raise HTTPException(status_code=400, detail="Invalid signature") | |
return "ok" | |
def handle_message(event): | |
if event.type != "message" or event.message.type != "text": | |
return | |
message = TextMessage(text=event.message.text) | |
line_bot_api.reply_message(event.reply_token, message) |