diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..ef47b93 --- /dev/null +++ b/.env.example @@ -0,0 +1,2 @@ +ACCOUNT_EMAIL=email@domain.fr +ACCOUNT_PASSWORD=your_password \ No newline at end of file diff --git a/.gitignore b/.gitignore index 423b7fe..0df679c 100644 --- a/.gitignore +++ b/.gitignore @@ -12,4 +12,5 @@ wheels/ # Application specific data/ -logs/ \ No newline at end of file +logs/ +.env diff --git a/src/app.py b/src/app.py index 2ad2a11..2def5ec 100644 --- a/src/app.py +++ b/src/app.py @@ -1,18 +1,51 @@ -from loguru import logger -from pathlib import Path - -from .lib.utils import EveryRunRotator - -def setup(): - logs_folder = Path(__file__).parents[1].joinpath("logs") - logs_folder.mkdir(exist_ok=True) - logger.debug("log file can be found in: {}", logs_folder.as_posix()) - logger.info("Starting the bot...") - - logger.add(logs_folder.joinpath("app.log"), retention=4, rotation=EveryRunRotator()) - logger.add(logs_folder.joinpath("app.log.json"), serialize=True, rotation=EveryRunRotator(), retention=1, compression="gz") - -async def main(): - setup() - logger.info("Starting the bot...") - logger.trace("ho yeah") +from os import environ + +from loguru import logger +from pathlib import Path + +# from sqlalchemy import create_engine +from .lib.utils import EveryRunRotator +from .services.wiki_masters import WikiMastersClient +from dotenv import load_dotenv + + +class App: + def __init__(self): + self.engine = None + + self.setup() + + def setup(self): + logs_folder = Path(__file__).parents[1].joinpath("logs") + + logs_folder.mkdir(exist_ok=True) + logger.debug("log file can be found in: {}", logs_folder.as_posix()) + + logger.add( + logs_folder.joinpath("app.log"), retention=4, rotation=EveryRunRotator() + ) + logger.add( + logs_folder.joinpath("app.log.json"), + serialize=True, + rotation=EveryRunRotator(), + retention=1, + compression="gz", + ) + + db_folder = Path(__file__).parents[1].joinpath("data") + db_folder.mkdir(exist_ok=True) + + # self.engine = create_engine(f"sqlite+pysqlite:///{db_folder}/app.db", echo=True) + + def start(self): + logger.info("Starting the bot...") + + load_dotenv() + + truc = WikiMastersClient() + truc.login(environ.get("ACCOUNT_EMAIL"), environ.get("ACCOUNT_PASSWORD")) + + +async def main(): + app = App() + app.start() diff --git a/src/lib/supabase_cookie.py b/src/lib/supabase_cookie.py new file mode 100644 index 0000000..35af4d7 --- /dev/null +++ b/src/lib/supabase_cookie.py @@ -0,0 +1,43 @@ +import base64 +from urllib.parse import quote_plus + +APP_ID = "cyrxjeppjqsxxjayfrur" + +BASE64_PREFIX = b"base64-" +MAX_CHUNK_SIZE = 3180 + + +def get_auth_cookies(json_auth_answer: str) -> dict[str, str]: + content = BASE64_PREFIX + base64.urlsafe_b64encode(json_auth_answer).strip(b"=") + return create_chunks(f"sb-{APP_ID}-auth-token", content) + + +def create_chunks( + key: str, value: bytes, chunk_size: int = MAX_CHUNK_SIZE +) -> dict[str, str]: + value_encoded = quote_plus(value) + + # We do not need to split in chunks + if len(value_encoded) <= chunk_size: + return {key: value} + + chunks = {} + i = 0 + while len(value_encoded) > 0: + chunk_content = value_encoded[0 : min(len(value_encoded), chunk_size)] + + try: + last_escape_idx = len(chunk_content) - 1 - chunk_content[::-1].index("%") + + if last_escape_idx > chunk_size - 3: + # incomplete escape + chunk_content = value_encoded[0:last_escape_idx] + except ValueError: + pass + + chunks[f"{key}.{i}"] = chunk_content + i += 1 + + value_encoded = value_encoded[len(chunk_content) :] + + return chunks diff --git a/src/services/wiki_masters.py b/src/services/wiki_masters.py new file mode 100644 index 0000000..53526b4 --- /dev/null +++ b/src/services/wiki_masters.py @@ -0,0 +1,69 @@ +import httpx +from supabase import create_client, Client +from loguru import logger +from ..lib.supabase_cookie import get_auth_cookies +# 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 + + +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" + + def __init__(self) -> None: + self.logger = logger.bind(service=self.__class__.__name__) + + self.logger.info("creating client for supabase") + self.sb_client: Client = create_client(self.SUPABASE_URL, self.SUPABASE_KEY) + self.logger.info("done creating client for supabase: {}", self.sb_client) + + def open_pack(self, cookies): + 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.debug("trying to pull: {}", response.json()) + else: + logger.error( + "error while opening [{}] {}", response.status_code, response.content + ) + + def login(self, email, password): + url = f"{self.SUPABASE_URL}/auth/v1/token?grant_type=password" + response = httpx.post( + url, + json={"email": email, "password": password, "gotrue_meta_security": {}}, + headers={ + "ApiKey": self.SUPABASE_KEY, + "Authorization": f"Bearer {self.SUPABASE_KEY}", + }, + ) + + logger.debug("trying to login {}", response) + if response.status_code == 200: + cookies_compute = get_auth_cookies(response.content) + # TODO: save cookies for the current user + + content = response.json() + session = self.sb_client.auth.set_session( + content["access_token"], content["refresh_token"] + ) + username = session.user.user_metadata["username"] + self.logger.debug("connected with supabase as {}", username) + + response = self.sb_client.postgrest.rpc( + "sync_profile_packs", {"user_id": session.user.id} + ).execute() + self.logger.info("info sync: {}", response)