1  import aiohttp
 2  from bs4 import BeautifulSoup
 3  from pathlib import Path
 4  from .user import SoundgasmUser
 5  from ..base.decorators import dual_mode
 6  
 7  
 8  class SoundgasmSession:
 9      """
10      Authenticated session for Soundgasm operations such as login and upload.
11      Holds an AsyncHttpClient instance that must be closed by calling close().
12      """
13  
14      def __init__(self, http_client, username: str):
15          """
16          Initialize the session with an HTTP client and a username.
17          """
18          self.http = http_client
19          self.logged_in = False
20          self.username = username
21  
22      async def _login(self, password: str) -> None:
23          """
24          Authenticate the user by submitting credentials and CSRF token.
25          Raises an exception if login fails.
26          """
27          login_page_resp = await self.http.get("https://soundgasm.net/login")
28          html = await login_page_resp.text()
29          soup = BeautifulSoup(html, "html.parser")
30          csrf_tag = soup.find("input", {"name": "authenticity_token"})
31          if not csrf_tag or not csrf_tag.get("value"):
32              raise Exception("Login CSRF token not found")
33          csrf_token = csrf_tag["value"]
34          payload = {
35              "authenticity_token": csrf_token,
36              "user[login]": self.username,
37              "user[password]": password,
38          }
39          headers = {"Referer": "https://soundgasm.net/login"}
40          resp = await self.http.post(url="https://soundgasm.net/session", data=payload, headers=headers)
41          if "/session" in str(resp.url):
42              raise Exception("Login failed")
43          self.logged_in = True
44  
45      @dual_mode
46      async def upload(self, title: str, description: str, file_path: str) -> None:
47          """
48          Upload an audio file to the authenticated account with given title and description.
49          Requires a successful login.
50          """
51          if not self.logged_in:
52              raise Exception("Not logged in")
53          upload_page_resp = await self.http.get("https://soundgasm.net/upload")
54          html = await upload_page_resp.text()
55          soup = BeautifulSoup(html, "html.parser")
56          csrf_tag = soup.find("input", {"name": "authenticity_token"})
57          if not csrf_tag or not csrf_tag.get("value"):
58              raise Exception("Upload CSRF token not found")
59          csrf_token = csrf_tag["value"]
60          with open(file_path, "rb") as f:
61              data = aiohttp.FormData()
62              data.add_field("authenticity_token", csrf_token)
63              data.add_field("audio[title]", title)
64              data.add_field("audio[description]", description)
65              data.add_field("audio[audio_file]", f, filename=Path(file_path).name)
66              headers = {"Referer": "https://soundgasm.net/upload", "X-Requested-With": "XMLHttpRequest"}
67              resp = await self.http.post(url="https://soundgasm.net/audio", data=data, headers=headers)
68              if resp.status != 200:
69                  raise Exception(f"Upload failed with status {resp.status}")
70  
71      def get_user(self, username: str) -> SoundgasmUser:
72          """
73          Return a user handle for read-only operations such as fetching tracks.
74          """
75          return SoundgasmUser(username)
76  
77      @dual_mode
78      async def close(self) -> None:
79          """
80          Close the underlying HTTP client.
81          """
82          await self.http.close()