feat(notifications): new match format with Gmaps + lazyflat deep-link

New Telegram match layout:
  Karl-Ziegler-Straße 7           (linked → Google Maps)
  12489 Treptow-Köpenick
  Miete: 944.12 (18.51 €/m²)
  Fläche: 51.0
  Zimmer: 2.0
  WBS: nicht erforderlich

  Zur original Anzeige            (→ flat URL)
  Zur lazyflat Seite              (→ /?flat=<id>)

Deep-link behavior on lazyflat: ?flat=<id> expands the matching row,
scrolls it into view, and pulses a yellow highlight for 3s. The query
param is stripped from history afterwards so reload stays clean.
Unknown flat IDs drop the param silently.

Helpers: _address_lines splits the scraper's "Street, PLZ, District"
into two display lines; _gmaps_url falls back to a maps.google query
when the payload has no explicit link; _wbs_label normalises the
German WBS variants to "erforderlich" / "nicht erforderlich".

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
EiSiMo 2026-04-23 10:28:04 +02:00
parent 77246d1381
commit 81d6b65eae
3 changed files with 108 additions and 7 deletions

View file

@ -7,6 +7,7 @@ Channels:
"""
import logging
from typing import Optional
from urllib.parse import quote
import requests
@ -70,15 +71,75 @@ def notify_user(user_id: int, event: EventType, *,
# -- Convenience builders -----------------------------------------------------
def _address_lines(address: str) -> tuple[str, str]:
"""Split "Street Nr, PLZ, District" into (line1, line2) for the two-line
layout in notifications. Falls back gracefully for unexpected shapes."""
parts = [p.strip() for p in (address or "").split(",") if p.strip()]
if len(parts) >= 3:
return parts[0], f"{parts[1]} {', '.join(parts[2:])}"
if len(parts) == 2:
return parts[0], parts[1]
return (address or "").strip(), ""
def _gmaps_url(address: str, lat, lng) -> str:
if lat is not None and lng is not None:
return f"https://www.google.com/maps/search/?api=1&query={lat},{lng}"
return f"https://www.google.com/maps/search/?api=1&query={quote(address or '')}"
def _wbs_label(wbs: str) -> str:
w = (wbs or "").strip().lower()
if w == "erforderlich":
return "erforderlich"
if w in ("nicht erforderlich", "kein", "nein", "no", "ohne", "-", ""):
return "nicht erforderlich"
return wbs # pass through unrecognised literals
def _fmt_num(v) -> str:
return "" if v is None else f"{v}"
def on_match(user_id: int, flat: dict) -> None:
addr = flat.get("address") or flat.get("link")
rent = flat.get("total_rent")
rooms = flat.get("rooms")
address = (flat.get("address") or "").strip()
line1, line2 = _address_lines(address)
link = flat.get("link", "")
body = f"Neue passende Wohnung: {addr}\nMiete: {rent}\nZimmer: {rooms}\n{link}"
md = (f"*Neue passende Wohnung*\n[{addr}]({link})\n"
f"Miete: {rent} € · Zimmer: {rooms}\n{PUBLIC_URL}")
notify_user(user_id, "match", subject="[lazyflat] passende Wohnung", body_plain=body, body_markdown=md)
rent = flat.get("total_rent")
sqm_price = flat.get("sqm_price")
rooms = flat.get("rooms")
size = flat.get("size")
wbs_txt = _wbs_label(flat.get("wbs", ""))
gmaps = flat.get("address_link_gmaps") or _gmaps_url(
address, flat.get("lat"), flat.get("lng"),
)
flat_id = str(flat.get("id", ""))
deep_link = f"{PUBLIC_URL}/?flat={quote(flat_id, safe='')}"
rent_str = _fmt_num(rent)
if sqm_price:
rent_str += f" ({sqm_price} €/m²)"
body = (
f"{line1}\n{line2}\n"
f"Miete: {rent_str}\n"
f"Fläche: {_fmt_num(size)}\n"
f"Zimmer: {_fmt_num(rooms)}\n"
f"WBS: {wbs_txt}\n\n"
f"Original: {link}\n"
f"lazyflat: {deep_link}"
)
md = (
f"[{line1}]({gmaps})\n{line2}\n"
f"Miete: {rent_str}\n"
f"Fläche: {_fmt_num(size)}\n"
f"Zimmer: {_fmt_num(rooms)}\n"
f"WBS: {wbs_txt}\n\n"
f"[Zur original Anzeige]({link})\n"
f"[Zur lazyflat Seite]({deep_link})"
)
notify_user(user_id, "match", subject="[lazyflat] passende Wohnung",
body_plain=body, body_markdown=md)
def on_apply_ok(user_id: int, flat: dict, message: str) -> None: