From d9b48b6f08c33ae44cc3f55e1ff717d8453516bf Mon Sep 17 00:00:00 2001 From: faretek Date: Fri, 23 Jan 2026 17:50:19 +0000 Subject: [PATCH 1/3] fix: decode_session_id: detect compression This probably doesn't affect anyone but it is more accurate Closes #540 --- scratchattach/site/session.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/scratchattach/site/session.py b/scratchattach/site/session.py index 6a3fda08..0710944c 100644 --- a/scratchattach/site/session.py +++ b/scratchattach/site/session.py @@ -1133,7 +1133,7 @@ def decode_session_id(session_id: str) -> tuple[dict[str, str], datetime.datetim Session id is in the format: :: - p1 contains a base64-zlib compressed JSON string + p1 contains a base64 JSON string (if it starts with `.`, then it is zlib compressed) p2 is a base 62 encoded timestamp p3 might be a `synchronous signature` for the first 2 parts (might be useless for us) @@ -1148,10 +1148,13 @@ def decode_session_id(session_id: str) -> tuple[dict[str, str], datetime.datetim - django_timezone - _auth_user_hash """ - p1, p2, p3 = session_id.split(':') + p1, p2, _ = session_id.split(':') + p1_bytes = base64.urlsafe_b64decode(p1 + "==") + if p1.startswith('.'): + p1_bytes = zlib.decompress(p1_bytes) return ( - json.loads(zlib.decompress(base64.urlsafe_b64decode(p1 + "=="))), + json.loads(p1_bytes), datetime.datetime.fromtimestamp(commons.b62_decode(p2)) ) From 5f52d9dad4503827c7b8f7420d4bd38fe2af3d47 Mon Sep 17 00:00:00 2001 From: faretek Date: Fri, 23 Jan 2026 17:58:49 +0000 Subject: [PATCH 2/3] fix: incorrect zlib check --- scratchattach/site/session.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scratchattach/site/session.py b/scratchattach/site/session.py index 0710944c..d22d9597 100644 --- a/scratchattach/site/session.py +++ b/scratchattach/site/session.py @@ -1150,7 +1150,7 @@ def decode_session_id(session_id: str) -> tuple[dict[str, str], datetime.datetim """ p1, p2, _ = session_id.split(':') p1_bytes = base64.urlsafe_b64decode(p1 + "==") - if p1.startswith('.'): + if p1.startswith('".'): p1_bytes = zlib.decompress(p1_bytes) return ( From 1c6b6b8b9e1213f1b8b5e9f573b2716c0a9c6c12 Mon Sep 17 00:00:00 2001 From: faretek Date: Fri, 23 Jan 2026 18:29:00 +0000 Subject: [PATCH 3/3] fix: cat ear detection --- scratchattach/site/user.py | 453 +++++++++++++++++++++++++------------ tests/test_memberships.py | 5 +- 2 files changed, 310 insertions(+), 148 deletions(-) diff --git a/scratchattach/site/user.py b/scratchattach/site/user.py index 070db4fb..a36506f7 100644 --- a/scratchattach/site/user.py +++ b/scratchattach/site/user.py @@ -1,4 +1,5 @@ """User class""" + from __future__ import annotations import json @@ -31,18 +32,22 @@ from . import typed_dicts from . import session + class Rank(Enum): """ Possible ranks in scratch """ + NEW_SCRATCHER = 0 SCRATCHER = 1 SCRATCH_TEAM = 2 + class _OcularStatusMeta(TypedDict): updated: str updatedBy: str + class _OcularStatus(TypedDict): _id: str name: str @@ -50,21 +55,36 @@ class _OcularStatus(TypedDict): color: str meta: _OcularStatusMeta + class Verificator: def __init__(self, user: User, project_id: int): - self.project = user._make_linked_object("id", project_id, project.Project, exceptions.ProjectNotFound) + self.project = user._make_linked_object( + "id", project_id, project.Project, exceptions.ProjectNotFound + ) self.projecturl = self.project.url - self.code = ''.join(random.choices(string.ascii_letters + string.digits, k=8)) + self.code = "".join(random.choices(string.ascii_letters + string.digits, k=8)) self.username = user.username def check(self) -> bool: - return bool(list(filter(lambda x : x.author_name == self.username and (x.content == self.code or x.content.startswith(self.code) or x.content.endswith(self.code)), self.project.comments()))) + return bool( + list( + filter( + lambda x: x.author_name == self.username + and ( + x.content == self.code + or x.content.startswith(self.code) + or x.content.endswith(self.code) + ), + self.project.comments(), + ) + ) + ) + @dataclass class User(BaseSiteComponent[typed_dicts.UserDict]): - - ''' + """ Represents a Scratch user. Attributes: @@ -84,7 +104,8 @@ class User(BaseSiteComponent[typed_dicts.UserDict]): :.scratchteam: Retuns True if the user is in the Scratch team :.update(): Updates the attributes - ''' + """ + username: str = field(kw_only=True, default="") join_date: str = field(kw_only=True, default="") about_me: str = field(kw_only=True, default="") @@ -95,7 +116,9 @@ class User(BaseSiteComponent[typed_dicts.UserDict]): scratchteam: bool = field(kw_only=True, repr=False, default=False) is_member: bool = field(kw_only=True, repr=False, default=False) has_ears: bool = field(kw_only=True, repr=False, default=False) - _classroom: tuple[bool, Optional[classroom.Classroom]] = field(init=False, default=(False, None)) + _classroom: tuple[bool, Optional[classroom.Classroom]] = field( + init=False, default=(False, None) + ) _headers: dict[str, str] = field(init=False, default_factory=headers.copy) _cookies: dict[str, str] = field(init=False, default_factory=dict) _json_headers: dict[str, str] = field(init=False, default_factory=dict) @@ -163,7 +186,8 @@ def _assert_permission(self): self._assert_auth() if self._session.username != self.username: raise exceptions.Unauthorized( - "You need to be authenticated as the profile owner to do this.") + "You need to be authenticated as the profile owner to do this." + ) @property def url(self): @@ -177,10 +201,10 @@ def __rich__(self): featured_data = self.featured_data() or {} ocular_data = self.ocular_status() - ocular = 'No ocular status' + ocular = "No ocular status" if status := ocular_data.get("status"): - color_str = '' + color_str = "" color_data = ocular_data.get("color") if color_data is not None: color_str = f"[{color_data}] ⬤ [/]" @@ -197,13 +221,17 @@ def __rich__(self): info.add_row("Joined", escape(self.join_date)) info.add_row("Country", escape(self.country)) info.add_row("Messages", str(self.message_count())) - info.add_row("Class", str(_classroom.title if _classroom is not None else 'None')) + info.add_row( + "Class", str(_classroom.title if _classroom is not None else "None") + ) desc = Table("Profile", ocular, box=box.SIMPLE) desc.add_row("About me", escape(self.about_me)) desc.add_row("Wiwo", escape(self.wiwo)) - desc.add_row(escape(featured_data.get("label", "Featured Project")), - escape(str(self.connect_featured_project()))) + desc.add_row( + escape(featured_data.get("label", "Featured Project")), + escape(str(self.connect_featured_project())), + ) ret = Table.grid(expand=True) @@ -245,12 +273,14 @@ def classroom(self) -> classroom.Classroom | None: continue href = str(a.get("href")) if re.match(r"/classes/\d*/", href): - class_name = a.text.strip()[len("Student of: "):] - is_closed = bool(re.search(r"\n *\(ended\)", class_name))# as this has a \n, we can be sure + class_name = a.text.strip()[len("Student of: ") :] + is_closed = bool( + re.search(r"\n *\(ended\)", class_name) + ) # as this has a \n, we can be sure if is_closed: class_name = re.sub(r"\n *\(ended\)", "", class_name).strip() - class_id = int(href.split('/')[2]) + class_id = int(href.split("/")[2]) break if class_name: @@ -258,7 +288,7 @@ def classroom(self) -> classroom.Classroom | None: _session=self._session, id=class_id or 0, title=class_name, - is_closed=is_closed + is_closed=is_closed, ) else: self._classroom = True, None @@ -271,7 +301,9 @@ def does_exist(self) -> Optional[bool]: boolean : True if the user exists, False if the user is deleted, None if an error occured """ with requests.no_error_handling(): - status_code = requests.get(f"https://scratch.mit.edu/users/{self.username}/").status_code + status_code = requests.get( + f"https://scratch.mit.edu/users/{self.username}/" + ).status_code if status_code == 200: return True elif status_code == 404: @@ -279,9 +311,8 @@ def does_exist(self) -> Optional[bool]: return None - # Will maybe be deprecated later, but for now still has its own purpose. - #@deprecated("This function is partially deprecated. Use user.rank() instead.") + # @deprecated("This function is partially deprecated. Use user.rank() instead.") def is_new_scratcher(self): """ Returns: @@ -289,8 +320,10 @@ def is_new_scratcher(self): """ try: with requests.no_error_handling(): - res = requests.get(f"https://scratch.mit.edu/users/{self.username}/").text - group = res[res.rindex(''):][:70] + res = requests.get( + f"https://scratch.mit.edu/users/{self.username}/" + ).text + group = res[res.rindex('') :][:70] return "new scratcher" in group.lower() except Exception as e: @@ -298,7 +331,14 @@ def is_new_scratcher(self): return None def message_count(self): - return json.loads(requests.get(f"https://api.scratch.mit.edu/users/{self.username}/messages/count/?cachebust={random.randint(0,10000)}", headers = {'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.3c6 (KHTML, like Gecko) Chrome/75.0.3770.142 Safari/537.36',}).text)["count"] + return json.loads( + requests.get( + f"https://api.scratch.mit.edu/users/{self.username}/messages/count/?cachebust={random.randint(0,10000)}", + headers={ + "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.3c6 (KHTML, like Gecko) Chrome/75.0.3770.142 Safari/537.36", + }, + ).text + )["count"] def featured_data(self): """ @@ -306,17 +346,19 @@ def featured_data(self): dict: Gets info on the user's featured project and featured label (like "Featured project", "My favorite things", etc.) """ try: - response = requests.get(f"https://scratch.mit.edu/site-api/users/all/{self.username}/").json() + response = requests.get( + f"https://scratch.mit.edu/site-api/users/all/{self.username}/" + ).json() return { - "label":response["featured_project_label_name"], - "project": - dict( - id=str(response["featured_project_data"]["id"]), - author=response["featured_project_data"]["creator"], - thumbnail_url="https://"+response["featured_project_data"]["thumbnail_url"][2:], - title=response["featured_project_data"]["title"] - ) - } + "label": response["featured_project_label_name"], + "project": dict( + id=str(response["featured_project_data"]["id"]), + author=response["featured_project_data"]["creator"], + thumbnail_url="https://" + + response["featured_project_data"]["thumbnail_url"][2:], + title=response["featured_project_data"]["title"], + ), + } except Exception: return None @@ -333,14 +375,21 @@ def unfollowers(self) -> list[User]: usernames = [] for i in range(1, 2 + follower_count // 60): with requests.no_error_handling(): - resp = requests.get(f"https://scratch.mit.edu/users/{self.username}/followers/", params={"page": i}) + resp = requests.get( + f"https://scratch.mit.edu/users/{self.username}/followers/", + params={"page": i}, + ) soup = BeautifulSoup(resp.text, "html.parser") usernames.extend(span.text.strip() for span in soup.select("span.title")) # api response contains all-time followers, including deleted and unfollowed unfollowers = [] for offset in range(0, follower_count, 40): - unfollowers.extend(user for user in self.followers(offset=offset, limit=40) if user.username not in usernames) + unfollowers.extend( + user + for user in self.followers(offset=offset, limit=40) + if user.username not in usernames + ) return unfollowers @@ -351,7 +400,7 @@ def follower_count(self): with requests.no_error_handling(): text = requests.get( f"https://scratch.mit.edu/users/{self.username}/followers/", - headers = self._headers + headers=self._headers, ).text return commons.webscrape_count(text, "Followers (", ")") @@ -359,7 +408,7 @@ def following_count(self): with requests.no_error_handling(): text = requests.get( f"https://scratch.mit.edu/users/{self.username}/following/", - headers = self._headers + headers=self._headers, ).text return commons.webscrape_count(text, "Following (", ")") @@ -369,7 +418,10 @@ def followers(self, *, limit=40, offset=0): list: The user's followers as list of scratchattach.user.User objects """ response = commons.api_iterative( - f"https://api.scratch.mit.edu/users/{self.username}/followers/", limit=limit, offset=offset) + f"https://api.scratch.mit.edu/users/{self.username}/followers/", + limit=limit, + offset=offset, + ) return commons.parse_object_list(response, User, self._session, "username") def follower_names(self, *, limit=40, offset=0): @@ -385,7 +437,10 @@ def following(self, *, limit=40, offset=0): list: The users that the user is following as list of scratchattach.user.User objects """ response = commons.api_iterative( - f"https://api.scratch.mit.edu/users/{self.username}/following/", limit=limit, offset=offset) + f"https://api.scratch.mit.edu/users/{self.username}/following/", + limit=limit, + offset=offset, + ) return commons.parse_object_list(response, User, self._session, "username") def following_names(self, *, limit=40, offset=0): @@ -443,7 +498,7 @@ def project_count(self): with requests.no_error_handling(): text = requests.get( f"https://scratch.mit.edu/users/{self.username}/projects/", - headers = self._headers + headers=self._headers, ).text return commons.webscrape_count(text, "Shared Projects (", ")") @@ -451,7 +506,7 @@ def studio_count(self): with requests.no_error_handling(): text = requests.get( f"https://scratch.mit.edu/users/{self.username}/studios/", - headers = self._headers + headers=self._headers, ).text return commons.webscrape_count(text, "Studios I Curate (", ")") @@ -459,16 +514,19 @@ def studios_following_count(self): with requests.no_error_handling(): text = requests.get( f"https://scratch.mit.edu/users/{self.username}/studios_following/", - headers = self._headers + headers=self._headers, ).text return commons.webscrape_count(text, "Studios I Follow (", ")") def studios(self, *, limit=40, offset=0) -> list[studio.Studio]: _studios = commons.api_iterative( - f"https://api.scratch.mit.edu/users/{self.username}/studios/curate", limit=limit, offset=offset) + f"https://api.scratch.mit.edu/users/{self.username}/studios/curate", + limit=limit, + offset=offset, + ) studios = [] for studio_dict in _studios: - _studio = studio.Studio(_session = self._session, id = studio_dict["id"]) + _studio = studio.Studio(_session=self._session, id=studio_dict["id"]) _studio._update_from_dict(studio_dict) studios.append(_studio) return studios @@ -479,12 +537,18 @@ def projects(self, *, limit=40, offset=0) -> list[project.Project]: list: The user's shared projects """ _projects = commons.api_iterative( - f"https://api.scratch.mit.edu/users/{self.username}/projects/", limit=limit, offset=offset, _headers= self._headers) + f"https://api.scratch.mit.edu/users/{self.username}/projects/", + limit=limit, + offset=offset, + _headers=self._headers, + ) for p in _projects: - p["author"] = {"username":self.username} + p["author"] = {"username": self.username} return commons.parse_object_list(_projects, project.Project, self._session) - def loves(self, *, limit=40, offset=0, get_full_project: bool = False) -> list[project.Project]: + def loves( + self, *, limit=40, offset=0, get_full_project: bool = False + ) -> list[project.Project]: """ Returns: list: The user's loved projects @@ -504,8 +568,7 @@ def loves(self, *, limit=40, offset=0, get_full_project: bool = False) -> list[p # So the page number for this is 1 + (offset + limit - 1) // 40 # But this is a range so we have to add another 1 for the second argument - pages = range(1 + offset // 40, - 2 + (offset + limit - 1) // 40) + pages = range(1 + offset // 40, 2 + (offset + limit - 1) // 40) _projects = [] for page in pages: @@ -513,13 +576,13 @@ def loves(self, *, limit=40, offset=0, get_full_project: bool = False) -> list[p first_idx = (page - 1) * 40 with requests.no_error_handling(): - page_content = requests.get(f"https://scratch.mit.edu/projects/all/{self.username}/loves/" - f"?page={page}", headers=self._headers).content + page_content = requests.get( + f"https://scratch.mit.edu/projects/all/{self.username}/loves/" + f"?page={page}", + headers=self._headers, + ).content - soup = BeautifulSoup( - page_content, - "html.parser" - ) + soup = BeautifulSoup(page_content, "html.parser") # We need to check if we are out of bounds # If we are, we can jump out early @@ -537,7 +600,8 @@ def loves(self, *, limit=40, offset=0, get_full_project: bool = False) -> list[p # Each project element is a list item with the class name 'project thumb item' so we can just use that for i, project_element in enumerate( - soup.find_all("li", {"class": "project thumb item"})): + soup.find_all("li", {"class": "project thumb item"}) + ): # Remember we only want certain projects: # The current project idx = first_idx + i # We want to start at {offset} and end at {offset + limit} @@ -564,26 +628,27 @@ def loves(self, *, limit=40, offset=0, get_full_project: bool = False) -> list[p assert isinstance(first_anchor, Tag) assert isinstance(second_anchor, Tag) assert isinstance(third_anchor, Tag) - project_id = commons.webscrape_count(first_anchor.attrs["href"], - "/projects/", "/") + project_id = commons.webscrape_count( + first_anchor.attrs["href"], "/projects/", "/" + ) title = second_anchor.contents[0] author = third_anchor.contents[0] # Instantiating a project with the properties that we know # This may cause issues (see below) - _project = project.Project(id=project_id, - _session=self._session, - title=title, - author_name=author, - url=f"https://scratch.mit.edu/projects/{project_id}/") + _project = project.Project( + id=project_id, + _session=self._session, + title=title, + author_name=author, + url=f"https://scratch.mit.edu/projects/{project_id}/", + ) if get_full_project: # Put this under an if statement since making api requests for every single # project will cause the function to take a lot longer _project.update() - _projects.append( - _project - ) + _projects.append(_project) return _projects @@ -591,7 +656,7 @@ def loves_count(self): with requests.no_error_handling(): text = requests.get( f"https://scratch.mit.edu/projects/all/{self.username}/loves/", - headers=self._headers + headers=self._headers, ).text # If there are no loved projects, then Scratch doesn't actually display the number - so we have to catch this @@ -609,14 +674,18 @@ def favorites(self, *, limit=40, offset=0): list: The user's favorite projects """ _projects = commons.api_iterative( - f"https://api.scratch.mit.edu/users/{self.username}/favorites/", limit=limit, offset=offset, _headers= self._headers) + f"https://api.scratch.mit.edu/users/{self.username}/favorites/", + limit=limit, + offset=offset, + _headers=self._headers, + ) return commons.parse_object_list(_projects, project.Project, self._session) def favorites_count(self): with requests.no_error_handling(): text = requests.get( f"https://scratch.mit.edu/users/{self.username}/favorites/", - headers=self._headers + headers=self._headers, ).text return commons.webscrape_count(text, "Favorites (", ")") @@ -633,20 +702,19 @@ def has_badge(self) -> bool: return False for child in head.children: if child.name == "img": - if child["src"] == "//cdn.scratch.mit.edu/scratchr2/static/__ff7229f036c458728e45c39b0751aa44__/membership/membership-badge.svg": + if "membership-badge.svg" in child["src"]: return True return False - - def toggle_commenting(self): """ You can only use this function if this object was created using :meth:`scratchattach.session.Session.connect_user` """ self._assert_permission() - requests.post(f"https://scratch.mit.edu/site-api/comments/user/{self.username}/toggle-comments/", - headers = headers, - cookies = self._cookies + requests.post( + f"https://scratch.mit.edu/site-api/comments/user/{self.username}/toggle-comments/", + headers=headers, + cookies=self._cookies, ) def viewed_projects(self, limit=24, offset=0): @@ -658,7 +726,11 @@ def viewed_projects(self, limit=24, offset=0): """ self._assert_permission() _projects = commons.api_iterative( - f"https://api.scratch.mit.edu/users/{self.username}/projects/recentlyviewed", limit=limit, offset=offset, _headers= self._headers) + f"https://api.scratch.mit.edu/users/{self.username}/projects/recentlyviewed", + limit=limit, + offset=offset, + _headers=self._headers, + ) return commons.parse_object_list(_projects, project.Project, self._session) def set_pfp(self, image: bytes): @@ -671,7 +743,8 @@ def set_pfp(self, image: bytes): f"https://scratch.mit.edu/site-api/users/all/{self.username}/", headers=self._headers, cookies=self._cookies, - files={"file": image}) + files={"file": image}, + ) def set_bio(self, text): """ @@ -683,7 +756,8 @@ def set_bio(self, text): f"https://scratch.mit.edu/site-api/users/all/{self.username}/", headers=self._json_headers, cookies=self._cookies, - json={"bio": text}) + json={"bio": text}, + ) def set_wiwo(self, text): """ @@ -695,7 +769,8 @@ def set_wiwo(self, text): f"https://scratch.mit.edu/site-api/users/all/{self.username}/", headers=self._json_headers, cookies=self._cookies, - json={"status": text}) + json={"status": text}, + ) def set_featured(self, project_id, *, label=""): """ @@ -712,7 +787,7 @@ def set_featured(self, project_id, *, label=""): f"https://scratch.mit.edu/site-api/users/all/{self.username}/", headers=self._json_headers, cookies=self._cookies, - json={"featured_project": int(project_id), "featured_project_label": label} + json={"featured_project": int(project_id), "featured_project_label": label}, ) def set_forum_signature(self, text): @@ -721,18 +796,23 @@ def set_forum_signature(self, text): """ self._assert_permission() headers = { - 'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7', - 'content-type': 'application/x-www-form-urlencoded', - 'origin': 'https://scratch.mit.edu', - 'referer': 'https://scratch.mit.edu/discuss/settings/TimMcCool/', - 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36', + "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7", + "content-type": "application/x-www-form-urlencoded", + "origin": "https://scratch.mit.edu", + "referer": "https://scratch.mit.edu/discuss/settings/TimMcCool/", + "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36", } data = { - 'csrfmiddlewaretoken': 'a', - 'signature': text, - 'update': '', + "csrfmiddlewaretoken": "a", + "signature": text, + "update": "", } - response = requests.post(f'https://scratch.mit.edu/discuss/settings/{self.username}/', cookies=self._cookies, headers=headers, data=data) + response = requests.post( + f"https://scratch.mit.edu/discuss/settings/{self.username}/", + cookies=self._cookies, + headers=headers, + data=data, + ) def post_comment(self, content, *, parent_id="", commentee_id=""): """ @@ -750,9 +830,9 @@ def post_comment(self, content, *, parent_id="", commentee_id=""): """ self._assert_auth() data = { - "commentee_id": commentee_id, - "content": str(content), - "parent_id": parent_id, + "commentee_id": commentee_id, + "content": str(content), + "parent_id": parent_id, } r = requests.post( f"https://scratch.mit.edu/site-api/comments/user/{self.username}/add/", @@ -769,27 +849,52 @@ def post_comment(self, content, *, parent_id="", commentee_id=""): text = r.text try: data = { - 'id': text.split('
')[1].split('
')[0].strip(), - 'reply_count': 0, - 'cached_replies': [] + "id": text.split('
')[1] + .split("
")[0] + .strip(), + "reply_count": 0, + "cached_replies": [], } - _comment = comment.Comment(source=comment.CommentSource.USER_PROFILE, parent_id=None if parent_id=="" else parent_id, commentee_id=commentee_id, source_id=self.username, id=data["id"], _session = self._session, datetime = datetime.now()) + _comment = comment.Comment( + source=comment.CommentSource.USER_PROFILE, + parent_id=None if parent_id == "" else parent_id, + commentee_id=commentee_id, + source_id=self.username, + id=data["id"], + _session=self._session, + datetime=datetime.now(), + ) _comment._update_from_dict(data) return _comment except Exception as e: if '{"error": "isFlood"}' in text: - raise(exceptions.CommentPostFailure( - "You are being rate-limited for running this operation too often. Implement a cooldown of about 10 seconds.")) from e + raise ( + exceptions.CommentPostFailure( + "You are being rate-limited for running this operation too often. Implement a cooldown of about 10 seconds." + ) + ) from e elif '')[0] + raw_error_data = text.split( + '")[0] error_data = json.loads(raw_error_data) - expires = error_data['mute_status']['muteExpiresAt'] + expires = error_data["mute_status"]["muteExpiresAt"] expires = datetime.fromtimestamp(expires, timezone.utc) - raise(exceptions.CommentPostFailure(f"You have been muted. Mute expires on {expires}")) from e + raise ( + exceptions.CommentPostFailure( + f"You have been muted. Mute expires on {expires}" + ) + ) from e else: - raise(exceptions.FetchError(f"Couldn't parse API response: {r.text!r}")) from e + raise ( + exceptions.FetchError(f"Couldn't parse API response: {r.text!r}") + ) from e def reply_comment(self, content, *, parent_id, commentee_id=""): """ @@ -807,7 +912,9 @@ def reply_comment(self, content, *, parent_id, commentee_id=""): :param parent_id: ID of the comment you want to reply to :param commentee_id: ID of the user that will be mentioned in your comment and will receive a message about your comment. If you don't want to mention a user, don't put the argument. """ - return self.post_comment(content, parent_id=parent_id, commentee_id=commentee_id) + return self.post_comment( + content, parent_id=parent_id, commentee_id=commentee_id + ) def activity(self, *, limit=1000): """ @@ -815,27 +922,32 @@ def activity(self, *, limit=1000): list: The user's activity data as parsed list of scratchattach.activity.Activity objects """ with requests.no_error_handling(): - soup = BeautifulSoup(requests.get(f"https://scratch.mit.edu/messages/ajax/user-activity/?user={self.username}&max={limit}").text, 'html.parser') + soup = BeautifulSoup( + requests.get( + f"https://scratch.mit.edu/messages/ajax/user-activity/?user={self.username}&max={limit}" + ).text, + "html.parser", + ) activities = [] source = soup.find_all("li") for data in source: - _activity = activity.Activity(_session = self._session, raw=data) + _activity = activity.Activity(_session=self._session, raw=data) _activity._update_from_html(data) activities.append(_activity) return activities - def activity_html(self, *, limit=1000): """ Returns: str: The raw user activity HTML data """ with requests.no_error_handling(): - return requests.get(f"https://scratch.mit.edu/messages/ajax/user-activity/?user={self.username}&max={limit}").text - + return requests.get( + f"https://scratch.mit.edu/messages/ajax/user-activity/?user={self.username}&max={limit}" + ).text def follow(self): """ @@ -844,8 +956,8 @@ def follow(self): self._assert_auth() requests.put( f"https://scratch.mit.edu/site-api/users/followers/{self.username}/add/?usernames={self._session._username}", - headers = headers, - cookies = self._cookies, + headers=headers, + cookies=self._cookies, ) def unfollow(self): @@ -855,8 +967,8 @@ def unfollow(self): self._assert_auth() requests.put( f"https://scratch.mit.edu/site-api/users/followers/{self.username}/remove/?usernames={self._session._username}", - headers = headers, - cookies = self._cookies, + headers=headers, + cookies=self._cookies, ) def delete_comment(self, *, comment_id): @@ -870,9 +982,9 @@ def delete_comment(self, *, comment_id): with requests.no_error_handling(): return requests.post( f"https://scratch.mit.edu/site-api/comments/user/{self.username}/del/", - headers = headers, - cookies = self._cookies, - data = json.dumps({"id":str(comment_id)}) + headers=headers, + cookies=self._cookies, + data=json.dumps({"id": str(comment_id)}), ) def report_comment(self, *, comment_id): @@ -885,9 +997,9 @@ def report_comment(self, *, comment_id): self._assert_auth() return requests.post( f"https://scratch.mit.edu/site-api/comments/user/{self.username}/rep/", - headers = headers, - cookies = self._cookies, - data = json.dumps({"id":str(comment_id)}) + headers=headers, + cookies=self._cookies, + data=json.dumps({"id": str(comment_id)}), ) def comments(self, *, page=1, limit=None) -> list[comment.Comment]: @@ -915,18 +1027,22 @@ def comments(self, *, page=1, limit=None) -> list[comment.Comment]: return [] for entity in _comments: - comment_id = entity.find("div", {"class": "comment"})['data-comment-id'] - user = entity.find("a", {"id": "comment-user"})['data-comment-user'] + comment_id = entity.find("div", {"class": "comment"})["data-comment-id"] + user = entity.find("a", {"id": "comment-user"})["data-comment-user"] content = str(entity.find("div", {"class": "content"}).text).strip() - time = entity.find("span", {"class": "time"})['title'] + time = entity.find("span", {"class": "time"})["title"] main_comment = { - 'id': comment_id, - 'author': {"username":user}, - 'content': content, - 'datetime_created': time, + "id": comment_id, + "author": {"username": user}, + "content": content, + "datetime_created": time, } - _comment = comment.Comment(source=comment.CommentSource.USER_PROFILE, source_id=self.username, _session = self._session) + _comment = comment.Comment( + source=comment.CommentSource.USER_PROFILE, + source_id=self.username, + _session=self._session, + ) _comment._update_from_dict(main_comment) ALL_REPLIES = [] @@ -936,20 +1052,31 @@ def comments(self, *, page=1, limit=None) -> list[comment.Comment]: else: hasReplies = False for reply in replies: - r_comment_id = reply.find("div", {"class": "comment"})['data-comment-id'] - r_user = reply.find("a", {"id": "comment-user"})['data-comment-user'] - r_content = str(reply.find("div", {"class": "content"}).text).strip().replace("\n", "").replace( - " ", " ") - r_time = reply.find("span", {"class": "time"})['title'] + r_comment_id = reply.find("div", {"class": "comment"})[ + "data-comment-id" + ] + r_user = reply.find("a", {"id": "comment-user"})["data-comment-user"] + r_content = ( + str(reply.find("div", {"class": "content"}).text) + .strip() + .replace("\n", "") + .replace(" ", " ") + ) + r_time = reply.find("span", {"class": "time"})["title"] reply_data = { - 'id': r_comment_id, - 'author': {'username': r_user}, - 'content': r_content, - 'datetime_created': r_time, + "id": r_comment_id, + "author": {"username": r_user}, + "content": r_content, + "datetime_created": r_time, "parent_id": comment_id, "cached_parent_comment": _comment, } - _r_comment = comment.Comment(source=comment.CommentSource.USER_PROFILE, source_id=self.username, _session = self._session, cached_parent_comment=_comment) + _r_comment = comment.Comment( + source=comment.CommentSource.USER_PROFILE, + source_id=self.username, + _session=self._session, + cached_parent_comment=_comment, + ) _r_comment._update_from_dict(reply_data) ALL_REPLIES.append(_r_comment) @@ -973,9 +1100,14 @@ def comment_by_id(self, comment_id) -> comment.Comment: page = 1 page_content = self.comments(page=page) while page_content != []: - results = list(filter(lambda x : str(x.id) == str(comment_id), page_content)) + results = list(filter(lambda x: str(x.id) == str(comment_id), page_content)) if results == []: - results = list(filter(lambda x : str(x.id) == str(comment_id), [item for x in page_content for item in x.cached_replies])) + results = list( + filter( + lambda x: str(x.id) == str(comment_id), + [item for x in page_content for item in x.cached_replies], + ) + ) if results != []: return results[0] else: @@ -999,15 +1131,24 @@ def stats(self): dict: A dict containing the user's stats. If the stats aren't available, all values will be -1. """ try: - stats= requests.get( + stats = requests.get( f"https://scratchdb.lefty.one/v3/user/info/{self.username}" ).json()["statistics"] stats.pop("ranks") except Exception: - stats = {"loves":-1,"favorites":-1,"comments":-1,"views":-1,"followers":-1,"following":-1} + stats = { + "loves": -1, + "favorites": -1, + "comments": -1, + "views": -1, + "followers": -1, + "following": -1, + } return stats - @deprecated("Warning: ScratchDB is down indefinitely, therefore this method is deprecated.") + @deprecated( + "Warning: ScratchDB is down indefinitely, therefore this method is deprecated." + ) def ranks(self): """ Gets information about the user's ranks. Fetched from ScratchDB. @@ -1023,7 +1164,22 @@ def ranks(self): f"https://scratchdb.lefty.one/v3/user/info/{self.username}" ).json()["statistics"]["ranks"] except Exception: - return {"country":{"loves":0,"favorites":0,"comments":0,"views":0,"followers":0,"following":0},"loves":0,"favorites":0,"comments":0,"views":0,"followers":0,"following":0} + return { + "country": { + "loves": 0, + "favorites": 0, + "comments": 0, + "views": 0, + "followers": 0, + "following": 0, + }, + "loves": 0, + "favorites": 0, + "comments": 0, + "views": 0, + "followers": 0, + "following": 0, + } def ocular_status(self) -> _OcularStatus: """ @@ -1032,7 +1188,9 @@ def ocular_status(self) -> _OcularStatus: Returns: dict """ - return requests.get(f"https://my-ocular.jeffalo.net/api/user/{self.username}").json() + return requests.get( + f"https://my-ocular.jeffalo.net/api/user/{self.username}" + ).json() def verify_identity(self, *, verification_project_id=395330233): """ @@ -1059,7 +1217,7 @@ def rank(self) -> Rank: if self.is_new_scratcher(): return Rank.NEW_SCRATCHER - + if not self.scratchteam: return Rank.SCRATCHER @@ -1068,6 +1226,7 @@ def rank(self) -> Rank: # ------ # + def get_user(username) -> User: """ Gets a user without logging in. @@ -1088,6 +1247,6 @@ def get_user(username) -> User: "To ignore this warning, use warnings.filterwarnings('ignore', category=scratchattach.UserAuthenticationWarning).\n" "To ignore all warnings of the type GetAuthenticationWarning, which includes this warning, use " "`warnings.filterwarnings('ignore', category=scratchattach.GetAuthenticationWarning)`.", - exceptions.UserAuthenticationWarning + exceptions.UserAuthenticationWarning, ) return commons._get_object("username", username, User, exceptions.UserNotFound) diff --git a/tests/test_memberships.py b/tests/test_memberships.py index ffa0561d..b1dd6471 100644 --- a/tests/test_memberships.py +++ b/tests/test_memberships.py @@ -2,11 +2,13 @@ import warnings warnings.filterwarnings("ignore", category=sa.UserAuthenticationWarning) + + def test_memberships(): # unfortunately we don't have amazingly robust test-cases here. u1 = sa.get_user("-KittyMax-") assert u1.is_member - assert not u1.has_ears + assert u1.has_ears assert u1.has_badge() u2 = sa.get_user("ceebee") @@ -29,5 +31,6 @@ def test_memberships(): assert u5.has_ears assert not u5.has_badge() + if __name__ == "__main__": test_memberships()