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>
83 lines
3.9 KiB
Python
83 lines
3.9 KiB
Python
import logging
|
|
|
|
from actions import open_page
|
|
from classes.application_result import ApplicationResult
|
|
from language import _
|
|
from providers._provider import ApplyContext, Provider
|
|
|
|
logger = logging.getLogger("flat-apply")
|
|
|
|
|
|
class Wbm(Provider):
|
|
@property
|
|
def domain(self) -> str:
|
|
return "wbm.de"
|
|
|
|
async def apply_for_flat(self, url: str, ctx: ApplyContext) -> ApplicationResult:
|
|
r = ctx.recorder
|
|
p = ctx.profile
|
|
async with open_page(url, recorder=r) as page:
|
|
r.step("page.404_check")
|
|
if await page.get_by_role("heading", name="Page Not Found").is_visible():
|
|
return ApplicationResult(False, message=_("not_found"))
|
|
|
|
r.step("cookies.check")
|
|
cookie_accept_btn = page.get_by_text("Alle zulassen")
|
|
if await cookie_accept_btn.is_visible():
|
|
await cookie_accept_btn.click()
|
|
r.step("cookies.accepted")
|
|
else:
|
|
r.step("cookies.absent")
|
|
|
|
r.step("chatbot.remove")
|
|
try:
|
|
await page.locator('#removeConvaiseChat').click()
|
|
except Exception as e:
|
|
r.step("chatbot.remove_failed", "warn", str(e))
|
|
|
|
r.step("ad.offline_check")
|
|
if page.url == "https://www.wbm.de/wohnungen-berlin/angebote/":
|
|
return ApplicationResult(False, message=_("ad_offline"))
|
|
|
|
r.step("form.open"); await page.locator('.openimmo-detail__contact-box-button').click()
|
|
|
|
r.step("form.wbs_section")
|
|
if p.is_possessing_wbs:
|
|
await page.locator('label[for="powermail_field_wbsvorhanden_1"]').click()
|
|
await page.locator("input[name*='[wbsgueltigbis]']").fill(p.wbs_valid_till_dt().strftime("%Y-%m-%d"))
|
|
await page.locator("select[name*='[wbszimmeranzahl]']").select_option(str(p.wbs_rooms))
|
|
await page.locator("#powermail_field_einkommensgrenzenacheinkommensbescheinigung9").select_option(p.wbs_type)
|
|
if p.is_prio_wbs:
|
|
await page.locator("#powermail_field_wbsmitbesonderemwohnbedarf_1").check(force=True)
|
|
else:
|
|
await page.locator('label[for="powermail_field_wbsvorhanden_2"]').click()
|
|
|
|
r.step("form.personal")
|
|
await page.locator("#powermail_field_anrede").select_option(p.salutation)
|
|
await page.locator("#powermail_field_name").fill(p.lastname)
|
|
await page.locator("#powermail_field_vorname").fill(p.firstname)
|
|
await page.locator("#powermail_field_strasse").fill(p.street)
|
|
await page.locator("#powermail_field_plz").fill(p.postcode)
|
|
await page.locator("#powermail_field_ort").fill(p.city)
|
|
await page.locator("#powermail_field_e_mail").fill(p.email)
|
|
await page.locator("#powermail_field_telefon").fill(p.telephone)
|
|
await page.locator("#powermail_field_datenschutzhinweis_1").check(force=True)
|
|
|
|
if not ctx.submit_forms:
|
|
r.step("submit.dry_run")
|
|
return ApplicationResult(success=True, message=_("application_success_dry"))
|
|
|
|
r.step("submit.click"); await page.get_by_role("button", name="Anfrage absenden").click()
|
|
|
|
r.step("success.check")
|
|
if await page.get_by_text("Wir haben Ihre Anfrage für das Wohnungsangebot erhalten.").is_visible():
|
|
return ApplicationResult(True)
|
|
elif await self._missing_fields_warning(page):
|
|
r.step("missing_fields_detected", "warn")
|
|
return ApplicationResult(False, _("missing_fields"))
|
|
r.step("success.not_found", "warn")
|
|
return ApplicationResult(success=False, message=_("submit_conformation_msg_not_found"))
|
|
|
|
async def _missing_fields_warning(self, page) -> bool:
|
|
missing_field_msg = page.get_by_text("Dieses Feld muss ausgefüllt werden!").first
|
|
return await missing_field_msg.first.is_visible()
|