|
from fastapi import FastAPI |
|
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["ZwnEWwK46AUmRDOFXJOwUh1TqAELahP+kxHY0iBevF8JiEUe0CBQpkW2Gpo3C38LOaIX7hpW4IuTOSOjVof+DX9XgZg8MTeh5Iz+NDLlw6KKMtqNsyR6zt9CMnsso21ieifC/S8dcgXx3iKlhdS7UAdB04t89/1O/w1cDnyilFU="]) |
|
handler = WebhookHandler(os.environ["7b9984bbc5c62445a79b95b15e34b727"]) |
|
|
|
@app.get("/") |
|
def root(): |
|
return {"title": "Echo Bot"} |
|
|
|
@app.post("/callback") |
|
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" |
|
|
|
@handler.add(MessageEvent) |
|
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) |
