diff --git a/worker/src/main.py b/worker/src/main.py index c20ed73..a52955b 100644 --- a/worker/src/main.py +++ b/worker/src/main.py @@ -18,8 +18,9 @@ from telegram.request import HTTPXRequest from db import close_pool, get_pool from health import create_health_app, record_scheduler_run, set_start_time, set_telegram_polling -from scraper import extract_ad_fields, fetch_ads +from scraper import extract_ad_fields, fetch_ads, get_network_status from notifier import log_notification, notify_new_ad, notify_price_drop, is_user_muted, buffer_for_digest +from settings import get_turbo_mode logger = logging.getLogger(__name__) @@ -206,6 +207,18 @@ async def flush_digests(pool: asyncpg.Pool, bot: ExtBot) -> int: async def scheduler_task(pool: object, bot: ExtBot) -> None: while True: + proxy_enabled = bool(get_network_status().get("proxy_enabled")) + turbo_mode = False + try: + turbo_mode = await get_turbo_mode(pool) + except Exception: + logger.exception("Could not load turbo mode setting") + + turbo_active = turbo_mode and proxy_enabled + speed_divisor = 10 if turbo_active else 1 + inter_keyword_sleep_s = 5.0 / speed_divisor + loop_sleep_s = 30.0 / speed_divisor + record_scheduler_run() # mark this cycle as started try: processed = await process_notification_queue(pool, bot) @@ -222,14 +235,33 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None: logger.exception("Error flushing digests") try: + now_utc = datetime.now(tz=timezone.utc) rows = await pool.fetch( - "SELECT id, keyword, interval_minutes, initial_loaded, ads_cursor, " + "SELECT id, keyword, interval_minutes, last_scraped_at, initial_loaded, ads_cursor, " "price_min, price_max, allowed_postcodes FROM keywords " - "WHERE is_active = true " - "AND (last_scraped_at IS NULL OR last_scraped_at < now() - (interval_minutes || ' minutes')::interval)" + "WHERE is_active = true" ) + due_rows = [] for row in rows: + interval_minutes = float(row.get("interval_minutes") or 1) + effective_interval_minutes = max(interval_minutes / speed_divisor, 0.1) + last_scraped_at = row.get("last_scraped_at") + + if last_scraped_at is None: + due_rows.append(row) + continue + + if last_scraped_at.tzinfo is None: + last_scraped_at = last_scraped_at.replace(tzinfo=timezone.utc) + + if last_scraped_at <= now_utc - timedelta(minutes=effective_interval_minutes): + due_rows.append(row) + + if turbo_active and due_rows: + logger.info("Turbo mode active via proxy: %d due keyword(s), x10 speed", len(due_rows)) + + for row in due_rows: kw_id = str(row["id"]) keyword = row["keyword"] initial_loaded = row["initial_loaded"] @@ -371,12 +403,12 @@ async def scheduler_task(pool: object, bot: ExtBot) -> None: kw_id, str(sys.exc_info()[1]), ) - await asyncio.sleep(5) + await asyncio.sleep(inter_keyword_sleep_s) except Exception: logger.exception("Scheduler iteration error") - await asyncio.sleep(30) + await asyncio.sleep(loop_sleep_s) async def main() -> None: diff --git a/worker/src/migrations/07-app-settings.sql b/worker/src/migrations/07-app-settings.sql new file mode 100644 index 0000000..6ebed8e --- /dev/null +++ b/worker/src/migrations/07-app-settings.sql @@ -0,0 +1,9 @@ +CREATE TABLE IF NOT EXISTS app_settings ( + key text PRIMARY KEY, + value text NOT NULL, + updated_at timestamptz NOT NULL DEFAULT now() +); + +INSERT INTO app_settings (key, value) +VALUES ('turbo_mode', 'false') +ON CONFLICT (key) DO NOTHING; diff --git a/worker/src/settings.py b/worker/src/settings.py new file mode 100644 index 0000000..6d9c78d --- /dev/null +++ b/worker/src/settings.py @@ -0,0 +1,32 @@ +import asyncpg + + +async def ensure_app_settings(pool: asyncpg.Pool) -> None: + await pool.execute( + """ + CREATE TABLE IF NOT EXISTS app_settings ( + key text PRIMARY KEY, + value text NOT NULL, + updated_at timestamptz NOT NULL DEFAULT now() + ); + """ + ) + + +async def get_turbo_mode(pool: asyncpg.Pool) -> bool: + await ensure_app_settings(pool) + raw = await pool.fetchval("SELECT value FROM app_settings WHERE key = 'turbo_mode'") + return str(raw).lower() in {"1", "true", "yes", "on"} + + +async def set_turbo_mode(pool: asyncpg.Pool, enabled: bool) -> None: + await ensure_app_settings(pool) + await pool.execute( + """ + INSERT INTO app_settings (key, value, updated_at) + VALUES ('turbo_mode', $1, now()) + ON CONFLICT (key) + DO UPDATE SET value = EXCLUDED.value, updated_at = now() + """, + "true" if enabled else "false", + ) diff --git a/worker/src/templates/dashboard.html b/worker/src/templates/dashboard.html index 23362bd..13d935d 100644 --- a/worker/src/templates/dashboard.html +++ b/worker/src/templates/dashboard.html @@ -82,6 +82,28 @@ +
+
Turbo Mode (Admin)
+
+ {{ 'Active' if data.turbo_active else ('Armed' if data.turbo_mode else 'Off') }} +
+
+ {% if data.proxy_enabled %} + Scheduler runs every ~{{ '%.1f'|format(data.turbo_effective_sleep_s) }}s + {% else %} + Enable proxy first to activate turbo + {% endif %} +
+ {% if data.seconds_until_next is not none %} +
Next cycle in ~{{ data.seconds_until_next }}s
+ {% endif %} +
+ +
+
+
Delivery Queue
diff --git a/worker/src/web.py b/worker/src/web.py index 947c16a..fc054e3 100644 --- a/worker/src/web.py +++ b/worker/src/web.py @@ -2,15 +2,17 @@ import os import logging import time from contextlib import asynccontextmanager +from datetime import datetime, timezone import httpx from fastapi import FastAPI, Request -from fastapi.responses import HTMLResponse, JSONResponse +from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from fastapi.templating import Jinja2Templates import asyncpg from db import get_pool from scraper import get_network_status, refresh_network_status +from settings import get_turbo_mode, set_turbo_mode logger = logging.getLogger(__name__) @@ -141,6 +143,7 @@ app = FastAPI(title="Willhaben Tracker Web UI", lifespan=lifespan) @app.get("/", response_class=HTMLResponse) async def dashboard(request: Request): + pool = await get_pool() try: total_keywords = await value("SELECT COUNT(*) FROM keywords") active_keywords = await value("SELECT COUNT(*) FROM keywords WHERE is_active = true") @@ -166,8 +169,26 @@ async def dashboard(request: Request): network = get_network_status() system_ip = network.get("system_public_ip") last_used_ip = network.get("last_used_public_ip") + proxy_enabled = bool(network.get("proxy_enabled")) system_country = await _country_code_for_ip(system_ip if isinstance(system_ip, str) else None) last_used_country = await _country_code_for_ip(last_used_ip if isinstance(last_used_ip, str) else None) + turbo_mode = await get_turbo_mode(pool) + turbo_active = turbo_mode and proxy_enabled + + turbo_base_sleep_s = 30.0 + turbo_effective_sleep_s = turbo_base_sleep_s / (10 if turbo_active else 1) + + next_refresh_at = None + if last_scheduler and last_scheduler.get("scraped_at"): + next_refresh_at = last_scheduler["scraped_at"] + if next_refresh_at.tzinfo is None: + next_refresh_at = next_refresh_at.replace(tzinfo=timezone.utc) + next_refresh_at = next_refresh_at.timestamp() + turbo_effective_sleep_s + + now_epoch = datetime.now(tz=timezone.utc).timestamp() + seconds_until_next = None + if next_refresh_at is not None: + seconds_until_next = max(0, int(round(next_refresh_at - now_epoch))) data = { "total_keywords": total_keywords or 0, @@ -178,17 +199,29 @@ async def dashboard(request: Request): "queue_pending": queue_pending or 0, "queue_dead": queue_dead or 0, "last_scheduler": last_scheduler["scraped_at"] if last_scheduler else None, - "proxy_enabled": _proxy_enabled(), + "proxy_enabled": proxy_enabled, "system_public_ip": system_ip, "last_used_public_ip": last_used_ip, "system_country_code": system_country, "last_used_country_code": last_used_country, "system_country_flag": _flag_from_country_code(system_country), "last_used_country_flag": _flag_from_country_code(last_used_country), + "turbo_mode": turbo_mode, + "turbo_active": turbo_active, + "turbo_effective_sleep_s": turbo_effective_sleep_s, + "seconds_until_next": seconds_until_next, } return templates.TemplateResponse("dashboard.html", {"request": request, "error": None, "data": data}) +@app.post("/admin/turbo") +async def toggle_turbo(request: Request): + pool = await get_pool() + current = await get_turbo_mode(pool) + await set_turbo_mode(pool, not current) + return RedirectResponse(url="/", status_code=303) + + @app.get("/keywords", response_class=HTMLResponse) async def keywords_list(request: Request): try: