diff --git a/README.md b/README.md index 2e807ed..5007d6a 100644 --- a/README.md +++ b/README.md @@ -116,6 +116,143 @@ with open(data.filename, 'wb') as f: f.writelines(data.content) ``` +### Kudos of activity +Get a list of athletes who kudoed a given activity. +The returned data is more detailed information about the athlete in comparison with the API call. + +```python +from stravaweblib import WebClient + +# Log in (requires API token and email/password for the site) +client = WebClient(access_token=OAUTH_TOKEN, email=EMAIL, password=PASSWORD) + +# Get the id of the first activity of the current athlete +activities = client.get_activities() +activity_id = activities.next().id + +# Get kudos data for activity of the current athlete +my_activity_kudos_data = client.get_activity_kudos(activity_id=activity_id) + +# Get kudos data for other activities, it means we can retrieve kudos data of any activity +activity_id = 12345678 +other_activity_kudos_data = client.get_activity_kudos(activity_id=activity_id) + +``` + +### Mentionable entities +Get a list of mentionable entities. +The returned data is a list of athletes and clubs that the current athlete is following. + +```python +from stravaweblib import WebClient + +# Log in (requires API token and email/password for the site) +client = WebClient(access_token=OAUTH_TOKEN, email=EMAIL, password=PASSWORD) + +# Get all mentionable entities of the current athlete +mentionable_entities = client.get_mentionable_entities() +``` + +### Give kudos +Give kudos for a given activity. + +```python +from stravaweblib import WebClient + +# Log in (requires API token and email/password for the site) +client = WebClient(access_token=OAUTH_TOKEN, email=EMAIL, password=PASSWORD) + +# Give kudos +activity_id = 12345678 +client.give_kudos(activity_id=activity_id) +``` + +### Post comment +Post a comment for a given activity. + +```python +from stravaweblib import WebClient + +# Log in (requires API token and email/password for the site) +client = WebClient(access_token=OAUTH_TOKEN, email=EMAIL, password=PASSWORD) + +# Post comment. Also, it is possible to mention somebody or any club +activity_id = 12345678 +mentionable_athlete_ids = [12345, 12354] # optional +mentionable_club_ids = [1234] # optional +comment = "Wow, awesome result!" +client.post_comment(activity_id=activity_id, + mentionable_athlete_ids=mentionable_athlete_ids, + mentionable_club_ids=mentionable_club_ids, + comment=comment) +``` +Example of the comment: **AthleteName1** **AthleteName2** **ClubName** Wow, awesome result! + +### Like, unlike comment +Give or remove a like for a given comment. + +```python +from stravaweblib import WebClient + +# Log in (requires API token and email/password for the site) +client = WebClient(access_token=OAUTH_TOKEN, email=EMAIL, password=PASSWORD) + +# Like comment +comment_id = 12345678 +client.like_comment(comment_id=comment_id) + +# Unlike comment +client.unlike_comment(comment_id=comment_id) +``` + +### Get feed entries +Return a list of feed entries. + +```python +from stravaweblib import WebClient + +# Log in (requires API token and email/password for the site) +client = WebClient(access_token=OAUTH_TOKEN, email=EMAIL, password=PASSWORD) + +# Return a list of My Activity feed entries. +my_activity_feed_entries = client.get_my_activity_feed() + +# Return a list of the Following feed entries. +following_feed_entries = client.get_following_feed() + +# Return a list of the given Club feed entries. +club_id=1234 +club_feed_entries = client.get_club_feed(club_id=club_id) +``` + +### Parse athlete profile following tab +Return a list of athletes. + +```python +from stravaweblib import WebClient + +# Log in (requires API token and email/password for the site) +client = WebClient(access_token=OAUTH_TOKEN, email=EMAIL, password=PASSWORD) + +# Return a list of followers athletes of the authorised athlete or given athlete_id +my_followers = client.get_followers_athletes() + +athlete_id=1234567 +athlete_followers = client.get_followers_athletes(athlete_id=athlete_id) + +# Return a list of following athletes of the authorised athlete or given athlete_id +my_following = client.get_following_athletes() + +athlete_id=1234567 +athlete_following = client.get_following_athletes(athlete_id=athlete_id) + +# Return a list of suggested athletes of the authorised athlete. +suggested_athletes = client.get_suggested_athletes() + +# Return a list of both following athletes. Both - it means the current and other athlete are following. +athlete_id=1234567 +both_following_athletes = client.get_both_following_athletes(athlete_id=athlete_id) +``` License ======= diff --git a/setup.py b/setup.py index 57e290e..5922d3b 100644 --- a/setup.py +++ b/setup.py @@ -14,7 +14,7 @@ setup( name="stravaweblib", - version="0.0.8", + version="0.0.9", description="Extends the Strava v3 API using web scraping", long_description=long_description, long_description_content_type="text/markdown", @@ -37,7 +37,7 @@ packages=["stravaweblib"], python_requires=">=3.4.0", install_requires=[ - "stravalib>=0.6.6,<1.0.0", + "stravalib>=0.6.6,<2.0.0", "beautifulsoup4>=4.6.0,<5.0.0", ], ) diff --git a/stravaweblib/models.py b/stravaweblib/models.py new file mode 100644 index 0000000..86137d7 --- /dev/null +++ b/stravaweblib/models.py @@ -0,0 +1,38 @@ +from typing import List + +from pydantic import BaseModel, Field + + +class Athlete(BaseModel): + avatar_url: str + firstname: str + id: int + is_following: bool + is_private: bool + location: str + member_type: str + name: str + url: str + + +class Kudos(BaseModel): + athletes: List[Athlete] = Field(default_factory=list) + is_owner: bool + kudosable: bool + + +class MentionableAthlete(BaseModel): + display: str + id: str + location: str + member_type: str + profile: str + type: str + + +class MentionableClub(BaseModel): + display: str + id: str + image: str + location: str + type: str diff --git a/stravaweblib/webclient.py b/stravaweblib/webclient.py index f988111..1efb2b4 100644 --- a/stravaweblib/webclient.py +++ b/stravaweblib/webclient.py @@ -6,6 +6,9 @@ import functools import json import time +from copy import copy +from json import JSONDecodeError +from typing import List, Union, Dict from bs4 import BeautifulSoup import requests @@ -14,6 +17,9 @@ __all__ = ["WebClient", "FrameType", "DataFormat", "ExportFile", "ActivityFile"] +from pydantic import parse_obj_as + +from .models import Kudos, MentionableAthlete, MentionableClub BASE_URL = "https://www.strava.com" @@ -49,6 +55,19 @@ def __str__(self): return str(self.name).replace("_", " ").title() +class FeedType(str, enum.Enum): + FOLLOWING = "following" + CLUB = "club" + MY_ACTIVITY = "my_activity" + + +class FollowsType(str, enum.Enum): + FOLLOWING = "following" + FOLLOWERS = "followers" + SUGGESTED = "suggested" + BOTH_FOLLOWING = "both_following" + + class WebClient(stravalib.Client): """ An extension to the stravalib Client that fills in some of the gaps in @@ -89,6 +108,7 @@ def __init__(self, *args, **kwargs): # REST API does not have an access_token (yet). Should we verify the match after # exchange_code_for_token()? pass + self.athlete_id = self._session.cookies.get('strava_remember_id') @property def jwt(self): @@ -363,6 +383,348 @@ def get_route_data(self, route_id, fmt=DataFormat.GPX): return self._make_export_file(resp, route_id) + def get_activity_kudos(self, activity_id: int) -> Kudos: + """ + Get a list of athletes who kudoed a given activity. + + The returned data is more detailed information about the athlete in comparison with the API call. + + :param activity_id: a target activity ID for which to fetch kudos. + :type activity_id: int + + :return: List of athletes who kudoed activity. + :rtype: list + """ + url = "{}/feed/activity/{}/kudos".format(BASE_URL, activity_id) + resp = self._session.get(url, allow_redirects=False) + if resp.status_code != 200: + raise stravalib.exc.Fault("Status code '{}' received when trying " + "to fetch detailed kudos." + "".format(resp.status_code)) + + return Kudos(**resp.json()) + + def get_mentionable_entities(self) -> List[Union[MentionableAthlete, MentionableClub]]: + """ + Get a list of mentionable entities. + + The returned data is a list of athletes and clubs that the authorised athlete is following. + + :return: List of mentionable entities. + :rtype: list + """ + url = "{}/athlete/mentionable_entities".format(BASE_URL) + resp = self._session.get(url, allow_redirects=False) + if resp.status_code != 200: + raise stravalib.exc.Fault("Status code '{}' received when trying " + "to fetch mentionable entities." + "".format(resp.status_code)) + + return parse_obj_as(List[Union[MentionableAthlete, MentionableClub]], resp.json()) + + def give_kudos(self, activity_id: int) -> Dict: + """ + Give kudos for a given activity. + + :param activity_id: a target activity ID to be kudoed. + :type activity_id: int + + :return: Status: {"success":"true"} + :rtype: dict + """ + url = "{}/feed/activity/{}/kudo".format(BASE_URL, activity_id) + resp = self._session.post(url, allow_redirects=True, data=self.csrf) + if resp.status_code != 200: + raise stravalib.exc.Fault("Status code '{}' received when trying " + "to give kudos for activity." + "".format(resp.status_code)) + try: + return resp.json() + except JSONDecodeError: + raise ValueError(resp.content) + + def post_comment(self, + activity_id: int, + comment: str, + mentionable_athlete_ids: List[int] = None, + mentionable_club_ids: List[int] = None + ): + """ + Post a comment for a given activity. + + :param activity_id: a target activity ID to be commented. + :type activity_id: int + :param comment: a string to be posted. + :type comment: str + :param mentionable_athlete_ids: set ids of athletes to be mentioned, even one id should be passed as a list + :type mentionable_athlete_ids: list + :param mentionable_club_ids: set ids of clubs to be mentioned, even one id should be passed as a list + :type mentionable_club_ids: list + """ + athlete_str = "" + club_str = "" + if mentionable_athlete_ids: + _athlete_str_pattern = "[strava://athletes/{}] " + for athlete_id in mentionable_athlete_ids: + athlete_str += copy(_athlete_str_pattern).format(athlete_id) + + if mentionable_club_ids: + _club_str_pattern = "[strava://clubs/{}] " + for club_id in mentionable_club_ids: + club_str += copy(_club_str_pattern).format(club_id) + + comment = "{}{}{}".format(athlete_str, club_str, comment) + + url = "{}/feed/activity/{}/comment".format(BASE_URL, activity_id) + resp = self._session.post(url, + allow_redirects=True, + data={ + "comment": comment, + **self.csrf + }) + if resp.status_code != 200: + raise stravalib.exc.Fault("Status code '{}' received when trying " + "to post comment for activity." + "".format(resp.status_code)) + + def like_comment(self, comment_id: int) -> Dict: + """ + Give a like for a given comment. + + :param comment_id: a target comment ID to be liked. + :type comment_id: int + + :return: Status: {"success":"true"} + :rtype: dict + """ + url = "{}/comments/{}/reactions".format(BASE_URL, comment_id) + resp = self._session.post(url, allow_redirects=True, data=self.csrf) + if resp.status_code != 201: + raise stravalib.exc.Fault("Status code '{}' received when trying " + "to like comment.".format(resp.status_code)) + try: + return resp.json() + except JSONDecodeError: + raise ValueError(resp.content) + + def unlike_comment(self, comment_id: int): + """ + Remove a like for a given comment. + + :param comment_id: a target comment ID to be unliked. + :type comment_id: int + """ + url = "{}/comments/{}/reactions".format(BASE_URL, comment_id) + resp = self._session.post(url, + allow_redirects=False, + data={ + "_method": "delete", + **self.csrf + }) + if resp.status_code != 204: + raise stravalib.exc.Fault("Status code '{}' received when trying " + "to unlike comment.".format(resp.status_code)) + + def _get_activity_feed(self, + feed_type: FeedType, + club_id: int = None, + athlete_id: int = None, + before: int = None, + cursor: int = None + ) -> List[dict]: + """ + Return a list of feed entries of given feed type. + + :param feed_type: my_activity, following, club types are allowed + :type feed_type: FeedType + :param athlete_id: a target club ID + :type athlete_id: int + :param athlete_id: authorised athlete ID + :type athlete_id: int + :param before: an epoch timestamp to use for filtering activities that have taken place before a certain time + :type before: int + :param cursor: cursor of the last item in the previous page of results, used to request the subsequent page of results + :type cursor: int + + :return: List of feed entries + :rtype: List[dict] + """ + entries = [] + + if feed_type == FeedType.CLUB and club_id is None: + raise ValueError("`club_id` param must be set.") + + url = "{}/dashboard/feed?feed_type={}".format(BASE_URL, feed_type) + if feed_type == FeedType.CLUB: + url = "{}&club_id={}".format(url, club_id) + if athlete_id: + url = "{}&athlete_id={}".format(url, athlete_id) + if before: + url = "{}&before={}".format(url, before) + if cursor: + url = "{}&cursor={}".format(url, cursor) + + resp = self._session.get(url, allow_redirects=False) + if resp.status_code != 200: + raise stravalib.exc.Fault("Status code '{}' received when trying " + "to retrieve dashboard feed." + "".format(resp.status_code)) + + try: + resp = resp.json() + except JSONDecodeError: + raise ValueError(resp.content) + + entries.extend(resp["entries"]) + + if resp.get("pagination").get("hasMore"): + last_entry = entries[-1] + cursor_data = last_entry["cursorData"] + athlete_id = int(last_entry["viewingAthlete"]["id"]) + before = int(cursor_data["updated_at"]) + cursor = int(cursor_data["rank"]) if cursor_data.get("rank") else cursor_data.get("rank") + entries.extend(self._get_activity_feed(feed_type=feed_type, + club_id=club_id, + athlete_id=athlete_id, + before=before, + cursor=cursor + ) + ) + + return entries + + def get_my_activity_feed(self) -> List[dict]: + """ + Return a list of My Activity feed entries. + + :return: List of feed entries + :rtype: List[dict] + """ + return self._get_activity_feed(feed_type=FeedType.MY_ACTIVITY, athlete_id=self.athlete_id) + + def get_following_feed(self) -> List[dict]: + """ + Return a list of the Following feed entries. + + :return: List of feed entries + :rtype: List[dict] + """ + return self._get_activity_feed(feed_type=FeedType.FOLLOWING, athlete_id=self.athlete_id) + + def get_club_feed(self, club_id: int) -> List[dict]: + """ + Return a list of the given Club feed entries. + + :return: List of feed entries + :rtype: List[dict] + """ + return self._get_activity_feed(feed_type=FeedType.CLUB, club_id=club_id) + + def _parse_athlete_profile_following_tab(self, + athlete_id: int, + follows_type: FollowsType, + pagination_url: str = None + ) -> List[dict]: + """ + Parse the profile following tab and return a list of athletes. + + :param athlete_id: a target athlete ID + :type athlete_id: int + :param follows_type: following, followers, suggested, both_following types are allowed + :type follows_type: FollowsType + :param pagination_url: parsing will process from the given pagination url + :type pagination_url: str + + :return: List of athletes + :rtype: List[dict] + """ + + if pagination_url: + url = "{}{}".format(BASE_URL, pagination_url) + else: + url = "{}/athletes/{}/follows?type={}".format(BASE_URL, athlete_id, follows_type) + + resp = self._session.get(url, allow_redirects=False) + if resp.status_code != 200: + raise stravalib.exc.Fault( + "Failed to load athlete profile page (status code: {})".format(resp.status_code), + ) + + soup = BeautifulSoup(resp.text, 'html.parser') + tab_content = soup.find("div", attrs={"class": "tab-content"}) + list_athletes = tab_content.find("ul", attrs={"class": "list-athletes"}) + if not list_athletes: + raise ValueError("Current athlete doesn't have any {}.".format(follows_type)) + + athletes = [] + for athlete in list_athletes.find_all("li", recursive=False): + athletes.append( + { + "athlete_id": athlete.get("data-athlete-id"), + "athlete_name": athlete.find("div", attrs={"class": "text-callout"}).find("a").text.strip(), + "avatar_img": athlete.find("img", attrs={"class": "avatar-img"}).get("src"), + "location": athlete.find("div", attrs={"class": "location"}).text.strip() + } + ) + + if tab_content.find("nav"): + next_page = tab_content.find("li", attrs={"class": "next_page"}).find("a") + if next_page: + next_page_url = next_page.get("href") + athletes.extend(self._parse_athlete_profile_following_tab( + athlete_id=athlete_id, + follows_type=follows_type, + pagination_url=next_page_url) + ) + + return athletes + + def get_followers_athletes(self, athlete_id: int = None) -> List[dict]: + """ + Return a list of followers athletes of the authorised athlete or given athlete_id + + :param athlete_id: a target athlete ID + :type athlete_id: int + + :return: List of followers athletes + :rtype: List[dict] + """ + return self._parse_athlete_profile_following_tab(athlete_id=athlete_id or self.athlete_id, + follows_type=FollowsType.FOLLOWERS) + + def get_following_athletes(self, athlete_id: int = None) -> List[dict]: + """ + Return a list of following athletes of the authorised athlete or given athlete_id + + :param athlete_id: a target athlete ID + :type athlete_id: int + + :return: List of following athletes + :rtype: List[dict] + """ + return self._parse_athlete_profile_following_tab(athlete_id=athlete_id or self.athlete_id, + follows_type=FollowsType.FOLLOWING) + + def get_suggested_athletes(self) -> List[dict]: + """ + Return a list of suggested athletes of the authorised athlete. + + :return: List of suggested athletes + :rtype: List[dict] + """ + return self._parse_athlete_profile_following_tab(athlete_id=self.athlete_id, follows_type=FollowsType.SUGGESTED) + + def get_both_following_athletes(self, athlete_id: int) -> List[dict]: + """ + Return a list of both following athletes of given athlete_id + + :param athlete_id: a target athlete ID + :type athlete_id: int + + :return: List of following athletes + :rtype: List[dict] + """ + return self._parse_athlete_profile_following_tab(athlete_id=athlete_id, follows_type=FollowsType.BOTH_FOLLOWING) # Inherit parent documentation for WebClient.__init__