Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import locale
import logging
import requests
from photini.configstore import key_store
from photini.photinimap import GeocoderBase, PhotiniMap
from photini.pyqt import (
Busy, catch_all, CompactButton, QtCore, QtGui, QtWidgets)
logger = logging.getLogger(__name__)
translate = QtCore.QCoreApplication.translate
class MapboxGeocoder(GeocoderBase):
api_key = key_store.get('mapboxmap', 'api_key')
def do_geocode(self, query, params={}):
params['access_token'] = self.api_key
params['autocomplete '] = 'false'
lang, encoding = locale.getdefaultlocale()
if lang:
params['language'] = lang
query += '.json'
url = 'https://api.mapbox.com/geocoding/v5/mapbox.places/' + query
with Busy():
self.rate_limit()
try:
rsp = requests.get(url, params=params, timeout=5)
except Exception as ex:
logger.error(str(ex))
return []
def connect(self, token=None):
self.cached_data = {}
refresh_token = self.get_password()
if not refresh_token:
return False
if not token:
# create expired token
token = {
'access_token' : 'xxx',
'refresh_token': refresh_token,
'expires_in' : -30,
}
client_id = key_store.get('googlephotos', 'client_id')
client_secret = key_store.get('googlephotos', 'client_secret')
auto_refresh_kwargs = {
'client_id' : client_id,
'client_secret': client_secret,
}
token_url = self.oauth_url + 'v4/token'
self.api = OAuth2Session(
client_id=client_id, token=token, token_updater=self.save_token,
auto_refresh_kwargs=auto_refresh_kwargs,
auto_refresh_url=token_url)
if token['expires_in'] < 0:
# refresh token
self.api.refresh_token(token_url, **auto_refresh_kwargs)
self.connection_changed.emit(self.api.authorized)
return self.api.authorized
def connect(self, token=None):
self.cached_data = {}
refresh_token = self.get_password()
if not refresh_token:
return False
if not token:
# create expired token
token = {
'access_token' : 'xxx',
'refresh_token': refresh_token,
'expires_in' : -30,
}
client_id = key_store.get('googlephotos', 'client_id')
client_secret = key_store.get('googlephotos', 'client_secret')
auto_refresh_kwargs = {
'client_id' : client_id,
'client_secret': client_secret,
}
token_url = self.oauth_url + 'v4/token'
self.api = OAuth2Session(
client_id=client_id, token=token, token_updater=self.save_token,
auto_refresh_kwargs=auto_refresh_kwargs,
auto_refresh_url=token_url)
if token['expires_in'] < 0:
# refresh token
self.api.refresh_token(token_url, **auto_refresh_kwargs)
self.connection_changed.emit(self.api.authorized)
return self.api.authorized
def permitted(self, level):
refresh_token = self.get_password()
if not refresh_token:
self.api = None
return False
if not self.api:
self.current_feed = None
# create expired token
token = {
'access_token' : 'xxx',
'refresh_token' : refresh_token,
'expires_in' : -30,
}
# create new session
client_id = key_store.get('picasa', 'client_id')
client_secret = key_store.get('picasa', 'client_secret')
auto_refresh_kwargs = {
'client_id' : client_id,
'client_secret': client_secret,
}
if self.auto_refresh:
self.api = OAuth2Session(
client_id, token=token, token_updater=self._save_token,
auto_refresh_kwargs=auto_refresh_kwargs,
auto_refresh_url=self.token_url,
)
else:
self.api = OAuth2Session(client_id, token=token)
self.api.verify = certifi.old_where()
# refresh manually to get a valid token now
token = self.api.refresh_token(self.token_url, **auto_refresh_kwargs)
def get_auth_url(self, level):
app_id = key_store.get('facebook', 'app_id')
client = oauthlib.oauth2.MobileApplicationClient(app_id)
self.api = OAuth2Session(
client=client, scope=','.join(self.scope[level]),
redirect_uri='https://www.facebook.com/connect/login_success.html',
)
result = self.api.authorization_url(
'https://www.facebook.com/dialog/oauth',
display='popup', auth_type='rerequest')[0]
# use unquote to prevent "redirect_uri URL is not properly
# formatted" error on Windows
return unquote(result)
lang, encoding = locale.getdefaultlocale()
if lang:
url += '&mkt={0},ngt'.format(lang.replace('_', '-'))
else:
url += '&mkt=ngt'
if self.app.test_mode:
url += '&branch=experimental'
return {
'head': '''
'''.format(key_store.get('bing', 'api_key'), url),
'body': '',
}
def get_auth_url(self, redirect_uri):
code_verifier = ''
while len(code_verifier) < 43:
code_verifier += OAuth2Session().new_state()
self.auth_params = {
'client_id' : key_store.get('googlephotos', 'client_id'),
'client_secret': key_store.get('googlephotos', 'client_secret'),
'code_verifier': code_verifier,
'redirect_uri' : redirect_uri,
}
url = 'https://accounts.google.com/o/oauth2/v2/auth'
url += '?client_id=' + self.auth_params['client_id']
url += '&redirect_uri=' + self.auth_params['redirect_uri']
url += '&response_type=code'
url += '&scope=' + urllib.parse.quote(' '.join(self.scope))
url += '&code_challenge=' + self.auth_params['code_verifier']
return url
def get_auth_url(self, redirect_uri):
code_verifier = ''
while len(code_verifier) < 43:
code_verifier += OAuth2Session().new_state()
self.auth_params = {
'client_id' : key_store.get('googlephotos', 'client_id'),
'client_secret': key_store.get('googlephotos', 'client_secret'),
'code_verifier': code_verifier,
'redirect_uri' : redirect_uri,
}
url = 'https://accounts.google.com/o/oauth2/v2/auth'
url += '?client_id=' + self.auth_params['client_id']
url += '&redirect_uri=' + self.auth_params['redirect_uri']
url += '&response_type=code'
url += '&scope=' + urllib.parse.quote(' '.join(self.scope))
url += '&code_challenge=' + self.auth_params['code_verifier']
return url
def search_terms(self):
widget = CompactButton(
translate('MapTabMapbox', 'Search powered by Mapbox'))
widget.clicked.connect(self.load_mapbox_tos)
return [widget]
@QtCore.pyqtSlot()
@catch_all
def load_mapbox_tos(self):
QtGui.QDesktopServices.openUrl(
QtCore.QUrl('https://www.mapbox.com/tos/'))
class TabWidget(PhotiniMap):
api_key = key_store.get('mapboxmap', 'api_key')
@staticmethod
def tab_name():
return translate('MapTabMapbox', 'Map (&Mapbox)')
def get_geocoder(self):
return MapboxGeocoder(parent=self)
def get_head(self):
return '''