为推特注册 webhook 的程序是什么？

Question

我正在尝试通过 twitter 文档中给出的 curl 命令将 url 注册为 twitter 上的 webhook。我认为 Twitter 文档有点过时或不正确。

 curl --request POST --url "https://api.twitter.com/1.1/account_activity/all/prod/webhooks.json?url=https%3A%2F%2FXXXX.com%2Fwebhook%2Ftwitter" --header "authorization: OAuth oauth_consumer_key='XXXXXXXXXXXXXXXXXXXXXX', oauth_nonce='ODgyNjc5NjQ0MTM3NzI4NTcwMjY4NDQ0', oauth_signature='7daMyzB1JClE4xv8hXNCimWpGtA%3D', oauth_signature_method='HMAC-SHA1', oauth_timestamp='1568620293', oauth_token='XXXXXXXXXXXXXXXXXXXXX', oauth_version='1.0'"

运行此代码后，我收到错误

{"errors":[{"code":215,"message":"Bad Authentication data."}]}

下面给出了我用来生成时间戳、oauth_nonce、oauth_signature 的代码。这些代码在python3中。我不太确定他们是否给出了正确的输出。

oauth_timestamp = str(int(time.time()))


#oauth_nonce
def get_nonce():
    nonce = ''.join(random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits)for x in range(32))
return nonce


#oauth_signature

def escape(s):
"""Percent Encode the passed in string"""
    return urllib.parse.quote_plus(s, safe='~')


def create_auth_header(parameters):
"""For all collected parameters, order them and create auth header"""
    ordered_parameters = {}
    ordered_parameters =  collections.OrderedDict(sorted(parameters.items()))
    auth_header = (
    '%s="%s"' % (k, v) for k, v in ordered_parameters.items())
    val = "OAuth " + ', '.join(auth_header)
    return val


def generate_signature(method, url, url_parameters, oauth_parameters, oauth_consumer_key, oauth_consumer_secret, oauth_token_secret=None, status=None):
    """Create the signature base string"""

    #Combine parameters into one hash
    temp = collect_parameters(oauth_parameters, status, url_parameters)

    #Create string of combined url and oauth parameters
    parameter_string = stringify_parameters(temp)

    #Create your Signature Base String
    signature_base_string = (
    method.upper() + '&' +
    escape(str(url)) + '&' +
    escape(parameter_string)
    )

    #Get the signing key
    signing_key = create_signing_key(oauth_consumer_secret, oauth_token_secret)

    return calculate_signature(signing_key, signature_base_string)



 def stringify_parameters(parameters):
     """Orders parameters, and generates string representation of parameters"""
     output = ''
     ordered_parameters = {}
     ordered_parameters =  collections.OrderedDict(sorted(parameters.items()))

     counter = 1
     for k, v in ordered_parameters.items():
          output += escape(str(k)) + '=' + escape(str(v))
          if counter < len(ordered_parameters):
              output += '&'
              counter += 1

     return output



def collect_parameters(oauth_parameters, status, url_parameters):
    """Combines oauth, url and status parameters"""
    #Add the oauth_parameters to temp hash
    temp = oauth_parameters.copy()

    #Add the status, if passed in.  Used for posting a new tweet
    if status is not None:
        temp['status'] = status

    #Add the url_parameters to the temp hash
    for k, v in url_parameters.items():
        temp[k] = v

    return temp




def calculate_signature(signing_key, signature_base_string):
    """Calculate the signature using SHA1"""
    hashed = hmac.new(signing_key,    signature_base_string.encode('utf-8'), hashlib.sha1)


    sig = binascii.b2a_base64(hashed.digest())[:-1]

    return escape(sig)


def create_signing_key(oauth_consumer_secret, oauth_token_secret):
    """Create key to sign request with"""
    signing_key = escape(oauth_consumer_secret) + '&'

    signing_key += escape(oauth_token_secret)

    return signing_key.encode('utf-8')




 oauth_parameters = {
    'oauth_timestamp': str(int(time.time())),
    'oauth_signature_method': "HMAC-SHA1",
    'oauth_version': "1.0",
    'oauth_token': "vvvvvvvvvvvvvvv",
    'oauth_nonce': get_nonce(),
    'oauth_consumer_key': 'bbbbbbbbbbbbbbb'
     }


 oauth_parameters['oauth_signature'] = generate_signature(
    method,
    url,
    url_parameters, oauth_parameters,
    consumer_key,
    consumer_secret,
    access_token_secret
     )


 auth_headers = {'Authorization': create_auth_header(oauth_parameters),
            'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'
                 }


 auth_data = {
       'grant_type': 'client_credentials'
  }


 webhook_endpoint ='https%3A%2F%2Fmy-site.com%2Fwebhook%2Ftwitter'

 url = 'https://api.twitter.com/1.1/account_activity/all/env/webhooks.json?url={}'.format(webhook_endpoint)

 method = "post"
 url_parameters = {
    'exclude_replies': 'true'
     }

 r = requests.post(url, headers=auth_headers, data=auth_data)

请告诉我哪个代码不正确，或者我做错了什么？如果您可以在答案中编写下面的代码，那将是有益的。

Answer 1

好的，所以我终于能够注册我的 webhook，以下是我遵循的步骤

1）首先我在rbenv的帮助下安装了ruby

2）然后我安装了twurl使用

gem install twurl

3）现在使用以下命令授权您的应用程序

twurl authorize --consumer-key key --consumer-secret secret

4）在命令提示符下运行上述命令后，您将获得输出为

Go to https://api.twitter.com/oauth/authorize?oauth_consumer_key=xxxxxxxxxxxxx&oauth_nonce=xxxxxxxxxxx&oauth_signature=xxxxxxxxxxxx&oauth_signature_method=HMAC-SHA1&oauth_timestamp=1568986465&oauth_token=xxxxxxx&oauth_version=1.0 and paste in the supplied PIN

5) 只需转到 url ，twitter 页面将打开 9876506 之类的 pin，只需复制 pin 并粘贴到命令行中即可。

6）然后你会在你的cmd中得到一个授权成功的输出。

7) 现在您已授权您的应用程序。

8) 现在在注册 webhook 之前，您在 twitter 应用程序的详细信息部分中提供的回调 url 应该生成一个 crc 令牌。

9) 如果您的域是“www.my_domain.com”，则创建一个端点为“www.my_domain.com/webhook/twitter”并添加以下代码以在您的 Web 应用程序中生成 crc。

def twitterCrcValidation():    
    crc = request.args['crc_token']    
    validation = hmac.new(key=bytes(CONSUMER_SECRET, 'utf-8'),msg=bytes(crc, 'utf-8'),
    digestmod = hashlib.sha256
    )
    digested = base64.b64encode(validation.digest())
    response = {
               'response_code':200,
               'response_token': 'sha256=' + format(str(digested)[2:-1])
                }
    print('responding to CRC call')    
    return json.dumps(response)

10) 现在运行以下命令来注册 webhook，代替放置您的环境名称，例如“prod”

twurl -X POST "/1.1/account_activity/all/<environment>/webhooks.json?url=https://www.my_domain.com/webhooks/twitter"

11) 在此之后，您将得到如下的 json 响应

{"id":"1174326336837472256",
 "url":"https://www.my_domain.com/webhooks/twitter",
 "valid":true,
  "created_timestamp":"2019-09-18 14:16:31 +0000"
 }

12) 最后你的 url 被注册为 webhook。

PS ----Twitter 每 24 小时向您的端点发送 crc 令牌，以检查 url 端点是否处于活动状态。因此，包含 url 端点的服务器应该处于运行状态。我在 heroku 服务器上部署了包含 url 端点的 Flask 应用程序。因此，与 ngrok 不同，heruko 为您提供了一个带有单个 url 的免费服务器。在 ngrok 中，每当您运行服务器时，您都会获得一个新的 url，这对于 twitter 来说不是必需的。但是如果您不刷新网站，heroku 服务器每 10 分钟就会休眠一次。所以 twitter 可能已经在我的网站上运行了 crc 检查，由于 heroku 的政策，我的网站处于睡眠状态，我的 webhook id 无效。

您可以通过运行以下给出的命令使您的 webhook id 再次有效。

twurl -X PUT "/1.1/account_activity/all/prod/webhooks/11743xxxxxxxxxx6.json"

然后您可以通过运行以下命令来获取有关您的 webhook 的信息

twurl -X GET "/1.1/account_activity/all/webhooks.json" --header authorization:bearer token