iThome 鐵人賽 Azure 上

sudo apt-get update
sudo apt-get install ca-certificates \
curl apt-transport-https lsb-release gnupg

curl -sL https://packages.microsoft.com/keys/microsoft.asc |
    gpg --dearmor |
    sudo tee /etc/apt/trusted.gpg.d/microsoft.gpg > /dev/null

AZ_REPO=$(lsb_release -cs)
echo "deb [arch=amd64] https://packages.microsoft.com/repos/azure-cli/ $AZ_REPO main" |
    sudo tee /etc/apt/sources.list.d/azure-cli.list

sudo apt-get update
sudo apt-get install azure-cli

brew update && brew install azure-cli

az login

sudo apt-get update
sudo apt-get install python3.7

sudo apt install python3-pip

alias pip3.7='python3.7 -m pip'

pip3.7 install jupyter
sudo apt install jupyter-notebook
pip3.7 install ipykernel --upgrade
python3.7 -m ipykernel install --user

jupyter notebook --ip 0.0.0.0

sudo apt-get update
sudo apt-get install python3.7

sudo apt install python3-pip

alias pip3.7='python3.7 -m pip'

pip3.7 install jupyter
sudo apt install jupyter-notebook
pip3.7 install ipykernel --upgrade
python3.7 -m ipykernel install --user

jupyter notebook --ip 0.0.0.0

# 在終端機中輸入
az webapp deployment user set \
--user-name <username> --password <password>

# 取得在Azure Web App內的git url
az webapp deployment source config-local-git \
--name <webappname> --resource-group <yourResourceGroup>

# 在終端機中輸入
az webapp deployment source config-local-git \
--name <webappname> --resource-group <yourResourceGroup>

# 得到 URL
{
  "url": "https://<username>@<webappname>.scm.azurewebsites.net/linecv.git"
}

git remote add azure <your_git_url>

git push azure master

az webapp log tail \
--name <webappname> --resource-group <yourResourceGroup>

az webapp create-remote-connection \
-n <你的Web App名稱> --resource-group <你的資源群組> &

ssh root@127.0.0.1 -p 57281

from flask import Flask
app = Flask(__name__)

@app.route("/")
def hello():
    "hello world"
    return "Hello World!!!!!"

Flask

pip3 install line-bot-sdk

from flask import Flask, request, abort
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
app = Flask(__name__)

LINE_SECRET = "YOUR line secret"]
LINE_TOKEN = "YOU line_token"
LINE_BOT = LineBotApi(LINE_TOKEN)
HANDLER = WebhookHandler(LINE_SECRET)


@app.route("/callback", methods=["POST"])
def callback():
    # X-Line-Signature: 數位簽章
    signature = request.headers["X-Line-Signature"]
    print(signature)
    body = request.get_data(as_text=True)
    print(body)
    try:
        HANDLER.handle(body, signature)
    except InvalidSignatureError:
        print("Check the channel secret/access token.")
        abort(400)
    return "OK"

from linebot.models import (
    MessageEvent,
    TextMessage,
    TextSendMessage,
)

# message 可以針對收到的訊息種類
@HANDLER.add(MessageEvent, message=TextMessage)
def handle_message(event):
    url_dict = {
      "ITHOME":"https://www.ithome.com.tw/", 
      "HELP":"https://developers.line.biz/zh-hant/docs/messaging-api/"}
# 將要發出去的文字變成TextSendMessage
    try:
        url = url_dict[event.message.text.upper()]
        message = TextSendMessage(text=url)
    except:
        message = TextSendMessage(text=event.message.text)
# 回覆訊息
    LINE_BOT.reply_message(event.reply_token, message)

with open("templates/detect_result.json", "r") as f_r:
   bubble = json.load(f_r)
f_r.close()
# 依情況更動 components
bubble["body"]["contents"][0]["contents"][0]["text"] = output
bubble["header"]["contents"][0]["contents"][0]["url"] = link
LINE_BOT.reply_message(
   event.reply_token, 
   [
     FlexSendMessage(alt_text="Report", contents=bubble)
     ]
)

pip3 install azure-cognitiveservices-vision-computervision

"""
Azure object detection
"""
import os
from io import BytesIO
import requests
from PIL import Image, ImageDraw, ImageFont
from azure.cognitiveservices.vision.computervision import ComputerVisionClient
from msrest.authentication import CognitiveServicesCredentials

# 匯入必要套件，主要都是跟讀檔、繪圖和 Azure 的相關套件 

# 一開始除了匯入套件以外，還需要利用金鑰SUBSCRIPTION_KEY和端點ENDPOINT，取得使用電腦視覺服務的權限。

SUBSCRIPTION_KEY = os.getenv("SUBSCRIPTION_KEY")
ENDPOINT = os.getenv("ENDPOINT")
CV_CLIENT = ComputerVisionClient(
    ENDPOINT, CognitiveServicesCredentials(SUBSCRIPTION_KEY)
)


def main():
    """
    Azure object detection
    """
    
    # 透過圖片的URL取得圖片
    url = "https://i.imgur.com/Js5H6Qa.jpg"
    response = requests.get(url)
    img = Image.open(BytesIO(response.content))
    
    # 開始設定繪圖相關的部分，由於會需要在圖片上寫字，需要準備字型檔
    draw = ImageDraw.Draw(img)
    font_size = int(5e-2 * img.size[1])
    fnt = ImageFont.truetype("../static/TaipeiSansTCBeta-Regular.ttf", size=font_size)
    # 透過電腦視覺的功能取得物件，偵測的結果會包含匡出物體的左上角座標(x, y)，以及方匡的寬跟高(w, h)，過這四個值即可畫出方匡，並且標示辨識結果以及辨識的信心程度。
    object_detection = CV_CLIENT.detect_objects(url)
    if len(object_detection.objects) > 0:
        for obj in object_detection.objects:
            left = obj.rectangle.x
            top = obj.rectangle.y
            right = obj.rectangle.x + obj.rectangle.w
            bot = obj.rectangle.y + obj.rectangle.h
            name = obj.object_property
            confidence = obj.confidence
            print("{} at location {}, {}, {}, {}".format(name, left, right, top, bot))
            draw.rectangle([left, top, right, bot], outline=(255, 0, 0), width=3)
            draw.text(
                [left, top + font_size],
                "{0} {1:0.1f}".format(name, confidence * 100),
                fill=(255, 0, 0),
                font=fnt,
            )
    # 最後存檔
    img.save("output.png")
    print("Done!")
    print("Please check ouptut.png")


if __name__ == "__main__":
    main()

pip3 install azure-storage-blob

import os
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient
import json
import requests
from azure.storage.blob import AccessPolicy, ContainerSasPermissions


# 這邊就是填剛剛複製的連接字串
connect_str = 'Your connect string'

blob_service_client = BlobServiceClient.from_connection_string(connect_str)

# 自己命名 container 名稱，建立 container client
container_client = blob_service_client.create_container(<container name>)

# 設定 blob 讀寫權限，方便之後使用檔案的URL
access_policy = AccessPolicy(permission=ContainerSasPermissions(read=True, write=True))
identifiers = {'both': access_policy}
container_client.set_container_access_policy(identifiers, public_access='blob')

with open('your.jpg', "rb") as data:
    blob_client.upload_blob(data)
data.close()

# 印出檔案的 URL
print(blob_client.url)

# 套件：azure-cognitiveservices-vision-computervision
from azure.cognitiveservices.vision.computervision \
import ComputerVisionClient
from msrest.authentication import (
  CognitiveServicesCredentials
)
# 利用金鑰SUBSCRIPTION_KEY和端點ENDPOINT，取得使用電腦視覺服務的權限。
SUBSCRIPTION_KEY = "YOUR SUBSCRIPTION_KEY"
ENDPOINT = "YOUR ENDPOINT"
CV_CLIENT = ComputerVisionClient(
    ENDPOINT, CognitiveServicesCredentials(SUBSCRIPTION_KEY)
)

# 利用 describe_image 取得描述影像的句子與信心程度

description_results = CV_CLIENT.describe_image(url)
output = ""
for caption in description_results.captions:
    output += "'{}' with confidence {:.2f}% \n".format(
        caption.text, caption.confidence * 100
    )
print(output)

{
    "line": {
        "line_secret": "your line secret",
        "line_token": "your line token",
    },
    "azure": {
        "cv_key": "your subscription key of computer vision",
        "cv_end": "your endpoint of computer vision",
        "blob_connect": "your connect string",
        "blob_container": "your blob container name"
}

az webapp create-remote-connection \
-n linecv --resource-group Tibame &

scp -P <port> config.json ben@127.0.0.1:/home/config.json

Flask==1.0.2
line-bot-sdk
azure-cognitiveservices-vision-computervision
azure-storage-blob
Pillow

import os
import json

from flask import Flask, request, abort
from azure.cognitiveservices.vision.computervision import ComputerVisionClient
from azure.storage.blob import BlobServiceClient

from msrest.authentication import CognitiveServicesCredentials
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import (
    MessageEvent,
    FlexSendMessage,
    ImageMessage,
)

from PIL import Image, ImageDraw, ImageFont

APP = Flask(__name__)


CONFIG = json.load(open("/home/config.json", "r"))

SUBSCRIPTION_KEY = CONFIG["azure"]["cv_key"]
ENDPOINT = CONFIG["azure"]["cv_end"]
CV_CLIENT = ComputerVisionClient(
    ENDPOINT, CognitiveServicesCredentials(SUBSCRIPTION_KEY)
)

CONNECT_STR = CONFIG["azure"]["blob_connect"]
CONTAINER = CONFIG["azure"]["blob_container"]
BLOB_SERVICE = BlobServiceClient.from_connection_string(CONNECT_STR)


LINE_SECRET = CONFIG["line"]["line_secret"]
LINE_TOKEN = CONFIG["line"]["line_token"]
LINE_BOT = LineBotApi(LINE_TOKEN)
HANDLER = WebhookHandler(LINE_SECRET)




@APP.route("/")
def hello():
    "hello world"
    return "Hello World!!!!!"


# 為了上傳圖片到 Azure blob

def upload_blob(container, path):
    """
    Upload files to Azure blob
    """
    blob_client = BLOB_SERVICE.get_blob_client(container=container, blob=path)
    with open(path, "rb") as data:
        blob_client.upload_blob(data, overwrite=True)
    data.close()
    return blob_client.url

# 影像描述

def azure_describe(url):
    """
    Output azure image description result
    """
    description_results = CV_CLIENT.describe_image(url)
    output = ""
    for caption in description_results.captions:
        output += "'{}' with confidence {:.2f}% \n".format(
            caption.text, caption.confidence * 100
        )
    return output

# 物體偵測

def azure_object_detection(url, filename):
    """
    Azure object detection, and output images with bounding boxes
    """
    img = Image.open(filename)
    draw = ImageDraw.Draw(img)
    font_size = int(5e-2 * img.size[1])
    fnt = ImageFont.truetype("static/TaipeiSansTCBeta-Regular.ttf", size=font_size)
    object_detection = CV_CLIENT.detect_objects(url)
    if len(object_detection.objects) > 0:
        for obj in object_detection.objects:
            left = obj.rectangle.x
            top = obj.rectangle.y
            right = obj.rectangle.x + obj.rectangle.w
            bot = obj.rectangle.y + obj.rectangle.h
            name = obj.object_property
            confidence = obj.confidence
            print("{} at location {}, {}, {}, {}".format(name, left, right, top, bot))
            draw.rectangle([left, top, right, bot], outline=(255, 0, 0), width=3)
            draw.text(
                [left, top + font_size],
                "{} {}".format(name, confidence),
                fill=(255, 0, 0),
                font=fnt,
            )
    img.save(filename)
    # 把畫完方框的圖片傳至 blob
    link = upload_blob(CONTAINER, filename)
    # 為了避免一堆圖檔塞爆 Web App，用完就刪掉
    os.remove(filename)
    return link

# Callback for Line chatbot

@APP.route("/callback", methods=["POST"])
def callback():
    """
    LINE bot webhook callback
    """
    # get X-Line-Signature header value
    signature = request.headers["X-Line-Signature"]
    print(signature)
    body = request.get_data(as_text=True)
    print(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(MessageEvent, message=ImageMessage)
def handle_content_message(event):
    """
    Reply Image message with results of image description and objection detection
    """
    # event 是使用者與 Line 之間的互動事件，可以印出 event 的物件，觀察 event。
    # message ID 會作為後續圖片存檔的名稱， user ID 則是作為之後人臉登入的依據。
    print(event.message)
    print(event.source.user_id)
    print(event.message.id)

    # 讀取為了產生 flex message 的 json 檔
    with open("templates/detect_result.json", "r") as f_h:
        bubble = json.load(f_h)
    f_h.close()
    
    # 以 Line message ID 作為檔案名稱
    filename = "{}.jpg".format(event.message.id)
    
    # 圖片訊息以 binary 的形式傳輸，取得之後先存成圖檔
    message_content = LINE_BOT.get_message_content(event.message.id)
    with open(filename, "wb") as f_h:
        for chunk in message_content.iter_content():
            f_h.write(chunk)
    f_h.close()
    
    # 開啟檔案，為了取得影像指寸
    img = Image.open(filename)
    
    # 上傳圖片到 Blob
    link = upload_blob(CONTAINER, filename)
    
    # 執行物體偵測和影像描述
    link_ob = azure_object_detection(link, filename)
    output = azure_describe(link)
    link = link_ob

    # 以 flex message 輸出結果
    bubble["body"]["contents"][0]["text"] = output
    bubble["header"]["contents"][0]["url"] = link
    # 以影像原尺寸，作為 Flex Message 輸出的依據
    bubble["header"]["contents"][0]["aspectRatio"] = "{}:{}".format(
        img.size[0], img.size[1]
    )

    LINE_BOT.reply_message(
        event.reply_token, [FlexSendMessage(alt_text="Report", contents=bubble)]
    )

# 套件：azure-cognitiveservices-vision-computervision
from azure.cognitiveservices.vision.computervision \
import ComputerVisionClient
from msrest.authentication import (
  CognitiveServicesCredentials
)
from azure.cognitiveservices.vision.computervision.models \
import OperationStatusCodes
from io import BytesIO
import requests
from PIL import Image, ImageDraw, ImageFont

# 利用金鑰SUBSCRIPTION_KEY和端點ENDPOINT，取得使用電腦視覺服務的權限。

SUBSCRIPTION_KEY = "YOUR SUBSCRIPTION_KEY"
ENDPOINT = "YOUR ENDPOINT"
CV_CLIENT = ComputerVisionClient(
    ENDPOINT, CognitiveServicesCredentials(SUBSCRIPTION_KEY)
)

# 讀取 URL 得到圖片

url = "https://i.imgur.com/qyWiqQv.jpg"
response = requests.get(url)
img = Image.open(BytesIO(response.content))


draw = ImageDraw.Draw(img)
font_size = int(5e-2 * img.size[1])
fnt = ImageFont.truetype(
  "../static/TaipeiSansTCBeta-Regular.ttf",
  size=font_size)

# 開始利用 Azure 電腦視覺執行 OCR

ocr_results = CV_CLIENT.read(url, raw=True)
operation_location_remote = \
ocr_results.headers["Operation-Location"]
operation_id = operation_location_remote.split("/")[-1]

# 因為讀取文字有多有少，所以時間會不一，透過 operation_id 可以確認目前進度
status = ["notStarted", "running"]
while True:
    get_handw_text_results = \
    CV_CLIENT.get_read_result(operation_id)
    if get_handw_text_results.status not in status:
        break
    time.sleep(1)

# 當執行狀態 status 為 succeeded ，就可以把結果標示在原本的照片上了

succeeded = OperationStatusCodes.succeeded

if get_handw_text_results.status == succeeded:
    res = get_handw_text_results.analyze_result.read_results
    for text_result in res:
        for line in text_result.lines:
            bounding_box = line.bounding_box
            bounding_box += bounding_box[:2]
            draw.line(
                line.bounding_box, 
                fill=(255, 0, 0), 
                width=int(font_size / 10)
            )
            
            left = line.bounding_box[0]
            top = line.bounding_box[1]
            draw.text(
                [left, top - font_size],
                line.text,
                fill=(0, 255, 255),
                font=fnt,
            )

# bounding_box是四邊形的頂點 [x1, y1, x2, y2, x3, y3, x4, y4]，這邊的四邊形並非長方形，要使用 draw.line 畫出封閉四邊形。
# draw.line 需要知道起點位置，才能畫出封閉形狀。


img.save("output.png")

import requests    

string = "무엇을 합니까?"

TRANS_KEY = "YOUR TRANSLATOR SUBSCRIPTION KEY"
trans_url = "https://api.cognitive.microsofttranslator.com/translate"

params = {"api-version": "2.0", "to": ["zh-Hant"]}

headers = {
    "Ocp-Apim-Subscription-Key": TRANS_KEY,
    "Content-type": "application/json",
    "Ocp-Apim-Subscription-Region": "eastus2",
}

# 字串需要包成 dict ，可以把不同的字串分別包裝成 dict，放入 list 之中 

body = [{"text": string}]

req = requests.post(trans_url, params=params, headers=headers, json=body)
response = req.json()

ans = []
for i in response:
    ans.append(i["translations"][0]["text"])
language = response[0]["detectedLanguage"]["language"]

print("language is {}".format(language))
print("translation is {}".format(ans))

language is ko
translation is 做什麼？

from azure.cognitiveservices.speech import (
    SpeechConfig,
    SpeechSynthesizer,
)

# 輸入金鑰和區域，設定config
SPEECH_KEY = "你的金鑰"
speech_config = SpeechConfig(subscription=SPEECH_KEY, region="eastus2")

# 因為後續要針對韓文發音，所以設定成"ko-KR"
speech_config.speech_synthesis_language = "ko-KR"

# 設定轉換後存檔路徑
file_name = "your_path.wav"
audio_config = AudioOutputConfig(filename=file_name)

synthesizer = SpeechSynthesizer(
    speech_config=speech_config, audio_config=audio_config
)

# 開始轉換成語音
synthesizer.speak_text_async(string)

{
    "line": {
        "line_secret": "your line secret",
        "line_token": "your line token",
    },
    "azure": {
        "cv_key": "your subscription key of computer vision",
        "cv_end": "your endpoint of computer vision",
        "blob_connect": "your connect string",
        "blob_container": "your blob container name",
        "trans_key": "your subscription key of translator",
        "speech_key": "your subscription key of speech"
}

Flask==1.0.2
line-bot-sdk
azure-cognitiveservices-vision-computervision
azure-cognitiveservices-speech
azure-storage-blob
Pillow
langdetect

import os
import json
import time
import requests
from flask import Flask, request, abort
from azure.cognitiveservices.vision.computervision import ComputerVisionClient
from azure.cognitiveservices.vision.computervision.models import OperationStatusCodes
from azure.storage.blob import BlobServiceClient
from azure.cognitiveservices.speech import (
    SpeechConfig,
    SpeechSynthesizer,
)
from azure.cognitiveservices.speech.audio import AudioOutputConfig

from msrest.authentication import CognitiveServicesCredentials
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import (
    MessageEvent,
    TextMessage,
    TextSendMessage,
    FlexSendMessage,
    ImageMessage,
)
from PIL import Image
from langdetect import detect

app = Flask(__name__)


CONFIG = json.load(open("/home/config.json", "r"))

# 取得電腦視覺的用戶權限
SUBSCRIPTION_KEY = CONFIG["azure"]["cv_key"]
ENDPOINT = CONFIG["azure"]["cv_end"]
CV_CLIENT = ComputerVisionClient(
    ENDPOINT, CognitiveServicesCredentials(SUBSCRIPTION_KEY)
)

# 連結 blob service
CONNECT_STR = CONFIG["azure"]["blob_connect"]
CONTAINER = CONFIG["azure"]["blob_container"]
BLOB_SERVICE = BlobServiceClient.from_connection_string(CONNECT_STR)


# 取得翻譯工具的金鑰
TRANS_KEY = CONFIG["azure"]["trans_key"]
# 設定 Azure 語音的 config
SPEECH_KEY = CONFIG["azure"]["speech_key"]
SPEECH_CONFIG = SpeechConfig(subscription=SPEECH_KEY, region="eastus2")
SPEECH_CONFIG.speech_synthesis_language = "ko-KR"

LINE_SECRET = CONFIG["line"]["line_secret"]
LINE_TOKEN = CONFIG["line"]["line_token"]
LINE_BOT = LineBotApi(LINE_TOKEN)
HANDLER = WebhookHandler(LINE_SECRET)


@app.route("/")
def hello():
    "hello world"
    return "Hello World!!!!!"

# 上傳檔案到 Azure blob
def upload_blob(container, path):
    """
    Upload files to Azure blob
    """
    blob_client = BLOB_SERVICE.get_blob_client(container=container, blob=path)
    with open(path, "rb") as data:
        blob_client.upload_blob(data, overwrite=True)
    data.close()
    return blob_client.url


# 透過 Azure 電腦視覺執行 OCR
def azure_ocr(url):
    """
    Azure OCR: get characters from image url
    """
    ocr_results = CV_CLIENT.read(url, raw=True)
    operation_location_remote = ocr_results.headers["Operation-Location"]
    operation_id = operation_location_remote.split("/")[-1]
    while True:
        get_handw_text_results = CV_CLIENT.get_read_result(operation_id)
        if get_handw_text_results.status not in ["notStarted", "running"]:
            break
        time.sleep(1)

    text = []
    if get_handw_text_results.status == OperationStatusCodes.succeeded:
        for text_result in get_handw_text_results.analyze_result.read_results:
            for line in text_result.lines:
                text.append(line.text)
    if len(text) == 0:
        return text
    return []

# 將字串翻譯成中文
def azure_translation(string, message_id):
    """
    Translation with azure API
    """
    trans_url = "https://api.cognitive.microsofttranslator.com/translate"

    params = {"api-version": "2.0", "to": ["zh-Hant"]}

    headers = {
        "Ocp-Apim-Subscription-Key": TRANS_KEY,
        "Content-type": "application/json",
        "Ocp-Apim-Subscription-Region": "eastus2",
    }

    body = [{"text": string}]

    req = requests.post(trans_url, params=params, headers=headers, json=body)
    response = req.json()
    output = ""
    speech_button = ""
    ans = []
    for i in response:
        ans.append(i["translations"][0]["text"])
    language = response[0]["detectedLanguage"]["language"]
    # 如果是韓文的話，就透過 Azure Speech 取得發音
    if language == "ko":
        output = " ".join(string) + "\n" + " ".join(ans)
        speech_button = azure_speech(string, message_id)
    return output, speech_button

# 將字串轉換成音訊檔，並且上傳到 Azure blob
def azure_speech(string, message_id):
    """
    Azure speech: text to speech, and save wav file to azure blob
    """

    file_name = "{}.wav".format(message_id)
    audio_config = AudioOutputConfig(filename=file_name)

    synthesizer = SpeechSynthesizer(
        speech_config=SPEECH_CONFIG, audio_config=audio_config
    )
    synthesizer.speak_text_async(string)
    # 上傳 Azure blob，並取得 URL
    link = upload_blob(CONTAINER, file_name)
    # 將 URL 包裝成 Flex message的按扭，以便最後輸出
    output = {
        "type": "button",
        "flex": 2,
        "style": "primary",
        "color": "#1E90FF",
        "action": {"type": "uri", "label": "Voice", "uri": link},
        "height": "sm",
    }
    os.remove(file_name)
    return output





@app.route("/callback", methods=["POST"])
def callback():
    """
    LINE bot webhook callback
    """
    # get X-Line-Signature header value
    signature = request.headers["X-Line-Signature"]
    print(signature)
    body = request.get_data(as_text=True)
    print(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(MessageEvent, message=TextMessage)
def handle_message(event):
    """
    Reply text message
    """
    with open("templates/detect_result.json", "r") as f_h:
        bubble = json.load(f_h)
    f_h.close()
    # 利用 langdetect 此套件的 detect 判斷是否為韓文
    if detect(event.message.text) == "ko":
        output, speech_button = azure_translation(event.message.text, event.message.id)
        # header 是放圖片的部分，沒有圖片的話，就先去除
        bubble.pop("header")
        # 放入翻譯的結果
        bubble["body"]["contents"][0]["text"] = output
        # 放入語音連結的按鈕
        bubble["body"]["contents"].append(speech_button)
        # 調整 body 的高度
        bubble["body"]["height"] = "{}px".format(150)
        message = FlexSendMessage(alt_text="Report", contents=bubble)
    else:
        message = TextSendMessage(text=event.message.text)
    LINE_BOT.reply_message(event.reply_token, message)


@HANDLER.add(MessageEvent, message=ImageMessage)
def handle_content_message(event):
    """
    Reply Image message with results of image description and objection detection
    """
    print(event.message)
    print(event.source.user_id)
    print(event.message.id)

    with open("templates/detect_result.json", "r") as f_h:
        bubble = json.load(f_h)
    f_h.close()
    filename = "{}.jpg".format(event.message.id)
    message_content = LINE_BOT.get_message_content(event.message.id)
    with open(filename, "wb") as f_h:
        for chunk in message_content.iter_content():
            f_h.write(chunk)
    f_h.close()

    img = Image.open(filename)
    link = upload_blob(CONTAINER, filename)

    text = azure_ocr(link)
    output, speech_button = azure_translation(" ".join(text), event.message.id)
    
    # 整合圖片、翻譯結果和音訊按鈕到 flex message
    bubble["body"]["contents"].append(speech_button)
    bubble["body"]["height"] = "{}px".format(150)
    bubble["body"]["contents"][0]["text"] = output
    bubble["header"]["contents"][0]["url"] = link
    bubble["header"]["contents"][0]["aspectRatio"] = "{}:{}".format(
        img.size[0], img.size[1]
    )

    LINE_BOT.reply_message(
        event.reply_token, [FlexSendMessage(alt_text="Report", contents=bubble)]
    )

pip3 install azure-cognitiveservices-vision-face

import sys
import time
from azure.cognitiveservices.vision.face import FaceClient
from msrest.authentication import CognitiveServicesCredentials
from azure.cognitiveservices.vision.face.models import TrainingStatusType
KEY = "YOUR KEY"
ENDPOINT = "YOR ENDPOINT"
FACE_CLIENT = FaceClient(
  ENDPOINT, CognitiveServicesCredentials(KEY))

# 先建立 Person Group
PERSON_GROUP_ID = "請自行命名"
FACE_CLIENT.person_group.create(
  person_group_id=PERSON_GROUP_ID, name=PERSON_GROUP_ID)
  
  
# 在 Person Group 中新增一人

person = FACE_CLIENT.person_group_person.create(
  PERSON_GROUP_ID, "YOUR NAME")

# 新增照片給指定之人

my_pic = open("YOUR PHOTO", 'r+b')
FACE_CLIENT.person_group_person.add_face_from_stream(
  PERSON_GROUP_ID, person.person_id, my_pic)

# 開始訓練，一開始放進去的照片量較大的話，會需要較長時間，可以透過 get_training_status 確認當下狀態

FACE_CLIENT.person_group.train(PERSON_GROUP_ID)

while (True):
    training_status = FACE_CLIENT.person_group.get_training_status(PERSON_GROUP_ID)
    print("Training status: {}.".format(training_status.status))
    print()
    if (training_status.status is TrainingStatusType.succeeded):
        break
    elif (training_status.status is TrainingStatusType.failed):
        sys.exit('Training the person group has failed.')
    time.sleep(5)

img = open("YOUR PHOTO", "r+b")
detected_face = FACE_CLIENT.face.detect_with_stream(
    img, detection_model="detection_01"
)
# 臉部服務會給每一張偵測到的臉一個face ID
results = FACE_CLIENT.face.identify([detected_face[0].face_id], PERSON_GROUP_ID)
result = results[0].as_dict()

# 如果在資料庫中有找到相像的人，會給予person ID
# 再拿此person ID去查詢名字
person = FACE_CLIENT.person_group_person.get(
        PERSON_GROUP_ID, result["candidates"][0]["person_id"]
)
print(person.name)

from pymongo import MongoClient
uri = "mongodb://mongoit:...."
client = MongoClient(uri, retryWrites=False)

# 取得名為 face_register 的資料庫，若原本沒有此資料庫，此舉也能產生此資料庫
db = client['face_register']
# 取得名為 line 的 collection，若原本沒有此 collection，此舉也能產生此 collection
collect = db['line']
post = {"name": "Triathlon", "userId": "1234567890"}}
collect.insert_one(post)

collect.find_one({"name": "Triathlon"})

from datetime import datetime
collect_login = db["daily_login"]
now = datetime.now()
post = {"userId": "1234567890", "time": now.timestamp()}
collect_login.insert_one(post)

yesterday = datetime.now() - timedelta(days=1)
# 檢查符合條件的有多少筆：要符合User ID 且 timestamp 在一天之內的
result = collect_login.count_documents(
    {"$and": [{"userId": "1234567890"}, {"time": {"$gte": yesterday.timestamp()}}]}
)

{
    "line": {
        "line_secret": "your line secret",
        "line_token": "your line token",
    },
    "azure": {
        "face_key": "your subscription key of Azure Face service",
        "face_end": "your endpoint of Azure Face service",
        "blob_connect": "your connect string",
        "blob_container": "your blob container name",
        "trans_key": "your subscription key of translator",
        "speech_key": "your subscription key of speech",
        "mongo_uri":"your mongon uri"
}

Flask==1.0.2
line-bot-sdk
azure-cognitiveservices-vision-face
azure-cognitiveservices-speech
azure-storage-blob
Pillow
pymongo
langdetect

from datetime import datetime, timezone, timedelta
import os
import json
import requests
from flask import Flask, request, abort
from azure.cognitiveservices.vision.face import FaceClient
from azure.storage.blob import BlobServiceClient
from azure.cognitiveservices.speech import (
    SpeechConfig,
    SpeechSynthesizer,
)
from azure.cognitiveservices.speech.audio import AudioOutputConfig

from msrest.authentication import CognitiveServicesCredentials
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import (
    MessageEvent,
    TextMessage,
    TextSendMessage,
    FlexSendMessage,
    ImageMessage,
)
from pymongo import MongoClient
from PIL import Image
from langdetect import detect

app = Flask(__name__)


CONFIG = json.load(open("/home/config.json", "r"))

# 取得 Azure Face 的權限
FACE_KEY = CONFIG["azure"]["face_key"]
FACE_END = CONFIG["azure"]["face_end"]
FACE_CLIENT = FaceClient(FACE_END, CognitiveServicesCredentials(FACE_KEY))
PERSON_GROUP_ID = "triathlon"

# 連接MongoDB
MONGO = MongoClient(CONFIG["azure"]["mongo_uri"], retryWrites=False)
DB = MONGO["face_register"]


CONNECT_STR = CONFIG["azure"]["blob_connect"]
CONTAINER = CONFIG["azure"]["blob_container"]
BLOB_SERVICE = BlobServiceClient.from_connection_string(CONNECT_STR)

TRANS_KEY = CONFIG["azure"]["trans_key"]
SPEECH_KEY = CONFIG["azure"]["speech_key"]
SPEECH_CONFIG = SpeechConfig(subscription=SPEECH_KEY, region="eastus2")
SPEECH_CONFIG.speech_synthesis_language = "ko-KR"

LINE_SECRET = CONFIG["line"]["line_secret"]
LINE_TOKEN = CONFIG["line"]["line_token"]
LINE_BOT = LineBotApi(LINE_TOKEN)
HANDLER = WebhookHandler(LINE_SECRET)


@app.route("/")
def hello():
    "hello world"
    return "Hello World!!!!!"


# 查詢名為 line 的 collection 之中，是否有某個人名
def check_registered(name):
    """
    Check if a specific name is in the database
    """
    collect_register = DB["line"]
    return collect_register.find_one({"name": name})

# 確認從 MongoDB 對應到的 User ID 是否與從 Line 取得的 User ID 相符
# 若相符，則將當下的 timestamp 連同 User ID 紀錄於 MongoDB
def face_login(name, user_id):
    """
    Insert face recognition result to MongoDB
    """
    result = check_registered(name)
    if result:
        if result["userId"] == user_id:
            collect_login = DB["daily_login"]
            now = datetime.now()
            post = {"userId": user_id, "time": now.timestamp()}
            collect_login.insert_one(post)

# 檢查該 User ID 最近一天內是否有登入
def is_login(user_id):
    """
    Check login status from MongoDB
    """
    collect_login = DB["daily_login"]
    yesterday = datetime.now() - timedelta(days=1)
    result = collect_login.count_documents(
        {"$and": [{"userId": user_id}, {"time": {"$gte": yesterday.timestamp()}}]}
    )

    return result > 0

# 上傳檔案到 Azure Blob
def upload_blob(container, path):
    """
    Upload files to Azure blob
    """
    blob_client = BLOB_SERVICE.get_blob_client(container=container, blob=path)
    with open(path, "rb") as data:
        blob_client.upload_blob(data, overwrite=True)
    data.close()
    return blob_client.url


# 文字轉換成語音
def azure_speech(string, message_id):
    """
    Azure speech: text to speech, and save wav file to azure blob
    """

    file_name = "{}.wav".format(message_id)
    audio_config = AudioOutputConfig(filename=file_name)

    synthesizer = SpeechSynthesizer(
        speech_config=SPEECH_CONFIG, audio_config=audio_config
    )
    synthesizer.speak_text_async(string)
    link = upload_blob(CONTAINER, file_name)
    output = {
        "type": "button",
        "flex": 2,
        "style": "primary",
        "color": "#1E90FF",
        "action": {"type": "uri", "label": "Voice", "uri": link},
        "height": "sm",
    }
    os.remove(file_name)
    return output

# 翻譯韓文成中文
def azure_translation(string, message_id):
    """
    Translation with azure API
    """
    trans_url = "https://api.cognitive.microsofttranslator.com/translate"

    params = {"api-version": "2.0", "to": ["zh-Hant"]}

    headers = {
        "Ocp-Apim-Subscription-Key": TRANS_KEY,
        "Content-type": "application/json",
        "Ocp-Apim-Subscription-Region": "eastus2",
    }

    body = [{"text": string}]

    req = requests.post(trans_url, params=params, headers=headers, json=body)
    response = req.json()
    output = ""
    speech_button = ""
    ans = []
    for i in response:
        ans.append(i["translations"][0]["text"])
    language = response[0]["detectedLanguage"]["language"]
    if language == "ko":
        output = " ".join(string) + "\n" + " ".join(ans)
        speech_button = azure_speech(string, message_id)
    return output, speech_button

# 人臉辨識
def azure_face_recognition(filename):
    """
    Azure face recognition
    """
    img = open(filename, "r+b")
    detected_face = FACE_CLIENT.face.detect_with_stream(
        img, detection_model="detection_01"
    )
    # 如果偵測不到人臉，或人臉太多，直接回傳空字串
    if len(detected_face) != 1:
        return ""
    results = FACE_CLIENT.face.identify([detected_face[0].face_id], PERSON_GROUP_ID)
    # 找不到相對應的人臉，回傳 unknown
    if len(results) == 0:
        return "unknown"
    result = results[0].as_dict()
    if len(result["candidates"]) == 0:
        return "unknown"
    # 如果信心程度低於 0.5，也當作不認識
    if result["candidates"][0]["confidence"] < 0.5:
        return "unknown"
    # 前面的 result 只會拿到 person ID，要進一步比對，取得人名
    person = FACE_CLIENT.person_group_person.get(
        PERSON_GROUP_ID, result["candidates"][0]["person_id"]
    )
    return person.name


@app.route("/callback", methods=["POST"])
def callback():
    """
    LINE bot webhook callback
    """
    # get X-Line-Signature header value
    signature = request.headers["X-Line-Signature"]
    print(signature)
    body = request.get_data(as_text=True)
    print(body)
    try:
        HANDLER.handle(body, signature)
    except InvalidSignatureError:
        print(
            "Invalid signature. Please check your channel access token/channel secret."
        )
        abort(400)
    return "OK"



# Line chatbot 接收影像後，開始執行人臉辨識
@HANDLER.add(MessageEvent, message=ImageMessage)
def handle_content_message(event):
    """
    Reply Image message with results of image description and objection detection
    """
    print(event.message)
    print(event.source.user_id)
    print(event.message.id)

    with open("templates/detect_result.json", "r") as f_h:
        bubble = json.load(f_h)
    f_h.close()
    filename = "{}.jpg".format(event.message.id)
    message_content = LINE_BOT.get_message_content(event.message.id)
    with open(filename, "wb") as f_h:
        for chunk in message_content.iter_content():
            f_h.write(chunk)
    f_h.close()
    img = Image.open(filename)
    link = upload_blob(CONTAINER, filename)
    # 人臉辨識後取得人名
    name = azure_face_recognition(filename)
    output = ""
    if name != "":
        now = datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%d %H:%M")
        output = "{0}, {1}".format(name, now)
        # 取得人名後，進行登入
        face_login(name, event.source.user_id)
    # 包裝成 flex message
    bubble["body"]["contents"][0]["text"] = output
    bubble["header"]["contents"][0]["url"] = link
    bubble["header"]["contents"][0]["aspectRatio"] = "{}:{}".format(
        img.size[0], img.size[1]
    )

    LINE_BOT.reply_message(
        event.reply_token, [FlexSendMessage(alt_text="Report", contents=bubble)]
    )
    

@HANDLER.add(MessageEvent, message=TextMessage)
def handle_message(event):
    """
    Reply text message
    """
    with open("templates/detect_result.json", "r") as f_h:
        bubble = json.load(f_h)
    f_h.close()
    # 如果傳來的文字是韓文，且此使用者一天之內曾經登入的話，就可使用翻譯韓文的功能
    if (detect(event.message.text) == "ko") and is_login(event.source.user_id):
        output, speech_button = azure_translation(event.message.text, event.message.id)
        bubble.pop("header")
        bubble["body"]["contents"][0]["text"] = output
        bubble["body"]["contents"].append(speech_button)
        bubble["body"]["height"] = "{}px".format(150)
        message = FlexSendMessage(alt_text="Report", contents=bubble)
    else:
        message = TextSendMessage(text=event.message.text)
    LINE_BOT.reply_message(event.reply_token, message)

{
    "ENDPOINT": "<your endpoint>",
    "training_key": "<your training key>",
    "prediction_key": "<your prediction key>",
    "prediction_resource_id": "<your prediction resource id>",
    "publish_iteration_name": "publish iteration name can be defined by yourself",
    "project_name": "project name can be defined by yourself",
    "annotation_file": "annotation.json",
    "label": [
        "fork",
        "scissors"
    ],
    "image_folder": "your image folder"
}

{
  "<label>": {
      "<file name>": [
        <left>, <top>, <width>, <height>
      ],
      ...
      ...},
  "scissors": {
      "scissors_1": [
        0.4007353, 0.194068655, 0.259803921, 0.6617647
      ],
      ...,
      ...,}
}

args = parse_args()
config = json.load(open(args.config, "r"))
credentials = ApiKeyCredentials(i
n_headers={"Training-key": config["training_key"]})
trainer = CustomVisionTrainingClient(
  config["ENDPOINT"], credentials)
obj_detection_domain = next(
    domain
    for domain in trainer.get_domains()
    if domain.type == "ObjectDetection" and \
    domain.name == "General"
)
project = trainer.create_project(
    config["project_name"], domain_id=obj_detection_domain.id
)

def add_image(
  trainer, label, project_id, annotation, image_folder):
    tagged_images_with_regions = []
    tag = trainer.create_tag(project_id, label)
    for file_name in annotation.keys():
        left, top, width, height = annotation[file_name]
        regions = [
            Region(tag_id=tag.id, 
            left=left, top=top, width=width, height=height)
        ]

        file_path = os.path.join(
          image_folder, label, file_name + ".jpg")
        with open(file_path, "rb") as image_contents:
            tagged_images_with_regions.append(
                ImageFileCreateEntry(
                    name=file_name, 
                    contents=image_contents.read(), 
                    regions=regions
                )
            )
        image_contents.close()
    return tagged_images_with_regions

image_folder = config["image_folder"]
annotations = json.load(open("annotation.json", "r"))
tagged_images_with_regions = []
for label in annotations.keys():
    tagged_images_with_regions += add_image(
        trainer, label, project.id, 
        annotations[label], image_folder)
upload_result = trainer.create_images_from_files(
    project.id, 
    ImageFileCreateBatch(images=tagged_images_with_regions)
)
if not upload_result.is_batch_successful:
    print("Image batch upload failed.")
    for image in upload_result.images:
        print("Image status: ", image.status)

iteration = trainer.train_project(project.id)
while iteration.status != "Completed":
    iteration = trainer.get_iteration(
      project.id, iteration.id)
    print("Training status: " + iteration.status)
    time.sleep(1)

publish_iteration_name = config["publish_iteration_name"]
prediction_resource_id = config["prediction_resource_id"]
trainer.publish_iteration(
    project.id, iteration.id, 
    publish_iteration_name, prediction_resource_id
)

def get_project_id(config):
    credentials = ApiKeyCredentials(
      in_headers={"Training-key": config["training_key"]})
    trainer = CustomVisionTrainingClient(
      config["ENDPOINT"], credentials)
    project_id = next(
        proj.id
        for proj in trainer.get_projects()
        if proj.name == config["project_name"]
    )
    return project_id
 
prediction_credentials = ApiKeyCredentials(
        in_headers={"Prediction-key": config["prediction_key"]}
    )
predictor = CustomVisionPredictionClient(
  config["ENDPOINT"], prediction_credentials)
project_id = get_project_id(config)
 
with open(args.image, "rb") as image_contents:
    results = predictor.classify_image(
        project_id,
        config["publish_iteration_name"],
        image_contents.read(),
    )
image_contents.close()

img = Image.open(args.image)
draw = ImageDraw.Draw(img)
font = ImageFont.truetype(
    "../static/TaipeiSansTCBeta-Regular.ttf", size=int(5e-2 * img.size[1])
)

for prediction in results.predictions:
    if prediction.probability > 0.5:
        bbox = prediction.bounding_box.as_dict()
        left = bbox['left'] * img.size[0]
        top = bbox['top'] * img.size[1]
        right = left + bbox['width'] * img.size[0]
        bot = top + bbox['height'] * img.size[1]
        draw.rectangle(
          [left, top, right, bot], 
          outline=(255, 0, 0), width=3)
        draw.text(
            [left, abs(top - 5e-2 * img.size[1])],
            "{0} {1:0.2f}".format(
                prediction.tag_name, 
                prediction.probability * 100
            ),
            fill=(255, 0, 0),
            font=font,
        )
img.show()