fix: sync scraper.py with server version (proxy support + ID-based fetch)

- Remove initial_loaded/baseline logic from scheduler
- Remove initial_loaded from keywords SELECT query
- Replace synchronous notification loops with asyncio.create_task
  (fire-and-forget) for both new ads and price drops
- Add safe_notify_new_ad/safe_notify_price_drop wrapper functions
- Scraper already simplified (single page, 30 ads, no cursor)

worker/src/main.py

@@ -19,11 +19,33 @@ logger = logging.getLogger(__name__)
 load_dotenv()
 
 
+async def safe_notify_new_ad(bot, pool, tg_id, notify_fields, ad_uuid):
+    try:
+        msg_id_val = await notify_new_ad(bot, tg_id, notify_fields)
+        if msg_id_val:
+            user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
+            if user_row:
+                await log_notification(pool, str(user_row["id"]), ad_uuid, msg_id_val)
+    except Exception:
+        logger.exception("Failed to send new ad notification for %s", ad_uuid)
+
+
+async def safe_notify_price_drop(bot, pool, tg_id, notify_fields, ad_uuid):
+    try:
+        msg_id_val = await notify_price_drop(bot, tg_id, notify_fields)
+        if msg_id_val:
+            user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
+            if user_row:
+                await log_notification(pool, str(user_row["id"]), ad_uuid, msg_id_val)
+    except Exception:
+        logger.exception("Failed to send price drop notification for %s", ad_uuid)
+
+
 async def scheduler_task(pool: object, bot: ExtBot) -> None:
     while True:
         try:
             rows = await pool.fetch(
-                "SELECT id, keyword, interval_minutes, initial_loaded FROM keywords "
+                "SELECT id, keyword, interval_minutes FROM keywords "
                 "WHERE is_active = true "
                 "AND (last_scraped_at IS NULL OR last_scraped_at < now() - (interval_minutes || ' minutes')::interval)"
             )
@@ -31,7 +53,6 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
             for row in rows:
                 kw_id = str(row["id"])
                 keyword = row["keyword"]
-                initial_loaded = row["initial_loaded"]
 
                 subs = await pool.fetch(
                     "SELECT telegram_id FROM users u JOIN keyword_subscriptions ks ON u.id = ks.user_id "
@@ -50,15 +71,9 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
                     ads_raw, total_hits = await fetch_ads(keyword)
                     new_count = 0
 
-                    if not initial_loaded and len(ads_raw) > 0:
-                        logger.info("Initial baseline load for '%s' — indexing %d ads, no notifications", keyword, len(ads_raw))
-
                     for ad_data in ads_raw:
                         fields = extract_ad_fields(ad_data)
                         wh_ad_id = fields["wh_ad_id"]
-                        is_price_drop = False
-                        old_price = None
-                        new_price = None
 
                         existing = await pool.fetchrow(
                             "SELECT id, price FROM ads WHERE wh_ad_id = $1",
@@ -66,6 +81,7 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
                         )
 
                         if not existing:
+                            # NEW AD - insert and fire-and-forget notify
                             ad_row = await pool.fetchrow(
                                 "INSERT INTO ads (wh_ad_id, raw_json, title, price, location, url, published_at, main_image_url, postcode, modified_at) "
                                 "VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) RETURNING id",
@@ -75,20 +91,12 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
                             )
                             ad_uuid = str(ad_row["id"])
 
-                            # Only notify for genuinely new ads after baseline load is done
-                            if initial_loaded:
-                                notify_fields = {**fields, "keyword": keyword}
-                                for tg_id in telegram_ids:
-                                    msg_id_val = await notify_new_ad(bot, tg_id, notify_fields)
-                                    if msg_id_val:
-                                        user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
-                                        if user_row:
-                                            try:
-                                                await log_notification(pool, str(user_row["id"]), ad_uuid, msg_id_val)
-                                            except Exception:
-                                                logger.exception("Failed to log new ad notification")
-                                new_count += 1
+                            notify_fields = {**fields, "keyword": keyword}
+                            for tg_id in telegram_ids:
+                                asyncio.create_task(safe_notify_new_ad(bot, pool, tg_id, notify_fields, ad_uuid))
+                            new_count += 1
                         else:
+                            # EXISTING AD - check price drop
                             ad_uuid = str(existing["id"])
                             old_price = existing["price"]
                             new_price = fields["price"]
@@ -102,29 +110,17 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None:
                                     "INSERT INTO price_history (ad_id, old_price, new_price) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
                                     ad_uuid, old_price, new_price,
                                 )
-                                is_price_drop = True
+                                notify_fields = {**fields, "keyword": keyword}
+                                for tg_id in telegram_ids:
+                                    asyncio.create_task(safe_notify_price_drop(bot, pool, tg_id, notify_fields, ad_uuid))
                             else:
+                                # Update metadata if missing
                                 if fields.get("main_image_url") or fields.get("postcode"):
                                     await pool.execute(
                                         "UPDATE ads SET main_image_url = COALESCE($1, main_image_url), postcode = COALESCE($2, postcode) WHERE id = $3 AND (main_image_url IS NULL OR postcode IS NULL)",
                                         fields.get("main_image_url"), fields.get("postcode"), ad_uuid,
                                     )
 
-                        if is_price_drop:
-                            notify_fields = {**fields, "keyword": keyword}
-                            for tg_id in telegram_ids:
-                                msg_id_val = await notify_price_drop(bot, tg_id, notify_fields)
-                                if msg_id_val:
-                                    user_row = await pool.fetchrow("SELECT id FROM users WHERE telegram_id = $1", tg_id)
-                                    if user_row:
-                                        try:
-                                            await log_notification(pool, str(user_row["id"]), ad_uuid, msg_id_val)
-                                        except Exception:
-                                            logger.exception("Failed to log price drop notification")
-
-                    if not initial_loaded:
-                        await pool.execute("UPDATE keywords SET initial_loaded = true WHERE id = $1", kw_id)
-
                     await pool.execute("UPDATE keywords SET last_scraped_at = now() WHERE id = $1", kw_id)
 
                     await pool.execute(
@@ -203,4 +199,4 @@ async def main() -> None:
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    asyncio.run(main())

worker/src/scraper.py

@@ -1,13 +1,195 @@
 import asyncio
 import logging
-from datetime import datetime, timezone
+import os
+from datetime import datetime
 from typing import Any
-from urllib.parse import quote_plus
 
 import httpx
 
+from db import get_pool
+from settings import get_effective_proxy_url, get_proxy_url_from_env, proxy_available
+
 logger = logging.getLogger(__name__)
 
+_client: httpx.AsyncClient | None = None
+_client_proxy_url: str | None = None
+_proxy_ip_logged = False
+_system_public_ip: str | None = None
+_proxy_public_ip: str | None = None
+_last_used_public_ip: str | None = None
+_proxy_enabled_effective: bool | None = None
+
+
+def _get_proxy_url() -> str | None:
+    return get_proxy_url_from_env()
+
+
+async def _get_effective_proxy_url() -> str | None:
+    global _proxy_enabled_effective
+
+    pool = await get_pool()
+    proxy_url = await get_effective_proxy_url(pool)
+    _proxy_enabled_effective = proxy_url is not None
+    return proxy_url
+
+
+def _redact_proxy_url(proxy_url: str) -> str:
+    try:
+        parsed = httpx.URL(proxy_url)
+        host = parsed.host or "unknown-host"
+        port = parsed.port or 80
+        user = parsed.username or "unknown-user"
+        return f"{host}:{port} (user={user}, credentials=set)"
+    except Exception:
+        return "<unparseable proxy>"
+
+
+async def _fetch_public_ip(client: httpx.AsyncClient) -> str | None:
+    try:
+        resp = await client.get("https://api.ipify.org?format=json")
+        resp.raise_for_status()
+        data = resp.json()
+        ip = data.get("ip")
+        return str(ip) if ip else None
+    except Exception as exc:
+        logger.warning("Could not resolve public IP: %s", exc)
+        return None
+
+
+async def _log_proxy_ip_comparison(proxy_url: str) -> None:
+    global _proxy_ip_logged, _system_public_ip, _proxy_public_ip, _last_used_public_ip
+
+    if _proxy_ip_logged:
+        return
+
+    _proxy_ip_logged = True
+
+    direct_client = httpx.AsyncClient(timeout=10.0, trust_env=False)
+    proxy_client = httpx.AsyncClient(timeout=10.0, trust_env=False, proxy=proxy_url)
+
+    try:
+        direct_ip = await _fetch_public_ip(direct_client)
+        proxy_ip = await _fetch_public_ip(proxy_client)
+        _system_public_ip = direct_ip
+        _proxy_public_ip = proxy_ip
+        _last_used_public_ip = proxy_ip or direct_ip
+        logger.info(
+            "HTTPS proxy enabled: %s | public_ip_without_proxy=%s | public_ip_with_proxy=%s",
+            _redact_proxy_url(proxy_url),
+            direct_ip or "unknown",
+            proxy_ip or "unknown",
+        )
+    finally:
+        await direct_client.aclose()
+        await proxy_client.aclose()
+
+
+async def get_client() -> httpx.AsyncClient:
+    """Return a shared AsyncClient with keepalive connection pool."""
+    global _client, _client_proxy_url, _system_public_ip, _last_used_public_ip
+
+    proxy_url = await _get_effective_proxy_url()
+
+    if _client is not None and not _client.is_closed and _client_proxy_url != proxy_url:
+        await _client.aclose()
+        logger.info(
+            "Recreated httpx client because proxy changed: %s -> %s",
+            "enabled" if _client_proxy_url else "disabled",
+            "enabled" if proxy_url else "disabled",
+        )
+        _client = None
+
+    if _client is None or _client.is_closed:
+        max_conns = int(os.getenv("HTTP_MAX_CONNECTIONS", "10"))
+        max_keepalive = int(os.getenv("HTTP_KEEPALIVE_CONNECTIONS", "5"))
+
+        if proxy_url:
+            await _log_proxy_ip_comparison(proxy_url)
+        elif _system_public_ip is None:
+            direct_client = httpx.AsyncClient(timeout=10.0, trust_env=False)
+            try:
+                _system_public_ip = await _fetch_public_ip(direct_client)
+                _last_used_public_ip = _system_public_ip
+            finally:
+                await direct_client.aclose()
+
+        _client = httpx.AsyncClient(
+            timeout=float(os.getenv("HTTP_TIMEOUT_S", "30.0")),
+            limits=httpx.Limits(
+                max_connections=max_conns,
+                max_keepalive_connections=max_keepalive,
+                keepalive_expiry=60,
+            ),
+            proxy=proxy_url,
+            trust_env=False,
+        )
+        _client_proxy_url = proxy_url
+        logger.info(
+            "Created httpx client: max_conns=%d, keepalive=%d, proxy=%s",
+            max_conns,
+            max_keepalive,
+            "enabled" if proxy_url else "disabled",
+        )
+
+    return _client
+
+
+async def refresh_network_status() -> dict[str, str | bool | None]:
+    """Ensure network status has best-effort values even before first scrape."""
+    global _system_public_ip, _proxy_public_ip, _last_used_public_ip, _proxy_enabled_effective
+
+    proxy_url = await _get_effective_proxy_url()
+    _proxy_enabled_effective = proxy_url is not None
+
+    if _system_public_ip is None:
+        direct_client = httpx.AsyncClient(timeout=10.0, trust_env=False)
+        try:
+            _system_public_ip = await _fetch_public_ip(direct_client)
+        finally:
+            await direct_client.aclose()
+
+    if proxy_url and _proxy_public_ip is None:
+        proxy_client = httpx.AsyncClient(timeout=10.0, trust_env=False, proxy=proxy_url)
+        try:
+            _proxy_public_ip = await _fetch_public_ip(proxy_client)
+        finally:
+            await proxy_client.aclose()
+
+    if proxy_url:
+        _last_used_public_ip = _proxy_public_ip or _system_public_ip
+    else:
+        _last_used_public_ip = _system_public_ip
+
+    return get_network_status()
+
+
+def get_network_status() -> dict[str, str | bool | None]:
+    """Return best-effort network status for UI display."""
+    available = proxy_available()
+    proxy_enabled = bool(_proxy_enabled_effective) if _proxy_enabled_effective is not None else available
+    last_used = _last_used_public_ip
+    if last_used is None:
+        last_used = _proxy_public_ip if proxy_enabled else _system_public_ip
+
+    return {
+        "proxy_available": available,
+        "proxy_enabled": proxy_enabled,
+        "system_public_ip": _system_public_ip,
+        "proxy_public_ip": _proxy_public_ip,
+        "last_used_public_ip": last_used,
+    }
+
+
+async def close_client() -> None:
+    """Close the shared AsyncClient. Call during shutdown."""
+    global _client, _client_proxy_url
+    if _client and not _client.is_closed:
+        await _client.aclose()
+        logger.info("Closed httpx client")
+    _client = None
+    _client_proxy_url = None
+
+
 _API_URL = (
     "https://www.willhaben.at/webapi/ad-search/search/atz/seo/"
     "kaufen-und-verkaufen/marktplatz"
@@ -21,25 +203,27 @@ _HEADERS = {
 
 
 async def fetch_ads(keyword: str) -> tuple[list[dict[str, Any]], int]:
+    """Fetch the latest 30 ads for a keyword (single page, newest first)."""
     params = {
         "keyword": keyword,
         "rows": 30,
         "sort": 1,
     }
 
-    async with httpx.AsyncClient(timeout=30.0) as client:
-        for attempt in range(1, 4):
-            try:
-                resp = await client.get(_API_URL, headers=_HEADERS, params=params)
-                resp.raise_for_status()
-                data = resp.json()
-                break
-            except Exception as exc:
-                logger.warning("fetch_ads attempt %d failed: %s", attempt, exc)
-                if attempt < 3:
-                    await asyncio.sleep(2 ** attempt)
-                    continue
-                raise
+    client = await get_client()
+
+    for attempt in range(1, 4):
+        try:
+            resp = await client.get(_API_URL, headers=_HEADERS, params=params)
+            resp.raise_for_status()
+            data = resp.json()
+            break
+        except Exception as exc:
+            logger.warning("fetch_ads attempt %d failed for '%s': %s", attempt, keyword, exc)
+            if attempt < 3:
+                await asyncio.sleep(2 ** attempt)
+                continue
+            raise
 
     ads_raw = data.get("advertSummaryList", {}).get("advertSummary", [])
     total_hits = int(data.get("rowsFound", 0))
@@ -111,4 +295,4 @@ def extract_ad_fields(ad_dict: dict[str, Any]) -> dict[str, Any]:
         "main_image_url": main_image_url,
         "postcode": postcode,
         "modified_at": modified_at,
-    }
+    }