main.py

import json, os
import gradio as gr
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():
    def get_gemini_response(input):
        response = model.generate_content(input)
        return response.text
    # Define the Gradio interface
    iface = gr.Interface(
        fn=get_gemini_response,
        inputs=[
            gr.Textbox(label="Enter your question here:"),
        ],
        outputs=gr.Textbox(label="The response is"),
        title="Gemini Bot",
        description="Ask Gemini!!!",
    )
    # Launch the Gradio interface
    iface.launch()

    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)