def add_page(arg):
"""
add specific information from facebook page to DB
connecting facebook api and request selected fields.
:param arg: page name, id or full url
"""
if '/' in arg:
fb_url = arg.split('/')[-1]
else:
fb_url = arg
try:
new_page = graph.get_object(REQUEST_FORMAT.format(fb_url, FIELDS_FORMAT))
except facebook.GraphAPIError:
print('Not a page')
exit()
# relpace cover with cover link
new_page['cover'] = new_page['cover']['source'] if 'cover' in new_page.keys() else 'default'
# add field values to database
upsert(pages, new_page)
print(SUCCESS_ADD_MSG.format(**new_page))
python类GraphAPIError()的实例源码
def generate_long_lived_fb_token(access_token, graph_api):
"""Generate a long lived facebook token."""
params = {
'grant_type': 'fb_exchange_token',
'client_id': settings.FACEBOOK_APP_ID,
'client_secret': settings.FACEBOOK_APP_SECRET,
'fb_exchange_token': access_token
}
connection_name = 'access_token?{}'.format(urllib.parse.urlencode(params))
try:
long_lived_fb_token = graph_api.get_connections(id='oauth', connection_name=connection_name)
except facebook.GraphAPIError:
raise FacebookInvalidTokenException
return long_lived_fb_token['access_token']
def assert_raises_multi_regex(
self, expected_exception, expected_regexp, callable_obj=None,
*args, **kwargs):
"""
Custom function to backport assertRaisesRegexp to all supported
versions of Python.
"""
self.assertRaises(expected_exception, callable_obj, *args, **kwargs)
try:
callable_obj(*args)
except facebook.GraphAPIError as error:
self.assertEqual(error.message, expected_regexp)
def test_get_deleted_app_access_token(self):
deleted_app_id = '174236045938435'
deleted_secret = '0073dce2d95c4a5c2922d1827ea0cca6'
deleted_error_message = (
"Error validating application. Application has been deleted.")
self.assert_raises_multi_regex(
facebook.GraphAPIError,
deleted_error_message,
facebook.GraphAPI().get_app_access_token,
deleted_app_id,
deleted_secret)
def test_invalid_version(self):
self.assertRaises(facebook.GraphAPIError,
facebook.GraphAPI, version=1.2)
def test_invalid_format(self):
self.assertRaises(facebook.GraphAPIError,
facebook.GraphAPI, version="2.a")
self.assertRaises(facebook.GraphAPIError,
facebook.GraphAPI, version="a.1")
self.assertRaises(facebook.GraphAPIError,
facebook.GraphAPI, version=2.23)
self.assertRaises(facebook.GraphAPIError,
facebook.GraphAPI, version="2.23")
def test_extend_access_token(self):
"""
Test if extend_access_token requests the correct endpoint.
Note that this only tests whether extend_access_token returns the
correct error message when called without a proper user-access token.
"""
try:
facebook.GraphAPI().extend_access_token(self.app_id, self.secret)
except facebook.GraphAPIError as e:
self.assertEqual(
e.message, "fb_exchange_token parameter not specified")
def test_access_with_expired_access_token(self):
expired_token = (
"AAABrFmeaJjgBAIshbq5ZBqZBICsmveZCZBi6O4w9HSTkFI73VMtmkL9jLuWs"
"ZBZC9QMHvJFtSulZAqonZBRIByzGooCZC8DWr0t1M4BL9FARdQwPWPnIqCiFQ")
graph = facebook.GraphAPI(access_token=expired_token)
self.assertRaises(facebook.GraphAPIError, graph.get_object, 'me')
def get_user(self):
"""
Override this method by sublcassing the class.
"""
if not current.session.token:
return None
return dict(first_name='Pinco',
last_name='Pallino',
username='pincopallino')
raise NotImplementedError("Must override get_user()")
# Following code is never executed. It can be used as example
# for overriding in subclasses.
if not self.accessToken():
return None
if not self.graph:
self.graph = GraphAPI((self.accessToken()))
user = None
try:
user = self.graph.get_object("me")
except GraphAPIError:
current.session.token = None
self.graph = None
if user:
return dict(first_name=user['first_name'],
last_name=user['last_name'],
username=user['id'])
def get_user(self):
"""
Override this method by sublcassing the class.
"""
if not current.session.token:
return None
return dict(first_name='Pinco',
last_name='Pallino',
username='pincopallino')
raise NotImplementedError("Must override get_user()")
# Following code is never executed. It can be used as example
# for overriding in subclasses.
if not self.accessToken():
return None
if not self.graph:
self.graph = GraphAPI((self.accessToken()))
user = None
try:
user = self.graph.get_object("me")
except GraphAPIError:
current.session.token = None
self.graph = None
if user:
return dict(first_name=user['first_name'],
last_name=user['last_name'],
username=user['id'])
def get_birthday_reminders(self):
"""
Responds to user-input, typically speech text, by listing the user's
Facebook friends with birthdays today.
Arguments:
text -- user-input, typically transcribed speech
self.assistant -- used to interact with the user (for both input and output)
profile -- contains information related to the user (e.g., phone
number)
"""
try:
results = self.graph.request("me/friends",
args={'fields': 'id,name,birthday'})
except facebook.GraphAPIError:
response = ("I have not been authorized to query your Facebook. If you " +
"would like to check birthdays in the future, please visit " +
"the Jasper dashboard.")
return response
except:
return "I apologize, there's a problem with that service at the moment."
needle = datetime.datetime.now(tz=pytz.utc).strftime("%m/%d")
people = []
for person in results['data']:
try:
if needle in person['birthday']:
people.append(person['name'])
except:
continue
if len(people) > 0:
if len(people) == 1:
output = people[0] + " has a birthday today."
else:
output = "Your friends with birthdays today are " + \
", ".join(people[:-1]) + " and " + people[-1] + "."
else:
output = "None of your friends have birthdays today."
return output
def get_notifications(self):
"""
Not working since facebooks new update which doesn't allow notifications to be fetched. :(
Responds to user-input, typically speech text, with a summary of
the user's Facebook notifications, including a count and details
related to each individual notification.
Arguments:
text -- user-input, typically transcribed speech
self.assistant -- used to interact with the user (for both input and output)
profile -- contains information related to the user (e.g., phone
number)
"""
try:
results = self.graph.request("me/notifications")
except facebook.GraphAPIError:
response = ("I have not been authorized to query your Facebook. If you " +
"would like to check your notifications in the future, " +
"please visit the Stephanie facebook module configuraton.")
return response
except:
return "I apologize, there's a problem with that service at the moment."
if not len(results['data']):
return "You have no Facebook notifications."
updates = []
for notification in results['data']:
updates.append(notification['title'])
count = len(results['data'])
response = ("You have " + str(count) +
" Facebook notifications. " + " ".join(updates) + ". ")
return response
def status_update(self):
self.assistant.say("What's in your mind?")
text = self.assistant.listen().decipher()
try:
self.graph.put_wall_post(text)
self.assistant.say("You have successully put up a wall post.")
except facebook.GraphAPIError:
response = ("I have not been authorized to query your Facebook. If you " +
"would like to check your notifications in the future, " +
"please visit the Stephanie facebook module configuraton.")
return response
except:
return "I apologize, there's a problem with that service at the moment."
def get_user(self):
"""
Override this method by sublcassing the class.
"""
if not current.session.token:
return None
return dict(first_name='Pinco',
last_name='Pallino',
username='pincopallino')
raise NotImplementedError("Must override get_user()")
# Following code is never executed. It can be used as example
# for overriding in subclasses.
if not self.accessToken():
return None
if not self.graph:
self.graph = GraphAPI((self.accessToken()))
user = None
try:
user = self.graph.get_object("me")
except GraphAPIError:
current.session.token = None
self.graph = None
if user:
return dict(first_name=user['first_name'],
last_name=user['last_name'],
username=user['id'])
def autopost_facebook(self):
posts = FeedPost.objects.filter(for_network=conf.NETWORK_FACEBOOK)[:MAX_POSTS_PER_CALL]
for post in posts:
try:
api.facebook.post(post.text, url=post.url)
except facebook.GraphAPIError as e:
logger.error("Facebook Autopost: error on #{0.pk}: {1.args}".format(post, e))
else:
logger.info("Facebook Autopost: posted #{0.pk} ('{0}')".format(post))
post.scheduled = False
post.posted = now()
post.save()
def get_user(self):
"""
Override this method by sublcassing the class.
"""
if not current.session.token:
return None
return dict(first_name='Pinco',
last_name='Pallino',
username='pincopallino')
raise NotImplementedError("Must override get_user()")
# Following code is never executed. It can be used as example
# for overriding in subclasses.
if not self.accessToken():
return None
if not self.graph:
self.graph = GraphAPI((self.accessToken()))
user = None
try:
user = self.graph.get_object("me")
except GraphAPIError:
current.session.token = None
self.graph = None
if user:
return dict(first_name=user['first_name'],
last_name=user['last_name'],
username=user['id'])
def handle(text, mic, profile):
"""
Responds to user-input, typically speech text, with a summary of
the user's Facebook notifications, including a count and details
related to each individual notification.
Arguments:
text -- user-input, typically transcribed speech
mic -- used to interact with the user (for both input and output)
profile -- contains information related to the user (e.g., phone
number)
"""
oauth_access_token = profile['keys']['FB_TOKEN']
graph = facebook.GraphAPI(oauth_access_token)
try:
results = graph.request("me/notifications")
except facebook.GraphAPIError:
mic.say("I have not been authorized to query your Facebook. If you " +
"would like to check your notifications in the future, " +
"please visit the Jasper dashboard.")
return
except:
mic.say(
"I apologize, there's a problem with that service at the moment.")
if not len(results['data']):
mic.say("You have no Facebook notifications. ")
return
updates = []
for notification in results['data']:
updates.append(notification['title'])
count = len(results['data'])
mic.say("You have " + str(count) +
" Facebook notifications. " + " ".join(updates) + ". ")
return
oauth20_account.py 文件源码
项目：rekall-agent-server
作者: rekall-innovations
项目源码
文件源码
阅读 31
收藏 0
点赞 0
评论 0
def get_user(self):
"""
Override this method by sublcassing the class.
"""
if not current.session.token:
return None
return dict(first_name='Pinco',
last_name='Pallino',
username='pincopallino')
raise NotImplementedError("Must override get_user()")
# Following code is never executed. It can be used as example
# for overriding in subclasses.
if not self.accessToken():
return None
if not self.graph:
self.graph = GraphAPI((self.accessToken()))
user = None
try:
user = self.graph.get_object("me")
except GraphAPIError:
current.session.token = None
self.graph = None
if user:
return dict(first_name=user['first_name'],
last_name=user['last_name'],
username=user['id'])
def get_user(self):
"""
Override this method by sublcassing the class.
"""
if not current.session.token:
return None
return dict(first_name='Pinco',
last_name='Pallino',
username='pincopallino')
raise NotImplementedError("Must override get_user()")
# Following code is never executed. It can be used as example
# for overriding in subclasses.
if not self.accessToken():
return None
if not self.graph:
self.graph = GraphAPI((self.accessToken()))
user = None
try:
user = self.graph.get_object("me")
except GraphAPIError:
current.session.token = None
self.graph = None
if user:
return dict(first_name=user['first_name'],
last_name=user['last_name'],
username=user['id'])
def handle(text, mic, profile):
"""
Responds to user-input, typically speech text, with a summary of
the user's Facebook notifications, including a count and details
related to each individual notification.
Arguments:
text -- user-input, typically transcribed speech
mic -- used to interact with the user (for both input and output)
profile -- contains information related to the user (e.g., phone
number)
"""
oauth_access_token = profile['keys']['FB_TOKEN']
graph = facebook.GraphAPI(oauth_access_token)
try:
results = graph.request("me/notifications")
except facebook.GraphAPIError:
mic.say("I have not been authorized to query your Facebook. If you " +
"would like to check your notifications in the future, " +
"please visit the Jasper dashboard.")
return
except:
mic.say(
"I apologize, there's a problem with that service at the moment.")
if not len(results['data']):
mic.say("You have no Facebook notifications. ")
return
updates = []
for notification in results['data']:
updates.append(notification['title'])
count = len(results['data'])
mic.say("You have " + str(count) +
" Facebook notifications. " + " ".join(updates) + ". ")
return
def get_facebook_user(graph_api):
"""Get the user information from Facebook."""
params = ['id', 'first_name', 'last_name', 'gender', 'email']
graph_object_id = 'me?fields={}'.format(','.join(params))
try:
facebook_user = graph_api.get_object(id=graph_object_id)
except facebook.GraphAPIError:
raise FacebookInvalidTokenException
return facebook_user
def get_user(self):
"""
Override this method by sublcassing the class.
"""
if not current.session.token:
return None
return dict(first_name='Pinco',
last_name='Pallino',
username='pincopallino')
raise NotImplementedError("Must override get_user()")
# Following code is never executed. It can be used as example
# for overriding in subclasses.
if not self.accessToken():
return None
if not self.graph:
self.graph = GraphAPI((self.accessToken()))
user = None
try:
user = self.graph.get_object("me")
except GraphAPIError:
current.session.token = None
self.graph = None
if user:
return dict(first_name=user['first_name'],
last_name=user['last_name'],
username=user['id'])
def handle(text, mic, profile):
"""
Responds to user-input, typically speech text, by listing the user's
Facebook friends with birthdays today.
Arguments:
text -- user-input, typically transcribed speech
mic -- used to interact with the user (for both input and output)
profile -- contains information related to the user (e.g., phone
number)
"""
oauth_access_token = profile['keys']["FB_TOKEN"]
graph = facebook.GraphAPI(oauth_access_token)
try:
results = graph.request("me/friends",
args={'fields': 'id,name,birthday'})
except facebook.GraphAPIError:
mic.say("I have not been authorized to query your Facebook. If you " +
"would like to check birthdays in the future, please visit " +
"the Jasper dashboard.")
return
except:
mic.say(
"I apologize, there's a problem with that service at the moment.")
return
needle = datetime.datetime.now(tz=get_timezone(profile)).strftime("%m/%d")
people = []
for person in results['data']:
try:
if needle in person['birthday']:
people.append(person['name'])
except:
continue
if len(people) > 0:
if len(people) == 1:
output = people[0] + " has a birthday today."
else:
output = "Your friends with birthdays today are " + \
", ".join(people[:-1]) + " and " + people[-1] + "."
else:
output = "None of your friends have birthdays today."
mic.say(output)
def handle(self, text, mic):
"""
Responds to user-input, typically speech text, by listing the user's
Facebook friends with birthdays today.
Arguments:
text -- user-input, typically transcribed speech
mic -- used to interact with the user (for both input and output)
"""
oauth_access_token = self.profile['keys']["FB_TOKEN"]
graph = facebook.GraphAPI(oauth_access_token)
try:
results = graph.request("me/friends",
args={'fields': 'id,name,birthday'})
except facebook.GraphAPIError:
mic.say(self.gettext(
"I have not been authorized to query your Facebook. If you " +
"would like to check birthdays in the future, please visit " +
"the Jasper dashboard."))
return
except:
mic.say(self.gettext("I apologize, there's a problem with that " +
"service at the moment."))
return
needle = datetime.datetime.now(
tz=app_utils.get_timezone(self.profile)).strftime("%m/%d")
people = []
for person in results['data']:
try:
if needle in person['birthday']:
people.append(person['name'])
except:
continue
if len(people) > 0:
if len(people) == 1:
output = self.gettext("%s has a birthday today.") % people[0]
else:
output = (self.gettext(
"Your friends with birthdays today are %s and %s.") %
(", ".join(people[:-1]), people[-1]))
else:
output = self.gettext("None of your friends have birthdays today.")
mic.say(output)
def handle(self, text, mic):
"""
Responds to user-input, typically speech text, with a summary of
the user's Facebook notifications, including a count and details
related to each individual notification.
Arguments:
text -- user-input, typically transcribed speech
mic -- used to interact with the user (for both input and output)
"""
oauth_access_token = self.profile['keys']['FB_TOKEN']
graph = facebook.GraphAPI(oauth_access_token)
try:
results = graph.request("me/notifications")
except facebook.GraphAPIError:
mic.say(self.gettext(
"I have not been authorized to query your Facebook. If " +
"you would like to check your notifications in the " +
"future, please visit the Jasper dashboard."))
return
except:
mic.say(self.gettext(
"I apologize, I can't access Facebook at the moment."))
if not len(results['data']):
mic.say(self.gettext("You have no Facebook notifications."))
return
updates = []
for notification in results['data']:
updates.append(notification['title'])
count = len(results['data'])
if count == 0:
mic.say(self.gettext("You have no Facebook notifications."))
elif count == 1:
mic.say(self.gettext("You have one Facebook notification."))
else:
mic.say(
self.gettext("You have %d Facebook notifications.") % count)
if count > 0:
mic.say("%s." % " ".join(updates))
return
def handle(text, mic, profile):
"""
Responds to user-input, typically speech text, by listing the user's
Facebook friends with birthdays today.
Arguments:
text -- user-input, typically transcribed speech
mic -- used to interact with the user (for both input and output)
profile -- contains information related to the user (e.g., phone
number)
"""
oauth_access_token = profile['keys']["FB_TOKEN"]
graph = facebook.GraphAPI(oauth_access_token)
try:
results = graph.request("me/friends",
args={'fields': 'id,name,birthday'})
except facebook.GraphAPIError:
mic.say("I have not been authorized to query your Facebook. If you " +
"would like to check birthdays in the future, please visit " +
"the Jasper dashboard.")
return
except:
mic.say(
"I apologize, there's a problem with that service at the moment.")
return
needle = datetime.datetime.now(tz=getTimezone(profile)).strftime("%m/%d")
people = []
for person in results['data']:
try:
if needle in person['birthday']:
people.append(person['name'])
except:
continue
if len(people) > 0:
if len(people) == 1:
output = people[0] + " has a birthday today."
else:
output = "Your friends with birthdays today are " + \
", ".join(people[:-1]) + " and " + people[-1] + "."
else:
output = "None of your friends have birthdays today."
mic.say(output)
