This repository has been archived by the owner on Jan 23, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 8
/
Copy pathstandalone_bot.py
153 lines (115 loc) · 4.75 KB
/
standalone_bot.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
"""gchatautorespond standalone (self-hosted) bot.
Use `auth` once to perform oauth and store a credentials file to the working directory.
Then, use `run` to start a bot from that credentials file.
Use SIGTERM to gracefully quiet or SIGQUIT to quit immediately.
Will exit and delete a credentials file upon revocation.
Usage:
standalone_bot.py auth
standalone_bot.py run <email> <autoresponse>
standalone_bot.py notify <email> [<autoresponse>]
standalone_bot.py (-h | --help)
Options:
-h --help Show this screen.
"""
from __future__ import print_function
from builtins import input
import os
from docopt import docopt
import httplib2 # included with oauth2client
from oauth2client.client import OAuth2WebServerFlow, AccessTokenRefreshError
import oauth2client.file
import warnings
warnings.filterwarnings('ignore', 'test worker credentials unavailable',)
os.environ['DJANGO_SETTINGS_MODULE'] = 'gchatautorespond.settings_standalone'
import django # noqa
django.setup()
from django.conf import settings # noqa
from gchatautorespond.lib.chatworker.bot import AutoRespondBot # noqa
class StandaloneBot(AutoRespondBot):
"""An AutoResponseBot meant to be run from a shell.
It's able to manage its own auth.
"""
def __init__(self, *args, **kwargs):
super(StandaloneBot, self).__init__(*args, **kwargs)
self.reconnect_delay = 5
self.add_event_handler('failed_auth', self.failed_auth)
def failed_auth(self, _):
"""This event handler is triggered in two cases:
* expired auth -> attempt to refresh and reconnect
* revoked auth -> shutdown
"""
credentials = self.get_oauth_credentials()
try:
self.logger.info("refreshing credentials")
credentials.refresh(httplib2.Http())
except AccessTokenRefreshError:
self.logger.warning("credentials revoked?")
self.oauth_storage.delete()
self.disconnect()
else:
self.logger.info("credentials refreshed")
self.password = credentials.access_token
self.credentials['access_token'] = credentials.access_token
self.oauth_storage.put(credentials)
self.disconnect(reconnect=True)
@property
def oauth_filename(self):
return "%s.oauth_credentials" % self.email
@property
def oauth_storage(self):
return oauth2client.file.Storage(self.oauth_filename)
def get_oauth_credentials(self):
oauth_credentials = self.oauth_storage.get()
if oauth_credentials is None:
raise IOError("could not retrieve oauth credentials from %r. Have you run `auth`?" % self.oauth_filename)
return oauth_credentials
def connect(self):
oauth_credentials = self.get_oauth_credentials()
self.password = oauth_credentials.access_token
return super(StandaloneBot, self).connect()
def perform_oauth():
"""Provides a series of prompts for a user to follow to authenticate.
Returns ``oauth2client.client.OAuth2Credentials`` when successful.
In most cases, this should only be run once per machine to store
credentials to disk, then never be needed again.
If the user refuses to give access,
``oauth2client.client.FlowExchangeError`` is raised.
"""
# This is a separate project from the hosted version, since we can't protect the client secret.
flow = OAuth2WebServerFlow(
'996043626456-32qgmel3mi3t27k32veicv1dmmf3t06s.apps.googleusercontent.com',
'gnmF3W2Q8Ull9AO09Z-1esqd',
' '.join(['https://www.googleapis.com/auth/googletalk', 'email']),
'urn:ietf:wg:oauth:2.0:oob',
)
auth_uri = flow.step1_get_authorize_url()
print()
print("Visit the following url:\n %s" % auth_uri)
code = input("Follow the prompts, then paste the auth code here and hit enter: ")
credentials = flow.step2_exchange(code)
storage = StandaloneBot(credentials.id_token['email'], None, None, None, None, None).oauth_storage
storage.put(credentials)
return credentials
def main(bot_cls=None):
arguments = docopt(__doc__)
if bot_cls is None:
bot_cls = StandaloneBot
if arguments['auth']:
perform_oauth()
elif arguments['run']:
bot = bot_cls(
arguments['<email>'], None, None, arguments['<autoresponse>'], False, None
)
bot.connect()
bot.process(block=True)
elif arguments['notify']:
bot = bot_cls(
arguments['<email>'], None, None, arguments['<autoresponse>'],
send_email_notifications=True,
notify_email=arguments['<email>'],
disable_responses=(not bool(arguments['<autoresponse>']))
)
bot.connect()
bot.process(block=True)
if __name__ == '__main__':
main()