Spaces:
Running
Running
from fastapi.middleware.cors import CORSMiddleware | |
from fastapi import FastAPI, Request, Header, BackgroundTasks, HTTPException, status | |
import json | |
import os | |
from linebot import ( | |
LineBotApi, WebhookHandler | |
) | |
from linebot.exceptions import ( | |
InvalidSignatureError | |
) | |
from linebot.models import ( | |
MessageEvent, TextMessage, TextSendMessage, ImageSendMessage, AudioMessage | |
) | |
app = FastAPI() | |
app.add_middleware( | |
CORSMiddleware, | |
allow_origins=["*"], | |
allow_credentials=True, | |
allow_methods=["*"], | |
allow_headers=["*"], | |
) | |
line_bot_api = LineBotApi(os.environ["CHANNEL_ACCESS_TOKEN"]) | |
line_handler = WebhookHandler(os.environ["CHANNEL_SECRET"]) | |
working_status = os.getenv("DEFALUT_TALKING", default = "true").lower() == "true" | |
def root(): | |
return {"title": "Echo Bot"} | |
async def webhook( | |
request: Request, | |
background_tasks: BackgroundTasks, | |
x_line_signature=Header(None), | |
): | |
body = await request.body() | |
try: | |
background_tasks.add_task( | |
line_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): | |
global working_status | |
if event.message.text == "再見": | |
working_status = True | |
line_bot_api.reply_message( | |
event.reply_token, | |
TextSendMessage(text="Bye!")) | |
return | |
if working_status: | |
out = event.message.text | |
if not out: | |
out = "不懂您的意思!" | |
line_bot_api.reply_message( | |
event.reply_token, | |
TextSendMessage(text=out)) | |
if __name__ == "__main__": | |
uvicorn.run("main:app", host="0.0.0.0", port=7860, reload=True) |