revil-O/platypush
1
0
Fork 0
forked from platypush/platypush
Code Pull requests Activity

Merge branch '286/merge-foursquare-backend'

Browse source
This commit is contained in:
Fabio Manganiello 2024-01-01 23:00:55 +01:00
commit b72fb83d18
Signed by untrusted user: blacklight
GPG key ID: D90FBA7F76362774
6 changed files with 85 additions and 105 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
docs/source/backends.rst
View file

@ -10,7 +10,6 @@ Backends
platypush/backend/button.flic.rst
platypush/backend/camera.pi.rst
platypush/backend/chat.telegram.rst
platypush/backend/foursquare.rst
platypush/backend/google.fit.rst
platypush/backend/google.pubsub.rst
platypush/backend/gps.rst

5
docs/source/platypush/backend/foursquare.rst
View file

@ -1,5 +0,0 @@
``foursquare``
================================
.. automodule:: platypush.backend.foursquare
:members:

54
platypush/backend/foursquare/__init__.py
View file

@ -1,54 +0,0 @@
from typing import Optional
from platypush.backend import Backend
from platypush.context import get_plugin
from platypush.message.event.foursquare import FoursquareCheckinEvent
class FoursquareBackend(Backend):
"""
This backend polls for new check-ins on the user's Foursquare account and triggers an event when a new check-in
occurs.
Requires:
* The :class:`platypush.plugins.foursquare.FoursquarePlugin` plugin configured and enabled.
"""
_last_created_at_varname = '_foursquare_checkin_last_created_at'
def __init__(self, poll_seconds: Optional[float] = 60.0, *args, **kwargs):
"""
:param poll_seconds: How often the backend should check for new check-ins (default: one minute).
"""
super().__init__(*args, poll_seconds=poll_seconds, **kwargs)
self._last_created_at = None
def __enter__(self):
self._last_created_at = int(
get_plugin('variable')
.get(self._last_created_at_varname)
.output.get(self._last_created_at_varname)
or 0
)
self.logger.info('Started Foursquare backend')
def loop(self):
checkins = get_plugin('foursquare').get_checkins().output
if not checkins:
return
last_checkin = checkins[0]
last_checkin_created_at = last_checkin.get('createdAt', 0)
if self._last_created_at and last_checkin_created_at <= self._last_created_at:
return
self.bus.post(FoursquareCheckinEvent(checkin=last_checkin))
self._last_created_at = last_checkin_created_at
get_plugin('variable').set(
**{self._last_created_at_varname: self._last_created_at}
)
# vim:sw=4:ts=4:et:

8
platypush/backend/foursquare/manifest.yaml
View file

@ -1,8 +0,0 @@
manifest:
events:
platypush.message.event.foursquare.FoursquareCheckinEvent: when a new check-in
occurs.
install:
pip: []
package: platypush.backend.foursquare
type: backend

119
platypush/plugins/foursquare/__init__.py
View file

@ -1,14 +1,19 @@
import datetime
import requests
from typing import List, Dict, Any, Optional, Union, Tuple
from platypush.plugins import Plugin, action
import requests
from platypush.context import Variable
from platypush.message.event.foursquare import FoursquareCheckinEvent
from platypush.plugins import RunnablePlugin, action
class FoursquarePlugin(Plugin):
class FoursquarePlugin(RunnablePlugin):
"""
Plugin to interact with the `Foursquare Places API <https://developer.foursquare.com/docs/api>`_.
It also raises events when a new check-in occurs on the user's account.
In order to enable the Foursquare API on your account you need to:
- Create a new app on the `Foursquare developers website <https://foursquare.com/developers/signup>`_.
@ -24,20 +29,36 @@ class FoursquarePlugin(Plugin):
"""
api_base_url = 'https://api.foursquare.com/v2'
_last_created_at_varname = '_foursquare_checkin_last_created_at'
_http_timeout = 10
def __init__(self, access_token: str, **kwargs):
def __init__(self, access_token: str, poll_interval: float = 120, **kwargs):
"""
:param access_token: The access token to use to authenticate to the Foursquare API.
"""
super().__init__(**kwargs)
super().__init__(poll_interval=poll_interval, **kwargs)
self.access_token = access_token
self._last_created_at = Variable(self._last_created_at_varname)
def _get_url(self, endpoint):
return '{url}/{endpoint}?oauth_token={token}&v={version}'.format(
url=self.api_base_url,
endpoint=endpoint,
token=self.access_token,
version=datetime.date.today().strftime('%Y%m%d'),
return (
self.api_base_url
+ '/'
+ endpoint
+ '?oauth_token='
+ self.access_token
+ '&v='
+ datetime.date.today().strftime('%Y%m%d')
)
def _get_checkins(self) -> List[Dict[str, Any]]:
url = self._get_url('users/self/checkins')
return (
requests.get(url, timeout=self._http_timeout)
.json()
.get('response', {})
.get('checkins', {})
.get('items', [])
)
@action
@ -46,16 +67,8 @@ class FoursquarePlugin(Plugin):
Get the list of check-ins of the current user.
:return: A list of checkins, as returned by the Foursquare API.
"""
url = self._get_url('users/self/checkins')
return (
requests.get(url)
.json()
.get('response', {})
.get('checkins', {})
.get('items', [])
)
return self._get_checkins()
# noinspection DuplicatedCode
@action
def search(
self,
@ -67,7 +80,7 @@ class FoursquarePlugin(Plugin):
near: Optional[str] = None,
query: Optional[str] = None,
limit: Optional[int] = None,
url: Optional[int] = None,
url: Optional[str] = None,
categories: Optional[List[str]] = None,
radius: Optional[int] = None,
sw: Optional[Union[Tuple[float], List[float]]] = None,
@ -123,12 +136,15 @@ class FoursquarePlugin(Plugin):
if ne:
args['ne'] = ne
url = self._get_url('venues/search')
return (
requests.get(url, params=args).json().get('response', {}).get('venues', [])
requests.get(
self._get_url('venues/search'), params=args, timeout=self._http_timeout
)
.json()
.get('response', {})
.get('venues', [])
)
# noinspection DuplicatedCode
@action
def explore(
self,
@ -223,7 +239,10 @@ class FoursquarePlugin(Plugin):
url = self._get_url('venues/explore')
return (
requests.get(url, params=args).json().get('response', {}).get('venues', [])
requests.get(url, params=args, timeout=self._http_timeout)
.json()
.get('response', {})
.get('venues', [])
)
@action
@ -263,7 +282,10 @@ class FoursquarePlugin(Plugin):
url = self._get_url('venues/trending')
return (
requests.get(url, params=args).json().get('response', {}).get('venues', [])
requests.get(url, params=args, timeout=self._http_timeout)
.json()
.get('response', {})
.get('venues', [])
)
@staticmethod
@ -275,7 +297,7 @@ class FoursquarePlugin(Plugin):
assert isinstance(
t, datetime.datetime
), 'Cannot parse object of type {} into datetime: {}'.format(type(t), t)
), f'Cannot parse object of type {type(t)} into datetime: {t}'
return t
@action
@ -306,7 +328,10 @@ class FoursquarePlugin(Plugin):
url = self._get_url('venues/timeseries')
return (
requests.get(url, params=args).json().get('response', {}).get('venues', [])
requests.get(url, params=args, timeout=self._http_timeout)
.json()
.get('response', {})
.get('venues', [])
)
@action
@ -326,13 +351,16 @@ class FoursquarePlugin(Plugin):
:return: A list of venues, as returned by the Foursquare API.
"""
args = {
'startAt': self._parse_time(start_at),
'endAt': self._parse_time(end_at),
'startAt': self._parse_time(start_at).isoformat(),
'endAt': self._parse_time(end_at).isoformat(),
}
url = self._get_url('venues/{}/stats'.format(venue_id))
url = self._get_url(f'venues/{venue_id}/stats')
return (
requests.get(url, params=args).json().get('response', {}).get('venues', [])
requests.get(url, params=args, timeout=self._http_timeout)
.json()
.get('response', {})
.get('venues', [])
)
@action
@ -343,7 +371,7 @@ class FoursquarePlugin(Plugin):
"""
url = self._get_url('venues/managed')
return (
requests.get(url)
requests.get(url, timeout=self._http_timeout)
.json()
.get('response', {})
.get('venues', [])
@ -386,11 +414,11 @@ class FoursquarePlugin(Plugin):
if latitude and longitude:
args['ll'] = ','.join([str(latitude), str(longitude)])
if altitude:
args['alt'] = altitude
args['alt'] = str(altitude)
if latlng_accuracy:
args['llAcc'] = latlng_accuracy
args['llAcc'] = str(latlng_accuracy)
if altitude_accuracy:
args['altAcc'] = altitude_accuracy
args['altAcc'] = str(altitude_accuracy)
if shout:
args['shout'] = shout
if broadcast:
@ -400,8 +428,27 @@ class FoursquarePlugin(Plugin):
url = self._get_url('checkins/add')
return (
requests.post(url, data=args).json().get('response', {}).get('checkin', {})
requests.post(url, data=args, timeout=self._http_timeout)
.json()
.get('response', {})
.get('checkin', {})
)
def main(self):
while not self.should_stop():
checkins = self._get_checkins()
if not checkins:
return
last_checkin = checkins[0]
last_checkin_created_at = last_checkin.get('createdAt', 0)
last_created_at = self._last_created_at.get() or 0
if last_created_at and last_checkin_created_at <= last_created_at:
return
self._bus.post(FoursquareCheckinEvent(checkin=last_checkin))
self._last_created_at.set(last_checkin_created_at)
self.wait_stop(self.poll_interval)
# vim:sw=4:ts=4:et:

3
platypush/plugins/foursquare/manifest.yaml
View file

@ -1,5 +1,6 @@
manifest:
events: {}
events:
- platypush.message.event.foursquare.FoursquareCheckinEvent
install:
pip: []
package: platypush.plugins.foursquare