我如何利用 Python、搜索推文 API 和 Twilio 解决纽约停车问题

2025-06-11

我如何利用 Python、搜索推文 API 和 Twilio 解决纽约停车问题

车

本教程最初发布于developer.twitter.com。

由于我在纽约市有车，所以我的车必须遵守该市“路边交替停车”的规定。这意味着大多数晚上，我都需要在清晨社区街道清扫之前把车移走。我已经养成了每晚睡觉前把车移走的习惯。有时我在这方面做得有点过头了，经常在不需要的时候也把车移走。由于路边交替停车在节假日或恶劣天气下经常被取消，所以我注册了一个推特账号@NYCASP，它会每天以及在紧急情况下发布信息。

为了解决这个问题，我使用了 Twitter 数据和 Twilio。现在，每当我不需要移动车辆时，我都会收到一条短信。我使用了Python 的 Search Tweets 包装器从前面提到的 Twitter 账号和 Twilio 中抓取数据，这样每当我不需要移动车辆时，我都会收到一条短信。我编写了一个脚本，它会检查“暂停”和“明天”这两个词是否出现在推文中，如果满足这些条件，就会向我发送短信。本教程将引导您了解我是如何创建此解决方案的。

确定我们的要求

我编写的代码是用 Python 3.6 编写的。我建议也使用这个版本。要使用 pip，您需要先安装 pip。在本教程中，我们将使用Atom。您需要确保已安装命令行工具。

我们还需要在命令行中运行以下命令，以确保设置了正确的依赖项：

pip install pandas
pip install twilio
pip install searchtweets
Enter fullscreen mode Exit fullscreen mode

现在让我们在命令行上为该项目创建一个新目录并进入该目录。

mkdir parking
cd parking
Enter fullscreen mode Exit fullscreen mode

设置 Jupyter 笔记本

我使用 Jupyter Notebook 编写代码，这样就可以以交互方式处理数据。这对我来说在这个项目中非常重要，因为我可以看到我搜索的推文。您可以使用预装了 Jupyter 的 Anaconda Python 发行版，或者使用以下语法：

pip install jupyter
Enter fullscreen mode Exit fullscreen mode

安装 Jupyter 后，您可以通过在命令行中输入以下内容来运行笔记本：

jupyter notebook
Enter fullscreen mode Exit fullscreen mode

如果您需要随时停止 Jupyter 笔记本，您可以按ctrl+c来停止。

如果设置正确，浏览器中应该会显示目录中的内容列表。点击右上角的new。然后，点击Python 3下拉菜单中显示的 ，启动一个笔记本。启动后，点击 ，将untitled笔记本名称更改为 parking。

设置与 Twilio 的连接

您需要创建一个脚本，以便我们能够连接到Twilio。为此，您需要一个 Twilio 帐户，并查看该主题的入门文档。要创建一个脚本来连接 Twilio，请打开我们正在处理的目录。

在我们的命令行中输入：

atom .
Enter fullscreen mode Exit fullscreen mode

我们还需要为我们的 Twilio 帐户设置环境变量。创建 Twilio 帐户后，您可以在控制台中找到这些变量：

export 'TWILIO_ACCOUNT_SID'='xxxxxxxxxxxxxxxxxxx'
export 'TWILIO_AUTH_TOKEN'='xxxxxxxxxxxxxxxxxxxxxxx'
export 'TWILIO_PHONE_NUMBER'='xxxxxxxxxxx'
export 'CELL_PHONE_NUMBER'='xxxxxxxxxxx'
Enter fullscreen mode Exit fullscreen mode

当我们将电话号码添加为环境变量时，我们需要将电话号码与国家代码、区号和号码放在一个字符串中。例如，如果我在美国，电话号码是 (555) 555-5555，那么我的电话号码变量的字符串应该是“15555555555”。

从这里我们可以添加一个名为的文件twilio_connect_demo.py：

我们首先添加我们需要的两个导入语句：

import os
from twilio.rest import Client
Enter fullscreen mode Exit fullscreen mode

现在我们可以编写一个函数来帮助我们连接到 Twilio API。

def twilio_connect():
    account_sid = os.environ.get('TWILIO_ACCOUNT_SID')
    auth_token = os.environ.get('TWILIO_AUTH_TOKEN')
    client = Client(account_sid, auth_token)
    return client
Enter fullscreen mode Exit fullscreen mode

现在我们可以编写另一个函数来发送短信：

def send_message(client):
    return client.messages.create(from_=os.environ.get('TWILIO_PHONE_NUMBER'),
                       to=os.environ.get('CELL_PHONE_NUMBER'),
                       body="You don't have to move your car tonight. Enjoy your night!")
Enter fullscreen mode Exit fullscreen mode

现在我们有了发送短信的脚本，这在本教程的后面会很有帮助。

导入我们需要的内容

在笔记本的第一个单元格中，我们需要导入要使用的库。我们将使用datetime，它是 Python 标准库的一部分，允许我们使用日期来运行搜索。我们还需要使用 pandas，我们将使用它来获取数据并将其转换为数据框。要使用搜索推文 API，我们将导入搜索推文 Python 包装器。我们需要此库中的ResultStream、gen_rule_playload和load_credentials函数。我们还将从刚刚编写的脚本中导入两个函数，以便稍后在代码中发送短信。

import datetime
import pandas as pd
from searchtweets import ResultStream, gen_rule_payload, load_credentials

from twilio_connect_demo import twilio_connect, send_message
Enter fullscreen mode Exit fullscreen mode

要运行此代码，我们可以按播放按钮或使用键盘快捷键shift-enter。

连接到 Twitter API

您需要创建一个可以连接到 API 的 Twitter 应用。在此之前，您需要一个已获批准的开发者账户。完成初始设置后，您需要创建一个开发者账户。您可以点击此处申请开发者账户。

您的帐户获得批准后，您将需要创建一个应用程序。然后，为“搜索推文：30 天”环境设置该应用程序，并设置开发环境标签名称。

完成初始设置后，您需要在 Twitter 应用中找到您的消费者 API 密钥和 API 私钥。有关如何设置和查找密钥的更多信息，请参阅我们相关的文档。

在 Atom 中，我们需要添加一个包含应用程序密钥和机密的新文件。让我们创建一个名为 的文件，secret.yaml其中包含以下内容：

search_tweets_api:
  account_type: premium
  endpoint: https://api.twitter.com/1.1/tweets/search/30day/env_name.json
  consumer_key: xxxxxxxxxxxxxxxxxxx
  consumer_secret: xxxxxxxxxxxxxxxxxxx
Enter fullscreen mode Exit fullscreen mode

端点中env_name的 是您在developer.twitter.com上创建的开发环境的名称。您需要将其更改为您自己的开发环境的名称。

在将此文件推送到 GitHub 之前，请务必将其添加到您的gitignore 文件中.gitignore。有关使用 gitignore 文件的更多信息，请查看此页面。您可能还需要查看我们关于如何确保令牌安全的指南。

使用 Twitter 数据

回到运行 Jupyter 笔记本的浏览器，我们可以创建一个名为 search_args 的变量，我们可以在其中加载我们的凭据以连接到 Twitter API。

search_args = load_credentials(filename="secret.yaml",
                              yaml_key="search_tweets_api",
                              env_overwrite=False)
Enter fullscreen mode Exit fullscreen mode

由于我们希望搜索的日期是动态的，这意味着我们不必输入日期，所以我们需要创建一些变量。首先，我们创建一个名为的变量，它从我们之前导入的库today中获取今天的日期。datetime

today = datetime.date.today()
print(today)
Enter fullscreen mode Exit fullscreen mode

运行这行代码，我们应该返回今天的日期。为了获取距今天 30 天后的开始日期，我们需要创建一个名为 start date 的变量。它将使用 datetime 库中内置的 timedelta 方法，获取今天的日期并减去 30 天。

start_date = today + datetime.timedelta(-30)
print(start_date)
Enter fullscreen mode Exit fullscreen mode

运行此命令后，我们应该获取到从今天起 30 天后的开始日期。现在，我们可以将日期传入搜索推文的 Python 包装器中，该包装器允许我们创建一条规则，允许我们从@NYCASP 的 Twitter 账号gen_rule_payload中提取过去 30 天的数据。

rule = gen_rule_payload("from:NYCASP",
                        from_date=str(start_date),
                        to_date=str(today),
                        results_per_call=500,
                       )
print(rule)
Enter fullscreen mode Exit fullscreen mode

运行此行代码后，您应该会得到可以提前粘贴到 REST 客户端（例如Postman或Insomnia）正文中以查看数据的规则。

{"query": "from:NYCASP", "maxResults": 500, "toDate": "201811060000", "fromDate": "201810070000"}
Enter fullscreen mode Exit fullscreen mode

现在让我们创建一个名为 的变量rs，它是 ResultStream 的缩写，因为我们将创建一个推文结果流。为此，我们可以将刚刚创建的规则以及我们的凭证和其他一些参数传递给此变量，如下所示：

rs = ResultStream(rule_payload=rule,
                 max_results=500,
                 max_pages=1,
                 **search_args)

print(rs)
Enter fullscreen mode Exit fullscreen mode

一旦我们打印出名为 rs 的变量，我们就可以看到传递给它的信息。它看起来应该像这样：

ResultStream: 
    {
    "username": null,
    "endpoint": "https://api.twitter.com/1.1/tweets/search/30day/env_name.json",
    "rule_payload": {
        "query": "from:NYCASP",
        "maxResults": 500,
        "toDate": "201811060000",
        "fromDate": "201810070000"
    },
    "tweetify": true,
    "max_results": 500
}
Enter fullscreen mode Exit fullscreen mode

我们可以使用属性将结果流传递到.stream()名为的变量中tweets。

tweets = rs.stream()
Enter fullscreen mode Exit fullscreen mode

从这里开始，我们可以将结果流转换为列表，以便以更健壮的方式处理这些数据。这时，我们就可以开始查看正在处理的推文了。让我们使用列表推导式打印出前 5 行，以确保我们走在正确的轨道上。

list_tweets = list(tweets)
[print(tweet.all_text, end='\n\n') for tweet in list_tweets[0:5]];
Enter fullscreen mode Exit fullscreen mode

当我们运行此代码时，我们将得到如下结果：

#NYCASP rules will be suspended tomorrow, Wednesday, November 7 for Diwali. Parking meters will be in effect.

#NYCASP rules are suspended today, November 6 for Election Day. Parking meters are in effect.

#NYCASP rules will be suspended tomorrow, Tuesday, November 6 for Election Day. Parking meters will be in effect.

#NYCASP rules are in effect today, November 5.

#NYCASP rules will be in effect tomorrow, Monday, November 5.
Enter fullscreen mode Exit fullscreen mode

现在我们知道数据接收正确了，可以创建两个列表，一个用于存储日期，一个用于存储推文文本。我们将使用两个空列表，并使用 for 循环来迭代数据。

tweet_text = []
tweet_date = []

for tweet in list_tweets:
    tweet_text.append(tweet['text'])
    tweet_date.append(tweet['created_at'])
Enter fullscreen mode Exit fullscreen mode

现在我们可以使用 pandas 创建一个由这两列组成的数据框。

df = pd.DataFrame({'tweet':tweet_text, 'date':tweet_date})
Enter fullscreen mode Exit fullscreen mode

我们可以使用 head 属性查看数据框的前 5 行。

df.head()
Enter fullscreen mode Exit fullscreen mode

我们将看到类似这样的内容：
df 图像

发送消息

现在我们已经拥有了正确格式的 Twitter 数据。我们可以设置当满足正确条件时发送消息。首先，我们需要创建一个名为 的变量client，用于连接到 Twilio。

client = twilio_connect()
Enter fullscreen mode Exit fullscreen mode

从这里我们可以设置逻辑，如果最后一条推文中出现“暂停”和“明天”这两个词，则向我们发送短信。

if 'suspended' in df['tweet'].values[0]:
    if 'tomorrow' in df['tweet'].values[0]:
        send_message(client=client)
        print('text sent')
    else:
        print('suspended but not tomorrow, no text sent')
else: 
    print('not suspended, no text sent')
Enter fullscreen mode Exit fullscreen mode

因此，我们应该根据明天街道两侧停车是否暂停的情况，获得相应的打印消息。这样我们就可以从命令行进行调试。

如果最后一条推文中出现“暂停”和“明天”这两个词，我们就会收到如下短信：
文本

要将此脚本下载为.py文件，请点击右上角的“文件”部分，然后选择“下载为”选项。然后，系统将提示您选择格式，请务必选择Python。

部署

目前情况下，除非您将短信部署到服务器，否则您只能在此时收到这条短信。如果您想在不需要移动车辆时收到短信，可以将此脚本设置为每天在服务器上使用 cron 作业运行。我已将其设置为每天晚上 7:30 运行。

后续步骤Next steps

完整代码可在此处找到。我在伦敦办公室的同事提到，这段代码可以轻松调整，以便与此句柄配合使用，以便她查看牛津地铁何时出现延误。在与其他人讨论这个项目时，他们提到，如果它能告知何时需要移动车辆，而不是何时不需要移动车辆，那么这个项目可能更有意义。您可以轻松修改代码来实现这一点。就目前而言，这段代码可以作为其他类似想法的模板。

如果这能激发您构建任何东西的灵感，请在论坛上告诉我们或在推特上@TwitterDev告诉我们。

链接：https://dev.to/xdevs/how-i-solved-my-nyc-parking-problem-with-python-the-search-tweets-api-and-twilio-1chp
Devto 翻译
PREV
掌握 Git Stash 工作流程
NEXT
教程 - 企业模块联合指南