忍者工坊

from flask import request
from flask_restful import Resource
import requests
import webbrowser
import os
import jwt
import json


class LineLoginController(Resource):
	def get(self):
		r = requests.post(
			"https://api.line.me/oauth2/v2.1/token",
			data={
				"grant_type": "authorization_code",
				"code": request.args.get('CODE'),
				"redirect_uri": os.environ.get('LINE_LOGIN_URI'),
				"client_id": os.environ.get('LINE_LOGIN_CLIENT_ID'),
				"client_secret": os.environ.get('LINE_LOGIN_SECRET'),
			}, headers={"Content-Type": "application/x-www-form-urlencoded"})
		payload = json.loads(r.text)
		print(payload)
		token = payload.get("id_token")
		if token is None:
			return {'result': payload['error_description']}, 400

		state = request.args.get('state')
		token = token.encode()
		dt = jwt.decode(token, state, None, algorithms=['HS256'])
		if dt:
			return {'result': dt}, 200

	def post(self):
		r_uri = os.environ.get("LINE_LOGIN_URI")
		client = os.environ.get("LINE_LOGIN_CLIENT_ID")
		state = "nostate" # it will be random value
		uri = f"https://access.line.me/oauth2/v2.1/authorize?response_type=code&client_id={client}&redirect_uri={r_uri}&scope=profile%20openid%20email&state={state}"
		return {'result': uri}

from controller.line_login_controller import LineLoginController
api.add_resource(LineLoginController, '/line/auth')

import json
import requests

def alert_handler(event, context):
    air = requests.get(
        'http://opendata.epa.gov.tw/webapi/Data/REWIQA/?$orderby=SiteName&$skip=0&$top=1000&format=json')
    air_body = json.loads(air.text)
    air_status = ["良好",	"普通",	"對敏感族群不健康",	"對所有族群不健康",	"非常不健康",	"危害", "資料有誤"]
    total, count = 0, 0
    for fetch in air_body:
        if (fetch['County'] == "臺中市" and (fetch['SiteName'] == "西屯" or fetch['SiteName'] == "沙鹿")):
            count += 1
            if fetch['AQI'] != 0:
                total = total + int(fetch['AQI'])
    switch = 0
    pm = int(total / count)
    print(pm)
    if pm in range(0, 50):
        switch = 0
    elif pm in range(51, 100):
        switch = 1
    elif pm in range(101, 150):
        switch = 2
    elif pm in range(151, 200):
        switch = 3
    elif pm in range(201, 250):
        switch = 4
    elif pm in range(251, 300):
        switch = 5
    else:
        switch = 6
    payload = f"\n空氣品質: {air_status[switch]}"
    r = requests.post('https://1uv2o723o4.execute-api.us-east-1.amazonaws.com/dev/notify/sqs',
                      data={'message': payload})
    print(r)

if usage == 'create':
            rich_menu_to_create = None
            if request.args.get('state') == 'up':
                rich_menu_to_create = RichMenu(
                    size=RichMenuSize(width=2500, height=843),
                    selected=False,
                    name="Nice richmenu",  # display name
                    chat_bar_text="我是第一頁",
                    areas=[RichMenuArea(  # 這邊是陣列的格式，可以動態設定自己要的區域想要有什麼功能
                        bounds=RichMenuBounds(
                            x=0, y=0, width=2500, height=843),
                        action=MessageAction(
                            label='message',
                            text='下一頁'
                        ))]
                )
            elif request.args.get('state') == 'down':
                rich_menu_to_create = RichMenu(
                    size=RichMenuSize(width=2500, height=843),
                    selected=False,
                    name="Nice richmenu",  # display name
                    chat_bar_text="我是第二頁",
                    areas=[RichMenuArea(
                        bounds=RichMenuBounds(
                            x=0, y=0, width=2500, height=843),
                        action=MessageAction(
                            label='message',
                            text='上一頁'
                        ))]
                )
            else:
                rich_menu_to_create = RichMenu(
                    size=RichMenuSize(width=2500, height=843),
                    selected=False,
                    name="Nice richmenu",  # display name
                    chat_bar_text="我是測試使用",
                    areas=[RichMenuArea(  # 這邊是陣列的格式，可以動態設定自己要的區域想要有什麼功能
                        bounds=RichMenuBounds(
                            x=0, y=0, width=2500, height=843),
                        action=URIAction(label='Go to line.me', uri='https://line.me'))]
                )
            rich_menu_id = self.line_bot_api.create_rich_menu(
                rich_menu=rich_menu_to_create)
            print(rich_menu_id)
            return {'id': rich_menu_id}, 200

message = event['events'][0]['message']['text']
if message == "上一頁" or message == "下一頁":
    try:
        rich_menu_id = self.line_bot_api.get_rich_menu_id_of_user(id)
    except:
        # link default rich menu
        self.line_bot_api.link_rich_menu_to_user(id, "richmenu-269cc28b8e8497d76c2df062b274a2ce")

    if rich_menu_id == "richmenu-269cc28b8e8497d76c2df062b274a2ce":
        self.line_bot_api.link_rich_menu_to_user(id, "richmenu-e31be74ad7e577b4752ab70c9c2a3fba")
    else:
        self.line_bot_api.link_rich_menu_to_user(id, "richmenu-269cc28b8e8497d76c2df062b274a2ce")
else:
    self.line_bot_api.reply_message(token, TextSendMessage(text=message))

if usage == 'create':
    rich_menu_to_create = RichMenu(
        size=RichMenuSize(width=2500, height=843),
        selected=False,
        name="Nice richmenu",  # display name
        chat_bar_text="我是測試使用",
        areas=[RichMenuArea(  # 這邊是陣列的格式，可以動態設定自己要的區域想要有什麼功能
            bounds=RichMenuBounds(x=0, y=0, width=2500, height=843),
            action=URIAction(label='Go to line.me', uri='https://line.me'))]
    )
    rich_menu_id = self.line_bot_api.create_rich_menu(
        rich_menu=rich_menu_to_create)
    print(rich_menu_id)
    return {'id': rich_menu_id}, 200

elif usage == 'upload':
    file = request.files['the_file'].read()
    rich_menu_id = request.form['richmenu_id']
    content_type = "image/png"
    try:
        self.line_bot_api.set_rich_menu_image(
            rich_menu_id, content_type, file)
    except Exception as e:
        print("===Upload Exception===")
        raise BadRequest(e)
    return {'result': 'upload ok'}, 200

elif usage == 'set':
    rich_menu_id = request.form['richmenu_id']
    try:
        self.line_bot_api.set_default_rich_menu(rich_menu_id)
    except:
        raise BadRequest("Maybe your richmenu id error.")
    return {'result': 'set default ok!'}, 200

elif usage == 'get':
    rich_menu_list = self.line_bot_api.get_rich_menu_list()
    total = []
    for rich_menu in rich_menu_list:
        total.append(rich_menu.rich_menu_id)
    return {'result': total}, 200

elif usage == 'delete':
    rich_menu_id = request.form['richmenu_id']
    try:
        self.line_bot_api.delete_rich_menu(rich_menu_id)
    except:
        raise BadRequest("Maybe your richmenu id error.")
    return {'result': f'{rich_menu_id} is delete!'}, 200

profile = line_bot_api.get_profile(event['events'][0]['source']['userId'])

try:
    with Database() as db, db.connect() as conn, conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
        cur.execute(
            f"INSERT INTO users(id, name, picture) VALUES ('{id}', '{name}', '{picture}')")
except:
    pass # 這邊簡單處理相同 id 的問題(id我有設 Private key 哦！)
state = f"Hello 👉 `{profile.display_name}` 👈"
line_bot_api.reply_message(token, [
  TextSendMessage(text=state),
  ImageSendMessage(original_content_url=profile.picture_url, preview_image_url=profile.picture_url)
  ])

line_bot_api = LineBotApi(os.getenv('LINE_CHANNEL_TOKEN'))
handler = WebhookHandler(os.getenv('LINE_CHANNEL_SECRET_KEY'))

def get(self):
    msg = request.args.get('msg')
    with Database() as db, db.connect() as conn, conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
        cur.execute("SELECT * FROM users")
        fetch = cur.fetchall()
    for user in fetch:
        self.line_bot_api.push_message(
            user['id'], TextSendMessage(text=msg))
    return {'message': msg}, 200

self.line_bot_api.reply_message(token, LocationSendMessage(
    title='my location',
    address='Tokyo',
    latitude=35.65910807942215,
    longitude=139.70372892916203
))

self.line_bot_api.reply_message(token, StickerSendMessage(package_id='1',sticker_id='1'))

buttons_template_message = TemplateSendMessage(
    alt_text='Buttons template',
    template=ButtonsTemplate(
        thumbnail_image_url=f'{picture}.jpg',
        title='Menu',
        text='Please select',
        actions=[
            PostbackAction(
                label='postback',
                display_text='postback text',
                data='action=buy&itemid=1'
            ),
            MessageAction(
                label='message',
                text='message text'
            ),
            URIAction(
                label='uri',
                uri='http://example.com/'
            )
        ]
    )
)
self.line_bot_api.reply_message(token, buttons_template_message)

from flask import Flask, request, abort
from flask_restful import Resource
import json
import os
from linebot import (
    LineBotApi, WebhookHandler
)
from linebot.exceptions import (
    InvalidSignatureError
)
from linebot.models import (
    MessageEvent, TextMessage, TextSendMessage,
)


class LineMessageApiWebhookController(Resource):
    def post(self):
        line_bot_api = LineBotApi(os.getenv('LINE_CHANNEL_TOKEN'))
        handler = WebhookHandler(os.getenv('LINE_CHANNEL_SECRET_KEY'))

        # get X-Line-Signature header value
        signature = request.headers['X-Line-Signature']

        body = request.get_data(as_text=True)
        event = json.loads(body)
        print(event)
        try:
            handler.handle(body, signature)
        except InvalidSignatureError:
            print(
                "Invalid signature. Please check your channel access token/channel secret.")
            abort(400)

        token = event['events'][0]['replyToken']
        if token == "00000000000000000000000000000000":
            pass
        else:
            line_bot_api.reply_message(token, TextSendMessage(
                text=event['events'][0]['message']['text']))
        return 'OK'

from controller.message_api_controller import LineMessageApiWebhookController
api.add_resource(LineMessageApiWebhookController, '/webhook')