|
import json, os |
|
from fastapi.middleware.cors import CORSMiddleware |
|
from fastapi import FastAPI, Request, Header, BackgroundTasks, HTTPException, status |
|
import google.generativeai as genai |
|
|
|
from linebot import LineBotApi, WebhookHandler |
|
from linebot.exceptions import InvalidSignatureError |
|
from linebot.models import MessageEvent, TextMessage, TextSendMessage, ImageSendMessage, AudioMessage |
|
|
|
genai.configure(api_key=os.environ["GOOGLE_API_KEY"]) |
|
generation_config = genai.types.GenerationConfig(max_output_tokens=2048, temperature=0.2, top_p=0.5, top_k=16) |
|
model = genai.GenerativeModel('gemini-1.5-flash') |
|
|
|
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" |
|
|
|
app = FastAPI() |
|
app.add_middleware( |
|
CORSMiddleware, |
|
allow_origins=["*"], |
|
allow_credentials=True, |
|
allow_methods=["*"], |
|
allow_headers=["*"], |
|
) |
|
|
|
@app.get("/") |
|
def root(): |
|
return {"title": "Line Bot"} |
|
|
|
@app.post("/webhook") |
|
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" |
|
|
|
@line_handler.add(MessageEvent, message=TextMessage) |
|
def handle_message(event): |
|
global working_status |
|
|
|
if event.type != "message" or event.message.type != "text": |
|
TextSendMessage(text="Event type error:[No message or the message does not contain text]") |
|
|
|
elif event.message.text == "再見": |
|
working_status = True |
|
line_bot_api.reply_message( |
|
event.reply_token, |
|
TextSendMessage(text="Bye!")) |
|
return |
|
|
|
elif working_status: |
|
try: |
|
prompt = event.message.text |
|
completion = model.generate_content(prompt, generation_config=generation_config) |
|
if (completion.parts[0].text != None): |
|
out = completion.parts[0].text |
|
else: |
|
out = "Gemini沒答案!請換個說法!" |
|
except: |
|
out = "Gemini執行出錯!請換個說法!" |
|
|
|
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) |