2016/12/28

[Python][Google App Engine][Line Bot]無料でLINEBotを作る その21 replyTokenを取得

Google App EngineのPythonで無料でLINE Botを作るというチャレンジをしているのですが、GitHubで公開されているLineのline-bot-sdk-pythonがflask前提となっていたので、flaskを使わないようなプログラムがほしいなと思い、急遽自作することに。

今回は、replyTokenを取得するメソッドを実装します。

linebot.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2007 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

import logging
import hashlib
import hmac
import base64
import json
import urllib2


class LineWebhookEventClass(object):
    def __init__(self,request_body):

        self._request_body = request_body.decode('utf-8').encode('utf-8')
        json_dict = json.loads(self._request_body)
        self.events = json_dict["events"]
        self.eventsLen = len(self.events)

    def _event(self,idx):

        if idx >= self.eventsLen:
            return None
        if idx < 0:
            return None

        return self.events[idx]

    def eventType(self,idx):

        _event = self._event(idx)
        if _event:
            return _event["type"]
        else:
            return None

    def _source(self,idx):
        _event = self._event(idx)
        if _event:
            return _event["source"]
        else:
            return None

   def userId(self,idx):

        _source = self._source(idx)
        if _source["type"] == "user":
            return _source["userId"]
        else:
            return None

    def groupId(self,idx):

        _source = self._source(idx)
        if _source["type"] == "group":
            return _source["groupId"]
        else:
            return None

    def messageType(self,idx):
        if self.eventType(idx) == "message":
            _event = self._event(idx)
            if _event:
                return _event["message"]["type"]
            else:
                return None
        else:
            return None


    def textMessage(self,idx):
        if self.messageType(idx) == "text":
            _event = self._event(idx)
            if _event:
                return _event["message"]["text"]
            else:
                return None
        else:
            return None

    def replyToken(self,idx):
        _event = self._event(idx)
        if _event:
            return _event["replyToken"]
        else:
            return None


class LineSendMessageClass(object):
    def __init__(self,access_token):
        self._reply_message_api = "https://api.line.me/v2/bot/message/reply"
        self._push_message_api = "https://api.line.me/v2/bot/message/push"

        self._access_token = access_token

    def _header(self):

        return {
            "Content-Type":"application/json",
            "Authorization":("Bearer <%s>" % self._access_token)
        }

    def replyMessage(self,reply_token,*messages):


        _form_data = {
            "replyToken":reply_token,
            "messages":messages
        }
        _form_data = json.dumps(_form_data)

        _req = urllib2.Request(self._reply_message_api,_form_data,self._header())

        _f = urllib2.urlopen(_req)
        _response= _f.read()
        _f.close()

    def pushMessage(self,to,*msg):
        
        _form_data = {
            "to":to,
            "messages":msg
        }

        _form_data = json.dumps(_form_data)

        _req = urllib2.Request(self._push_message_api,_form_data,self._header())

        _f = urllib2.urlopen(_req)
        _response= _f.read()
        logging.info(_response)
        _f.close()

class LineWebhookClass(object):
    def __init__(self):
        pass

    @classmethod
    def isValidSignature(cls,channel_secret,x_line_signature,request_body):
        _request_body = request_body.decode('utf-8').encode('utf-8')

        hsh = hmac.new(channel_secret.encode('utf-8'),_request_body,hashlib.sha256).digest()
        signature = base64.b64encode(hsh)
        if signature == x_line_signature:
            return True
        else:
            return False


class LineTemplateActionClass(object):
    def __init__(self):
        pass

    @classmethod
    def _text(cls,text):
        _text_max_length = 300

        #unicode型ではない場合
        if not isinstance(text,unicode):
            return None

        #300文字より大きい場合は、300文字に変更する
        if len(text) > _text_max_length:
            return text[0:_text_max_length]

        return text

    @classmethod
    def _label(cls,label):
        _label_max_length = 20

        #unicode型ではない場合
        if not isinstance(label,unicode):
            return None

        #20文字より大きい場合は、20文字に変更する
        if len(label) > _label_max_length:
            return label[0:_label_max_length]

        return label

    @classmethod
    def postbackAction(cls,label,data,text):

        _label = cls._label(label)
        if _label is None:
            return None

        _text = cls._text(text)
        if _text is None:
            return None

        _data = data
        _data_max_length = 300

        # unicode型ではない場合
        if not isinstance(_data,unicode):
            return None

        #300文字より大きい場合は、300文字に変更する
        if len(_data) > _data_max_length:
            _data = _data[0:_data_max_length]

        return {
            "type":"postback",
            "label":_label,
            "text":_text,
            "data":_data
        }

    @classmethod
    def uriAction(cls,label,uri):

        _label = cls._label(label)
        if _label is None:
            return None

        _uri = uri
        _uri_max_length = 1000

        #string型ではない場合
        if not isinstance(uri,unicode):
            return None

        #1000文字より大きい場合は Noneを返す
        if len(_uri) > _uri_max_length:
            return None

        return {
            "type":"uri",
            "label":_label,
            "uri":_uri
        }


    @classmethod
    def messageAction(cls,label,text):

        _label = cls._label(label)
        if _label is None:
            return None

        _text = cls._text(text)
        if _text is None:
            return None

        return {
            "type":"message",
            "label":_label,
            "text":_text
        }


class LineTemplateMessageClass(object):
    def __init__(self):
        pass

    @classmethod
    def _baseTemplateMessage(cls,alttext):
        _alttext = alttext
        _alttext_max_length = 400

        #unicode型ではない場合
        if not isinstance(_alttext,unicode):
            return None

        #400文字より大きい場合は、20文字に変更する
        if len(_alttext) > _alttext_max_length:
            _alttext = _alttext[0:_alttext_max_length]

        return {
            "type":"template",
            "altText":_alttext
        }

    @classmethod
    def _button(cls,thumbnailimageurl,title,text,*actions):
        _title_max_length = 40
        _title = title

        #unicode型ではない場合
        if not isinstance(_title,unicode):
            return None

        #40文字より大きい場合は、40文字に変更する
        if len(_title) > _title_max_length:
            _title = _title[0:_title_max_length]


        _text = text
        _text_max_length = 60
        if not isinstance(_text,unicode):
            return None

        #60文字より大きい場合は、60文字に変更する
        if len(_text) > _text_max_length:
            _text = _text[0:_text_max_length]

        return {
            "thumbnailImageUrl":thumbnailimageurl,
            "title":_title,
            "text":_text,
            "actions":actions
        }

    @classmethod
    def carousel(cls,alttext,*columns):
        _carousel = cls._baseTemplateMessage(alttext)
        if _carousel is None:
            return None

        _columns = []
        for i in range(0,len(columns)):
            _column = columns[i]
            _columns.append(cls._button(_column["thumbnailimageurl"],_column["title"],_column["text"],*_column["actions"]))

        _carousel["template"] = {
            "type":"carousel",
            "columns":_columns
        }

        return _carousel

    @classmethod
    def buttons(cls,alttext,thumbnailimageurl,title,text,*actions):
        _buttons = cls._baseTemplateMessage(alttext)
        if _buttons is None:
            return None

        _template = {
            "type":"buttons"
        }

        _template.update(
            cls._button(thumbnailimageurl,title,text,*actions)
        )
        
        _buttons["template"] = _template

        return _buttons

    @classmethod
    def confirm(cls,alttext,text,*actions):

        _confirm = cls._baseTemplateMessage(alttext)
        if _confirm is None:
            return None

        _text_max_length = 240
        _text = text
        #unicode型ではない場合
        if not isinstance(_text,unicode):
            return None

        #400文字より大きい場合は、20文字に変更する
        if len(_text) > _text_max_length:
            _text = _text[0:_text_max_length]


        _confirm["template"] = {
            "type":"confirm",
            "text":_text,
            "actions":actions
        }

        return _confirm

0 コメント:

コメントを投稿