twitter bots: Merge twitter-bot and twitter-search-bot.

This commit is contained in:
derAnfaenger 2017-08-24 13:32:44 +02:00 committed by Tim Abbott
parent 623557d189
commit 6630deda6a
2 changed files with 144 additions and 273 deletions

View file

@ -32,61 +32,99 @@ from six.moves.configparser import ConfigParser, NoSectionError, NoOptionError
import zulip import zulip
VERSION = "0.9" VERSION = "0.9"
CONFIGFILE = os.path.expanduser("~/.zulip_twitterrc") CONFIGFILE = os.path.expanduser("~/.zulip_twitterrc")
CONFIGFILE_INTERNAL = os.path.expanduser("~/.zulip_twitterrc_internal") INSTRUCTIONS = r"""
twitter-bot --site=https://zulip.example.com --search="@nprnews,quantum physics"
Send Twitter tweets to a Zulip stream.
Depends on: https://github.com/bear/python-twitter version 3.1
To use this script:
0. Use `pip install python-twitter` to install `python-twitter`
1. Set up Twitter authentication, as described below
2. Set up a Zulip bot user and download its `.zuliprc` config file to `~/.zuliprc`
3. Subscribe the bot to the stream that will receive Twitter updates (default stream: twitter)
4. Test the script by running it manually, like this:
twitter-bot --site=<your.zulip.server> --search="<search-query>"
or
twitter-bot --site=<your.zulip.server> --twitter-name="<your-twitter-handle>"
4. Configure a crontab entry for this script. A sample crontab entry
that will process tweets every 5 minutes is:
*/5 * * * * /usr/local/share/zulip/integrations/twitter/twitter-bot [options]
== Setting up Twitter authentications ==
Run this on a personal or trusted machine, because your API key is
visible to local users through the command line or config file.
This bot uses OAuth to authenticate with Twitter. Please create a
~/.zulip_twitterrc with the following contents:
[twitter]
consumer_key =
consumer_secret =
access_token_key =
access_token_secret =
In order to obtain a consumer key & secret, you must register a
new application under your Twitter account:
1. Go to http://dev.twitter.com
2. Log in
3. In the menu under your username, click My Applications
4. Create a new application
Make sure to go the application you created and click "create my
access token" as well. Fill in the values displayed.
"""
def write_config(config, configfile_path): def write_config(config, configfile_path):
# type: (ConfigParser, int, int) -> None # type: (ConfigParser, str) -> None
with open(configfile_path, 'w') as configfile: with open(configfile_path, 'w') as configfile:
config.write(configfile) config.write(configfile)
parser = zulip.add_default_arguments(argparse.ArgumentParser(r""" parser = zulip.add_default_arguments(argparse.ArgumentParser("Fetch tweets from Twitter."))
parser.add_argument('--instructions',
twitter-bot --user foo@example.com --api-key 0b1c2d3e4f5a6b7c8d9e0f1a2b3c4d5 --twitter-id twitter_handle --site=https://zulip.example.com action='store_true',
help='Show instructions for the twitter bot setup and exit'
Slurp tweets on your timeline into a specific zulip stream. )
Run this on your personal machine. Your API key and twitter id
are revealed to local users through the command line or config
file.
This bot uses OAuth to authenticate with twitter. Please create a
~/.zulip_twitterrc with the following contents:
[twitter]
consumer_key =
consumer_secret =
access_token_key =
access_token_secret =
In order to obtain a consumer key & secret, you must register a
new application under your twitter account:
1. Go to http://dev.twitter.com
2. Log in
3. In the menu under your username, click My Applications
4. Create a new application
Make sure to go the application you created and click "create my
access token" as well. Fill in the values displayed.
Depends on: https://github.com/bear/python-twitter version 3.1
(`pip install python-twitter`)
"""))
parser.add_argument('--twitter-id',
help='Twitter username to poll for new tweets from"',
metavar='URL')
parser.add_argument('--stream',
help='Default zulip stream to write tweets to')
parser.add_argument('--limit-tweets', parser.add_argument('--limit-tweets',
default=15, default=15,
type=int, type=int,
help='Maximum number of tweets to push at once') help='Maximum number of tweets to send at once')
parser.add_argument('--search',
dest='search_terms',
help='Terms to search on',
action='store')
parser.add_argument('--stream',
dest='stream',
help='The stream to which to send tweets',
default="twitter",
action='store')
parser.add_argument('--twitter-name',
dest='twitter_name',
help='Twitter username to poll new tweets from"')
options = parser.parse_args() opts = parser.parse_args()
if not options.twitter_id: if opts.instructions:
parser.error('You must specify --twitter-id') print(INSTRUCTIONS)
sys.exit()
if all([opts.search_terms, opts.twitter_name]):
parser.error('You must only specify either a search term or a username.')
if opts.search_terms:
client_type = 'ZulipTwitterSearch/'
CONFIGFILE_INTERNAL = os.path.expanduser("~/.zulip_twitterrc_fetchsearch")
elif opts.twitter_name:
client_type = 'ZulipTwitter/'
CONFIGFILE_INTERNAL = os.path.expanduser("~/.zulip_twitteruserrc_fetchuser")
else:
parser.error('You must either specify a search term or a username.')
try: try:
config = ConfigParser() config = ConfigParser()
@ -101,59 +139,88 @@ try:
except (NoSectionError, NoOptionError): except (NoSectionError, NoOptionError):
parser.error("Please provide a ~/.zulip_twitterrc") parser.error("Please provide a ~/.zulip_twitterrc")
if not consumer_key or not consumer_secret or not access_token_key or not access_token_secret: if not all([consumer_key, consumer_secret, access_token_key, access_token_secret]):
parser.error("Please provide a ~/.zulip_twitterrc") parser.error("Please provide a ~/.zulip_twitterrc")
try:
since_id = config_internal.getint('twitter', 'since_id')
except (NoOptionError, NoSectionError):
since_id = 0
try:
previous_twitter_name = config_internal.get('twitter', 'twitter_name')
except (NoOptionError, NoSectionError):
previous_twitter_name = ''
try:
previous_search_terms = config_internal.get('twitter', 'search_terms')
except (NoOptionError, NoSectionError):
previous_search_terms = ''
try: try:
import twitter import twitter
except ImportError: except ImportError:
parser.error("Please install twitter-python") parser.error("Please install python-twitter")
api = twitter.Api(consumer_key=consumer_key, api = twitter.Api(consumer_key=consumer_key,
consumer_secret=consumer_secret, consumer_secret=consumer_secret,
access_token_key=access_token_key, access_token_key=access_token_key,
access_token_secret=access_token_secret) access_token_secret=access_token_secret)
user = api.VerifyCredentials() user = api.VerifyCredentials()
if not user.id: if not user.id:
print("Unable to log in to twitter with supplied credentials. Please double-check and try again") print("Unable to log in to twitter with supplied credentials. Please double-check and try again")
sys.exit() sys.exit(1)
try:
since_id = config.getint('twitter', 'since_id')
except (NoOptionError, NoSectionError):
since_id = -1
try:
user_id = config.get('twitter', 'user_id')
except (NoOptionError, NoSectionError):
user_id = options.twitter_id
client = zulip.Client( client = zulip.Client(
email=options.zulip_email, email=opts.zulip_email,
api_key=options.zulip_api_key, api_key=opts.zulip_api_key,
site=options.zulip_site, site=opts.zulip_site,
client="ZulipTwitter/" + VERSION, client=client_type+VERSION,
verbose=True) verbose=True)
if since_id < 0 or options.twitter_id != user_id: if opts.search_terms:
# No since id yet, fetch the latest and then start monitoring from next time search_query = " OR ".join(opts.search_terms.split(","))
# Or, a different user id is being asked for, so start from scratch if since_id == 0 or opts.search_terms != previous_search_terms:
# Either way, fetch last 5 tweets to start off # No since id yet, fetch the latest and then start monitoring from next time
statuses = api.GetUserTimeline(screen_name=options.twitter_id, count=5) # Or, a different user id is being asked for, so start from scratch
else: # Either way, fetch last 5 tweets to start off
# We have a saved last id, so insert all newer tweets into the zulip stream statuses = api.GetSearch(search_query, count=5)
statuses = api.GetUserTimeline(screen_name=options.twitter_id, since_id=since_id) else:
# We have a saved last id, so insert all newer tweets into the zulip stream
statuses = api.GetSearch(search_query, since_id=since_id)
elif opts.twitter_name:
if since_id == 0 or opts.twitter_name != previous_twitter_name:
# Same strategy as for search_terms
statuses = api.GetUserTimeline(screen_name=opts.twitter_name, count=5)
else:
statuses = api.GetUserTimeline(screen_name=opts.twitter_name, since_id=since_id)
for status in statuses[::-1][:options.limit_tweets]: for status in statuses[::-1][:opts.limit_tweets]:
# https://twitter.com/eatevilpenguins/status/309995853408530432
composed = "%s (%s)" % (status.user.name, status.user.screen_name) composed = "%s (%s)" % (status.user.name, status.user.screen_name)
url = "https://twitter.com/%s/status/%s" % (status.user.screen_name, status.id)
content = status.text
if opts.search_terms:
search_term_used = None
for term in opts.search_terms.split(","):
if term.lower() in content.lower():
search_term_used = term
break
# For some reason (perhaps encodings or message tranformations we
# didn't anticipate), we don't know what term was used, so use a
# default.
if not search_term_used:
search_term_used = "mentions"
subject = search_term_used
elif opts.twitter_name:
subject = composed
message = { message = {
"type": "stream", "type": "stream",
"to": [options.stream], "to": [opts.stream],
"subject": composed, "subject": subject,
"content": status.text, "content": url
} }
ret = client.send_message(message) ret = client.send_message(message)
@ -168,6 +235,7 @@ for status in statuses[::-1][:options.limit_tweets]:
if 'twitter' not in config_internal.sections(): if 'twitter' not in config_internal.sections():
config_internal.add_section('twitter') config_internal.add_section('twitter')
config_internal.set('twitter', 'since_id', str(since_id)) config_internal.set('twitter', 'since_id', str(since_id))
config_internal.set('twitter', 'user_id', str(user)) config_internal.set('twitter', 'search_terms', str(opts.search_terms))
config_internal.set('twitter', 'twitter_name', str(opts.twitter_name))
write_config(config_internal, CONFIGFILE_INTERNAL) write_config(config_internal, CONFIGFILE_INTERNAL)

View file

@ -1,197 +0,0 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Twitter search integration for Zulip
#
# Copyright © 2014 Zulip, Inc.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
from __future__ import print_function
import os
import sys
import argparse
from six.moves.configparser import ConfigParser, NoSectionError, NoOptionError
import zulip
VERSION = "0.9"
CONFIGFILE = os.path.expanduser("~/.zulip_twitterrc")
CONFIGFILE_INTERNAL = os.path.expanduser("~/.zulip_twitterrc_internal")
def write_config(config, configfile_path):
# type: (ConfigParser, int) -> None
with open(configfile_path, 'w') as configfile:
config.write(configfile)
parser = zulip.add_default_arguments(argparse.ArgumentParser(r"""
twitter-search-bot --user username@example.com --api-key a0b1c2d3e4f5a6b7c8d9e0f1a2b3c4d5 \
--search="@nprnews,quantum physics"
Send Twitter search results to a Zulip stream.
Depends on: https://github.com/bear/python-twitter version 3.1
To use this script:
0. Use `pip install python-twitter` to install `python-twitter`.
1. Set up Twitter authentication, as described below
2. Subscribe to the stream that will receive Twitter updates (default stream: twitter)
3. Test the script by running it manually, like this:
/usr/local/share/zulip/integrations/twitter/twitter-search-bot \
--search="@nprnews,quantum physics" --site=https://zulip.example.com
4. Configure a crontab entry for this script. A sample crontab entry
that will process tweets every 5 minutes is:
*/5 * * * * /usr/local/share/zulip/integrations/twitter/twitter-search-bot [options]
== Setting up Twitter authentications ==
Run this on a personal or trusted machine, because your API key is
visible to local users through the command line or config file.
This bot uses OAuth to authenticate with Twitter. Please create a
~/.zulip_twitterrc with the following contents:
[twitter]
consumer_key =
consumer_secret =
access_token_key =
access_token_secret =
In order to obtain a consumer key & secret, you must register a
new application under your Twitter account:
1. Go to http://dev.twitter.com
2. Log in
3. In the menu under your username, click My Applications
4. Create a new application
Make sure to go the application you created and click "create my
access token" as well. Fill in the values displayed.
"""))
parser.add_argument('--search',
dest='search_terms',
help='Terms to search on',
action='store')
parser.add_argument('--stream',
dest='stream',
help='The stream to which to send tweets',
default="twitter",
action='store')
parser.add_argument('--limit-tweets',
default=15,
type=int,
help='Maximum number of tweets to send at once')
opts = parser.parse_args()
if not opts.search_terms:
parser.error('You must specify a search term.')
try:
config = ConfigParser()
config.read(CONFIGFILE)
config_internal = ConfigParser()
config_internal.read(CONFIGFILE_INTERNAL)
consumer_key = config.get('twitter', 'consumer_key')
consumer_secret = config.get('twitter', 'consumer_secret')
access_token_key = config.get('twitter', 'access_token_key')
access_token_secret = config.get('twitter', 'access_token_secret')
except (NoSectionError, NoOptionError):
parser.error("Please provide a ~/.zulip_twitterrc")
if not (consumer_key and consumer_secret and access_token_key and access_token_secret):
parser.error("Please provide a ~/.zulip_twitterrc")
try:
since_id = config_internal.getint('search', 'since_id')
except (NoOptionError, NoSectionError):
since_id = 0
try:
import twitter
except ImportError:
parser.error("Please install twitter-python")
api = twitter.Api(consumer_key=consumer_key,
consumer_secret=consumer_secret,
access_token_key=access_token_key,
access_token_secret=access_token_secret)
user = api.VerifyCredentials()
if not user.id:
print("Unable to log in to twitter with supplied credentials. Please double-check and try again")
sys.exit()
client = zulip.Client(
email=opts.zulip_email,
api_key=opts.zulip_api_key,
site=opts.zulip_site,
client="ZulipTwitterSearch/" + VERSION,
verbose=True)
search_query = " OR ".join(opts.search_terms.split(","))
statuses = api.GetSearch(search_query, since_id=since_id)
for status in statuses[::-1][:opts.limit_tweets]:
# https://twitter.com/eatevilpenguins/status/309995853408530432
composed = "%s (%s)" % (status.user.name,
status.user.screen_name)
url = "https://twitter.com/%s/status/%s" % (status.user.screen_name,
status.id)
content = status.text
search_term_used = None
for term in opts.search_terms.split(","):
if term.lower() in content.lower():
search_term_used = term
break
# For some reason (perhaps encodings or message tranformations we
# didn't anticipate), we don't know what term was used, so use a
# default.
if not search_term_used:
search_term_used = "mentions"
message = {
"type": "stream",
"to": [opts.stream],
"subject": search_term_used,
"content": url,
}
ret = client.send_message(message)
if ret['result'] == 'error':
# If sending failed (e.g. no such stream), abort and retry next time
print("Error sending message to zulip: %s" % ret['msg'])
break
else:
since_id = status.id
if 'search' not in config_internal.sections():
config_internal.add_section('search')
config_internal.set('search', 'since_id', str(since_id))
write_config(config_internal, CONFIGFILE_INTERNAL)