我编写的代码是使用 Python 3.6 编写的。建议同样使用此版本。若要使用 pip，你首先要安装 pip。在本教程中，我们将使用 Atom。你需要确保已安装命令行工具。

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

pip install pandas
pip install twilio
pip install searchtweets

现在，在命令行上为这个项目创建一个新目录，并将其更改为该目录。

mkdir parking
cd parking

我使用 Jupyter 笔记本编写了代码，因此我能够以交互方式使用数据。在此项目中，这对我非常重要，这样我可以查看所搜索的推文。你可以使用已预安装 Jupyter 的 Python 发行版 Anaconda，也可以使用以下语法：

pip install jupyter

安装 Jupyter 后，可通过将此内容键入命令行来运行笔记本：

jupyter notebook

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

如果已正确完成设置，目录中一系列项目应在浏览器中启动。在右上角，单击显示 new 的位置。然后，你需要在下拉选项中单击显示 Python 3 的位置，以启动笔记本。启动笔记本后，单击显示 untitled 的位置，将笔记本名称更改为 parking。

你需要创建一个脚本，以便我们可以连接到 Twilio。为此，你需要一个 Twilio 账号，并需要查看有关此主题的入门指南文档。若要创建一个用于连接 Twilio 的脚本，我们需要打开正在使用的目录。

atom .

我们还需要为 Twilio 账号设置环境变量。创建 Twilio 账号之后，可在你的控制台中找到以下内容：

export 'TWILIO_ACCOUNT_SID'='xxxxxxxxxxxxxxxxxxx'
export 'TWILIO_AUTH_TOKEN'='xxxxxxxxxxxxxxxxxxxxxxx'
export 'TWILIO_PHONE_NUMBER'='xxxxxxxxxxx'
export 'CELL_PHONE_NUMBER'='xxxxxxxxxxx'

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

在此，可添加名为 twilio_connect_demo.py 的文件：

import os
from twilio.rest import Client

现在，我们可以编写一个函数来帮助我们连接到 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

现在，我们可以编写另一个函数来发送短信：python def send_message(client):

python
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!")

现在我们已经有一个可发送短信的脚本了，本教程稍后会用到该脚本。

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

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

from twilio_connect_demo import twilio_connect, send_message

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

你需要创建一个 Twitter 应用，以用于连接到 API。在执行此操作之前，需要一个获得批准的开发者账号。完成初始设置后，需要创建一个开发者账号。你可以在此处申请开发者账号。

账号获得批准后，你需要创建一个应用。然后，为搜索推文：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

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

建议确保先将此文件添加到你的 .gitignore，然后再将其推送到 GitHub。有关使用 gitignore 文件的详细信息，请查看此页面。另外建议你查看有关确保令牌安全的指南。

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

search_args = load_credentials(filename="secret.yaml",
yaml_key="search_tweets_api",
env_overwrite=False)

由于我们希望搜索的日期是动态的，即我们无需输入所需的日期，因此我们需要创建一些变量。首先创建一个名为 today 的变量，该变量可从我们之前导入的 datetime 库中获取今天的日期。

today = datetime.date.today()
print(today)

当我们运行该行代码时，就可获得今天的日期。若要将开始日期设为今天之前的第 30 天，我们需要创建一个名为 start date 的变量。此过程将获取今天的日期，然后使用日期时间库中的内置时间间隔方法减去 30 天。

start_date = today + datetime.timedelta(-30)
print(start_date)

运行此代码后，就会获得今天之前的第 30 天的开始日期。现在，我们已获得日期并且可传递到从搜索推文 python 包装器导入的 gen_rule_payload 中，这样我们可以创建一个规则，实现从 @NYCASP twitter 用户名中拉取过去 30 天的数据。

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

运行该行代码后，你应获得此规则，还可将此规则粘贴到 Postman 或 Insomnia 等 REST 客户端的正文，以提前查看数据。

{"query": "from:NYCASP", "maxResults":500, "toDate":"201811060000", "fromDate":"201810070000"}

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

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

print(rs)

打印名为 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
}

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

tweets = rs.stream()

在此，我们可以将结果流转换为列表，以便以一种更可靠的方式使用此数据。此时我们将开始查看要使用的推文。我们来使用列表推导式打印前 5 行，只是为了确保我们的操作是正确的。

list_tweets = list(tweets)
[print(tweet.all_text, end=' ') for tweet in list_tweets[0:5]];

运行此代码时，我们将获得如下内容：

#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.

确定我们已正确接收数据之后，现在可以创建两个列表，一个用于日期，另一个用于推文文本。我们将创建两个空列表，并使用 for 循环迭代数据。

tweet_text = []
tweet_date = []

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

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

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

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

df.head()

我们将看到如下所示的内容：

如果要查看完整的推文列表，我们可以使用以下语法：

df

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

client = twilio_connect()

在此，我们可以设置这样的逻辑：如果上条推文中显示“suspended”和“tomorrow”两个词，则向我们发送短信。

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')

因此，根据明天是否暂停街道换边停车，我们应获得相应的打印消息。这样一来，我们就可以从所需的命令行进行调试。

如果上条推文中显示“suspended”和“tomorrow”两个词，我们将收到一条短信，如下所示：

若要将此脚本下载为 .py 文件，请单击右上角显示的文件，然后选择“下载为”选项。此时，系统会提示你选择格式，请务必选择 Python。

目前，你只能收到这一次短信，除非你已将脚本部署到服务器。如果你想要在每次无需挪车时都收到短信，你可以设置此脚本，使其通过 cron 作业每天在服务器上运行。我已进行设置，因此该脚本会在每天晚上 7:30 运行。