Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

new linkedin api endpoints. #4

Open
wants to merge 13 commits into
base: master
Choose a base branch
from
Open

new linkedin api endpoints. #4

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
5a3274b
[NEW] linkedin v2 new endpoints integrations.
qasimgulzar Sep 22, 2018
71871c8
[NEW] new version number.
qasimgulzar Sep 22, 2018
d283c79
[NEW] POST SHARE.
qasimgulzar Sep 22, 2018
9820214
[FIX] PR comments.
qasimgulzar Sep 23, 2018
644b0f4
[FIX] share company .
qasimgulzar Sep 23, 2018
ef48ea8
[FIX] organization api calls.
qasimgulzar Oct 6, 2018
a5e22ac
'HTTPError' object has no attribute 'message' FIX
luiscastillocr May 14, 2019
d975295
TypeError: list indices must be integers, not str FIX
luiscastillocr May 14, 2019
9f817b5
better contentEntities
luiscastillocr May 14, 2019
f8faeaa
refresh access token, some few improvements
luiscastillocr Sep 1, 2020
3c5511b
LinkedInApplication AccessToken.__new__ missing params
luiscastillocr Sep 1, 2020
010bbe1
Merge pull request #1 from WikiRealtyInc/master
qasimgulzar Oct 6, 2021
03be797
Merge branch 'master' into master
qasimgulzar Oct 6, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion linkedin_v2/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,2 +1,2 @@
__version__ = '0.9.2'
__version__ = '0.9.3'
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This should now be 0.9.6

VERSION = tuple(map(int, __version__.split('.')))
267 changes: 255 additions & 12 deletions linkedin_v2/linkedin.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@

from .models import AccessToken
from .utils import enum, to_utf8, raise_for_error, StringIO
import json
import urllib


__all__ = ['LinkedInAuthentication', 'LinkedInApplication', 'PERMISSIONS']
Expand All @@ -37,9 +39,10 @@
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',
COMPANY_SEARCH='https://api.linkedin.com/v2/search?q=companiesV2',
JOBS='https://api.linkedin.com/v2/jobs',
JOB_SEARCH='https://api.linkedin.com/v2/job-search')
JOB_SEARCH='https://api.linkedin.com/v2/job-search',
SHARE='https://api.linkedin.com/v2/shares')

NETWORK_UPDATES = enum('NetworkUpdate',
APPLICATION='APPS',
Expand Down Expand Up @@ -108,7 +111,8 @@ def last_error(self):

def _make_new_state(self):
return hashlib.md5(
'{}{}'.format(random.randrange(0, 2 ** 63), self.secret).encode("utf8")
'{}{}'.format(random.randrange(0, 2 ** 63),
self.secret).encode("utf8")
).hexdigest()

def get_access_token(self, timeout=60):
Expand All @@ -118,10 +122,12 @@ def get_access_token(self, timeout=60):
'redirect_uri': self.redirect_uri,
'client_id': self.key,
'client_secret': self.secret}
response = requests.post(self.ACCESS_TOKEN_URL, data=qd, timeout=timeout)
response = requests.post(
self.ACCESS_TOKEN_URL, data=qd, timeout=timeout)
raise_for_error(response)
response = response.json()
self.token = AccessToken(response['access_token'], response['expires_in'])
self.token = AccessToken(
response['access_token'], response['expires_in'])
return self.token


Expand Down Expand Up @@ -152,9 +158,11 @@ def __init__(self, authentication=None, token=None):
def make_request(self, method, url, data=None, params=None, headers=None,
timeout=60):
if headers is None:
headers = {'x-li-format': 'json', 'Content-Type': 'application/json'}
headers = {'x-li-format': 'json',
'Content-Type': 'application/json'}
else:
headers.update({'x-li-format': 'json', 'Content-Type': 'application/json'})
headers.update(
{'x-li-format': 'json', 'Content-Type': 'application/json'})

if params is None:
params = {}
Expand All @@ -167,7 +175,8 @@ def make_request(self, method, url, data=None, params=None, headers=None,
self.authentication.user_token, self.authentication.user_secret)
kw.update({'auth': auth})
else:
params.update({'oauth2_access_token': self.authentication.token.access_token})
params.update(
{'oauth2_access_token': self.authentication.token.access_token})

return requests.request(method.upper(), url, **kw)

Expand All @@ -176,8 +185,9 @@ def get_connections(self, totals_only=None, params=None, headers=None):
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)
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,
Expand All @@ -187,10 +197,243 @@ def get_profile(self, member_id=None, member_url=None, selectors=None,
connections_response = self.get_connections(totals_only=True)
connections_body = connections_response.get('paging', None)
connections = connections_body.get('total', 0)

url = '%s/me' % ENDPOINTS.BASE
response = self.make_request('GET', url, params=params, headers=headers)
response = self.make_request(
'GET', url, params=params, headers=headers)
raise_for_error(response)
json_response = response.json()
json_response.update({'numConnections': connections})
return json_response

def get_shares(self, totals_only=None, params=None, headers=None):
count = '100'
if totals_only:
count = '0'
url = '%s?q=owners&owners=urn:li:person:%s&sharesPerOwner=100' % (
ENDPOINTS.SHARE, self.get_profile()['id'])
response = self.make_request(
'GET', url, params=params, headers=headers)
raise_for_error(response)
return response.json()

def search_profile(self, params):
url = '%s/clientAwareMemberHandles?q=handleString&%s&projection=(elements*(member~))' % (
ENDPOINTS.BASE, urllib.parse.urlencode(params))
print(url)
response = self.make_request('GET', url)
raise_for_error(response)
return response.json()

def post_share(self, post_type='person', company_id=None, comment=None, title=None, description=None,
submitted_url=None, submitted_image_url=None,
visibility_code='anyone'):
post_owner = ''
if post_type == 'organization':
post_owner = "urn:li:organization:%s" % company_id
else:
post_owner = "urn:li:person:%s" % self.get_profile()['id']
post = {
"owner": post_owner,
"text": {
"text": description
},
"subject": title,
"distribution": {
"linkedInDistributionTarget": {}
},
"content": {
"contentEntities": [
{
"entityLocation": "",
"thumbnails": [
{
"resolvedUrl": ""
}
]
}
],
"title": ""
}
}
if comment is not None:
post['comment'] = comment
if title is not None:
post['content']['title'] = title
if submitted_url is not None:
post['content']['submitted-url'] = submitted_url
if submitted_image_url is not None:
post['content']['contentEntities']['thumbnails']['resolvedUrl'] = submitted_image_url
response = application.make_request(
'POST', ENDPOINTS.SHARE, data=json.dumps(post))
raise_for_error(response)
return response.json()

def search_company(self, params):
url = '%s/search?q=companiesV2&%s' % (
ENDPOINTS.BASE, urllib.parse.urlencode(params))
response = self.make_request('GET', url)
# raise_for_error(response)
return response.json()

def get_organization(self, organization_id, params=None, headers=None):
url = '%s/organizations/%s' % (ENDPOINTS.BASE, organization_id)
response = self.make_request(
'GET', url, params=params, headers=headers)
raise_for_error(response)
return response.json()

def get_brand(self, brand_id, params=None, headers=None):
url = '%s/organizationBrands/%s' % (ENDPOINTS.BASE, brand_id)
response = self.make_request(
'GET', url, params=params, headers=headers)
raise_for_error(response)
return response.json()

def send_invitation(self, invitee_email):
post = {
"invitee": "urn:li:email:%s" % invitee_email,
"message": {
"com.linkedin.invitations.InvitationMessage": {
"body": "Let's connect!"
}
}
}
response = application.make_request(
'POST', '%s/invitations' % ENDPOINTS.BASE, data=json.dumps(post))
raise_for_error(response)
return response.json()

def search_job(self):
url = '%s/recommendedJobs?q=byMember' % ENDPOINTS.BASE
response = self.make_request('GET', url)
# raise_for_error(response)
return response.json()

def get_job(self, **kwargs):
return self.search_job()

def get_post_comments(self, selectors, params=None, **kwargs):
url = '%s/socialActions/urn:li:share:%s/comments' % (
ENDPOINTS.BASE, kwargs['post_id'])
print(url)
response = self.make_request(
'GET', url, params=params)
# raise_for_error(response)
return response.json()

def get_groupss(self, group_id, params=None, headers=None):
reicolina marked this conversation as resolved.
Show resolved Hide resolved
url = '%s/groupDefinitions/%s' % (ENDPOINTS.BASE, group_id)
response = self.make_request(
'GET', url, params=params, headers=headers)
raise_for_error(response)
return response.json()

def get_groups(self, group_ids, params=None, headers=None):
url = '%s/groupDefinitions/?ids=List(%s)' % (ENDPOINTS.BASE, group_ids)
response = self.make_request(
'GET', url, params=params, headers=headers)
raise_for_error(response)
return response.json()

def get_memberships(self, group_id):
post = {
"action": "SEND_REQUEST",
"group": "urn:li:group:%s" % group_id,
"members": [
"urn:li:person:%s" % self.get_profile['id'],
]
}
response = self.make_request(
'POST', '%s/groupMemberships?action=membershipAction' % ENDPOINTS.BASE, data=json.dumps(post))
raise_for_error(response)
return response.json()

def submit_group_post(self, group_id, title, description,
shareCommentary):
post = {
"author": "urn:li:person:%s" % self.get_profile()['id'],
"containerEntity": "urn:li:group:%s" % group_id,
"lifecycleState": "PUBLISHED",
"specificContent": {
"com.linkedin.ugc.ShareContent": {
"media": [
{
"title": {
"attributes": [],
"text": title
},
"description": {
"attributes": [],
"text": description
},
"thumbnails": [],
"status": "READY"
}
],
"shareCommentary": {
"attributes": [],
"text": ""
}
}
},
"visibility": {
"com.linkedin.ugc.MemberNetworkVisibility": "CONTAINER"
}
}
if shareCommentary is not None:
post['specificContent']['shareCommentary'] = shareCommentary
response = self.make_request(
'POST', ENDPOINTS.SHARE, data=json.dumps(post))
raise_for_error(response)
return response.json()

def get_company_updates(self, organization_id, post):
url = '%s/people/id=%s/organizations/%s' % (
ENDPOINTS.SHARE, self.get_profile['id'], organization_id)
response = self.make_request(
'POST', url, data=json.dumps(post))
raise_for_error(response)
return response.json()

def get_group(self):
print(ENDPOINTS.BASE)
url = "%s/groupMemberships?q=member&member=urn:li:person:%s&membershipStatuses=List(MEMBER,OWNER)" % (ENDPOINTS.BASE, self.get_profile()['id'])
response = self.make_request('GET', url)
# raise_for_error(response)
return response.json()

def get_memberships(self):
pass

def submit_company_share(self, **kwargs):
params = kwargs.pop('params', True)
response = self.make_request(
'POST', ENDPOINTS.SHARE, data=json.dumps(params))
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'):
profile = self.get_profile()
post = {
'owner': 'urn:li:person:%s' % profile['id'],
"text": {
"text": description
},
"subject": title
}

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,
# }
reicolina marked this conversation as resolved.
Show resolved Hide resolved
if submitted_image_url:
post['content']['contentEntities']['thumbnails']['resolvedUrl'] = submitted_image_url
response = self.make_request('POST', 'https://api.linkedin.com/v2/shares', data=json.dumps(post))
return response.json()
reicolina marked this conversation as resolved.
Show resolved Hide resolved