---
titles: telegrambot
tags: bot
---
[回首頁](https://hackmd.io/@janice880624/total)
# telegrambot
### 使用的工具及服務：
* Python
* OLAMI
* KKBOX Open API

## telegram
> name:janice_music
> username:janiecmusic_bot
> Done! Congratulations on your new bot. You will find it at t.me/janiecmusic_bot. You can now add a description, about section and profile picture for your bot, see /help for a list of commands. By the way, when you've finished creating your cool bot, ping our Bot Support if you want a better username for it. Just make sure the bot is fully operational before you do this.
Use this token to access the HTTP API:
1015948599:AAEagT7-mlw_S6MFrgo6v0q5jxWKPoeMFPY
Keep your token secure and store it safely, it can be used by anyone to control your bot.
For a description of the Bot API, see this page: https://core.telegram.org/bots/api

### 建立檔案
1. `mkdir $project_name`
2. `cd $project_name`
3. `pipenv install --three python-telegram-bot flask gunicorn requests`

#### 新增`congig.ini`

[TELEGRAM]
ACCESS_TOKEN = 
WEBHOOK_URL =

touch config.ini

## OLAMI
* `mkdir nlp`
* `cd nlp`
* `touch __init__.py`
* `touch olami.py`
#### 修改`congig.ini`

[TELEGRAM]
ACCESS_TOKEN = 
WEBHOOK_URL =

[OLAMI]
APP_KEY = 
APP_SECRET = 

#### 修改`__init__.py`

from . import olami

#### 編輯`olami.py`

import configparser
import json
import logging
import time
from hashlib import md5
from api.kkbox import KKBOX

import requests

config = configparser.ConfigParser()
config.read('config.ini')

logger = logging.getLogger(__name__)


class NliStatusError(Exception):
    """The NLI result status is not 'ok'"""


class Olami:
    URL = 'https://tw.olami.ai/cloudservice/api'

    def __init__(self, app_key=config['OLAMI']['APP_KEY'], app_secret=config['OLAMI']['APP_SECRET'], input_type=1):
        self.app_key = app_key
        self.app_secret = app_secret
        self.input_type = input_type

    def nli(self, text, cusid=None):
        response = requests.post(self.URL, params=self._gen_parameters('nli', text, cusid))
        response.raise_for_status()
        response_json = response.json()
        if response_json['status'] != 'ok':
            raise NliStatusError("NLI responded status != 'ok': {}".format(response_json['status']))
        else:
            nli_obj = response_json['data']['nli'][0]
            return self.intent_detection(nli_obj)

    def _gen_parameters(self, api, text, cusid):
        timestamp_ms = (int(time.time() * 1000))
        params = {'appkey': self.app_key,
                  'api': api,
                  'timestamp': timestamp_ms,
                  'sign': self._gen_sign(api, timestamp_ms),
                  'rq': self._gen_rq(text)}
        if cusid is not None:
            params.update(cusid=cusid)
        return params

    def _gen_sign(self, api, timestamp_ms):
        data = self.app_secret + 'api=' + api + 'appkey=' + self.app_key + 'timestamp=' + \
               str(timestamp_ms) + self.app_secret
        return md5(data.encode('ascii')).hexdigest()

    def _gen_rq(self, text):
        obj = {'data_type': 'stt', 'data': {'input_type': self.input_type, 'text': text}}
        return json.dumps(obj)

    def intent_detection(self, nli_obj):
        def handle_selection_type(type):
            if type == 'news':
                return desc['result'] + '\n\n' + '\n'.join(
                    str(index + 1) + '. ' + el['title'] for index, el in enumerate(data))
            elif type == 'poem':
                return desc['result'] + '\n\n' + '\n'.join(
                    str(index + 1) + '. ' + el['poem_name'] + '，作者：' + el['author'] for index, el in enumerate(data))
            elif type == 'cooking':
                return desc['result'] + '\n\n' + '\n'.join(
                    str(index + 1) + '. ' + el['name'] for index, el in enumerate(data))
            else:
                return '對不起，你說的我還不懂，能換個說法嗎？'

        def handle_music_kkbox_type(semantic):
            music_type = semantic['modifier'][0].split('_')[2]
            slots = semantic['slots']
            kkbox = KKBOX()

            def get_slot_value_by_key(key):
                return next(filter(lambda el: el['name'] == key, slots))['value']

            key = 'keyword' if music_type == 'playlist' else (music_type + '_name')
            return kkbox.search(music_type, get_slot_value_by_key(key))

        type = nli_obj['type']
        desc = nli_obj['desc_obj']
        data = nli_obj.get('data_obj', [])

        if type == 'kkbox':
            id = data[0]['id']
            return ('https://widget.kkbox.com/v1/?type=song&id=' + id) if len(data) > 0 else desc['result']
        elif type == 'baike':
            return data[0]['description']
        elif type == 'joke':
            return data[0]['content']
        elif type == 'news':
            return data[0]['detail']
        elif type == 'cooking':
            return data[0]['content']
        elif type == 'selection':
            return handle_selection_type(desc['type'])
        elif type == 'ds':
            return desc['result'] + '\n請用 /help 指令看看我能怎麼幫助您'
        elif type == 'music_kkbox':
            return handle_music_kkbox_type(nli_obj['semantic'][0])
        else:
            return desc['result']

#### 新增`kkbox.py`

import configparser
import logging

import requests

config = configparser.ConfigParser()
config.read('config.ini')

logger = logging.getLogger(__name__)


class KKBOX:
    AUTH_URL = 'https://account.kkbox.com/oauth2/token'
    API_BASE_URL = 'https://api.kkbox.com/v1.1/'

    def __init__(self, id=config['KKBOX']['ID'], secret=config['KKBOX']['SECRET']):
        self.id = id
        self.secret = secret
        self.token = self._get_token()

    def _get_token(self):
        response = requests.post(self.AUTH_URL, data={'grant_type': 'client_credentials'}, auth=(self.id, self.secret))
        response.raise_for_status()
        return response.json()['access_token']

    def search(self, type, q, territory='TW'):
        response = requests.get(self.API_BASE_URL + 'search', params={'type': type, 'q': q, 'territory': territory},
                                headers={'Authorization': 'Bearer ' + self.token})
        response.raise_for_status()
        response_json = response.json()

        if type == 'artist':
            return response_json['artists']['data'][0]['url']
        else:
            id = response_json[type + 's']['data'][0]['id']
            return 'https://widget.kkbox.com/v1/?id=' + id \
                   + '&type=' + ('song' if type == 'track' else type)

## KKBOX Developer
* `mkdir api`
* `cd api`
* `touch __init__.py`
* `touch kkbox.py`

#### 修改`congig.ini`

[TELEGRAM]
ACCESS_TOKEN = 
WEBHOOK_URL =

[OLAMI]
APP_KEY = 
APP_SECRET = 

[KKBOX]
ID = 
SECRET = 

#### 編輯`api/__init__.py`

from . import kkbox

#### 修改 `main.py`

import configparser
import logging

import telegram
from flask import Flask, request
from telegram import ReplyKeyboardMarkup
from telegram.ext import Dispatcher, CommandHandler, MessageHandler, Filters

from nlp.olami import Olami

# Load data from config.ini file
config = configparser.ConfigParser()
config.read('config.ini')

# Enable logging
logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
                    level=logging.INFO)
logger = logging.getLogger(__name__)

# Initial Flask app
app = Flask(__name__)

# Initial bot by Telegram access token
bot = telegram.Bot(token=(config['TELEGRAM']['ACCESS_TOKEN']))

welcome_message = '親愛的主人，您可以問我\n' \
                  '天氣，例如：「今天天氣如何」\n' \
                  '百科，例如：「川普是誰」\n' \
                  '新聞，例如：「今日新聞」\n' \
                  '音樂，例如：「我想聽艾熱的星球墜落」\n' \
                  '日曆，例如：「現在時間」\n' \
                  '詩詞，例如：「我想聽水調歌頭這首詩」\n' \
                  '笑話，例如：「講個笑話」\n' \
                  '故事，例如：「說個故事」\n' \
                  '股票，例如：「台積電的股價」\n' \
                  '食譜，例如：「蛋炒飯怎麼做」\n' \
                  '聊天，例如：「你好嗎」'
reply_keyboard_markup = ReplyKeyboardMarkup([['今天天氣如何'],
                                             ['川普是誰'],
                                             ['今日新聞'],
                                             ['我想聽艾熱的星球墜落'],
                                             ['現在時間'],
                                             ['我想聽水調歌頭這首詩'],
                                             ['講個笑話'],
                                             ['說個故事'],
                                             ['台積電的股價'],
                                             ['蛋炒飯怎麼做'],
                                             ['你好嗎']])


@app.route('/hook', methods=['POST'])
def webhook_handler():
    """Set route /hook with POST method will trigger this method."""
    if request.method == "POST":
        update = telegram.Update.de_json(request.get_json(force=True), bot)
        dispatcher.process_update(update)
    return 'ok'


def start_handler(bot, update):
    """Send a message when the command /start is issued."""
    update.message.reply_text(welcome_message, reply_markup=reply_keyboard_markup)


def help_handler(bot, update):
    """Send a message when the command /help is issued."""
    update.message.reply_text(welcome_message, reply_markup=reply_keyboard_markup)


def reply_handler(bot, update):
    """Reply message."""
    text = update.message.text
    user_id = update.message.from_user.id
    reply = Olami().nli(text, user_id)
    update.message.reply_text(reply)


def error_handler(bot, update, error):
    """Log Errors caused by Updates."""
    logger.error('Update "%s" caused error "%s"', update, error)
    update.message.reply_text('對不起主人，我需要多一點時間來處理 Q_Q')


# New a dispatcher for bot
dispatcher = Dispatcher(bot, None)

# Add handler for handling message, there are many kinds of message. For this handler, it particular handle text
# message.
dispatcher.add_handler(MessageHandler(Filters.text, reply_handler))
dispatcher.add_handler(CommandHandler('start', start_handler))
dispatcher.add_handler(CommandHandler('help', help_handler))
dispatcher.add_error_handler(error_handler)

if __name__ == "__main__":
    # Running server
    app.run(debug=True)