c630b500ef
* DB: users + user_profiles/filters/notifications/preferences; applications gets user_id + forensics_json + profile_snapshot_json; new errors table with 14d retention; schema versioning via MIGRATIONS list * auth: password hashes in DB (argon2); env vars seed first admin; per-user sessions; CSRF bound to user id * apply: personal info/WBS moved out of env into the request body; providers take an ApplyContext with Profile + submit_forms; full Playwright recorder (step log, console, page errors, network, screenshots, final HTML) * web: five top-level tabs (Wohnungen/Bewerbungen/Logs/Fehler/Einstellungen); settings sub-tabs profil/filter/benachrichtigungen/account/benutzer; per-user matching, auto-apply and notifications (UI/Telegram/SMTP); red auto-apply switch on Wohnungen tab; forensics detail view for bewerbungen and fehler; retention background thread Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
import logging
|
|
import requests
|
|
|
|
from settings import APPLY_TIMEOUT, APPLY_URL, INTERNAL_API_KEY
|
|
|
|
logger = logging.getLogger("web.apply_client")
|
|
|
|
|
|
def _row_to_profile(profile_row) -> dict:
|
|
"""Convert a user_profiles row to the apply service Profile dict."""
|
|
if profile_row is None:
|
|
return {}
|
|
keys = [
|
|
"salutation", "firstname", "lastname", "email", "telephone",
|
|
"street", "house_number", "postcode", "city",
|
|
"is_possessing_wbs", "wbs_type", "wbs_valid_till",
|
|
"wbs_rooms", "wbs_adults", "wbs_children", "is_prio_wbs",
|
|
"immomio_email", "immomio_password",
|
|
]
|
|
d = {}
|
|
for k in keys:
|
|
try:
|
|
d[k] = profile_row[k]
|
|
except (KeyError, IndexError):
|
|
pass
|
|
for k in ("is_possessing_wbs", "is_prio_wbs"):
|
|
d[k] = bool(d.get(k) or 0)
|
|
return d
|
|
|
|
|
|
class ApplyClient:
|
|
def __init__(self):
|
|
self.base = APPLY_URL.rstrip("/")
|
|
self.timeout = APPLY_TIMEOUT
|
|
self.headers = {"X-Internal-Api-Key": INTERNAL_API_KEY}
|
|
|
|
def health(self) -> bool:
|
|
try:
|
|
r = requests.get(f"{self.base}/health", timeout=5)
|
|
return r.ok
|
|
except requests.RequestException:
|
|
return False
|
|
|
|
def apply(self, url: str, profile: dict, submit_forms: bool,
|
|
application_id: int | None = None) -> dict:
|
|
body = {
|
|
"url": url,
|
|
"profile": profile,
|
|
"submit_forms": bool(submit_forms),
|
|
"application_id": application_id,
|
|
}
|
|
try:
|
|
r = requests.post(
|
|
f"{self.base}/apply", json=body,
|
|
headers=self.headers, timeout=self.timeout,
|
|
)
|
|
if r.status_code >= 400:
|
|
return {"success": False,
|
|
"message": f"apply HTTP {r.status_code}: {r.text[:300]}",
|
|
"provider": "", "forensics": {}}
|
|
return r.json()
|
|
except requests.RequestException as e:
|
|
return {"success": False, "message": f"apply unreachable: {e}",
|
|
"provider": "", "forensics": {}}
|