Login and open functionnal

.env.example

@@ -0,0 +1,2 @@
+ACCOUNT_EMAIL=email@domain.fr
+ACCOUNT_PASSWORD=your_password

.gitignore

@@ -12,4 +12,5 @@ wheels/
 
 # Application specific
 data/
-logs/
+logs/
+.env

src/app.py

@@ -1,18 +1,51 @@
-from loguru import logger
+from os import environ
-from pathlib import Path
+
-
+from loguru import logger
-from .lib.utils import EveryRunRotator
+from pathlib import Path
-
+
-def setup():
+# from sqlalchemy import create_engine
-    logs_folder = Path(__file__).parents[1].joinpath("logs")
+from .lib.utils import EveryRunRotator
-    logs_folder.mkdir(exist_ok=True)
+from .services.wiki_masters import WikiMastersClient
-    logger.debug("log file can be found in: {}", logs_folder.as_posix())
+from dotenv import load_dotenv
-    logger.info("Starting the bot...")
+
-
+
-    logger.add(logs_folder.joinpath("app.log"), retention=4, rotation=EveryRunRotator())
+class App:
-    logger.add(logs_folder.joinpath("app.log.json"), serialize=True, rotation=EveryRunRotator(), retention=1, compression="gz")
+    def __init__(self):
-
+        self.engine = None
-async def main():
+
-    setup()
+        self.setup()
-    logger.info("Starting the bot...")
+
-    logger.trace("ho yeah")
+    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()

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

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)