Add community session file helpers and group membership state

Author: ja7den2006

examples/smoke_test.py

@@ -2,6 +2,7 @@
 
 import argparse
 import sys
+import tempfile
 import time
 from pathlib import Path
 
@@ -860,6 +861,10 @@ def run_community_suite(client: SteamClient, args) -> None:
         "Get Group Member Summaries Map",
         lambda: _format_group_member_summaries_map(client.get_group_member_summaries_map(args.group_url, limit=5)),
     )
+    run_check(
+        "Get Group Membership State",
+        lambda: _format_group_membership_state(client.get_group_membership_state(args.group_url)),
+    )
     if client.api_key:
         run_check(
             "Get Friend Bans",
@@ -875,6 +880,10 @@ def run_community_suite(client: SteamClient, args) -> None:
         "Community Cookie Mapping Roundtrip",
         lambda: _format_cookie_mapping_roundtrip(client, args),
     )
+    run_check(
+        "Community Cookie File Roundtrip",
+        lambda: _format_cookie_file_roundtrip(client, args),
+    )
     run_check(
         "Community Session Bundle Roundtrip",
         lambda: _format_bundle_roundtrip(client, args),
@@ -883,6 +892,10 @@ def run_community_suite(client: SteamClient, args) -> None:
         "Community Session Bundle JSON Roundtrip",
         lambda: _format_bundle_json_roundtrip(client, args),
     )
+    run_check(
+        "Community Session Bundle File Roundtrip",
+        lambda: _format_bundle_file_roundtrip(client, args),
+    )
     run_check(
         "Export Community Refresh Token",
         lambda: _format_refresh_token_export(client),
@@ -1249,6 +1262,15 @@ def _format_group_member_summaries_map(payload: dict) -> str:
     )
 
 
+def _format_group_membership_state(payload: dict) -> str:
+    return "state={0} can_join={1} can_leave={2} group_id={3}".format(
+        payload.get("membership_state"),
+        payload.get("can_join"),
+        payload.get("can_leave"),
+        payload.get("group_id", "<none>"),
+    )
+
+
 def _format_group_details(payload: dict) -> str:
     return "group={0} members={1} online={2}".format(
         _safe_console_text(payload.get("group_name") or "<none>"),
@@ -1636,6 +1658,23 @@ def _format_cookie_mapping_roundtrip(client: SteamClient, args) -> str:
         roundtrip_client.close()
 
 
+def _format_cookie_file_roundtrip(client: SteamClient, args) -> str:
+    with tempfile.TemporaryDirectory(prefix="sckit-cookie-") as temp_dir:
+        cookie_path = Path(temp_dir) / "community_cookie.txt"
+        client.save_community_cookie_string(cookie_path)
+        roundtrip_client = build_client(args)
+        try:
+            roundtrip_client.load_community_cookie_string(cookie_path)
+            state = roundtrip_client.get_community_session_state()
+            return "steamid={0} logged_in={1} path={2}".format(
+                state.get("steam_id", "<unknown>"),
+                state.get("logged_in", False),
+                cookie_path.name,
+            )
+        finally:
+            roundtrip_client.close()
+
+
 def _format_bundle_roundtrip(client: SteamClient, args) -> str:
     bundle = client.export_community_session_bundle()
     roundtrip_client = build_client(args)
@@ -1651,6 +1690,23 @@ def _format_bundle_roundtrip(client: SteamClient, args) -> str:
         roundtrip_client.close()
 
 
+def _format_bundle_file_roundtrip(client: SteamClient, args) -> str:
+    with tempfile.TemporaryDirectory(prefix="sckit-bundle-") as temp_dir:
+        bundle_path = Path(temp_dir) / "community_bundle.json"
+        client.save_community_session_bundle(bundle_path)
+        roundtrip_client = build_client(args)
+        try:
+            roundtrip_client.load_community_session_bundle(bundle_path)
+            state = roundtrip_client.get_community_session_state()
+            return "steamid={0} logged_in={1} path={2}".format(
+                state.get("steam_id", "<unknown>"),
+                state.get("logged_in", False),
+                bundle_path.name,
+            )
+        finally:
+            roundtrip_client.close()
+
+
 def _format_bundle_json_roundtrip(client: SteamClient, args) -> str:
     bundle_json = client.export_community_session_bundle_json()
     roundtrip_client = build_client(args)

src/steamcommunitykit/client.py

@@ -238,6 +238,9 @@ def get_group_id64(self, group_url: str) -> str:
     def get_group_id(self, group_url: str) -> str:
         return self.groups.fetch_group_id(group_url)
 
+    def get_group_membership_state(self, group_url: str) -> dict:
+        return self.groups.get_group_membership_state(group_url)
+
     def check_group_name_availability(self, name: str):
         return self.groups.check_name_availability(name)
 
@@ -1782,6 +1785,10 @@ def set_community_credentials_from_cookie_mapping(self, cookies: dict) -> Commun
         self.set_community_credentials(credentials)
         return credentials
 
+    def load_community_cookie_string(self, path: Union[str, Path]) -> CommunityCredentials:
+        cookie_string = Path(path).read_text(encoding="utf-8")
+        return self.set_community_credentials_from_cookie_string(cookie_string)
+
     def set_community_credentials_from_bundle(self, bundle: dict) -> CommunityCredentials:
         credentials = self.auth.community_credentials_from_bundle(bundle)
         self.set_community_credentials(credentials)
@@ -1792,6 +1799,10 @@ def set_community_credentials_from_bundle_json(self, bundle_json: str) -> Commun
         self.set_community_credentials(credentials)
         return credentials
 
+    def load_community_session_bundle(self, path: Union[str, Path]) -> CommunityCredentials:
+        bundle_json = Path(path).read_text(encoding="utf-8")
+        return self.set_community_credentials_from_bundle_json(bundle_json)
+
     def login_to_community_with_refresh_token(self, refresh_token: str) -> CommunityCredentials:
         credentials = self.auth.community_credentials_from_refresh_token(refresh_token)
         self.set_community_credentials(credentials)
@@ -1803,6 +1814,12 @@ def export_community_cookie_string(self) -> str:
     def export_community_cookie_mapping(self) -> dict:
         return self.auth.export_cookie_mapping(self._transport.require_community_credentials())
 
+    def save_community_cookie_string(self, path: Union[str, Path]) -> Path:
+        resolved_path = Path(path)
+        resolved_path.parent.mkdir(parents=True, exist_ok=True)
+        resolved_path.write_text(self.export_community_cookie_string(), encoding="utf-8")
+        return resolved_path
+
     def export_community_session_bundle(self) -> dict:
         return self.auth.export_credentials_bundle(self._transport.require_community_credentials())
 
@@ -1812,6 +1829,15 @@ def export_community_session_bundle_json(self, *, indent: Optional[int] = None)
             indent=indent,
         )
 
+    def save_community_session_bundle(self, path: Union[str, Path], *, indent: Optional[int] = 2) -> Path:
+        resolved_path = Path(path)
+        resolved_path.parent.mkdir(parents=True, exist_ok=True)
+        resolved_path.write_text(
+            self.export_community_session_bundle_json(indent=indent),
+            encoding="utf-8",
+        )
+        return resolved_path
+
     def export_community_refresh_token(self) -> str:
         credentials = self._transport.require_community_credentials()
         if not credentials.refresh_token:

src/steamcommunitykit/services/groups.py

@@ -143,6 +143,62 @@ def fetch_group_id(self, group_url: str) -> str:
         group_id = tail.split("<div class=\"formRowFields\">", 1)[1].split("</div>", 1)[0]
         return group_id.strip()
 
+    def get_group_membership_state(self, group_url: str) -> dict:
+        normalized_group_url = self._normalize_group_url_value(group_url)
+        response_text = self.transport.request(
+            "GET",
+            f"{COMMUNITY_BASE_URL}/groups/{normalized_group_url}",
+            cookies=self._community_cookies(),
+            expected="text",
+        )
+        state = self._parse_group_membership_state(response_text)
+        state["group_url"] = normalized_group_url
+        state["group_link"] = f"{COMMUNITY_BASE_URL}/groups/{normalized_group_url}/"
+        return state
+
+    @staticmethod
+    def _parse_group_membership_state(html_text: str) -> dict:
+        join_form_match = re.search(
+            r'<form name="join_group_form" id="join_group_form" method="POST" action="([^"]+)">.*?'
+            r'name="action" value="join".*?name="sessionID" value="([^"]+)"',
+            html_text,
+            re.S | re.I,
+        )
+        leave_form_match = re.search(
+            r'<form method="POST" id="leave_group_form" action="([^"]+)">.*?'
+            r'name="sessionID" value="([^"]+)".*?name="action" value="leaveGroup".*?'
+            r'name="groupId" value="([^"]+)"',
+            html_text,
+            re.S | re.I,
+        )
+        can_join = bool(re.search(r"<span>\s*Join Group\s*</span>", html_text, re.I))
+        can_leave = leave_form_match is not None
+        if can_leave and not can_join:
+            membership_state = "member"
+        elif can_join:
+            membership_state = "not_member"
+        else:
+            membership_state = "unknown"
+
+        payload = {
+            "membership_state": membership_state,
+            "is_member": membership_state == "member",
+            "can_join": can_join,
+            "can_leave": can_leave,
+            "join_action_url": None,
+            "leave_action_url": None,
+            "session_id": None,
+            "group_id": None,
+        }
+        if join_form_match:
+            payload["join_action_url"] = html.unescape(join_form_match.group(1))
+            payload["session_id"] = join_form_match.group(2)
+        if leave_form_match:
+            payload["leave_action_url"] = html.unescape(leave_form_match.group(1))
+            payload["session_id"] = leave_form_match.group(2)
+            payload["group_id"] = leave_form_match.group(3)
+        return payload
+
     def get_group_details(self, group_url: str, page: int = 1) -> dict:
         normalized_group_url = self._normalize_group_url_value(group_url)
         response_text = self.transport.request(

tests/test_client.py

@@ -1766,6 +1766,27 @@ def test_client_can_export_community_cookie_mapping() -> None:
     client.close()
 
 
+def test_client_can_save_and_load_community_cookie_file(tmp_path) -> None:
+    client = SteamClient(api_key="test")
+    client.set_community_credentials(
+        CommunityCredentials(
+            steam_id="76561197960435530",
+            session_id="session123",
+            steam_login_secure="76561197960435530%7C%7Ctoken123",
+        )
+    )
+
+    path = client.save_community_cookie_string(tmp_path / "session" / "community_cookie.txt")
+
+    restored_client = SteamClient(api_key="test")
+    restored = restored_client.load_community_cookie_string(path)
+
+    assert path.read_text(encoding="utf-8") == "sessionid=session123; steamLoginSecure=76561197960435530%7C%7Ctoken123"
+    assert restored.steam_id == "76561197960435530"
+    client.close()
+    restored_client.close()
+
+
 def test_client_can_export_and_restore_community_session_bundle() -> None:
     client = SteamClient(api_key="test")
     client.set_community_credentials(
@@ -1791,6 +1812,29 @@ def test_client_can_export_and_restore_community_session_bundle() -> None:
     restored_client.close()
 
 
+def test_client_can_save_and_load_community_session_bundle_file(tmp_path) -> None:
+    client = SteamClient(api_key="test")
+    client.set_community_credentials(
+        CommunityCredentials(
+            steam_id="76561197960435530",
+            session_id="session123",
+            access_token="access123",
+            refresh_token="refresh123",
+            steam_login_secure="76561197960435530%7C%7Ctoken123",
+        )
+    )
+
+    path = client.save_community_session_bundle(tmp_path / "session" / "community_bundle.json")
+
+    restored_client = SteamClient(api_key="test")
+    restored = restored_client.load_community_session_bundle(path)
+
+    assert '"steam_id": "76561197960435530"' in path.read_text(encoding="utf-8")
+    assert restored.refresh_token == "refresh123"
+    client.close()
+    restored_client.close()
+
+
 def test_client_can_export_and_restore_community_session_bundle_json() -> None:
     client = SteamClient(api_key="test")
     client.set_community_credentials(
@@ -1977,6 +2021,61 @@ def test_players_get_recently_played_games_summary_normalizes_games() -> None:
     client.close()
 
 
+def test_groups_membership_state_parses_join_form() -> None:
+    html = """
+    <form name="join_group_form" id="join_group_form" method="POST" action="https://steamcommunity.com/groups/SteamDB">
+        <input type="hidden" name="action" value="join">
+        <input type="hidden" name="sessionID" value="session123">
+    </form>
+    <div class="grouppage_join_area">
+        <span>Join Group</span>
+    </div>
+    """
+    client = SteamClient(session=RecordingSession(DummyResponse(text=html)))
+    client.set_community_credentials(
+        CommunityCredentials(
+            steam_id="76561197960435530",
+            session_id="session123",
+            steam_login_secure="76561197960435530%7C%7Ctoken123",
+        )
+    )
+
+    result = client.get_group_membership_state("SteamDB")
+
+    assert result["membership_state"] == "not_member"
+    assert result["can_join"] is True
+    assert result["can_leave"] is False
+    assert result["join_action_url"] == "https://steamcommunity.com/groups/SteamDB"
+    client.close()
+
+
+def test_groups_membership_state_parses_leave_form() -> None:
+    html = """
+    <form method="POST" id="leave_group_form" action="https://steamcommunity.com/profiles/76561197960435530/home_process">
+        <input type="hidden" name="sessionID" value="session123">
+        <input type="hidden" name="action" value="leaveGroup">
+        <input type="hidden" name="groupId" value="103582791429521412">
+    </form>
+    """
+    client = SteamClient(session=RecordingSession(DummyResponse(text=html)))
+    client.set_community_credentials(
+        CommunityCredentials(
+            steam_id="76561197960435530",
+            session_id="session123",
+            steam_login_secure="76561197960435530%7C%7Ctoken123",
+        )
+    )
+
+    result = client.get_group_membership_state("Valve")
+
+    assert result["membership_state"] == "member"
+    assert result["is_member"] is True
+    assert result["can_join"] is False
+    assert result["can_leave"] is True
+    assert result["group_id"] == "103582791429521412"
+    client.close()
+
+
 def test_players_find_owned_game_filters_by_app_id() -> None:
     session = RecordingSession(
         DummyResponse(