102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
import httpx
|
|
from dataclasses import dataclass
|
|
from supabase import create_client, Client
|
|
from supabase_auth import Session
|
|
from loguru import logger
|
|
from datetime import datetime, timedelta
|
|
from ..lib.supabase_cookie import get_auth_cookies
|
|
from typing import Self
|
|
# supabase ?
|
|
# https://cyrxjeppjqsxxjayfrur.supabase.co/auth/v1/token?grant_type=password
|
|
# https://cyrxjeppjqsxxjayfrur.supabase.co/auth/v1/user
|
|
|
|
# https://cyrxjeppjqsxxjayfrur.supabase.co/rest/v1/rpc/sync_profile_packs
|
|
|
|
|
|
@dataclass
|
|
class SyncDataSubset:
|
|
_id: str
|
|
username: str
|
|
packs_remaining: int
|
|
packs_last_regen_at: datetime
|
|
|
|
@staticmethod
|
|
def from_json(d: dict) -> Self:
|
|
return SyncDataSubset(
|
|
d["id"],
|
|
d["username"],
|
|
d["packs_remaining"],
|
|
datetime.fromisoformat(d["packs_last_regen_at"]),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class UserSession:
|
|
username: str
|
|
sb_client: Client
|
|
|
|
@property
|
|
def auth_cookies(self):
|
|
return get_auth_cookies(self.session.model_dump_json().encode("utf-8"))
|
|
|
|
@property
|
|
def session(self):
|
|
return self.sb_client.auth.get_session()
|
|
|
|
def sync(self) -> SyncDataSubset:
|
|
response = self.sb_client.postgrest.rpc(
|
|
"sync_profile_packs", {"user_id": self.session.user.id}
|
|
).execute()
|
|
|
|
return SyncDataSubset.from_json(response.data[0])
|
|
|
|
|
|
class WikiMastersClient:
|
|
BASE_URL = "https://www.wiki-masters.com"
|
|
|
|
# Extracted from website from createClient in js / also available in headers of requests to supabase
|
|
SUPABASE_URL = "https://cyrxjeppjqsxxjayfrur.supabase.co"
|
|
SUPABASE_KEY = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6ImN5cnhqZXBwanFzeHhqYXlmcnVyIiwicm9sZSI6ImFub24iLCJpYXQiOjE3NzM4ODAzMzksImV4cCI6MjA4OTQ1NjMzOX0.BZluyXygNxuQGDPxFX1zG5i-cqp10CVK-8GGtuak4Rg"
|
|
|
|
ONE_PACK_TIME = timedelta(minutes=10)
|
|
|
|
def __init__(self) -> None:
|
|
self.logger = logger.bind(service=self.__class__.__name__)
|
|
self.users: dict[str, UserSession] = {}
|
|
|
|
def open_pack(self, cookies) -> tuple[int, str]:
|
|
url = f"{self.BASE_URL}/api/packs/open"
|
|
|
|
logger.debug("trying to request a pull (cookies:{})", cookies)
|
|
response = httpx.post(
|
|
url,
|
|
headers={"referer": f"{self.BASE_URL}/pulls"},
|
|
cookies=cookies,
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
logger.error(
|
|
"error while opening [{}] {}", response.status_code, response.content
|
|
)
|
|
raise Exception("Could not pull :'(")
|
|
|
|
pull_data = response.json()
|
|
logger.debug("trying to pull: {}", pull_data)
|
|
|
|
return (pull_data["packs_remaining"], datetime.fromisoformat(pull_data["packs_last_regen_at"]))
|
|
|
|
def login(self, email, password) -> UserSession:
|
|
sb_client = create_client(self.SUPABASE_URL, self.SUPABASE_KEY)
|
|
response = sb_client.auth.sign_in_with_password(
|
|
{"email": email, "password": password}
|
|
)
|
|
|
|
if not response.session:
|
|
raise Exception(f"Login failed for {email}")
|
|
|
|
session = response.session
|
|
username = session.user.user_metadata["username"]
|
|
logger.debug("logged user {}", username)
|
|
|
|
return UserSession(username, sb_client)
|