Add OBA.gov.tr watch simulator

oba.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env
+# /// script
+# requires-python = ">=3.11"
+# dependencies = [
+#     "httpx",
+#     "bs4",
+# ]
+# ///
+import argparse
+import httpx
+from bs4 import BeautifulSoup
+import re
+import time
+import json
+import logging
+
+
+def get_next_url(client: httpx.Client):
+    current_url = "https://www.oba.gov.tr/egitim/oynatma/basogretmenlik-uzaktan-egitim-semineri-meb-personeli-2024-1146/19527"
+    soup = BeautifulSoup(client.get(current_url).text, "html.parser")
+
+    if "MEBBİS ile Giriş" in str(soup):
+        raise Exception("login required")
+
+    enabled_links = list(soup.select(".meta-info-unit a:not(.isDisabled)"))
+    all_links = list(soup.select(".meta-info-unit a"))
+    if enabled_links == all_links:
+        # we must have finished the course
+        return None
+    return "https://www.oba.gov.tr" + enabled_links[-1].attrs["href"]
+
+
+def simulate_video_watch(client: httpx.Client, url: str) -> None:
+    logging.info(f"simulating watch on page: {url}")
+
+    # Get the page
+    res = client.get(url)
+    res.raise_for_status()
+    soup = BeautifulSoup(res.text, "html.parser")
+
+    video_title = soup.select_one(".courses-top-bar").text.strip()
+    logging.info(f"video title: {video_title}")
+
+    # Find token and sector length
+    token_match = re.search(r"var token='(eyJ.+?)'", res.text)
+
+    if not token_match:
+        logging.warning("could not find token or sector length")
+        return
+
+    token = token_match.group(1)
+    logging.debug(f"initial token: {token}")
+
+    # Main loop
+    while True:
+        res = client.post(
+            "https://www.oba.gov.tr/content/save",
+            headers={"content-type": "application/x-www-form-urlencoded"},
+            data={"token": token},
+        )
+
+        try:
+            res = res.json()
+
+            # Check if completed
+            if res.get("result", {}).get("completed") is True:
+                logging.info("simulation completed")
+                return
+
+            # Update token if present
+            if "token" in res:
+                token = res["token"]
+                logging.debug(f"checkpoint reached. new token: {token}")
+
+            time.sleep(20)
+
+        except json.JSONDecodeError:
+            logging.error("error decoding json response")
+            raise
+
+
+def main():
+    logging.basicConfig(level=logging.INFO, format=f"%(asctime)s {logging.BASIC_FORMAT}")
+    logging.getLogger("httpx").setLevel(logging.WARNING)
+
+    arger = argparse.ArgumentParser()
+    arger.add_argument("--session", type=str, default="1n8gch3f7v231gvejcnbie480a", help="PHPSESSID cookie value", required=True)
+    args = arger.parse_args()
+
+    client = httpx.Client(
+        cookies={"PHPSESSID": args.session},
+        follow_redirects=True,
+        timeout=10,
+        base_url="https://www.oba.gov.tr",
+        headers={
+            "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
+        },
+    )
+    while True:
+        next_url = get_next_url(client)
+        if not next_url:
+            logging.info("no more pages to process")
+            break
+
+        simulate_video_watch(client=client, url=next_url)
+        time.sleep(2)
+
+
+if __name__ == "__main__":
+    main()