diff --git a/backend/app/bot_regional.py b/backend/app/bot_regional.py new file mode 100644 index 0000000..02e0f67 --- /dev/null +++ b/backend/app/bot_regional.py @@ -0,0 +1,175 @@ +"""BOT regional consumption factor — BOT (Bank of Thailand) regional private consumption index. + +Source: https://app.bot.or.th/BTWS_STAT/statistics/ReportPage.aspx?reportID= +The Bank of Thailand publishes regional breakdowns of the private consumption +index per region (North, Northeast, South, Central/Bangkok). Each region has its +own reportID. Server-rendered HTML table: + + header row: months (e.g. "มิ.ย. 2569 p พ.ค. 2569 r ..." — p/r are preliminary/revision) + rows | 1 | ดัชนีการอุปโภคบริโภคภาคเอกชน | 100.3 | 98.4 | ... + | 2 | ดัชนีการใช้จ่ายสินค้าไม่คงทน | ... + | 3 | ดัชนีสินค้าอุปโภคบริโภคหมุนเวียนเร็ว | ... + +We parse the main private-consumption index row per region so the tourism/ +consumption theme can sense regional demand shifts. + +Region reportIDs (from BOT BTWS_STAT): + - North: 954 + (others to be confirmed by browsing the BOT statistical menu) +""" + +from __future__ import annotations + +import html +import re +from dataclasses import dataclass +from typing import Optional +from urllib.request import Request, urlopen + +_USER_AGENT = ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" +) +_BASE_URL = "https://app.bot.or.th/BTWS_STAT/statistics/ReportPage.aspx?reportID=" + +# Default mapping of BOT regions; extend as other reportIDs are confirmed. +REGION_REPORT_IDS: dict[str, int] = { + "north": 954, +} + + +class BotRegionalError(Exception): + """Raised when a BOT regional report cannot be fetched or parsed.""" + + +@dataclass(frozen=True) +class BotRegionalSnapshot: + region: str + report_id: int + # Monthly private consumption index: {(year, month_label): value} + # For simplicity store the latest 6 month values as an ordered list along + # with their labels. + consumption_index: Optional[float] = None + nondurable_index: Optional[float] = None + months: Optional[list] = None # month labels in order (latest first) + consumption_series: Optional[list] = None # values parallel to months + source: str = "bot" + as_of: str = "" + + def to_dict(self) -> dict: + return { + "source": self.source, + "region": self.region, + "report_id": self.report_id, + "as_of": self.as_of, + "consumption_index": self.consumption_index, + "nondurable_index": self.nondurable_index, + "months": self.months or [], + "consumption_series": self.consumption_series or [], + } + + +def _fetch(report_id: int, timeout: float = 30.0) -> str: + url = f"{_BASE_URL}{report_id}" + req = Request(url, headers={"User-Agent": _USER_AGENT, "Accept": "text/html"}) + try: + with urlopen(req, timeout=timeout) as resp: + raw = resp.read() + except Exception as exc: + raise BotRegionalError(f"failed to fetch BOT report {report_id}: {exc}") from exc + try: + return raw.decode("utf-8") + except UnicodeDecodeError: + return raw.decode("latin-1", "ignore") + + +def _clean(event: str) -> str: + return html.unescape(re.sub(r"<[^>]+>", "", event)).strip() + + +def parse_bot_regional_html(html_text: str) -> dict: + """Parse the BOT regional consumption table into row label -> list of values. + + Returns {'title': ..., 'columns': [month labels], 'rows': {label: [values]}}. + """ + tables = re.findall(r"]*>(.*?)", html_text, re.S) + if not tables: + raise BotRegionalError("no tables found in BOT page") + + title = "" + columns: list[str] = [] + rows: dict[str, list[Optional[float]]] = {} + + m = re.search(r"(.*?)", html_text, re.S) + if m: + title = _clean(m.group(1)) + + for table in tables: + trs = re.findall(r"]*>(.*?)", table, re.S) + for tr in trs: + cells = [_clean(td) for td in re.findall(r"]*>(.*?)", tr, re.S)] + cells = [c for c in cells if c] + if not cells: + continue + # header row: month labels (e.g. 'มิ.ย. 2569 p') — keep only cells + # whose text carries a Thai month + Buddhist year (256X). + if "256" in " ".join(cells) and not re.search(r"\d\.\d", " ".join(cells)): + # keep only cells carrying a 4-digit BE year (256X) — drops + # column titles like 'ลำดับ' / 'รายการ'. + month_cells = [c for c in cells if re.search(r"25\d\d", c)] + if month_cells: + columns = month_cells + continue + # data row: [no, label, val1, val2, ...] — label is cells[1], values + # start at cells[2]. Some rows may omit the no/label (merged), in + # which case label stays from a previous row. + if len(cells) >= 2: + # Try to identify (no, label, values...) by checking cells[1] + label = cells[1].strip() + val_start = 2 + if not label or label in ("-", "N/A"): + # fallback: maybe [label, val1, val2...] with no 'no' column + label = cells[0].strip() + val_start = 1 + vals = [_to_float(c) for c in cells[val_start:]] + if any(v is not None for v in vals): + rows[label or f"row_{len(rows)}"] = vals + + if not rows: + raise BotRegionalError("no numeric rows found in BOT regional page") + return {"title": title, "columns": columns, "rows": rows} + + +def _to_float(text: str) -> Optional[float]: + text = text.replace(",", "").strip() + if not text or text in ("-", "N/A", ""): + return None + try: + return float(text) + except ValueError: + return None + + +def fetch_region(region: str = "north", timeout: float = 30.0) -> BotRegionalSnapshot: + region = region.strip().lower() + report_id = REGION_REPORT_IDS.get(region) + if report_id is None: + raise BotRegionalError(f"no reportID configured for region: {region}") + html_text = _fetch(report_id, timeout=timeout) + parsed = parse_bot_regional_html(html_text) + rows = parsed["rows"] + cols = parsed["columns"] + + # The main private consumption index is usually the first numeric row. + main_key = next((k for k in rows if "ดัชนีการอุปโภค" in k or "ดัชนี" in k), None) + nondurable_key = next((k for k in rows if "ไม่คงทน" in k), None) + + return BotRegionalSnapshot( + region=region, + report_id=report_id, + consumption_index=rows[main_key][0] if main_key and rows.get(main_key) else None, + nondurable_index=rows[nondurable_key][0] if nondurable_key and rows.get(nondurable_key) else None, + months=cols, + consumption_series=rows.get(main_key) if main_key else None, + as_of=cols[0] if cols else "", + ) diff --git a/backend/scripts/collect_bot_regional.py b/backend/scripts/collect_bot_regional.py new file mode 100644 index 0000000..fb2040d --- /dev/null +++ b/backend/scripts/collect_bot_regional.py @@ -0,0 +1,31 @@ +"""Collect BOT regional private-consumption index for a region to a JSON snapshot.""" + +from __future__ import annotations + +import argparse +import json +from datetime import datetime, timezone + +from app import bot_regional + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--region", default="north", + help="BOT region (reportID key), default 'north'") + parser.add_argument("--timeout", type=float, default=30.0) + args = parser.parse_args() + + snap = bot_regional.fetch_region(args.region, timeout=args.timeout) + payload = { + "source": "bot", + "retrieved_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + "factor": "consumption_regional", + "data": snap.to_dict(), + } + print(json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/tests/test_bot_regional.py b/backend/tests/test_bot_regional.py new file mode 100644 index 0000000..1f251ec --- /dev/null +++ b/backend/tests/test_bot_regional.py @@ -0,0 +1,59 @@ +"""Tests for the BOT regional consumption parser.""" + +from __future__ import annotations + +import unittest + +from app import bot_regional + + +def _bot_html() -> str: + return """ + RG_NR_042_S3 ดัชนีและเครื่องชี้การอุปโภคบริโภคภาคเอกชนของภาคเหนือ + + + + + + +
ลำดับรายการมิ.ย. 2569 pพ.ค. 2569 rเม.ย. 2569 r
1ดัชนีการอุปโภคบริโภคภาคเอกชน 1/100.398.4101.2
2ดัชนีการใช้จ่ายสินค้าไม่คงทน 2/98.195.8100.7
3ดัชนีสินค้าอุปโภคบริโภคหมุนเวียนเร็ว93.990.296.4
+ + """ + + +class BotRegionalParseTest(unittest.TestCase): + def test_parses_consumption_index(self) -> None: + parsed = bot_regional.parse_bot_regional_html(_bot_html()) + self.assertEqual(parsed["columns"][0], "มิ.ย. 2569 p") + self.assertIn("ดัชนีการอุปโภค", str(parsed["rows"].keys())) + # consumption index row value + key = next(k for k in parsed["rows"] if "ดัชนีการอุปโภค" in k) + self.assertEqual(parsed["rows"][key][0], 100.3) + + def test_fetch_region_builds_snapshot(self) -> None: + # Use a fake fetch to avoid network + actual_fetch = bot_regional._fetch + bot_regional._fetch = lambda rid, timeout=30: _bot_html() + try: + snap = bot_regional.fetch_region("north") + self.assertEqual(snap.region, "north") + self.assertEqual(snap.consumption_index, 100.3) + self.assertEqual(snap.nondurable_index, 98.1) + # series is parallel to months + self.assertIsNotNone(snap.months) + self.assertIsNotNone(snap.consumption_series) + self.assertEqual(len(snap.months), len(snap.consumption_series)) + finally: + bot_regional._fetch = actual_fetch + + def test_unknown_region_raises(self) -> None: + with self.assertRaises(bot_regional.BotRegionalError): + bot_regional.fetch_region("mars") + + def test_missing_table_raises(self) -> None: + with self.assertRaises(bot_regional.BotRegionalError): + bot_regional.parse_bot_regional_html("no data") + + +if __name__ == "__main__": + unittest.main()