moritz/lazyflat
1
0
Fork 0
Code Issues 3 Pull requests Projects Releases Packages Wiki Activity Actions

chore: sweep dead code across all three services

Browse source 
Per review §1 — verified no callers before each deletion:

- _next_scrape_utc (context dict key never read by any template)
- ALERT_SCRAPE_INTERVAL_SECONDS settings import (only _next_scrape_utc read it)
- alert/paths.py (imported by nothing)
- alert/settings.py LANGUAGE (alert doesn't use translations.toml)
- alert/main.py: the vestigial `c = {}` connectivity dict, the comment
  about re-enabling it, and the entire connectivity block in
  _flat_payload — the web-side columns stay NULL on insert now
- alert/maps.py: DESTINATIONS, calculate_score, _get_next_weekday,
  _calculate_transfers (only geocode is used in the scraper)
- alert/flat.py: connectivity + display_address properties,
  _connectivity field, unused datetime import
- apply/utils.py str_to_preview (no callers) — file removed
- web/matching.py: max_morning_commute + commute check
- web/app.py: don't pass connectivity dict into flat_matches_filter,
  don't write email_address through update_notifications
- web/db.py: get_error (no callers); drop kill_switch,
  max_morning_commute, email_address from their allowed-sets so they're
  not writable through update_* anymore
- web/settings.py + docker-compose.yml: SMTP_HOST/PORT/USERNAME/PASSWORD/
  FROM/STARTTLS (notifications.py is telegram-only now)

DB columns themselves (kill_switch, email_address, max_morning_commute,
connectivity_morning_time, connectivity_night_time) stay in the schema
— SQLite can't drop them cheaply and they're harmless.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
EiSiMo 2026-04-21 19:06:05 +02:00
parent 617c76cb54
commit ebb11178e7
11 changed files with 5 additions and 166 deletions
Download patch file Download diff file Expand all files Collapse all files

17
alert/flat.py
View file

@ -27,7 +27,6 @@ class Flat:
self.raw_data = data
self.id = self.link # we could use data.get('id', None) but link is easier to debug
self.gmaps = maps.Maps()
self._connectivity = None
self._coords = None
self.address_link_gmaps = f"https://www.google.com/maps/search/?api=1&query={quote(self.address)}"
@ -54,24 +53,8 @@ class Flat:
return self.total_rent / self.size
return 0.0
@property
def connectivity(self):
if not self._connectivity:
self._connectivity = self.gmaps.calculate_score(self.address)
return self._connectivity
@property
def coords(self):
if self._coords is None:
self._coords = self.gmaps.geocode(self.address) or (None, None)
return self._coords
@property
def display_address(self):
if ',' in self.address:
parts = self.address.split(',', 1)
street_part = parts[0].strip()
city_part = parts[1].replace(',', '').strip()
return f"{street_part}\n{city_part}"
else:
return self.address

11
alert/main.py
View file

@ -31,11 +31,6 @@ class FlatAlerter:
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,
@ -60,12 +55,6 @@ class FlatAlerter:
"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,
}

81
alert/maps.py
View file

@ -1,25 +1,12 @@
import logging
import googlemaps
from datetime import datetime, timedelta, time as dt_time
from settings import GMAPS_API_KEY
logger = logging.getLogger("flat-alert")
class Maps:
DESTINATIONS = {
"Hbf": "Berlin Hauptbahnhof",
"Friedrichstr": "Friedrichstraße, Berlin",
"Kotti": "Kottbusser Tor, Berlin",
"Warschauer": "Warschauer Straße, Berlin",
"Ostkreuz": "Ostkreuz, Berlin",
"Nollendorf": "Nollendorfplatz, Berlin",
"Zoo": "Zoologischer Garten, Berlin",
"Kudamm": "Kurfürstendamm, Berlin",
"Gesundbrunnen": "Gesundbrunnen, Berlin",
"Hermannplatz": "Hermannplatz, Berlin"
}
class Maps:
def __init__(self):
self.gmaps = googlemaps.Client(key=GMAPS_API_KEY)
@ -36,69 +23,3 @@ class Maps:
except Exception as e:
logger.warning("geocode failed for %r: %s", address, e)
return None
def _get_next_weekday(self, date, weekday):
days_ahead = weekday - date.weekday()
if days_ahead <= 0:
days_ahead += 7
return date + timedelta(days_ahead)
def _calculate_transfers(self, steps):
transit_count = sum(1 for step in steps if step['travel_mode'] == 'TRANSIT')
return max(0, transit_count - 1)
def calculate_score(self, origin_address):
now = datetime.now()
# Next Monday 8:00 AM
next_monday = self._get_next_weekday(now, 0)
morning_departure = datetime.combine(next_monday.date(), dt_time(8, 0))
# Next Sunday 2:00 AM
next_sunday = self._get_next_weekday(now, 6)
night_departure = datetime.combine(next_sunday.date(), dt_time(2, 0))
total_morning_minutes = 0
total_morning_transfers = 0
total_night_minutes = 0
total_night_transfers = 0
dest_count = 0
for key, dest_address in self.DESTINATIONS.items():
# Morning: Flat -> Center
routes_morning = self.gmaps.directions(
origin=origin_address,
destination=dest_address,
mode="transit",
departure_time=morning_departure
)
# Night: Center -> Flat
routes_night = self.gmaps.directions(
origin=dest_address,
destination=origin_address,
mode="transit",
departure_time=night_departure
)
if routes_morning:
leg = routes_morning[0]['legs'][0]
total_morning_minutes += leg['duration']['value'] / 60
total_morning_transfers += self._calculate_transfers(leg['steps'])
if routes_night:
leg = routes_night[0]['legs'][0]
total_night_minutes += leg['duration']['value'] / 60
total_night_transfers += self._calculate_transfers(leg['steps'])
dest_count += 1
avg_m_time = total_morning_minutes / dest_count if dest_count else 0
avg_m_trans = total_morning_transfers / dest_count if dest_count else 0
avg_n_time = total_night_minutes / dest_count if dest_count else 0
avg_n_trans = total_night_transfers / dest_count if dest_count else 0
return {
'morning_time': avg_m_time,
'morning_transfers': avg_m_trans,
'night_time': avg_n_time,
'night_transfers': avg_n_trans
}

7
alert/paths.py
View file

@ -1,7 +0,0 @@
import os
DATA_DIR = "data"
ALREADY_NOTIFIED_FILE = "data/already_notified.txt"
# create dirs if they do not exist yet.
os.makedirs(DATA_DIR, exist_ok=True)

1
alert/settings.py
View file

@ -13,7 +13,6 @@ def _required(key: str) -> str:
return val
LANGUAGE: str = getenv("LANGUAGE", "en")
TIME_INTERVALL: int = int(getenv("SLEEP_INTERVALL", "60"))
# web backend: alert POSTs discovered flats here

12
apply/utils.py
View file

@ -1,12 +0,0 @@
import logging
logger = logging.getLogger("flat-apply")
def str_to_preview(string, max_length):
if not max_length > 3:
raise ValueError('max_length must be greater than 3')
first_line = string.split('\n')[0]
if len(first_line) > max_length:
return first_line[:max_length-3] + '...'
return first_line

6
docker-compose.yml
View file

@ -22,12 +22,6 @@ services:
- RETENTION_DAYS=${RETENTION_DAYS:-14}
- RETENTION_RUN_INTERVAL_SECONDS=${RETENTION_RUN_INTERVAL_SECONDS:-3600}
- PUBLIC_URL=${PUBLIC_URL:-https://flat.lab.moritz.run}
- SMTP_HOST=${SMTP_HOST:-}
- SMTP_PORT=${SMTP_PORT:-587}
- SMTP_USERNAME=${SMTP_USERNAME:-}
- SMTP_PASSWORD=${SMTP_PASSWORD:-}
- SMTP_FROM=${SMTP_FROM:-wohnungsdidi@localhost}
- SMTP_STARTTLS=${SMTP_STARTTLS:-true}
- ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY:-}
- ANTHROPIC_MODEL=${ANTHROPIC_MODEL:-claude-haiku-4-5-20251001}
volumes:

15
web/app.py
View file

@ -53,7 +53,6 @@ from auth import (
)
from matching import flat_matches_filter, row_to_dict
from settings import (
ALERT_SCRAPE_INTERVAL_SECONDS,
APPLY_FAILURE_THRESHOLD,
INTERNAL_API_KEY,
PUBLIC_URL,
@ -200,16 +199,6 @@ def _auto_apply_allowed(prefs) -> bool:
return apply_client.health()
def _next_scrape_utc() -> str:
hb = db.get_state("last_alert_heartbeat")
dt = _parse_iso(hb)
if dt is None:
return ""
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return (dt + timedelta(seconds=ALERT_SCRAPE_INTERVAL_SECONDS)).astimezone(timezone.utc).isoformat(timespec="seconds")
def _last_scrape_utc() -> str:
hb = db.get_state("last_alert_heartbeat")
dt = _parse_iso(hb)
@ -425,7 +414,7 @@ def _wohnungen_context(user) -> dict:
continue
if not flat_matches_filter({
"rooms": f["rooms"], "total_rent": f["total_rent"], "size": f["size"],
"wbs": f["wbs"], "connectivity": {"morning_time": f["connectivity_morning_time"]},
"wbs": f["wbs"],
}, filters):
continue
last = db.last_application_for_flat(uid, f["id"])
@ -499,7 +488,6 @@ def _wohnungen_context(user) -> dict:
"apply_allowed": allowed,
"apply_block_reason": reason,
"apply_reachable": apply_client.health(),
"next_scrape_utc": _next_scrape_utc(),
"last_scrape_utc": _last_scrape_utc(),
"has_running_apply": has_running,
"poll_interval": 3 if has_running else 30,
@ -1063,7 +1051,6 @@ async def action_notifications(request: Request, user=Depends(require_user)):
"channel": channel,
"telegram_bot_token": form.get("telegram_bot_token", ""),
"telegram_chat_id": form.get("telegram_chat_id", ""),
"email_address": "",
"notify_on_match": _b("notify_on_match"),
"notify_on_apply_success": _b("notify_on_apply_success"),
"notify_on_apply_fail": _b("notify_on_apply_fail"),

10
web/db.py
View file

@ -435,7 +435,7 @@ def get_filters(user_id: int) -> sqlite3.Row:
def update_filters(user_id: int, data: dict) -> None:
_ensure_user_rows(user_id)
allowed = {"rooms_min", "rooms_max", "max_rent", "min_size",
"max_morning_commute", "wbs_required", "max_age_hours"}
"wbs_required", "max_age_hours"}
clean = {k: data.get(k) for k in allowed if k in data}
if not clean:
return
@ -453,7 +453,7 @@ def get_notifications(user_id: int) -> sqlite3.Row:
def update_notifications(user_id: int, data: dict) -> None:
_ensure_user_rows(user_id)
allowed = {
"channel", "telegram_bot_token", "telegram_chat_id", "email_address",
"channel", "telegram_bot_token", "telegram_chat_id",
"notify_on_match", "notify_on_apply_success", "notify_on_apply_fail",
}
clean = {k: v for k, v in data.items() if k in allowed}
@ -473,7 +473,7 @@ def get_preferences(user_id: int) -> sqlite3.Row:
def update_preferences(user_id: int, data: dict) -> None:
_ensure_user_rows(user_id)
allowed = {
"auto_apply_enabled", "submit_forms", "kill_switch",
"auto_apply_enabled", "submit_forms",
"apply_circuit_open", "apply_recent_failures",
}
clean = {k: v for k, v in data.items() if k in allowed}
@ -719,10 +719,6 @@ def recent_errors(user_id: Optional[int], limit: int = 100,
).fetchall())
def get_error(error_id: int) -> Optional[sqlite3.Row]:
return _get_conn().execute("SELECT * FROM errors WHERE id = ?", (error_id,)).fetchone()
# ---------------------------------------------------------------------------
# Audit log
# ---------------------------------------------------------------------------

3
web/matching.py
View file

@ -18,7 +18,6 @@ def flat_matches_filter(flat: dict, f: dict | None) -> bool:
rooms = flat.get("rooms") or 0.0
rent = flat.get("total_rent") or 0.0
size = flat.get("size") or 0.0
commute = (flat.get("connectivity") or {}).get("morning_time") or 0.0
wbs_str = str(flat.get("wbs", "")).strip().lower()
if f.get("rooms_min") is not None and rooms < float(f["rooms_min"]):
@ -29,8 +28,6 @@ def flat_matches_filter(flat: dict, f: dict | None) -> bool:
return False
if f.get("min_size") is not None and size < float(f["min_size"]):
return False
if f.get("max_morning_commute") is not None and commute > float(f["max_morning_commute"]):
return False
wbs_req = (f.get("wbs_required") or "").strip().lower()
if wbs_req == "yes":

8
web/settings.py
View file

@ -55,14 +55,6 @@ RETENTION_RUN_INTERVAL_SECONDS: int = int(getenv("RETENTION_RUN_INTERVAL_SECONDS
LOGIN_RATE_LIMIT: int = int(getenv("LOGIN_RATE_LIMIT", "5"))
LOGIN_RATE_WINDOW_SECONDS: int = int(getenv("LOGIN_RATE_WINDOW_SECONDS", "900"))
# --- Email (system-wide SMTP for notifications) -------------------------------
SMTP_HOST: str = getenv("SMTP_HOST", "")
SMTP_PORT: int = int(getenv("SMTP_PORT", "587"))
SMTP_USERNAME: str = getenv("SMTP_USERNAME", "")
SMTP_PASSWORD: str = getenv("SMTP_PASSWORD", "")
SMTP_FROM: str = getenv("SMTP_FROM", "wohnungsdidi@localhost")
SMTP_STARTTLS: bool = getenv("SMTP_STARTTLS", "true").lower() in ("true", "1", "yes", "on")
# --- App URL (used to build links in notifications) ---------------------------
PUBLIC_URL: str = getenv("PUBLIC_URL", "https://flat.lab.moritz.run")