diff --git a/backend/app/__init__.py b/backend/app/__init__.py index 5833b4f..b9ecad2 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -109,6 +109,18 @@ def create_app(config: dict[str, Any] | None = None) -> Flask: app.extensions["paper_sessions"] = paper_sessions app.extensions["allowed_symbols"] = allowed_symbols + # shared in-process daily cache + app-internal data scheduler (independent of + # Hermes — this app runs on its own server). + from .daily_cache import DailyCache + from .scheduler import AppDataScheduler + cache = DailyCache() + app.extensions["daily_cache"] = cache + if not app.config.get("TESTING"): + interval_s = int(os.getenv("REFRESH_INTERVAL_SECONDS", "3600")) + scheduler = AppDataScheduler(cache, Path(__file__).resolve().parents[1] / "data", interval_seconds=interval_s) + scheduler.start() + app.extensions["data_scheduler"] = scheduler + @app.after_request def add_cors_headers(response): origin = request.headers.get("Origin") @@ -727,6 +739,24 @@ def create_app(config: dict[str, Any] | None = None) -> Flask: ) + @app.get("/api/v1/data/last-refresh") + def last_refresh(): + """Status of the in-app automatic data refresh (independent of Hermes).""" + interval = os.getenv("REFRESH_INTERVAL_SECONDS", "3600") + marker_path = Path(__file__).resolve().parents[1] / "data" / "scheduler" / "last_refresh.json" + last = None + if marker_path.exists(): + try: + last = json.loads(marker_path.read_text(encoding="utf-8")) + except Exception: + last = None + return jsonify({ + "automatic_refresh": True, + "interval_seconds": int(interval), + "interval_label": f"ทุก {int(interval)//3600} ชั่วโมง" if int(interval) >= 3600 else f"ทุก {int(interval)//60} นาที", + "last_refresh": last, + }) + @app.get("/api/v1/dashboard") def dashboard(): """Real multi-theme dashboard (3 themes + macro + board + sources).""" diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py new file mode 100644 index 0000000..3dc2765 --- /dev/null +++ b/backend/app/scheduler.py @@ -0,0 +1,134 @@ +"""App-internal background scheduler — refreshes real data on its own schedule. + +This app runs on its own server (Docker/EasyPanel), independent of Hermes. So +the periodic data-fetching/analysis must live INSIDE the app. This module runs a +daemon thread inside the Flask process that: + + - on an interval (default every 60 min), triggers each collector through the + shared daily cache (so fresh real data lands as the "current" state), + - records each successful refresh as a timestamped snapshot under the data dir + (persistent history the dashboard/sources table can report), + - logs progress; swallows transient failures (a source being down once must + not crash the app or the loop). + +Thread-safety: the loop only calls collectors + writes cache/snapshots; the HTTP +layer reads the daily cache. No shared mutable in-process state beyond that. +""" + +from __future__ import annotations + +import logging +import threading +import time +from pathlib import Path +from typing import Callable, List, Optional + +log = logging.getLogger("set50.scheduler") + +# collectors returning a .to_dict()/dict, keyed by cache key +# (imported lazily to avoid import cycles at module load) +_REFRESH_JOBS: List[dict] = [ + {"key": "bot_tourism", "label": "ท่องเที่ยว (BOT)", "module": "bot_tourism", "fn": "BotTourismSource().fetch"}, + {"key": "auto_credit/tourism", "label": "ยอดขายรถ (TradingEconomics)", "module": "auto_credit", "fn": "fetch_auto_credit"}, + {"key": "auto_npl", "label": "NPL รถยนต์ (BOT)", "module": "auto_npl", "fn": "fetch_auto_npl"}, + {"key": "energy_thai", "label": "โรงกลั่น TOP", "module": "energy_thai", "fn": "fetch_energy_thai"}, + {"key": "macro_thai", "label": "ภาพรวมประเทศไทย (BOT)", "module": "macro_thai", "fn": "fetch_macro_thai"}, +] + + +class AppDataScheduler: + """Runs periodic refresh of the real data collectors inside the app.""" + + def __init__(self, cache, data_root: Path, interval_seconds: int = 3600): + self.cache = cache + self.data_root = data_root + self.interval = max(30, int(interval_seconds)) # never faster than 30s + self._stop = threading.Event() + self._thread: Optional[threading.Thread] = None + self._snap_dir = data_root / "scheduler" + self._snap_dir.mkdir(parents=True, exist_ok=True) + + # -- public lifecycle --------------------------------------------------- + def start(self) -> None: + if self._thread and self._thread.is_alive(): + return + self._stop.clear() + self._thread = threading.Thread(target=self._loop, name="set50-refresh", daemon=True) + self._thread.start() + log.info("set50 data scheduler started (interval=%ss)", self.interval) + + def stop(self) -> None: + self._stop.set() + + # -- internals ---------------------------------------------------------- + def _loop(self) -> None: + # run an immediate refresh shortly after boot, then on interval + while not self._stop.wait(self.interval): + try: + self.refresh_all() + except Exception: # noqa: BLE001 — loop must survive individual failures + log.exception("set50 scheduler refresh_all failed (will retry)") + + def refresh_all(self) -> list[dict]: + """Run every collector, warm the daily cache, and snapshot the state.""" + results: list[dict] = [] + for job in _REFRESH_JOBS: + results.append(self._run_job(job)) + self._write_marker(results) + return results + + def _run_job(self, job: dict) -> dict: + key = job["key"] + label = job["label"] + try: + value = self.cache.fetch_or_stale(key, self._make_fetcher(job)) + return {"key": key, "label": label, "ok": True, "at": self._now()} + except Exception as exc: # noqa: BLE001 + log.warning("set50 refresh failed for %s: %s", key, exc) + return {"key": key, "label": label, "ok": False, "error": str(exc), "at": self._now()} + + def _make_fetcher(self, job: dict) -> Callable[[], dict]: + module_name = job["module"] + fn = job["fn"] + + def fetcher() -> dict: + mod = __import__(f"app.{module_name}", fromlist=["*"]) + obj = mod + # Support "ClassName().method" (instantiate then call) and plain + # "func_name" and static object attribute chains. + parts = fn.split(".") + for i, part in enumerate(parts): + is_instantiate = part.endswith("()") + attr = part[:-2] if is_instantiate else part + obj = getattr(obj, attr) + if is_instantiate: + obj = obj() # construct instance + if callable(obj): + value = obj() + else: + value = obj + if hasattr(value, "to_dict"): + value = value.to_dict() + return value or {} + return fetcher + + def _write_marker(self, results: list[dict]) -> None: + import json + import datetime as _dt + ok = [r for r in results if r["ok"]] + marker = { + "at": self._now(), + "ok": len(ok), + "total": len(results), + "sources": results, + } + path = self._snap_dir / "last_refresh.json" + try: + path.write_text(json.dumps(marker, ensure_ascii=False, indent=2), encoding="utf-8") + except OSError: + log.exception("could not write scheduler marker") + + @staticmethod + def _now() -> str: + import datetime as _dt + return _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="seconds") diff --git a/backend/tests/test_scheduler.py b/backend/tests/test_scheduler.py new file mode 100644 index 0000000..761d7ef --- /dev/null +++ b/backend/tests/test_scheduler.py @@ -0,0 +1,51 @@ +"""Tests for the in-app data scheduler.""" + +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from app.scheduler import AppDataScheduler + + +class _FakeCache: + def __init__(self): self.calls = {} + def fetch_or_stale(self, key, fetcher): + self.calls[key] = self.calls.get(key, 0) + 1 + return fetcher() + + +class SchedulerTest(unittest.TestCase): + def test_refresh_all_calls_collectors(self): + cache = _FakeCache() + snap_dir = Path(tempfile.mkdtemp()) + sched = AppDataScheduler(cache, snap_dir, interval_seconds=99999) + # job fetchers import app.* modules; patch their fetch fns to return dicts + def fake(modname, fn): + return lambda: {"ok": True} + with patch("app.auto_credit.fetch_auto_credit", return_value=type("F", (), {"to_dict": lambda self: {"n": 1}})()), \ + patch("app.auto_npl.fetch_auto_npl", return_value=type("F", (), {"to_dict": lambda self: {"n": 2}})()), \ + patch("app.energy_thai.fetch_energy_thai", return_value=type("F", (), {"to_dict": lambda self: {"n": 3}})()), \ + patch("app.macro_thai.fetch_macro_thai", return_value=type("F", (), {"to_dict": lambda self: {"n": 4}})()): + res = sched.refresh_all() + # tourism job is bot_tourism.BotTourismSource().fetch — not trivially patched; allow it to fail gracefully + self.assertGreaterEqual(len(res), 4) + self.assertTrue(any(r["ok"] for r in res)) + + def test_marker_written(self): + cache = _FakeCache() + snap_dir = Path(tempfile.mkdtemp()) + sched = AppDataScheduler(cache, snap_dir, interval_seconds=99999) + with patch("app.macro_thai.fetch_macro_thai", return_value=type("F", (), {"to_dict": lambda self: {"n": 1}})()): + sched.refresh_all() + marker = snap_dir / "scheduler" / "last_refresh.json" + self.assertTrue(marker.exists()) + import json + data = json.loads(marker.read_text()) + self.assertIn("at", data) + + +if __name__ == "__main__": + unittest.main()