Main.py

# 1.引用套件
from flask import Flask, request, abort

from linebot import (
    LineBotApi, WebhookHandler
)
from linebot.exceptions import (
    InvalidSignatureError
)
from linebot.models import (
    MessageEvent, 
    TextMessage, 
    TextSendMessage, 
    AudioMessage, 
    ImageSendMessage,
    FollowEvent,
    PostbackEvent
)

import os.path
from pydub import AudioSegment
import tempfile
import Model.LineTemplate.RichMenuModel as RichMenuModel
import Model.LineTemplate.ButtonsTemplateModel as ButtonsTemplate
import Model.Signal.STFTModel as STFTModel
import Model.Signal.FFTModel as FFTModel
import Model.Imgur.ImgurModel as ImgurModel


# 2. 準備素材
app = Flask(__name__)
# line_bot_api = LineBotApi('EsyietS+/UNPYM9IpbOinAcgjH89znR+LK7dW0oxtSMqWSgxXAR64Yf5IYl7E88VrZ+2bI8OzXLiP3aiILi6sO8I3HVFHk3p9L6PwhxNWB05uB1ChzWayEnmMCtFQOTdWNhkmVowKedvygQUojZdJQdB04t89/1O/w1cDnyilFU=')
# handler = WebhookHandler('7b4a42778352e75cf922e71b71592ce3')
line_bot_api = LineBotApi('H9krsUUv9jc1bTl6gAZKEITBD2WT6f/kj6paZUvdm4DjZtlnge/rZiHJtBE6TFcFbNWShqvgnEm2nwvPAGlp0iNfeErDBEfcBqFs5vgSks2+TmTYftWRDU+3SGFTBSVNL0qkO2GFRFRIPfKAijYJuwdB04t89/1O/w1cDnyilFU=')
handler = WebhookHandler('b7b7fff7548aec0262044eb91ab78cf8')

global  TagsType, SetFs, SecWindow, SetSec, SetOverlap
TagsType = 0
SetFs=False    
SecWindow=False  
SetSec=False  
SetOverlap=False



@app.route("/callback", methods=['POST'])
def callback():
    # get X-Line-Signature header value
    signature = request.headers['X-Line-Signature']
    # get request body as text
    body = request.get_data(as_text=True)
    # print("Request body: " + body)
    # print("Signature: " + signature)

    # handle webhook body
    try:
        handler.handle(body, signature)

    except InvalidSignatureError:
        print("Invalid signature. Please check your channel access token/channel secret.")
        abort(400)
    return 'OK'


 

@handler.add(FollowEvent)
def handle_follow(event):
    try:
        # user_profile
        user_profile = line_bot_api.get_profile(event.source.user_id)
        print(user_profile)
        print(user_profile.user_id)
        print(user_profile.picture_url)

        # event 指用戶傳來的消息
        # 請Line_Bot_api 回復用戶，我們剛傳過來的文字
        line_bot_api.reply_message(
            event.reply_token,
            TextSendMessage(text="給我一段音訊，我來為您分析！"))
        
        # RichMenuModel
        rich = RichMenuModel.RichMenuFunction()
        rich.SetRichMenu(
            'LineTemplate/RichMenuBase.json', 
            'LineTemplate/myRich.png',
            'png',
            line_bot_api, 
            user_profile.user_id)
    except Exception as e:
        from traceback import format_exc
        print(format_exc())

@handler.add(PostbackEvent)
def handle_postback(event):
    global  TagsType, SetFs, SecWindow, SetSec, SetOverlap
    try:
        if event.postback.data == "TagsType=1":         
            TagsType = 1
     
        elif event.postback.data == "TagsType=2":        
            TagsType = 2
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text="請輸入自定義標籤名稱"))

        elif event.postback.data == "SetFs=True":      
            SetFs=True
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text="請輸入採樣頻率(Hz)"))

        elif event.postback.data == "SecWindow=True":      
            SecWindow=True
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text="請輸入欲使用之窗函數"))

        elif event.postback.data == "SetSec=True":      
            SetSec=True
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text="請輸入取樣秒數(Sec)"))

        elif event.postback.data == "SetOverlap=True":      
            SetOverlap=True
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text="請輸入Overlap百分比(%)"))
                
    except Exception as e:
        from traceback import format_exc
        print(format_exc())


@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
    try:
        global TagsType, SetFs, SecWindow, SetSec, SetOverlap

        # 全域變數控制處理
        if TagsType == 1:
            TagsType = 0
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text=event.message.text))
        if TagsType == 2:
            TagsType = 0
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text=f"這是[{event.message.text}]"))     
        elif SetFs==True:      
            SetFs = False
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text=f"採樣頻率(Hz)設定為:{event.message.text}(Hz)"))
        elif SecWindow==True:      
            SecWindow = False
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text=f"窗函數設定為:{event.message.text}"))
        elif SetSec==True:      
            SetSec = False
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text=f"取樣秒數(Sec)設定為:{event.message.text}(Sec)"))
        elif SetOverlap==True:      
            SetOverlap = False
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text=f"Overlap百分比(%)設定為:{event.message.text}(%)"))
        
        # 一般文字處理
        else:
            if event.message.text == "音訊處理":
                line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text="請傳送一段語音訊息"))  

            elif event.message.text == "進階分析":
                line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text="敬請期待XD"))   

            elif event.message.text == "參數設定":
                #buttonTemp
                buttonTemp = ButtonsTemplate.ButtonsTemplateFunctino()
                template_message_from_json = buttonTemp.SetTemplate('LineTemplate/DspParamSet.json')
                line_bot_api.reply_message(
                    event.reply_token,
                    template_message_from_json)        
            else:    
                # event 指用戶傳來的消息
                # 請Line_Bot_api 回復用戶，我們剛傳過來的文字
                line_bot_api.reply_message(
                    event.reply_token,
                    TextSendMessage(text=event.message.text))

    except Exception as e:
        from traceback import format_exc
        print(format_exc())

@handler.add(MessageEvent, message=AudioMessage)
def handle_aud(event): 
    message_content = line_bot_api.get_message_content(event.message.id)
    ext = 'mp3'

    tempFilePath = None
    try:

        # 保存音訊(tempfile)
        with tempfile.NamedTemporaryFile(prefix=ext + '-', delete=False) as tf:
            for chunk in message_content.iter_content():
                tf.write(chunk)
        tempFilePath = tf.name
 

        # 存起來(openFile)
        # 判斷資料夾是否存在 不存在就創建
        # tempFilePath = f'./File/Audio/{event.message.id}.mp3'
        # with open(tempFilePath, 'wb') as fd:
        #     for chunk in message_content.iter_content():
        #         fd.write(chunk)


        #轉檔
        # AudioSegment.converter = '/app/vendor/ffmpeg/ffmpeg' 
        AudioSegment.converter ='./ffmpeg.exe'
        sound = AudioSegment.from_file_using_temporary_files(tempFilePath)
        tempFilePath = os.path.splitext(tempFilePath)[0]+'.wav'
        sound.export(tempFilePath, format="wav")

        #DSP
        DspStft = FFTModel.processing()
        FftPath= DspStft.FFT(tempFilePath, True, True)
        fftImgUrl = ImgurModel.upload_image(FftPath)
        image_send_message_fft = ImageSendMessage(
            original_content_url=fftImgUrl,
            preview_image_url=fftImgUrl
        )

        # STFT
        # DspFft = STFTModel.processing()
        # StftPath = DspFft.STFT(tempFilePath, True, True)
        # stftImgUrl = ImgurModel.upload_image(StftPath)
        # image_send_message_stft = ImageSendMessage(
        #     original_content_url=stftImgUrl,
        #     preview_image_url=stftImgUrl
        # )

        #buttonTemp
        buttonTemp = ButtonsTemplate.ButtonsTemplateFunctino()
        template_message_from_json = buttonTemp.SetTemplate('LineTemplate/TegsTemplate.json')


        # 收到檔案
        textSendMessageStartProcessing = TextSendMessage("已收到檔案，正在分析中。")

        # 麻煩LineBotApi 把文字消息交給LINE
        # line_bot_api.reply_message(event.reply_token, follow_text_send_message)
        line_bot_api.reply_message(
            event.reply_token,
            [textSendMessageStartProcessing, image_send_message_fft, template_message_from_json])
            # [textSendMessageStartProcessing, image_send_message_fft, image_send_message_stft])

        
    except Exception as e:
        from traceback import format_exc
        ErrStr = format_exc()
        print(ErrStr)
        print(f"FilePath:{tempFilePath}")
        # line_bot_api.reply_message(event.reply_token, TextSendMessage(text=ErrStr))
    finally:
        return





# 本地端測試用
# 伺服器啟動
if __name__ == "__main__":
    app.run(host="0.0.0.0")


# heroku專用，偵測heroku給我的port
# import os
# if __name__ == "__main__":
#     app.run(host='0.0.0.0', port=os.environ['PORT'])
    # InitGlobalVar()