如何使用Python向Viber机器人发送消息? - python

我有以下HTTPS服务器:

from flask import Flask, request, Response
from viberbot import Api
from viberbot.api.bot_configuration import BotConfiguration
from viberbot.api.messages import VideoMessage
from viberbot.api.messages.text_message import TextMessage
import logging

from viberbot.api.viber_requests import ViberConversationStartedRequest
from viberbot.api.viber_requests import ViberFailedRequest
from viberbot.api.viber_requests import ViberMessageRequest
from viberbot.api.viber_requests import ViberSubscribedRequest
from viberbot.api.viber_requests import ViberUnsubscribedRequest

logger = logging.getLogger(__name__)
app = Flask(__name__)
viber = Api(BotConfiguration(
    name='PythonSampleBot',
    avatar='http://www.clker.com/cliparts/3/m/v/Y/E/V/small-red-apple-hi.png',
    auth_token='xxx-xxx-xxx'
))


@app.route('/', methods=['POST'])
def incoming():
    logger.debug("received request. post data: {0}".format(request.get_data()))
    # every viber message is signed, you can verify the signature using this method
    if not viber.verify_signature(request.get_data(), request.headers.get('X-Viber-Content-Signature')):
        return Response(status=403)

    # this library supplies a simple way to receive a request object
    viber_request = viber.parse_request(request.get_data())

    if isinstance(viber_request, ViberMessageRequest):
        message = viber_request.message
        # lets echo back
        viber.send_messages(viber_request.sender.id, [
            message
        ])
    elif isinstance(viber_request, ViberSubscribedRequest):
        viber.send_messages(viber_request.get_user.id, [
            TextMessage(text="thanks for subscribing!")
        ])
    elif isinstance(viber_request, ViberFailedRequest):
        logger.warn(
            "client failed receiving message. failure: {0}".format(viber_request))

    return Response(status=200)


def set_webhook(viber_bot):
    viber_bot.set_webhook('https://xxx.xxx.xxx.xxx:4443')
    logging.info("Web hook has been set")


if __name__ == "__main__":
    context = ('certificate.pem', 'key.pem')
    app.run(host='0.0.0.0', port=4443, debug=True, ssl_context=context)

并尝试发送消息:

import json
import requests

webhook_url = 'https://xxx.xxx.xxx.xxx:4443'

data = {
    "receiver": "xxx-xxx-xxx",
    "type": "text",
    "text": "Hello world!"
}

response = requests.post(
    webhook_url, data=json.dumps(data),
    headers={'Content-Type': 'application/json'},
    verify='E:\\Docs\\learn_py\\viberbot\\certificate.pem'
)
if response.status_code != 200:
    raise ValueError(
        'Request returned an error %s, the response is:\n%s'
        % (response.status_code, response.text)
    )

我收到403错误

ValueError:请求返回了错误403,响应为:

更新:

403来自:

if not viber.verify_signature(request.get_data(), request.headers.get('X-Viber-Content-Signature')):
        return Response(status=403)

python大神给出的解决方案

您收到403错误的原因有两个。要模拟来自Viber的Webhook请求,必须发送X-Viber-Content-Signature标头。此外,此值还必须是使用auth令牌和webhook有效负载(如其API文档中Callbacks中所述)计算的SHA256哈希。

我相信您在这里有2个选择。如果您只想验证代码是否正确接收了Webhook,则可以暂时将verify_signature()行注释掉。 Viber(或任何Webhook源)不需要验证Webhook请求。通常,开发人员会假定像Viber提供的那样的库可以正确地测试其代码,因此通常无需再次测试其功能。您还可以考虑对函数进行模拟,因为在这种情况下这非常简单。

如果您真的想测试Viber的签名验证,那么您将需要实现我首先提到的两个原因。基本上,这是在测试Webhook发送代码中需要执行的操作。请注意,我仅在下面包括了您需要的新代码,请合并到其他测试代码中。

import json
import hmac
import hashlib

# Compute SHA256 hex digest signature using auth token and payload.
auth_token = 'xxx-xxx-xxx'
signature = hmac.new(
    key=auth_token.encode('ascii'),
    msg=data.encode('ascii'),
    digestmod=hashlib.sha256
).hexdigest()

# Send test webhook request with computed signature in header.
response = requests.post(
    webhook_url,
    data=json.dumps(data),
    headers={
        'X-Viber-Content-Signature': signature,
        'Content-Type': 'application/json'
    },
    verify='E:\\Docs\\learn_py\\viberbot\\certificate.pem'
)

请注意,@ tukan指出了viber-bot-python存储库中的_calculate_message_signature()函数,该函数显示了如何计算签名。

在Flask中测试文件上传 - python

我在Flask集成测试中使用Flask-Testing。我有一个表单,该表单具有我要为其编写测试的徽标的文件上传,但是我不断收到错误消息:TypeError: 'str' does not support the buffer interface。我正在使用Python3。我找到的最接近的答案是this,但是它对我不起作用。这是我的许多尝…

对于DataFrame的每一行,在给定条件的情况下获取第一列的索引到新列中 - python

这是我的数据框的一部分。data = [ ['1245', np.nan, np.nan, 1.0, 1.0, ''], ['1246', np.nan, 1.0, 1.0, 1.0, ''], ['1247', 1.0, 1.0, 1.0, 1.0, �…

Spacy如何将标记标签整体化? - python

在包含#标签(例如tweet)的句子中,spacy的令牌生成器将标签分为两个令牌:import spacy nlp = spacy.load('en') doc = nlp(u'This is a #sentence.') [t for t in doc] 输出:[This, is, a, #, sentence, .…

如果__name__ =='__main__',则为Python的Powershell等效项: - python

我真的很喜欢python的功能,例如:if __name__ == '__main__': #setup testing code here #or setup a call a function with parameters and human format the output #etc... 很好,因为我可以将Python脚本文件…

在熊猫中,如何从单词列表或单词集中选择数据框中的短语? - python

在Python3和熊猫中,我具有数据框:df_projetos_api_final.info() <class 'pandas.core.frame.DataFrame'> Int64Index: 93631 entries, 1 to 93667 Data columns (total 21 columns): AnoMat…