157 lines
5.1 KiB
Python
157 lines
5.1 KiB
Python
import argparse
|
|
import asyncio
|
|
import contextlib
|
|
import dataclasses
|
|
import datetime
|
|
import logging
|
|
import random
|
|
import re
|
|
import subprocess
|
|
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)
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class ResidencyApplication:
|
|
citizenship_country: str
|
|
num_applicants: int
|
|
is_living_with_family: bool
|
|
residency_category: str
|
|
residency_purpose: str
|
|
|
|
|
|
async def check_slots(page: Page, values: ResidencyApplication, 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',
|
|
)
|
|
|
|
async with page.expect_navigation():
|
|
await page.click('.langBlock [name="txtEn"] a')
|
|
|
|
await page.check('[name="gelesen"]')
|
|
|
|
async with page.expect_navigation(url=re.compile('st=2'), wait_until='networkidle'):
|
|
await page.click('[name="applicationForm:managedForm:proceed"]')
|
|
|
|
await page.get_by_role("combobox", name="Citizenship *").select_option(label=values.citizenship_country)
|
|
await asyncio.sleep(0.3)
|
|
|
|
# how many applicants
|
|
await page.select_option('[name="personenAnzahl_normal"]', value=str(values.num_applicants))
|
|
await asyncio.sleep(0.3)
|
|
|
|
# living with family
|
|
await page.select_option('[name="lebnBrMitFmly"]', value='1' if values.is_living_with_family else '2')
|
|
await asyncio.sleep(0.3)
|
|
|
|
await page.get_by_text("Apply for a residence title").click()
|
|
await page.locator("label").filter(has_text=values.residency_category).click()
|
|
await page.get_by_text(values.residency_purpose).click()
|
|
|
|
stop_at = datetime.datetime.now() + datetime.timedelta(minutes=28)
|
|
|
|
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="Next").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:
|
|
dates = []
|
|
for el in await page.query_selector_all('.ui-datepicker-calendar td:not(.ui-datepicker-unselectable)'):
|
|
dates.append(
|
|
datetime.date(
|
|
year=int(await el.get_attribute('data-year')),
|
|
month=int(await el.get_attribute('data-month')),
|
|
day=int(await el.inner_text()),
|
|
)
|
|
)
|
|
|
|
alert("\n".join(['empty slots found', *[it.isoformat() for it in dates]]))
|
|
|
|
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')
|
|
|
|
input()
|
|
|
|
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:
|
|
values = ResidencyApplication(
|
|
citizenship_country='India',
|
|
num_applicants=1,
|
|
is_living_with_family=False,
|
|
residency_category='Educational purposes',
|
|
residency_purpose='Residence permit for the purpose of studying (sect. 16b)',
|
|
)
|
|
slots = await check_slots(page, values=values, retry_count=args.retry_count)
|
|
if not slots:
|
|
print('no slots yet')
|
|
return
|
|
except Exception as e:
|
|
logging.exception('got an error')
|
|
input()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(main())
|
|
logging.basicConfig(level=logging.INFO)
|