build.rsudp.c_tweet.Tweeter._when_alarm() - Code Metrics - Inspection of "Adding test coverage" - raspishake/rsudp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#24)

by Ian

created 2021-10-16 22:50 UTC

build.rsudp.c_tweet.Tweeter._when_alarm() B

↳ Parent: build.rsudp.c_tweet

Complexity

Conditions

Size

Total Lines	39
Code Lines	30

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	30
dl	0
loc	39
rs	8.6933
c	0
b	0
f	0
cc	5
nop	2

import os, sys, platform
import pkg_resources as pr
import time
import math
import numpy as np
from datetime import datetime, timedelta
import rsudp.raspberryshake as rs
from rsudp import printM, printW, printE, helpers
import rsudp
from twython import Twython


class Tweeter(rs.ConsumerThread):
	'''
	.. versionadded:: 1.0.2

		The option to add extra text to tweets using the :code:`"extra_text"`
		setting in settings files built by this version and later.

	.. |raspishakeq| raw:: html

		<a href="https://twitter.com/raspishakeq" target="_blank">Raspberry Shake</a>

	.. |usgsbigquakes| raw:: html

		<a href="https://twitter.com/USGSBigQuakes" target="_blank">USGS</a>

	.. |emsc_lastquake| raw:: html

		<a href="https://twitter.com/LastQuake" target="_blank">EMSC</a>

	Twitter is a social media platform sometimes used for quickly
	distributing public alert information. It is used by many agencies
	including |raspishakeq|, |usgsbigquakes|, and |emsc_lastquake|.

	.. |tw_api_bots| raw:: html

		<a href="https://developer.twitter.com/apps" target="_blank">API bots</a>

	.. note::
		Twitter is more difficult and stricter when it comes to making
		|tw_api_bots| than many services.
		First, you must go through a relatively rigorous process of applying for
		a developer account, then making a Twitter "app", and then giving the app
		permission to post on your behalf. See :ref:`setting-up-twitter` for details.

		Once you've gone through that process, Twitter limits posting and makes
		its rules on rate limiting relatively difficult to nail down.
		Generally, the early 2020 rate limit for posting is 300 in a 3-hour span,
		however that can vary depending on whether or not Twitter thinks there is
		suspicious activity coming from your account.

		In general, if you are looking for a simple multi-platform notification
		service, it may be easier and more reliable to use the Telegram service
		instead. rsudp has a telegram module at
		:py:class:`rsudp.c_telegram.Telegram`. See :ref:`setting-up-telegram` for details.

	:param str consumer_key: Twitter calls this the "consumer key"
	:param str consumer_secret: Twitter calls this the "consumer key secret"
	:param str access_token: Twitter calls this the "consumer access token"
	:param str access_secret: Twitter calls this the "consumer access secret"
	:param bool tweet_images: whether or not to send images. if False, only alerts will be sent.
	:type extra_text: bool or str
	:param extra_text: 103 additional characters to post as part of the twitter message.
	:param queue.Queue q: queue of data and messages sent by :class:`rsudp.c_consumer.Consumer`


	'''
	def __init__(self, consumer_key, consumer_secret, access_token, access_token_secret,
				 q=False, tweet_images=False, extra_text=False, testing=False,
				 ):
		"""
		Initialize the process
		"""
		super().__init__()
		self.sender = 'Tweeter'
		self.alive = True
		self.tweet_images = tweet_images
		self.testing = testing
		self.fmt = '%Y-%m-%d %H:%M:%S.%f'
		self.region = ' - region: %s' % rs.region.title() if rs.region else ''
		self.consumer_key = consumer_key
		self.consumer_secret = consumer_secret
		self.access_token = access_token
		self.access_token_secret = access_token_secret
		self.last_message = False

		self._resolve_extra_text(extra_text)

		if q:
			self.queue = q
		else:
			printE('no queue passed to consumer! Thread will exit now!', self.sender)
			sys.stdout.flush()
			self.alive = False
			sys.exit()

		if not self.testing:
			self.twitter = Twython(
				consumer_key,
				consumer_secret,
				access_token,
				access_token_secret
			)
		else:
			printW('The Twitter module will not post to Twitter in Testing mode.',
					self.sender, announce=False)


		self.livelink = 'live feed ➡️ https://stationview.raspberryshake.org/#?net=%s&sta=%s' % (rs.net, rs.stn)
		self.message0 = '(#RaspberryShake station %s.%s%s) Event detected at' % (rs.net, rs.stn, self.region)
		self.message1 = '(#RaspberryShake station %s.%s%s) Image of event detected at' % (rs.net, rs.stn, self.region)

		printM('Starting.', self.sender)


	def _resolve_extra_text(self, extra_text):
		allowable_len = 103	# length of string allowable given maximum message text & region
		if ((extra_text == '') or (extra_text == None) or (extra_text == False)):
			self.extra_text = ''
		else:
			extra_text = str(extra_text)
			len_ex_txt = len(extra_text)

			if len_ex_txt > allowable_len:
				printW('extra_text parameter is longer than allowable (%s chars) and will be truncated. Please keep extra_text at or below %s characters.' % (len_ex_txt, allowable_len), sender=self.sender)
				extra_text = extra_text[:allowable_len]

			self.extra_text =  ' %s' % (extra_text)


	def auth(self):
		self.twitter = Twython(
			self.consumer_key,
			self.consumer_secret,
			self.access_token,
			self.access_token_secret
		)


	def getq(self):
		d = self.queue.get()
		self.queue.task_done()

		if 'TERM' in str(d):
			self.alive = False
			printM('Exiting.', self.sender)
			sys.exit()
		else:
			return d


	def _when_alarm(self, d):
		'''
		Send a tweet when you get an ``ALARM`` message.

		:param bytes d: queue message
		'''
		event_time = helpers.fsec(helpers.get_msg_time(d))
		self.last_event_str = '%s' % (event_time.strftime(self.fmt)[:22])
		message = '%s %s UTC%s - %s' % (self.message0, self.last_event_str, self.extra_text, self.livelink)
		response = None
		try:
			printM('Tweet: %s' % (message), sender=self.sender)
			if not self.testing:
				response = self.twitter.update_status(status=message, lat=rs.inv[0][0].latitude,
														long=rs.inv[0][0].longitude,
														geo_enabled=True, display_coordinates=True)
														# location will only stick to tweets on accounts that have location enabled in Settings
				url = 'https://twitter.com/%s/status/%s' % (response['user']['screen_name'], response['id_str'])
				printM('Tweet URL: %s' % url)

		except Exception as e:
			printE('could not send alert tweet - %s' % (e))
			try:
				printE('Waiting 5 seconds and trying to send tweet again...', sender=self.sender, spaces=True)
				time.sleep(5.1)
				printM('Tweet: %s' % (message), sender=self.sender)
				if not self.testing:
					self.auth()
					response = self.twitter.update_status(status=message, lat=rs.inv[0][0].latitude,
															long=rs.inv[0][0].longitude,
															geo_enabled=True, display_coordinates=True)
															# location will only stick to tweets on accounts that have location enabled in Settings
					url = 'https://twitter.com/%s/status/%s' % (response['user']['screen_name'], response['id_str'])
					printM('Tweet URL: %s' % url)
			except Exception as e:
				printE('could not send alert tweet - %s' % (e))
				response = None

		self.last_message = message



	def _when_img(self, d):
		'''
		Send a tweet with an image in when you get an ``IMGPATH`` message.

		:param bytes d: queue message
		'''
		if self.tweet_images:
			imgpath = helpers.get_msg_path(d)
			imgtime = helpers.fsec(helpers.get_msg_time(d))
			message = '%s %s UTC%s' % (self.message1, imgtime.strftime(self.fmt)[:22], self.extra_text)
			response = None
			printM('Image tweet: %s' % (message), sender=self.sender)
			if not self.testing:
				if os.path.exists(imgpath):
					with open(imgpath, 'rb') as image:
						try:
							printM('Uploading image to Twitter %s' % (imgpath), self.sender)
							response = self.twitter.upload_media(media=image)
							time.sleep(5.1)
							printM('Sending tweet...', sender=self.sender)
							response = self.twitter.update_status(status=message, media_ids=response['media_id'],
																	lat=rs.inv[0][0].latitude, long=rs.inv[0][0].longitude,
																	geo_enabled=True, display_coordinates=True)
																	# location will only stick to tweets on accounts that have location enabled in Settings
							url = 'https://twitter.com/%s/status/%s' % (response['user']['screen_name'], response['id_str'])
							printM('Tweet URL: %s' % url)
						except Exception as e:
							printE('could not send multimedia tweet - %s' % (e))
							try:
								printM('Waiting 5 seconds and trying to send tweet again...', sender=self.sender)
								time.sleep(5.1)
								self.auth()
								printM('Uploading image to Twitter (2nd try) %s' % (imgpath), self.sender)
								response = self.twitter.upload_media(media=image)
								time.sleep(5.1)
								printM('Sending tweet...', sender=self.sender)
								response = self.twitter.update_status(status=message, media_ids=response['media_id'],
																		lat=rs.inv[0][0].latitude, long=rs.inv[0][0].longitude,
																		geo_enabled=True, display_coordinates=True)
																		# location will only stick to tweets on accounts that have location enabled in Settings
								url = 'https://twitter.com/%s/status/%s' % (response['user']['screen_name'], response['id_str'])
								printM('Tweet URL: %s' % url)

							except Exception as e:
								printE('could not send multimedia tweet (2nd try) - %s' % (e))
								response = None

				else:
					printM('Could not find image: %s' % (imgpath), sender=self.sender)
		
		self.last_message = message


	def run(self):

		"""
		Reads data from the queue and tweets a message if it sees an ALARM or IMGPATH message
		"""
		while True:
			d = self.getq()

			if 'ALARM' in str(d):
				self._when_alarm(d)

			elif 'IMGPATH' in str(d):
				self._when_img(d)


1		import os, sys, platform
2		import pkg_resources as pr
3		import time
4		import math
5		import numpy as np
6		from datetime import datetime, timedelta
7		import rsudp.raspberryshake as rs
8		from rsudp import printM, printW, printE, helpers
9		import rsudp
10		from twython import Twython
11
12
13		class Tweeter(rs.ConsumerThread):
14		'''
15		.. versionadded:: 1.0.2
16
17		The option to add extra text to tweets using the :code:`"extra_text"`
18		setting in settings files built by this version and later.
19
20		.. \|raspishakeq\| raw:: html
21
22		<a href="https://twitter.com/raspishakeq" target="_blank">Raspberry Shake</a>
23
24		.. \|usgsbigquakes\| raw:: html
25
26		<a href="https://twitter.com/USGSBigQuakes" target="_blank">USGS</a>
27
28		.. \|emsc_lastquake\| raw:: html
29
30		<a href="https://twitter.com/LastQuake" target="_blank">EMSC</a>
31
32		Twitter is a social media platform sometimes used for quickly
33		distributing public alert information. It is used by many agencies
34		including \|raspishakeq\|, \|usgsbigquakes\|, and \|emsc_lastquake\|.
35
36		.. \|tw_api_bots\| raw:: html
37
38		<a href="https://developer.twitter.com/apps" target="_blank">API bots</a>
39
40		.. note::
41		Twitter is more difficult and stricter when it comes to making
42		\|tw_api_bots\| than many services.
43		First, you must go through a relatively rigorous process of applying for
44		a developer account, then making a Twitter "app", and then giving the app
45		permission to post on your behalf. See :ref:`setting-up-twitter` for details.
46
47		Once you've gone through that process, Twitter limits posting and makes
48		its rules on rate limiting relatively difficult to nail down.
49		Generally, the early 2020 rate limit for posting is 300 in a 3-hour span,
50		however that can vary depending on whether or not Twitter thinks there is
51		suspicious activity coming from your account.
52
53		In general, if you are looking for a simple multi-platform notification
54		service, it may be easier and more reliable to use the Telegram service
55		instead. rsudp has a telegram module at
56		:py:class:`rsudp.c_telegram.Telegram`. See :ref:`setting-up-telegram` for details.
57
58		:param str consumer_key: Twitter calls this the "consumer key"
59		:param str consumer_secret: Twitter calls this the "consumer key secret"
60		:param str access_token: Twitter calls this the "consumer access token"
61		:param str access_secret: Twitter calls this the "consumer access secret"
62		:param bool tweet_images: whether or not to send images. if False, only alerts will be sent.
63		:type extra_text: bool or str
64		:param extra_text: 103 additional characters to post as part of the twitter message.
65		:param queue.Queue q: queue of data and messages sent by :class:`rsudp.c_consumer.Consumer`
66
67
68		'''
69		def __init__(self, consumer_key, consumer_secret, access_token, access_token_secret,
70		q=False, tweet_images=False, extra_text=False, testing=False,
71		):
72		"""
73		Initialize the process
74		"""
75		super().__init__()
76		self.sender = 'Tweeter'
77		self.alive = True
78		self.tweet_images = tweet_images
79		self.testing = testing
80		self.fmt = '%Y-%m-%d %H:%M:%S.%f'
81		self.region = ' - region: %s' % rs.region.title() if rs.region else ''
82		self.consumer_key = consumer_key
83		self.consumer_secret = consumer_secret
84		self.access_token = access_token
85		self.access_token_secret = access_token_secret
86		self.last_message = False
87
88		self._resolve_extra_text(extra_text)
89
90		if q:
91		self.queue = q
92		else:
93		printE('no queue passed to consumer! Thread will exit now!', self.sender)
94		sys.stdout.flush()
95		self.alive = False
96		sys.exit()
97
98		if not self.testing:
99		self.twitter = Twython(
100		consumer_key,
101		consumer_secret,
102		access_token,
103		access_token_secret
104		)
105		else:
106		printW('The Twitter module will not post to Twitter in Testing mode.',
107		self.sender, announce=False)
108
109
110		self.livelink = 'live feed ➡️ https://stationview.raspberryshake.org/#?net=%s&sta=%s' % (rs.net, rs.stn)
111		self.message0 = '(#RaspberryShake station %s.%s%s) Event detected at' % (rs.net, rs.stn, self.region)
112		self.message1 = '(#RaspberryShake station %s.%s%s) Image of event detected at' % (rs.net, rs.stn, self.region)
113
114		printM('Starting.', self.sender)
115
116
117		def _resolve_extra_text(self, extra_text):
118		allowable_len = 103 # length of string allowable given maximum message text & region
119		if ((extra_text == '') or (extra_text == None) or (extra_text == False)):
120		self.extra_text = ''
121		else:
122		extra_text = str(extra_text)
123		len_ex_txt = len(extra_text)
124
125		if len_ex_txt > allowable_len:
126		printW('extra_text parameter is longer than allowable (%s chars) and will be truncated. Please keep extra_text at or below %s characters.' % (len_ex_txt, allowable_len), sender=self.sender)
127		extra_text = extra_text[:allowable_len]
128
129		self.extra_text = ' %s' % (extra_text)
130
131
132		def auth(self):
133		self.twitter = Twython(
134		self.consumer_key,
135		self.consumer_secret,
136		self.access_token,
137		self.access_token_secret
138		)
139
140
141		def getq(self):
142		d = self.queue.get()
143		self.queue.task_done()
144
145		if 'TERM' in str(d):
146		self.alive = False
147		printM('Exiting.', self.sender)
148		sys.exit()
149		else:
150		return d
151
152
153		def _when_alarm(self, d):
154		'''
155		Send a tweet when you get an ``ALARM`` message.
156
157		:param bytes d: queue message
158		'''
159		event_time = helpers.fsec(helpers.get_msg_time(d))
160		self.last_event_str = '%s' % (event_time.strftime(self.fmt)[:22])
161		message = '%s %s UTC%s - %s' % (self.message0, self.last_event_str, self.extra_text, self.livelink)
162		response = None
163		try:
164		printM('Tweet: %s' % (message), sender=self.sender)
165		if not self.testing:
166		response = self.twitter.update_status(status=message, lat=rs.inv[0][0].latitude,
167		long=rs.inv[0][0].longitude,
168		geo_enabled=True, display_coordinates=True)
169		# location will only stick to tweets on accounts that have location enabled in Settings
170		url = 'https://twitter.com/%s/status/%s' % (response['user']['screen_name'], response['id_str'])
171		printM('Tweet URL: %s' % url)
172
173		except Exception as e:
174		printE('could not send alert tweet - %s' % (e))
175		try:
176		printE('Waiting 5 seconds and trying to send tweet again...', sender=self.sender, spaces=True)
177		time.sleep(5.1)
178		printM('Tweet: %s' % (message), sender=self.sender)
179		if not self.testing:
180		self.auth()
181		response = self.twitter.update_status(status=message, lat=rs.inv[0][0].latitude,
182		long=rs.inv[0][0].longitude,
183		geo_enabled=True, display_coordinates=True)
184		# location will only stick to tweets on accounts that have location enabled in Settings
185		url = 'https://twitter.com/%s/status/%s' % (response['user']['screen_name'], response['id_str'])
186		printM('Tweet URL: %s' % url)
187		except Exception as e:
188		printE('could not send alert tweet - %s' % (e))
189		response = None
190
191		self.last_message = message
192
193
194
195		def _when_img(self, d):
196		'''
197		Send a tweet with an image in when you get an ``IMGPATH`` message.
198
199		:param bytes d: queue message
200		'''
201		if self.tweet_images:
202		imgpath = helpers.get_msg_path(d)
203		imgtime = helpers.fsec(helpers.get_msg_time(d))
204		message = '%s %s UTC%s' % (self.message1, imgtime.strftime(self.fmt)[:22], self.extra_text)
205		response = None
206		printM('Image tweet: %s' % (message), sender=self.sender)
207		if not self.testing:
208		if os.path.exists(imgpath):
209		with open(imgpath, 'rb') as image:
210		try:
211		printM('Uploading image to Twitter %s' % (imgpath), self.sender)
212		response = self.twitter.upload_media(media=image)
213		time.sleep(5.1)
214		printM('Sending tweet...', sender=self.sender)
215		response = self.twitter.update_status(status=message, media_ids=response['media_id'],
216		lat=rs.inv[0][0].latitude, long=rs.inv[0][0].longitude,
217		geo_enabled=True, display_coordinates=True)
218		# location will only stick to tweets on accounts that have location enabled in Settings
219		url = 'https://twitter.com/%s/status/%s' % (response['user']['screen_name'], response['id_str'])
220		printM('Tweet URL: %s' % url)
221		except Exception as e:
222		printE('could not send multimedia tweet - %s' % (e))
223		try:
224		printM('Waiting 5 seconds and trying to send tweet again...', sender=self.sender)
225		time.sleep(5.1)
226		self.auth()
227		printM('Uploading image to Twitter (2nd try) %s' % (imgpath), self.sender)
228		response = self.twitter.upload_media(media=image)
229		time.sleep(5.1)
230		printM('Sending tweet...', sender=self.sender)
231		response = self.twitter.update_status(status=message, media_ids=response['media_id'],
232		lat=rs.inv[0][0].latitude, long=rs.inv[0][0].longitude,
233		geo_enabled=True, display_coordinates=True)
234		# location will only stick to tweets on accounts that have location enabled in Settings
235		url = 'https://twitter.com/%s/status/%s' % (response['user']['screen_name'], response['id_str'])
236		printM('Tweet URL: %s' % url)
237
238		except Exception as e:
239		printE('could not send multimedia tweet (2nd try) - %s' % (e))
240		response = None
241
242		else:
243		printM('Could not find image: %s' % (imgpath), sender=self.sender)
244
245		self.last_message = message
		0 ignored issues – show introduced 2021-10-16 22:46 UTC by Report Bug Copy Issue Report The variable `message` does not seem to be defined in case `self.tweet_images` on line `201` is `False`. Are you sure this can never be the case? Loading history...
246
247	View Code Duplication	def run(self):
		0 ignored issues – show Duplication introduced 2021-10-16 22:46 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
248		"""
249		Reads data from the queue and tweets a message if it sees an ALARM or IMGPATH message
250		"""
251		while True:
252		d = self.getq()
253
254		if 'ALARM' in str(d):
255		self._when_alarm(d)
256
257		elif 'IMGPATH' in str(d):
258		self._when_img(d)
259

raspishake / rsudp

Pull Request — master (#24)

build.rsudp.c_tweet.Tweeter._when_alarm() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like