def get_current_user(self):
cookies = dict((n, self.cookies[n].value) for n in self.cookies.keys())
cookie = facebook.get_user_from_cookie(
cookies, options.facebook_app_id, options.facebook_app_secret)
if not cookie:
return None
user = self.db.get(
"SELECT * FROM users WHERE id = %s", cookie["uid"])
if not user:
# TODO: Make this fetch async rather than blocking
graph = facebook.GraphAPI(cookie["access_token"])
profile = graph.get_object("me")
self.db.execute(
"REPLACE INTO users (id, name, profile_url, access_token) "
"VALUES (%s,%s,%s,%s)", profile["id"], profile["name"],
profile["link"], cookie["access_token"])
user = self.db.get(
"SELECT * FROM users WHERE id = %s", profile["id"])
elif user.access_token != cookie["access_token"]:
self.db.execute(
"UPDATE users SET access_token = %s WHERE id = %s",
cookie["access_token"], user.id)
return user
python类GraphAPI()的实例源码
def do_init(self):
app_id = self.get_configuration('facebook_app_id')
app_secret = self.get_configuration('facebook_app_secret')
app_access_token = self.get_configuration('facebook_access_token')
if app_id and app_secret and app_access_token:
params = {
'client_id': app_id,
'client_secret': app_secret,
'grant_type': 'fb_exchange_token',
'fb_exchange_token': app_access_token
}
r = self.requests.get("https://graph.facebook.com/oauth/access_token", params=params)
if r.ok:
oauth_access_token = r.json()['access_token']
self.oauth_access_token = self.write_configuration('facebook_oauth_token', oauth_access_token)
self.graph = facebook.GraphAPI(oauth_access_token)
return True
return False
def post_to_facebook(message, link=None):
"""Posts a message to the Facebook page using GraphAPI authenticated via
`FACEBOOK_PAGE_ACCESS_TOKEN`.
Args:
- message: str. The content of the message to be posted on Facebook.
- link: str. (Optional) Url of the attachment to be posted along
with message.
Returns:
- None
"""
graph = GraphAPI(access_token=FACEBOOK_PAGE_ACCESS_TOKEN)
attachment = {
'link': link, # link to visit on clicking on the attachment.
'picture': link # link of the attachment to be posted.
}
graph.put_wall_post(message=message, attachment=attachment)
def scrape(self, user_data):
try:
oauth = user_data.services['facebook']['access_token']
except KeyError:
return False
graph = GraphAPI(access_token=oauth)
events = yield facebook_paginate(
graph.get_connections('me', connection_name='events'),
max_results=self.num_events_per_user)
data = {"events": []}
for item in events:
event = {}
event['description'] = item.get('description', None)
event['name'] = item.get('name', None)
event['status'] = item.get('rsvp_status', None)
data['events'].append(event)
return data
def __init__(self, tokenpath):
self._token = open(tokenpath).read()
self._graph = facebook.GraphAPI(access_token=self._token, version='2.2')
self.group = self._graph.get_objects(ids=["728040737345863"])
print(self.group["728040737345863"])
def connect(self):
self.__client = facebook.GraphAPI(self.__page_access_token)
# Sends a start up message on Telegram
def test_get_app_access_token(self):
token = facebook.GraphAPI().get_app_access_token(
self.app_id, self.secret)
# Since "unicode" does not exist in Python 3, we cannot check
# the following line with flake8 (hence the noqa comment).
assert(isinstance(token, str) or isinstance(token, unicode)) # noqa
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_no_version(self):
graph = facebook.GraphAPI()
self.assertNotEqual(graph.version, None, "Version should not be None.")
self.assertNotEqual(
graph.version, "", "Version should not be an empty string.")
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_fql(self):
graph = facebook.GraphAPI(version=2.0)
graph.access_token = graph.get_app_access_token(
self.app_id, self.secret)
# Ensure that version is below 2.1. Facebook has stated that FQL is
# not present in this or future versions of the Graph API.
if graph.get_version() < 2.1:
# This is a tautology, but we are limited in what information
# we can retrieve with a proper OAuth access token.
fql_result = graph.fql(
"SELECT app_id from application where app_id = %s" %
self.app_id)
self.assertEqual(fql_result["data"][0]["app_id"], str(self.app_id))
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_bogus_access_token(self):
invalid_token_error_message = "Invalid OAuth access token."
graph = facebook.GraphAPI(access_token='wrong_token')
self.assert_raises_multi_regex(
facebook.GraphAPIError,
invalid_token_error_message,
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_api(cfg):
graph = facebook.GraphAPI(cfg['access_token'])
return graph
def __init__(self, graph=facebook.GraphAPI(FACEBOOK_ACCESS_TOKEN)):
print("Initializing Uploader")
self.events = None
#self.token = self.login()
self.graph = graph
print("Connected to Facebook")
def get_api(cfg):
graph = facebook.GraphAPI(cfg['access_token'])
# Get page token to post as the page. You can skip
# the following if you want to post as yourself.
resp = graph.get_object('me/accounts')
page_access_token = None
for page in resp['data']:
if page['id'] == cfg['page_id']:
page_access_token = page['access_token']
graph = facebook.GraphAPI(page_access_token)
return graph
# You can also skip the above if you get a page token:
# http://stackoverflow.com/questions/8231877/facebook-access-token-for-pages
# and make that long-lived token as in Step 3
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 set_graph(self, oauth_access_token=None):
if oauth_access_token:
self.oauth_access_token = oauth_access_token
self.graph = facebook.GraphAPI(self.oauth_access_token)
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 save_profile(backend, user, response, details, *args, **kwargs):
if backend.name == 'facebook':
# Fetches a list of friends who also use the app
graph = facebook.GraphAPI(access_token=response['access_token'], version='2.9')
friends = [friend for friend in graph.get_all_connections(id='me', connection_name='friends')]
# Update LucemUser with new list of friends using the app
try:
lucem_user = user.lucemuser
except ObjectDoesNotExist:
# Create LucemUser instance if it does not yet exist
lucem_user = LucemUser(user=user)
lucem_user.friends = friends
lucem_user.fb_id = response['id']
lucem_user.save()
def post(message, url=None):
"""
?????????? ? Facebook
"""
config = SocialConfig.get_solo()
if not message:
return
import facebook
facebook_api = facebook.GraphAPI(config.facebook_access_token)
attachment = {}
if url:
attachment['link'] = url
facebook_api.put_wall_post(message=message, attachment=attachment)
def get_api(cfg):
graph = facebook.GraphAPI(cfg['access_token'])
return graph
def __init__(self, debug):
self.debug = debug
self.settings = Settings()
try:
self.graph = facebook.GraphAPI(access_token=self._get_acces_token(), version='2.9')
self.logger = logging.basicConfig(level=logging.DEBUG) if self.debug else logging.basicConfig(
level=logging.INFO)
except KeyError:
exit("Check if configuration is set right")
try:
self.scoop = Reserve()
except:
pass
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 __init__(self, token):
"""
:param token: Facebook Page token
:param _api: Instance of the GraphAPI object
"""
self.token = token
self._api = facebook.GraphAPI(self.token)
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
facebook_utils.py 文件源码
项目：which-of-your-friends-are-on-tinder
作者: philipperemy
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
def query_profile_with_graph_api(profile_id, access_token):
graph = facebook.GraphAPI(access_token)
profile = graph.get_object(profile_id)
return profile
def authenticate_api():
auth_token = facepy.utils.get_application_access_token(FACEBOOK['APP_ID'],
FACEBOOK['APP_SECRET'],
api_version='2.9')
graph = facebook.GraphAPI(auth_token)
return graph
