140 lines
4.6 KiB
Python
140 lines
4.6 KiB
Python
import argparse
|
|
import asyncio
|
|
import contextlib
|
|
import datetime
|
|
import logging
|
|
import random
|
|
import re
|
|
import subprocess
|
|
import time
|
|
import urllib.parse
|
|
from pathlib import Path
|
|
|
|
from playwright.async_api import async_playwright, Browser, BrowserContext, Page
|
|
|
|
|
|
@contextlib.asynccontextmanager
|
|
async def launch_browser(headless: bool = False) -> Page:
|
|
async with async_playwright() as playwright:
|
|
browser = await playwright.chromium.launch(headless=headless)
|
|
browser: Browser
|
|
async with browser:
|
|
ctx = await browser.new_context()
|
|
ctx: BrowserContext
|
|
async with ctx:
|
|
page = await ctx.new_page()
|
|
async with page:
|
|
yield page
|
|
|
|
|
|
class Error(Exception):
|
|
pass
|
|
|
|
|
|
def alert(text: str):
|
|
subprocess.run(['open', f'alfred://runtrigger/dev.abdus.integrations/remind/?argument={urllib.parse.quote(text)}'])
|
|
|
|
|
|
dump_dir = Path('~/Desktop').expanduser() / 'immigration'
|
|
dump_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
async def check_slots(page: Page, retry_count: int = 1) -> list[datetime.datetime]:
|
|
await page.add_init_script('''Object.defineProperty(navigator, 'webdriver', { get: () => false })''')
|
|
window_id = random.randint(1000, 9999)
|
|
req_id = random.randint(0, 999)
|
|
await page.goto(
|
|
f'https://otv.verwalt-berlin.de/ams/TerminBuchen/wizardng?dswid={window_id}&dsrid={req_id}',
|
|
wait_until='networkidle',
|
|
)
|
|
await page.check('[name="gelesen"]')
|
|
async with page.expect_navigation(url=re.compile('st=2')):
|
|
await page.click('[name="applicationForm:managedForm:proceed"]')
|
|
|
|
await page.get_by_role("combobox", name="Staatsangehörigkeit *").select_option(label="Türkei")
|
|
await asyncio.sleep(0.3)
|
|
|
|
await page.get_by_role(
|
|
"combobox",
|
|
name="Anzahl der Personen, die einen Aufenthaltstitel beantragen (auch ausländische Ehepartner und Kinder) *",
|
|
).select_option(label="eine Person")
|
|
await asyncio.sleep(0.3)
|
|
|
|
await page.get_by_role(
|
|
"combobox", name="Leben Sie in Berlin zusammen mit einem Familienangehörigen (z.B. Ehepartner, Kind) *"
|
|
).select_option(label="nein")
|
|
await asyncio.sleep(0.3)
|
|
|
|
await page.get_by_text("Aufenthaltstitel - beantragen").click()
|
|
await page.locator("label").filter(has_text="Erwerbstätigkeit").click()
|
|
await page.get_by_text("Blaue Karte EU (§ 18b Abs. 2)").click()
|
|
|
|
stop_at = datetime.datetime.now() + datetime.timedelta(minutes=28)
|
|
|
|
async def go_forward():
|
|
async with page.expect_navigation(url=re.compile(r'st='), wait_until='networkidle', timeout=60_000):
|
|
await page.get_by_role("button", name="Weiter").click()
|
|
if 'st=2' in page.url:
|
|
error_message = await page.inner_text('.errorMessage')
|
|
if 'keine Termine frei' in error_message:
|
|
return False
|
|
raise Error(error_message)
|
|
return 'st=3' in page.url
|
|
|
|
while 'st=2' in page.url:
|
|
retry_count -= 1
|
|
|
|
async with page.expect_navigation(url=re.compile(r'st='), wait_until='networkidle', timeout=60_000):
|
|
await page.get_by_role("button", name="Weiter").click()
|
|
|
|
if datetime.datetime.now() > stop_at:
|
|
raise TimeoutError('could not find a slot')
|
|
|
|
if not retry_count:
|
|
return []
|
|
|
|
await asyncio.sleep(random.randint(4, 10))
|
|
continue
|
|
|
|
if 'st=3' in page.url:
|
|
alert('empty slot found')
|
|
now = int(datetime.datetime.now().timestamp())
|
|
html_path = dump_dir / f'immigration_{now}.html'
|
|
html_path.write_text(await page.content())
|
|
await page.screenshot(path=dump_dir/f'immigration_{now}.png')
|
|
|
|
await page.pause()
|
|
|
|
input()
|
|
|
|
|
|
def parse_args():
|
|
arger = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
|
|
arger.add_argument(
|
|
'--headless', '--silent', '--quiet', action='store_true', dest='headless', help='Run in headless mode'
|
|
)
|
|
arger.add_argument(
|
|
'--retry-count', '--retry', type=int, default=1, help='Run in headless mode'
|
|
)
|
|
|
|
args, _ = arger.parse_known_args()
|
|
return args
|
|
|
|
|
|
async def main():
|
|
args = parse_args()
|
|
async with launch_browser(headless=args.headless) as page:
|
|
try:
|
|
slots = await check_slots(page, retry_count=args.retry_count)
|
|
if not slots:
|
|
print('no slots yet')
|
|
return
|
|
except Exception as e:
|
|
logging.exception('got an error')
|
|
await page.pause()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(main())
|
|
logging.basicConfig(level=logging.INFO)
|
|
main()
|