112 lines
3.2 KiB
Python
Executable File
112 lines
3.2 KiB
Python
Executable File
#!/usr/bin/env -S uv run --script
|
|
# /// 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://oba.gov.tr/egitim/oynatma/yesil-vatan-seferberligi-aile-okul-ve-etkilesim-kooperatifcilik-bilinci-1530/35558"
|
|
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"tk:\s*'(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}")
|
|
|
|
logging.info("waiting before next checkpoint...")
|
|
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="dm5afs9fpibp8ekjrkp55le2bl", help="PHPSESSID cookie value")
|
|
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()
|