EiSiMo
3bb04210c4
1. Admin → Geheimnisse sub-tab lets you edit ANTHROPIC_API_KEY + BERLIN_WOHNEN_USERNAME/PASSWORD at runtime. Migration v7 adds a secrets(key,value,updated_at) table; startup seeds missing keys from env (idempotent). web reads secrets DB-first (env fallback) via llm._api_key(); alert fetches them from web /internal/secrets on each scan, passes them into Scraper(). Rotating creds no longer needs a redeploy. Masked display: 6 leading + 4 trailing chars, "…" in the middle. Blank form fields leave the stored value untouched. 2. Drop the max_morning_commute filter from UI + server + FILTER_KEYS + filter summary (the underlying Maps.calculate_score code stays for potential future re-enable). 3. /static/didi.webp wired as favicon via <link rel="icon"> in base.html. 4. apply.open_page wraps page.goto in try/except so a failed load still produces a "goto.failed" step + screenshot instead of returning an empty forensics blob. networkidle + post-submission sleep are also made best-effort. The error ZIP export already writes screenshot+HTML per step and final_html — with this change every apply run leaves a reconstructable trail even when the listing is already offline. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
import logging
|
|
import time
|
|
from rich.console import Console
|
|
from rich.logging import RichHandler
|
|
|
|
from flat import Flat
|
|
from scraper import Scraper
|
|
from settings import TIME_INTERVALL
|
|
from utils import hash_any_object
|
|
from web_client import WebClient
|
|
|
|
|
|
def setup_logging():
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(message)s",
|
|
datefmt="[%X]",
|
|
handlers=[RichHandler(markup=True, console=Console(width=110))],
|
|
)
|
|
logging.getLogger("googlemaps").setLevel(logging.WARNING)
|
|
logging.getLogger("urllib3").setLevel(logging.WARNING)
|
|
|
|
|
|
logger = logging.getLogger("alert")
|
|
setup_logging()
|
|
|
|
|
|
class FlatAlerter:
|
|
def __init__(self):
|
|
self.web = WebClient()
|
|
self.last_response_hash = ""
|
|
|
|
def _flat_payload(self, flat: Flat) -> dict:
|
|
# Transit-connectivity is disabled to save Google-Maps quota. The
|
|
# helper on Flat (flat.connectivity → Maps.calculate_score) is
|
|
# intentionally kept so it can be re-enabled without re-writing code —
|
|
# just replace the empty dict with `flat.connectivity` when needed.
|
|
c: dict = {}
|
|
lat, lng = flat.coords
|
|
return {
|
|
"id": flat.id,
|
|
"link": flat.link,
|
|
"address": flat.address,
|
|
"rooms": flat.rooms,
|
|
"size": flat.size,
|
|
"cold_rent": flat.cold_rent,
|
|
"utilities": flat.utilities,
|
|
"total_rent": flat.total_rent,
|
|
"sqm_price": flat.sqm_price,
|
|
"available_from": flat.available_from,
|
|
"published_on": flat.published_on,
|
|
"wbs": flat.wbs,
|
|
"floor": flat.floor,
|
|
"bathrooms": flat.bathrooms,
|
|
"year_built": flat.year_built,
|
|
"heating": flat.heating,
|
|
"energy_carrier": flat.energy_carrier,
|
|
"energy_value": flat.energy_value,
|
|
"energy_certificate": flat.energy_certificate,
|
|
"address_link_gmaps": flat.address_link_gmaps,
|
|
"lat": lat,
|
|
"lng": lng,
|
|
"connectivity": {
|
|
"morning_time": c.get("morning_time", 0),
|
|
"morning_transfers": c.get("morning_transfers", 0),
|
|
"night_time": c.get("night_time", 0),
|
|
"night_transfers": c.get("night_transfers", 0),
|
|
},
|
|
"raw_data": flat.raw_data,
|
|
}
|
|
|
|
def scan(self):
|
|
logger.info("starting scan")
|
|
# Pull fresh creds from web each scan so admin edits take effect
|
|
# without a redeploy.
|
|
secrets = self.web.fetch_secrets()
|
|
scraper = Scraper(
|
|
username=secrets.get("BERLIN_WOHNEN_USERNAME", ""),
|
|
password=secrets.get("BERLIN_WOHNEN_PASSWORD", ""),
|
|
)
|
|
if not scraper.login():
|
|
return
|
|
flats_data = scraper.get_flats()
|
|
|
|
response_hashed = hash_any_object(flats_data)
|
|
if response_hashed == self.last_response_hash:
|
|
logger.info("no change since last scan")
|
|
return
|
|
self.last_response_hash = response_hashed
|
|
|
|
for number, data in enumerate(flats_data, 1):
|
|
flat = Flat(data)
|
|
logger.info(f"{str(number).rjust(2)}: submitting {flat}")
|
|
payload = self._flat_payload(flat)
|
|
if not self.web.submit_flat(payload):
|
|
logger.warning(f"\tcould not submit {flat.id} to web, will retry next loop")
|
|
logger.info("scan finished")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
logger.info("starting wohnungsdidi alert service")
|
|
alerter = FlatAlerter()
|
|
while True:
|
|
try:
|
|
alerter.scan()
|
|
alerter.web.heartbeat()
|
|
logger.info(f"sleeping for {TIME_INTERVALL} seconds")
|
|
time.sleep(TIME_INTERVALL)
|
|
except KeyboardInterrupt:
|
|
break
|
|
except Exception:
|
|
logger.exception("unexpected error")
|
|
time.sleep(60)
|