290 lines
10 KiB
Python
290 lines
10 KiB
Python
"""HTTP client for the Visametric (Belgrade) appointment site (plan §1).
|
|
|
|
One :class:`VisametricClient` == one poll cycle == one fresh
|
|
``requests.Session``. The poller never reuses a client across cycles.
|
|
|
|
Request sequence (all verified live, plan §1):
|
|
|
|
1. ``GET /en`` → CSRF meta + captcha data URL
|
|
2. (captcha solved offsite)
|
|
3. ``POST /en/appointment-form`` → form page; success iff the body contains
|
|
the ``getavailablefirstdate`` marker (a wrong captcha answer returns the
|
|
landing page instead — do NOT detect failure via the word "incorrect")
|
|
4. ``POST /en/getcalendarstatus`` → plain text calendarType (fetched every
|
|
cycle, never hardcoded)
|
|
5. ``POST /en/getavailablefirstdate`` → JSON; ``isAvailable`` is the
|
|
authoritative signal
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import binascii
|
|
import logging
|
|
import re
|
|
from dataclasses import dataclass
|
|
|
|
import requests
|
|
|
|
from .config import BASE_URL, HTTP_TIMEOUT, LANG, PROFILE, USER_AGENT
|
|
|
|
log = logging.getLogger("visaapply.client")
|
|
|
|
# <meta name="csrf-token" content="..."> — present on every page; the token
|
|
# on the form page differs from the one on the landing page.
|
|
_CSRF_RE = re.compile(r'<meta\s+name="csrf-token"\s+content="([^"]+)"')
|
|
|
|
# The captcha image is embedded in inline JS:
|
|
# ...setAttribute("src", "data:image/png;base64," + "<BASE64>");
|
|
# The <img class="imageCaptcha"> tag itself has an EMPTY src — the image
|
|
# only exists in that JS string literal. The JS escapes '/' as '\/', so the
|
|
# literal must be unescaped before it is valid base64.
|
|
_CAPTCHA_ANCHOR = "data:image/png;base64"
|
|
|
|
# DD-MM-YYYY labels inside firstAvailableDate.
|
|
_DATE_RE = re.compile(r"(\d{2}-\d{2}-\d{4})")
|
|
|
|
# Marker that the appointment-form POST succeeded (plan §1).
|
|
_FORM_MARKER = "getavailablefirstdate"
|
|
|
|
|
|
class PollError(RuntimeError):
|
|
"""Base class for all poll-cycle errors (mapped to the error model)."""
|
|
|
|
|
|
class SiteBlockedError(PollError):
|
|
"""The site refused the request (403, Cloudflare challenge, ...)."""
|
|
|
|
|
|
class CaptchaRejectedError(PollError):
|
|
"""The appointment-form POST returned the landing page, not the form."""
|
|
|
|
|
|
class ParseError(PollError):
|
|
"""The page shape was not what we expected."""
|
|
|
|
|
|
class NetworkError(PollError):
|
|
"""Connection/timeout failure talking to the site."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LandingPage:
|
|
"""Step-1 result: CSRF token and the captcha image (valid base64)."""
|
|
|
|
csrf_token: str
|
|
captcha_b64: str # valid base64 of the captcha PNG
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FirstDateResult:
|
|
"""Step-5 result: authoritative availability flag and parsed dates."""
|
|
|
|
is_available: bool
|
|
dates: list[str]
|
|
|
|
|
|
class VisametricClient:
|
|
"""One instance == one poll cycle == one fresh ``requests.Session``."""
|
|
|
|
def __init__(self, proxies: dict[str, str] | None = None) -> None:
|
|
"""Create the session for one poll cycle.
|
|
|
|
Args:
|
|
proxies: optional requests proxies dict — wired now, unused by
|
|
default (plan §9: a future CF-ban countermeasure).
|
|
"""
|
|
self._session = requests.Session()
|
|
self._session.headers["User-Agent"] = USER_AGENT
|
|
self._proxies = proxies
|
|
self._landing_csrf: str | None = None
|
|
self._form_csrf: str | None = None
|
|
|
|
# -- Step 1 -------------------------------------------------------------
|
|
|
|
def fetch_landing(self) -> LandingPage:
|
|
"""Fetch the landing page; extract the CSRF token + captcha base64."""
|
|
url = f"{BASE_URL}/{LANG}"
|
|
resp = self._request("GET", url)
|
|
if resp.status_code != 200:
|
|
raise SiteBlockedError(f"GET {url} -> HTTP {resp.status_code}")
|
|
csrf = _extract_csrf(resp.text, "landing page")
|
|
b64 = _extract_captcha_b64(resp.text)
|
|
self._landing_csrf = csrf
|
|
return LandingPage(csrf_token=csrf, captcha_b64=b64)
|
|
|
|
# -- Step 3 -------------------------------------------------------------
|
|
|
|
def submit_captcha(self, code: str) -> str:
|
|
"""Post the solved captcha; returns the form page's (new) CSRF token."""
|
|
if self._landing_csrf is None:
|
|
raise ParseError("submit_captcha called before fetch_landing")
|
|
url = f"{BASE_URL}/{LANG}/appointment-form"
|
|
resp = self._request(
|
|
"POST",
|
|
url,
|
|
data={
|
|
"_token": self._landing_csrf,
|
|
"cpJvnsControl": "",
|
|
"mailConfirmCode": code,
|
|
},
|
|
headers={
|
|
"X-CSRF-TOKEN": self._landing_csrf,
|
|
"Referer": f"{BASE_URL}/{LANG}",
|
|
"Origin": BASE_URL,
|
|
},
|
|
)
|
|
if resp.status_code != 200 or _FORM_MARKER not in resp.text:
|
|
raise CaptchaRejectedError(
|
|
f"captcha rejected (HTTP {resp.status_code}, "
|
|
f"form marker {_FORM_MARKER!r} missing)",
|
|
)
|
|
csrf = _extract_csrf(resp.text, "appointment form page")
|
|
self._form_csrf = csrf
|
|
return csrf
|
|
|
|
# -- Step 4 -------------------------------------------------------------
|
|
|
|
def get_calendar_type(self) -> str:
|
|
"""Fetch calendarType from getcalendarstatus (never hardcoded)."""
|
|
if self._form_csrf is None:
|
|
raise ParseError("get_calendar_type called before submit_captcha")
|
|
url = f"{BASE_URL}/{LANG}/getcalendarstatus"
|
|
resp = self._request(
|
|
"POST",
|
|
url,
|
|
data={
|
|
"getvisaofficeid": str(PROFILE["office"]),
|
|
"getservicetypeid": str(PROFILE["service_type"]),
|
|
"getvisacountryid": str(PROFILE["consular"]),
|
|
},
|
|
# Laravel 419 (CSRF mismatch) without the token — verified live.
|
|
headers={
|
|
"X-CSRF-TOKEN": self._form_csrf,
|
|
"Referer": f"{BASE_URL}/{LANG}/appointment-form",
|
|
},
|
|
)
|
|
if resp.status_code != 200:
|
|
raise SiteBlockedError(f"getcalendarstatus -> HTTP {resp.status_code}")
|
|
value = resp.text.strip()
|
|
if not value:
|
|
raise ParseError("getcalendarstatus returned an empty body")
|
|
return value
|
|
|
|
# -- Step 5 -------------------------------------------------------------
|
|
|
|
def get_available_first_date(self, calendar_type: str) -> FirstDateResult:
|
|
"""Ask for first available dates; isAvailable is authoritative."""
|
|
if self._form_csrf is None:
|
|
raise ParseError("get_available_first_date called before submit_captcha")
|
|
url = f"{BASE_URL}/{LANG}/getavailablefirstdate"
|
|
resp = self._request(
|
|
"POST",
|
|
url,
|
|
data={
|
|
"serviceType": str(PROFILE["service_type"]),
|
|
"totalPerson": str(PROFILE["total_person"]),
|
|
"getOfficeID": str(PROFILE["office"]),
|
|
"calendarType": calendar_type,
|
|
"getConsular": str(PROFILE["consular"]),
|
|
},
|
|
headers={
|
|
"X-CSRF-TOKEN": self._form_csrf,
|
|
"X-Requested-With": "XMLHttpRequest",
|
|
"Referer": f"{BASE_URL}/{LANG}/appointment-form",
|
|
"Origin": BASE_URL,
|
|
},
|
|
)
|
|
if resp.status_code == 403:
|
|
raise SiteBlockedError(
|
|
f"getavailablefirstdate -> 403 {resp.text[:200]!r}",
|
|
)
|
|
if resp.status_code != 200:
|
|
raise SiteBlockedError(f"getavailablefirstdate -> HTTP {resp.status_code}")
|
|
try:
|
|
payload = resp.json()
|
|
except ValueError as e:
|
|
raise ParseError(
|
|
f"getavailablefirstdate returned non-JSON: {resp.text[:200]!r}",
|
|
) from e
|
|
is_available = payload.get("isAvailable")
|
|
if not isinstance(is_available, bool):
|
|
raise ParseError(f"missing boolean isAvailable in {payload!r}")
|
|
dates: list[str] = []
|
|
if is_available:
|
|
html = payload.get("firstAvailableDate") or ""
|
|
dates = _dedupe(_DATE_RE.findall(html))
|
|
if not dates:
|
|
raise ParseError("isAvailable=true but no dates parsed")
|
|
return FirstDateResult(is_available=is_available, dates=dates)
|
|
|
|
# -- plumbing -----------------------------------------------------------
|
|
|
|
def _request(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
data: dict[str, str] | None = None,
|
|
headers: dict[str, str] | None = None,
|
|
) -> requests.Response:
|
|
try:
|
|
return self._session.request(
|
|
method,
|
|
url,
|
|
data=data,
|
|
headers=headers,
|
|
timeout=HTTP_TIMEOUT,
|
|
allow_redirects=True,
|
|
proxies=self._proxies,
|
|
)
|
|
except requests.RequestException as e:
|
|
raise NetworkError(f"{method} {url} failed: {e}") from e
|
|
|
|
|
|
def _extract_csrf(html: str, context: str) -> str:
|
|
match = _CSRF_RE.search(html)
|
|
if not match:
|
|
raise ParseError(f"csrf-token meta not found on {context}")
|
|
return match.group(1)
|
|
|
|
|
|
def _extract_captcha_b64(html: str) -> str:
|
|
r"""Extract the captcha image base64 from the inline JS.
|
|
|
|
The JS source escapes '/' as '\/', so backslashes are stripped to
|
|
recover valid base64 (the base64 alphabet never contains backslashes).
|
|
The result is validated with a strict decode so malformed extraction
|
|
fails here instead of as a captcha-API error.
|
|
"""
|
|
idx = html.find(_CAPTCHA_ANCHOR)
|
|
if idx == -1:
|
|
raise ParseError("captcha base64 anchor not found on landing page")
|
|
plus_idx = html.find("+", idx)
|
|
if plus_idx == -1:
|
|
raise ParseError("captcha base64 not found after anchor")
|
|
open_quote = html.find('"', plus_idx)
|
|
if open_quote == -1:
|
|
raise ParseError("captcha base64 opening quote not found")
|
|
close_quote = html.find('"', open_quote + 1)
|
|
if close_quote == -1:
|
|
raise ParseError("captcha base64 closing quote not found")
|
|
b64 = html[open_quote + 1 : close_quote].replace("\\", "")
|
|
if not b64:
|
|
raise ParseError("empty captcha base64")
|
|
try:
|
|
base64.b64decode(b64, validate=True)
|
|
except (binascii.Error, ValueError) as e:
|
|
raise ParseError(f"malformed captcha base64: {e}") from e
|
|
return b64
|
|
|
|
|
|
def _dedupe(items: list[str]) -> list[str]:
|
|
"""Remove duplicates preserving first-seen order."""
|
|
seen: set[str] = set()
|
|
out: list[str] = []
|
|
for item in items:
|
|
if item not in seen:
|
|
seen.add(item)
|
|
out.append(item)
|
|
return out
|