From 586065b8c8d72076e8ce1cd540966e6527abf831 Mon Sep 17 00:00:00 2001 From: LonelyCube Date: Sat, 13 Feb 2016 15:28:40 -0500 Subject: [PATCH 1/2] python3 --- examples/authentication.py | 2 +- linkedin/server.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/authentication.py b/examples/authentication.py index cb1a22b..a43771d 100644 --- a/examples/authentication.py +++ b/examples/authentication.py @@ -8,5 +8,5 @@ RETURN_URL = 'http://localhost:8000' authentication = LinkedInAuthentication(API_KEY, API_SECRET, RETURN_URL, PERMISSIONS.enums.values()) - print authentication.authorization_url + print(authentication.authorization_url) application = LinkedInApplication(authentication) diff --git a/linkedin/server.py b/linkedin/server.py index 4032a2a..4a892bd 100644 --- a/linkedin/server.py +++ b/linkedin/server.py @@ -21,7 +21,7 @@ def quick_api(api_key, secret_key, port=8000): auth = LinkedInAuthentication(api_key, secret_key, 'http://localhost:8000/', PERMISSIONS.enums.values()) app = LinkedInApplication(authentication=auth) - print auth.authorization_url + print(auth.authorization_url) _wait_for_user_to_enter_browser(app, port) return app From 1e79a3c378b0a5479b940da7b87df757299705c6 Mon Sep 17 00:00:00 2001 From: Sungho Park Date: Wed, 11 Nov 2020 13:54:21 -0500 Subject: [PATCH 2/2] API url endpoint issue --- linkedin/linkedin.py | 413 +++++-------------------------------------- linkedin/server.py | 1 - 2 files changed, 40 insertions(+), 374 deletions(-) diff --git a/linkedin/linkedin.py b/linkedin/linkedin.py index 5c3a1b0..0d8a5e0 100644 --- a/linkedin/linkedin.py +++ b/linkedin/linkedin.py @@ -5,16 +5,15 @@ import random try: - from urllib.parse import quote, quote_plus + from urllib.parse import quote except ImportError: - from urllib import quote, quote_plus + from urllib import quote import requests from requests_oauthlib import OAuth1 -from .exceptions import LinkedInError -from .models import AccessToken, LinkedInInvitation, LinkedInMessage -from .utils import enum, to_utf8, raise_for_error, json, StringIO +from .models import AccessToken +from .utils import enum, to_utf8, raise_for_error, StringIO __all__ = ['LinkedInAuthentication', 'LinkedInApplication', 'PERMISSIONS'] @@ -31,14 +30,16 @@ MESSAGES='w_messages') ENDPOINTS = enum('LinkedInURL', - PEOPLE='https://api.linkedin.com/v1/people', - PEOPLE_SEARCH='https://api.linkedin.com/v1/people-search', - GROUPS='https://api.linkedin.com/v1/groups', - POSTS='https://api.linkedin.com/v1/posts', - COMPANIES='https://api.linkedin.com/v1/companies', - COMPANY_SEARCH='https://api.linkedin.com/v1/company-search', - JOBS='https://api.linkedin.com/v1/jobs', - JOB_SEARCH='https://api.linkedin.com/v1/job-search') + BASE='https://api.linkedin.com/v2', + CONNECTIONS='https://api.linkedin.com/v2/connections', + PEOPLE='https://api.linkedin.com/v2/people', + PEOPLE_SEARCH='https://api.linkedin.com/v2/people-search', + GROUPS='https://api.linkedin.com/v2/groups', + POSTS='https://api.linkedin.com/v2/posts', + COMPANIES='https://api.linkedin.com/v2/companies', + COMPANY_SEARCH='https://api.linkedin.com/v2/company-search', + JOBS='https://api.linkedin.com/v2/jobs', + JOB_SEARCH='https://api.linkedin.com/v2/job-search') NETWORK_UPDATES = enum('NetworkUpdate', APPLICATION='APPS', @@ -170,374 +171,40 @@ def make_request(self, method, url, data=None, params=None, headers=None, return requests.request(method.upper(), url, **kw) - def get_profile(self, member_id=None, member_url=None, selectors=None, - params=None, headers=None): - if member_id: - if type(member_id) is list: - # Batch request, ids as CSV. - url = '%s::(%s)' % (ENDPOINTS.PEOPLE, - ','.join(member_id)) - else: - url = '%s/id=%s' % (ENDPOINTS.PEOPLE, str(member_id)) - elif member_url: - url = '%s/url=%s' % (ENDPOINTS.PEOPLE, quote_plus(member_url)) - else: - url = '%s/~' % ENDPOINTS.PEOPLE - if selectors: - url = '%s:(%s)' % (url, LinkedInSelector.parse(selectors)) - - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def search_profile(self, selectors=None, params=None, headers=None): - if selectors: - url = '%s:(%s)' % (ENDPOINTS.PEOPLE_SEARCH, - LinkedInSelector.parse(selectors)) - else: - url = ENDPOINTS.PEOPLE_SEARCH - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def get_picture_urls(self, member_id=None, member_url=None, - params=None, headers=None): - if member_id: - url = '%s/id=%s/picture-urls::(original)' % (ENDPOINTS.PEOPLE, str(member_id)) - elif member_url: - url = '%s/url=%s/picture-urls::(original)' % (ENDPOINTS.PEOPLE, - quote_plus(member_url)) - else: - url = '%s/~/picture-urls::(original)' % ENDPOINTS.PEOPLE - - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def get_connections(self, member_id=None, member_url=None, selectors=None, - params=None, headers=None): - if member_id: - url = '%s/id=%s/connections' % (ENDPOINTS.PEOPLE, str(member_id)) - elif member_url: - url = '%s/url=%s/connections' % (ENDPOINTS.PEOPLE, - quote_plus(member_url)) - else: - url = '%s/~/connections' % ENDPOINTS.PEOPLE - if selectors: - url = '%s:(%s)' % (url, LinkedInSelector.parse(selectors)) - - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def get_memberships(self, member_id=None, member_url=None, group_id=None, - selectors=None, params=None, headers=None): - if member_id: - url = '%s/id=%s/group-memberships' % (ENDPOINTS.PEOPLE, str(member_id)) - elif member_url: - url = '%s/url=%s/group-memberships' % (ENDPOINTS.PEOPLE, - quote_plus(member_url)) - else: - url = '%s/~/group-memberships' % ENDPOINTS.PEOPLE - - if group_id: - url = '%s/%s' % (url, str(group_id)) - - if selectors: - url = '%s:(%s)' % (url, LinkedInSelector.parse(selectors)) - - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def get_group(self, group_id, selectors=None, params=None, headers=None): - url = '%s/%s' % (ENDPOINTS.GROUPS, str(group_id)) - - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def get_posts(self, group_id, post_ids=None, selectors=None, params=None, - headers=None): - url = '%s/%s/posts' % (ENDPOINTS.GROUPS, str(group_id)) - if post_ids: - url = '%s::(%s)' % (url, ','.join(map(str, post_ids))) - if selectors: - url = '%s:(%s)' % (url, LinkedInSelector.parse(selectors)) - - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def get_post_comments(self, post_id, selectors=None, params=None, headers=None): - url = '%s/%s/comments' % (ENDPOINTS.POSTS, post_id) - if selectors: - url = '%s:(%s)' % (url, LinkedInSelector.parse(selectors)) - - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def join_group(self, group_id): - url = '%s/~/group-memberships/%s' % (ENDPOINTS.PEOPLE, str(group_id)) - response = self.make_request('PUT', url, - data=json.dumps({'membershipState': {'code': 'member'}})) - raise_for_error(response) - return True - - def leave_group(self, group_id): - url = '%s/~/group-memberships/%s' % (ENDPOINTS.PEOPLE, str(group_id)) - response = self.make_request('DELETE', url) - raise_for_error(response) - return True - - def submit_group_post(self, group_id, title, summary, submitted_url, - submitted_image_url, content_title, description): - post = { - 'title': title, 'summary': summary, - 'content': { - 'submitted-url': submitted_url, - 'title': content_title, - 'description': description - } - } - if submitted_image_url: - post['content']['submitted-image-url'] = submitted_image_url - - url = '%s/%s/posts' % (ENDPOINTS.GROUPS, str(group_id)) - response = self.make_request('POST', url, data=json.dumps(post)) - raise_for_error(response) - return True - - def like_post(self, post_id, action): - url = '%s/%s/relation-to-viewer/is-liked' % (ENDPOINTS.POSTS, str(post_id)) - try: - self.make_request('PUT', url, data=json.dumps(action)) - except (requests.ConnectionError, requests.HTTPError) as error: - raise LinkedInError(error.message) - else: - return True - - def comment_post(self, post_id, comment): - post = { - 'text': comment - } - url = '%s/%s/comments' % (ENDPOINTS.POSTS, str(post_id)) - try: - self.make_request('POST', url, data=json.dumps(post)) - except (requests.ConnectionError, requests.HTTPError) as error: - raise LinkedInError(error.message) - else: - return True - - def get_company_by_email_domain(self, email_domain, params=None, headers=None): - url = '%s?email-domain=%s' % (ENDPOINTS.COMPANIES, email_domain) - + def get_connections(self, totals_only=None, params=None, headers=None): + count = '50' + if totals_only: + count = '0' + url = '%s?q=viewer&start=0&count=%s' % (ENDPOINTS.CONNECTIONS, count) response = self.make_request('GET', url, params=params, headers=headers) raise_for_error(response) return response.json() - def get_companies(self, company_ids=None, universal_names=None, selectors=None, - params=None, headers=None): - identifiers = [] - url = ENDPOINTS.COMPANIES - if company_ids: - identifiers += map(str, company_ids) - - if universal_names: - identifiers += ['universal-name=%s' % un for un in universal_names] - - if identifiers: - url = '%s::(%s)' % (url, ','.join(identifiers)) - - if selectors: - url = '%s:(%s)' % (url, LinkedInSelector.parse(selectors)) - - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def get_company_updates(self, company_id, params=None, headers=None): - url = '%s/%s/updates' % (ENDPOINTS.COMPANIES, str(company_id)) - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def get_company_products(self, company_id, selectors=None, params=None, - headers=None): - url = '%s/%s/products' % (ENDPOINTS.COMPANIES, str(company_id)) - if selectors: - url = '%s:(%s)' % (url, LinkedInSelector.parse(selectors)) - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def follow_company(self, company_id): - url = '%s/~/following/companies' % ENDPOINTS.PEOPLE - post = {'id': company_id} - response = self.make_request('POST', url, data=json.dumps(post)) - raise_for_error(response) - return True - - def unfollow_company(self, company_id): - url = '%s/~/following/companies/id=%s' % (ENDPOINTS.PEOPLE, str(company_id)) - response = self.make_request('DELETE', url) - raise_for_error(response) - return True - - def search_company(self, selectors=None, params=None, headers=None): - url = ENDPOINTS.COMPANY_SEARCH - if selectors: - url = '%s:(%s)' % (url, LinkedInSelector.parse(selectors)) - - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def submit_company_share(self, company_id, comment=None, title=None, description=None, - submitted_url=None, submitted_image_url=None, - visibility_code='anyone'): - - post = { - 'visibility': { - 'code': visibility_code, - }, - } - if comment is not None: - post['comment'] = comment - if title is not None and submitted_url is not None: - post['content'] = { - 'title': title, - 'submitted-url': submitted_url, - 'description': description, - } - if submitted_image_url: - post['content']['submitted-image-url'] = submitted_image_url - - url = '%s/%s/shares' % (ENDPOINTS.COMPANIES, company_id) - - response = self.make_request('POST', url, data=json.dumps(post)) - raise_for_error(response) - return response.json() - - def get_job(self, job_id, selectors=None, params=None, headers=None): - url = '%s/%s' % (ENDPOINTS.JOBS, str(job_id)) - url = '%s:(%s)' % (url, LinkedInSelector.parse(selectors)) - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def get_job_bookmarks(self, selectors=None, params=None, headers=None): - url = '%s/~/job-bookmarks' % ENDPOINTS.PEOPLE - if selectors: - url = '%s:(%s)' % (url, LinkedInSelector.parse(selectors)) - - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def search_job(self, selectors=None, params=None, headers=None): - url = ENDPOINTS.JOB_SEARCH - if selectors: - url = '%s:(%s)' % (url, LinkedInSelector.parse(selectors)) - - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def submit_share(self, comment=None, title=None, description=None, - submitted_url=None, submitted_image_url=None, - visibility_code='anyone'): - post = { - 'visibility': { - 'code': visibility_code, - }, - } - if comment is not None: - post['comment'] = comment - if title is not None and submitted_url is not None: - post['content'] = { - 'title': title, - 'submitted-url': submitted_url, - 'description': description, - } - if submitted_image_url: - post['content']['submitted-image-url'] = submitted_image_url - - url = '%s/~/shares' % ENDPOINTS.PEOPLE - response = self.make_request('POST', url, data=json.dumps(post)) - raise_for_error(response) - return response.json() - - def get_network_updates(self, types, member_id=None, - self_scope=True, params=None, headers=None): - if member_id: - url = '%s/id=%s/network/updates' % (ENDPOINTS.PEOPLE, - str(member_id)) - else: - url = '%s/~/network/updates' % ENDPOINTS.PEOPLE - - if not params: - params = {} - - if types: - params.update({'type': types}) - - if self_scope is True: - params.update({'scope': 'self'}) - - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() - - def get_network_update(self, types, update_key, - self_scope=True, params=None, headers=None): - url = '%s/~/network/updates/key=%s' % (ENDPOINTS.PEOPLE, str(update_key)) - - if not params: - params = {} - - if types: - params.update({'type': types}) - - if self_scope is True: - params.update({'scope': 'self'}) - - response = self.make_request('GET', url, params=params, headers=headers) - raise_for_error(response) - return response.json() + def get_profile(self, member_id=None, member_url=None, selectors=None, + params=None, headers=None): + connections = 0 + if selectors is not None and 'num-connections' in selectors: + connections_response = self.get_connections(totals_only=True) + connections_body = connections_response.get('paging', None) + connections = connections_body.get('total', 0) - def get_network_status(self, params=None, headers=None): - url = '%s/~/network/network-stats' % ENDPOINTS.PEOPLE + url = '%s/me' % ENDPOINTS.BASE response = self.make_request('GET', url, params=params, headers=headers) raise_for_error(response) - return response.json() - - def send_invitation(self, invitation): - assert type(invitation) == LinkedInInvitation, 'LinkedInInvitation required' - url = '%s/~/mailbox' % ENDPOINTS.PEOPLE - response = self.make_request('POST', url, - data=json.dumps(invitation.json)) - raise_for_error(response) - return True - - def send_message(self, message): - assert type(message) == LinkedInMessage, 'LinkedInInvitation required' - url = '%s/~/mailbox' % ENDPOINTS.PEOPLE - response = self.make_request('POST', url, - data=json.dumps(message.json)) - raise_for_error(response) - return True + json_response = response.json() + json_response.update({'numConnections': connections}) + return json_response - def comment_on_update(self, update_key, comment): - comment = {'comment': comment} - url = '%s/~/network/updates/key=%s/update-comments' % (ENDPOINTS.PEOPLE, update_key) - response = self.make_request('POST', url, data=json.dumps(comment)) + def get_email_address(self): + url = "https://api.linkedin.com/v2/emailAddress?q=members&projection=(elements*(handle~))" + response = self.make_request('GET', url) raise_for_error(response) - return True + json_response = response.json() + return json_response['elements'][0]['handle~']['emailAddress'] - def like_update(self, update_key, is_liked=True): - url = '%s/~/network/updates/key=%s/is-liked' % (ENDPOINTS.PEOPLE, update_key) - response = self.make_request('PUT', url, data=json.dumps(is_liked)) + def get_profile_picture(self): + url = "https://api.linkedin.com/v2/me?projection=(id,profilePicture(displayImage~:playableStreams))" + response = self.make_request('GET', url) raise_for_error(response) - return True + json_response = response.json() + return json_response['profilePicture']['displayImage~']['elements'][0]['identifiers'][0]['identifier'] \ No newline at end of file diff --git a/linkedin/server.py b/linkedin/server.py index 4a892bd..ffc6a75 100644 --- a/linkedin/server.py +++ b/linkedin/server.py @@ -21,7 +21,6 @@ def quick_api(api_key, secret_key, port=8000): auth = LinkedInAuthentication(api_key, secret_key, 'http://localhost:8000/', PERMISSIONS.enums.values()) app = LinkedInApplication(authentication=auth) - print(auth.authorization_url) _wait_for_user_to_enter_browser(app, port) return app