scheduler.py

# StockFlow — 更新時間：2026-07-24 15:08
"""
StockFlow — 獨立排程腳本 v3.2（多店管理 Phase 3）
新增：日報/週報改成「逐店」計算與發送 —— 每個 LINE 群組依照自己綁定的店家，
      只收到該店的庫存資料，不同店之間完全不會混在一起
沿用：schema 遷移與 main.py 保持同步（stores / store_id 過濾 / 一次性遷移保護）
沿用：LINE Messaging API 推播通知、90 天舊資料清除
"""

import os, time, logging
from datetime import datetime, timedelta, date, timezone

import psycopg2
import psycopg2.extras
import requests
from dotenv import load_dotenv

load_dotenv()

DATABASE_URL     = os.getenv("DATABASE_URL")
LINE_TOKEN       = os.getenv("LINE_CHANNEL_ACCESS_TOKEN", "")
CHECK_INTERVAL   = int(os.getenv("SCHEDULER_INTERVAL_SEC", 60))

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
log = logging.getLogger("stockflow.scheduler")


# ══════════════════════════════════════════════════════════════
# 資料庫
# ══════════════════════════════════════════════════════════════
def get_conn():
    return psycopg2.connect(DATABASE_URL, cursor_factory=psycopg2.extras.RealDictCursor)


# ══════════════════════════════════════════════════════════════
# 清除舊資料（跨店，依日期即可，不需要 store 過濾）
# ══════════════════════════════════════════════════════════════
def purge_old_records():
    TW     = timezone(timedelta(hours=8))
    cutoff = (datetime.now(TW).date() - timedelta(days=90)).isoformat()
    try:
        conn = get_conn(); cur = conn.cursor()
        cur.execute(
            "DELETE FROM inventory_records WHERE recorded_date < %s RETURNING id",
            (cutoff,)
        )
        deleted = cur.rowcount
        conn.commit(); cur.close(); conn.close()
        if deleted:
            log.info("已清除 %d 筆超過 90 天的舊盤點記錄", deleted)
    except Exception as e:
        log.error("清除舊資料失敗：%s", e)


# ══════════════════════════════════════════════════════════════
# 組裝報表（★ Phase 3：依 store_id 過濾，逐店獨立計算）
# ══════════════════════════════════════════════════════════════
def build_report(store_id: int):
    try:
        conn = get_conn(); cur = conn.cursor()

        cur.execute("""
            SELECT DISTINCT ON (p.id)
                   r.id AS record_id,
                   p.id AS product_id,
                   p.name AS product_name,
                   p.unit, p.unit_price::float,
                   p.min_threshold AS thresh,
                   r.recorded_date::text,
                   r.order_date::text,
                   r.staff_name
            FROM products p
            LEFT JOIN inventory_records r ON r.product_id = p.id AND r.store_id = %s
            WHERE p.active = TRUE AND p.store_id = %s
            ORDER BY p.id, r.recorded_date DESC NULLS LAST
        """, (store_id, store_id))
        records = cur.fetchall()

        if not records:
            cur.close(); conn.close()
            return None

        from collections import Counter
        dates = [r["recorded_date"] for r in records if r.get("recorded_date")]
        latest = Counter(dates).most_common(1)[0][0] if dates else datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%d")

        items = []
        for rec in records:
            if not rec["record_id"]:
                items.append({
                    "name": rec["product_name"], "pid": rec["product_id"],
                    "unit": rec["unit"], "price": rec["unit_price"],
                    "total": 0, "thresh": rec["thresh"] or 0,
                    "diff": -(rec["thresh"] or 0), "cost": 0,
                    "staff": "未設定", "order": "未設定", "batches": [],
                })
                continue
            cur.execute("""
                SELECT expiry_date::text, quantity
                FROM inventory_batches WHERE record_id=%s ORDER BY expiry_date
            """, (rec["record_id"],))
            batches   = cur.fetchall()
            total_qty = sum(b["quantity"] for b in batches)
            thresh    = rec["thresh"] or 0
            items.append({
                "name": rec["product_name"], "pid": rec["product_id"],
                "unit": rec["unit"], "price": rec["unit_price"],
                "total": total_qty, "thresh": thresh,
                "diff": total_qty - thresh, "cost": round(total_qty * rec["unit_price"], 0),
                "staff": rec["staff_name"] or "未設定",
                "order": rec["order_date"] or "未設定", "batches": batches,
            })

        cur.close(); conn.close()

        TW      = timezone(timedelta(hours=8))
        now     = datetime.now(TW)
        today   = now.date()
        now_str = now.strftime("%Y-%m-%d %H:%M")

        day7_start  = today - timedelta(days=7)
        day21_start = today - timedelta(days=21)
        conn2 = get_conn(); cur2 = conn2.cursor()
        cur2.execute("""
            SELECT p.id AS pid,
                   COALESCE(SUM(c.quantity) FILTER (WHERE c.recorded_date BETWEEN %s AND %s), 0) AS qty_7d,
                   COALESCE(SUM(c.quantity) FILTER (WHERE c.recorded_date BETWEEN %s AND %s), 0) AS qty_21d
            FROM products p
            LEFT JOIN consumption_records c ON c.product_id = p.id
            WHERE p.active = TRUE AND p.store_id = %s GROUP BY p.id
        """, (day7_start, today, day21_start, today, store_id))
        cons_map = {row["pid"]: row for row in cur2.fetchall()}
        cur2.close(); conn2.close()

        import math
        def calc_suggest(item):
            if item["thresh"] == 0:
                return 0
            c   = cons_map.get(item["pid"], {})
            q7  = float(c.get("qty_7d", 0))
            q21 = float(c.get("qty_21d", 0))
            if q21 > 0:
                target = math.ceil((q7 / 7) * 3 + item["thresh"])
            else:
                target = item["thresh"]
            return max(0, target - item["total"])

        exp_threshold = today + timedelta(days=2)
        expiring_soon = []
        for item in items:
            for b in item["batches"]:
                d = b["expiry_date"]
                if d and today.isoformat() <= d <= exp_threshold.isoformat():
                    expiring_soon.append({"name": item["name"], "unit": item["unit"], "expiry": d, "qty": b["quantity"]})
        expiring_soon.sort(key=lambda x: x["expiry"])

        for item in items:
            item["suggest"] = calc_suggest(item)
            item["need"]    = item["suggest"] > 0

        need_items   = [i for i in items if i["need"]]
        has_expiring = bool(expiring_soon)

        lines = [
            f"📦 庫存日報 {latest}",
            f"發送時間：{now_str}",
            "══════════════════",
        ]

        if need_items:
            lines.append("⚠️ 以下品項需要叫貨：")
            for item in need_items:
                c   = cons_map.get(item["pid"], {})
                q7  = float(c.get("qty_7d", 0))
                q21 = float(c.get("qty_21d", 0))
                hint = f"（近7天日均 {q7/7:.1f} / 近21天每3天均 {q21/7:.1f}）" if q21 > 0 else ""
                lines.append(
                    f"⚠️【{item['name']}】\n"
                    f"  現有：{item['total']} {item['unit']}（安全庫存 {item['thresh']}）\n"
                    f"  建議叫貨：{item['suggest']} {item['unit']} {hint}"
                )
        else:
            lines.append("✅ 所有品項庫存充足，無需叫貨。")

        if has_expiring:
            lines.append("══════════════════")
            lines.append("🕐 即將過期（2天內）：")
            for e in expiring_soon:
                lines.append(f"  ・{e['name']}　效期 {e['expiry']}　{e['qty']} {e['unit']}")

        lines.append("══════════════════")
        return "\n\n".join(lines) if need_items else "\n".join(lines)

    except Exception as e:
        log.error("組裝每日報表失敗（store_id=%s）：%s", store_id, e)
        return None


def build_weekly_report(store_id: int):
    """
    每週二消耗趨勢週報（依店過濾）。
    只顯示「本週消耗 ≥ 上週」（紅色趨勢）的品項，以及建議叫貨量。
    """
    try:
        TW    = timezone(timedelta(hours=8))
        today = datetime.now(TW).date()
        weekday    = today.weekday()
        week_start = today - timedelta(days=weekday)
        lw_start   = week_start - timedelta(days=7)
        lw_end     = week_start - timedelta(days=1)
        now_str    = datetime.now(TW).strftime("%Y-%m-%d %H:%M")

        conn = get_conn(); cur = conn.cursor()

        cur.execute("""
            SELECT
                p.id, p.name, p.unit, p.unit_price::float,
                COALESCE(SUM(c.quantity) FILTER (
                    WHERE c.recorded_date BETWEEN %s AND %s
                ), 0) AS this_week,
                COALESCE(SUM(c.quantity) FILTER (
                    WHERE c.recorded_date BETWEEN %s AND %s
                ), 0) AS last_week
            FROM products p
            LEFT JOIN consumption_records c ON c.product_id = p.id
            WHERE p.active = TRUE AND p.store_id = %s
            GROUP BY p.id, p.name, p.unit, p.unit_price
            ORDER BY p.id
        """, (week_start, today, lw_start, lw_end, store_id))
        consumption = cur.fetchall()

        cur.execute("""
            SELECT DISTINCT ON (p.id)
                   p.id AS product_id,
                   p.min_threshold,
                   COALESCE((
                       SELECT SUM(b.quantity)
                       FROM inventory_records r2
                       JOIN inventory_batches b ON b.record_id = r2.id
                       WHERE r2.product_id = p.id AND r2.store_id = %s
                       ORDER BY r2.recorded_date DESC
                       LIMIT 1
                   ), 0) AS current_qty
            FROM products p
            WHERE p.active = TRUE AND p.store_id = %s
            ORDER BY p.id
        """, (store_id, store_id))
        inv_map = {row["product_id"]: row for row in cur.fetchall()}
        cur.close(); conn.close()

        red_items = [c for c in consumption if int(c["this_week"]) > int(c["last_week"]) and int(c["this_week"]) > 0]

        if not red_items:
            return (
                f"📊 消耗週報 {today}\n"
                f"══════════════════\n"
                f"✅ 本週所有品項消耗量正常，無上升趨勢。"
            )

        line_parts = [
            f"📊 消耗週報 {today}",
            f"（每週二推播）",
            f"══════════════════",
            f"🔺 以下品項本週消耗上升：",
        ]
        for c in red_items:
            inv      = inv_map.get(c["id"])
            cur_qty  = int(inv["current_qty"])  if inv else 0
            thresh   = int(inv["min_threshold"]) if inv else 0
            suggest  = max(0, thresh - cur_qty)
            tw       = int(c["this_week"])
            lw       = int(c["last_week"])
            block    = [
                f"🔺【{c['name']}】",
                f"  本週 {tw} {c['unit']}（上週 {lw}）↑",
                f"  目前庫存：{cur_qty} {c['unit']}",
            ]
            if suggest > 0:
                block.append(f"  ⚠️ 建議叫貨：{suggest} {c['unit']}")
            line_parts.append("\n".join(block))

        line_parts.append("══════════════════")
        line_parts.append(f"發送時間：{now_str}")
        return "\n\n".join(line_parts)

    except Exception as e:
        log.error("組裝週報失敗（store_id=%s）：%s", store_id, e)
        return None


# ══════════════════════════════════════════════════════════════
# LINE Messaging API 推播
# ══════════════════════════════════════════════════════════════
def send_line(to_id: str, text: str) -> bool:
    if not LINE_TOKEN:
        log.warning("LINE_CHANNEL_ACCESS_TOKEN 未設定，跳過 LINE 通知")
        return False
    if not to_id:
        log.warning("LINE 接收對象 ID 未設定，跳過 LINE 通知")
        return False

    try:
        resp = requests.post(
            "https://api.line.me/v2/bot/message/push",
            headers={
                "Authorization": f"Bearer {LINE_TOKEN}",
                "Content-Type":  "application/json",
            },
            json={
                "to": to_id,
                "messages": [{"type": "text", "text": text}],
            },
            timeout=15,
        )
        if resp.status_code == 200:
            log.info("LINE 推播成功 → %s", to_id)
            return True
        log.error("LINE API 錯誤 %d：%s", resp.status_code, resp.text[:200])
        return False
    except Exception as e:
        log.error("LINE 連線異常：%s", e)
        return False


# ══════════════════════════════════════════════════════════════
# 主排程判斷（★ Phase 3：逐店計算報表，同店多群組共用一份算好的內容）
# ══════════════════════════════════════════════════════════════
def check_and_send():
    """
    每次輪詢呼叫一次。
    讀 line_groups 表，只處理「已綁定店家」（store_id 不為 NULL）的群組，
    每個群組依照自己的排程時間，收到自己綁定店家的日報/週報。
    """
    TW           = timezone(timedelta(hours=8))
    now          = datetime.now(TW)
    today_str    = now.strftime("%Y-%m-%d")
    current_hhmm = now.strftime("%H:%M")
    is_tuesday   = now.weekday() == 1  # 0=週一, 1=週二

    try:
        conn = get_conn(); cur = conn.cursor()
        cur.execute("""
            SELECT group_id, store_id,
                   schedule_time::text,  active,         last_sent_date::text,
                   weekly_time::text,    weekly_active,  last_weekly_date::text
            FROM line_groups
            WHERE store_id IS NOT NULL
        """)
        groups = cur.fetchall()
        cur.close(); conn.close()

        if not groups:
            return

        daily_reports  = {}   # store_id -> 已組好的日報文字（同店多群組共用，避免重複查詢）
        weekly_reports = {}   # store_id -> 已組好的週報文字

        for group in groups:
            gid      = group["group_id"]
            store_id = group["store_id"]

            # ── 每日日報 ─────────────────────────────────────────
            if group["active"]:
                sched_hhmm = group["schedule_time"][:5]
                last_sent  = group["last_sent_date"] or ""
                if current_hhmm >= sched_hhmm and last_sent != today_str:
                    if store_id not in daily_reports:
                        daily_reports[store_id] = build_report(store_id)
                    report = daily_reports[store_id]
                    if report:
                        ok = send_line(gid, report)
                        if ok:
                            conn = get_conn(); cur = conn.cursor()
                            cur.execute(
                                "UPDATE line_groups SET last_sent_date=%s, updated_at=NOW() WHERE group_id=%s",
                                (today_str, gid)
                            )
                            conn.commit(); cur.close(); conn.close()

            # ── 每週二週報 ───────────────────────────────────────
            if is_tuesday and group.get("weekly_active", True):
                weekly_hhmm  = (group.get("weekly_time") or "09:00")[:5]
                last_weekly  = group.get("last_weekly_date") or ""
                if current_hhmm >= weekly_hhmm and last_weekly != today_str:
                    if store_id not in weekly_reports:
                        weekly_reports[store_id] = build_weekly_report(store_id)
                    report = weekly_reports[store_id]
                    if report:
                        ok = send_line(gid, report)
                        if ok:
                            conn = get_conn(); cur = conn.cursor()
                            cur.execute(
                                "UPDATE line_groups SET last_weekly_date=%s, updated_at=NOW() WHERE group_id=%s",
                                (today_str, gid)
                            )
                            conn.commit(); cur.close(); conn.close()

    except Exception as e:
        log.error("LINE 群組排程異常：%s", e)


# ══════════════════════════════════════════════════════════════
# 資料表初始化（與 main.py init_db() 保持同步）
# ══════════════════════════════════════════════════════════════
def ensure_schema():
    try:
        conn = get_conn(); cur = conn.cursor()

        cur.execute("SELECT pg_advisory_lock(123456789)")

        cur.execute("""
            CREATE TABLE IF NOT EXISTS schema_migrations (
                name       TEXT PRIMARY KEY,
                applied_at TIMESTAMPTZ DEFAULT NOW()
            )
        """)

        # ── stores ──────────────────────────────────────────────
        cur.execute("""
            CREATE TABLE IF NOT EXISTS stores (
                id         SERIAL PRIMARY KEY,
                name       TEXT NOT NULL,
                code       TEXT UNIQUE NOT NULL,
                active     BOOLEAN NOT NULL DEFAULT TRUE,
                created_at TIMESTAMPTZ DEFAULT NOW()
            )
        """)
        cur.execute("SELECT COUNT(*) AS n FROM stores")
        if cur.fetchone()["n"] == 0:
            cur.execute(
                "INSERT INTO stores (name, code) VALUES (%s,%s) ON CONFLICT DO NOTHING",
                ("預設店", "DEFAULT")
            )
        cur.execute("SELECT id FROM stores ORDER BY id ASC LIMIT 1")
        default_store_id = cur.fetchone()["id"]

        # ── users ───────────────────────────────────────────────
        cur.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id            SERIAL PRIMARY KEY,
                username      TEXT UNIQUE NOT NULL,
                password_hash TEXT NOT NULL,
                role          TEXT NOT NULL DEFAULT 'staff'
                                  CHECK (role IN ('superadmin','admin','staff')),
                store_id      INTEGER REFERENCES stores(id) ON DELETE SET NULL,
                created_at    TIMESTAMPTZ DEFAULT NOW()
            )
        """)
        cur.execute("ALTER TABLE users ADD COLUMN IF NOT EXISTS store_id INTEGER REFERENCES stores(id) ON DELETE SET NULL")
        cur.execute("""
            DO $$
            BEGIN
                IF EXISTS (
                    SELECT 1 FROM pg_constraint
                    WHERE conname = 'users_role_check' AND conrelid = 'users'::regclass
                ) THEN
                    ALTER TABLE users DROP CONSTRAINT users_role_check;
                END IF;
                ALTER TABLE users ADD CONSTRAINT users_role_check
                    CHECK (role IN ('superadmin','admin','staff'));
            END$$
        """)
        cur.execute("SELECT 1 FROM schema_migrations WHERE name='v3_admin_to_superadmin_once'")
        if not cur.fetchone():
            cur.execute("UPDATE users SET role='superadmin', store_id=NULL WHERE role='admin'")
            cur.execute("INSERT INTO schema_migrations (name) VALUES ('v3_admin_to_superadmin_once')")
        cur.execute("UPDATE users SET store_id=%s WHERE role='staff' AND store_id IS NULL", (default_store_id,))

        # ── products ──────────────────────────────────────────
        cur.execute("""
            CREATE TABLE IF NOT EXISTS products (
                id                 SERIAL PRIMARY KEY,
                name               TEXT NOT NULL,
                unit               TEXT NOT NULL DEFAULT '瓶',
                unit_price         NUMERIC(10,2) NOT NULL DEFAULT 0,
                active             BOOLEAN NOT NULL DEFAULT TRUE,
                min_threshold      INTEGER NOT NULL DEFAULT 0,
                default_shelf_days INTEGER,
                category           TEXT,
                created_at         TIMESTAMPTZ DEFAULT NOW()
            )
        """)
        cur.execute("ALTER TABLE products ADD COLUMN IF NOT EXISTS min_threshold      INTEGER NOT NULL DEFAULT 0")
        cur.execute("ALTER TABLE products ADD COLUMN IF NOT EXISTS default_shelf_days INTEGER")
        cur.execute("ALTER TABLE products ADD COLUMN IF NOT EXISTS category           TEXT")
        cur.execute("ALTER TABLE products ADD COLUMN IF NOT EXISTS store_id INTEGER REFERENCES stores(id) ON DELETE SET NULL")
        cur.execute("UPDATE products SET store_id=%s WHERE store_id IS NULL", (default_store_id,))
        cur.execute("""
            DO $$
            BEGIN
                IF EXISTS (
                    SELECT 1 FROM pg_constraint
                    WHERE conname = 'products_name_key' AND conrelid = 'products'::regclass
                ) THEN
                    ALTER TABLE products DROP CONSTRAINT products_name_key;
                END IF;
                IF NOT EXISTS (
                    SELECT 1 FROM pg_constraint
                    WHERE conname = 'products_store_id_name_key' AND conrelid = 'products'::regclass
                ) THEN
                    ALTER TABLE products ADD CONSTRAINT products_store_id_name_key UNIQUE (store_id, name);
                END IF;
            END$$
        """)

        cur.execute("""
            CREATE TABLE IF NOT EXISTS inventory_records (
                id             SERIAL PRIMARY KEY,
                product_id     INTEGER REFERENCES products(id) ON DELETE SET NULL,
                recorded_date  DATE NOT NULL,
                min_threshold  INTEGER NOT NULL DEFAULT 0,
                order_date     DATE,
                staff_name     TEXT,
                updated_at     TIMESTAMPTZ DEFAULT NOW()
            )
        """)
        cur.execute("ALTER TABLE inventory_records ADD COLUMN IF NOT EXISTS product_id INTEGER REFERENCES products(id) ON DELETE SET NULL")
        cur.execute("ALTER TABLE inventory_records ADD COLUMN IF NOT EXISTS store_id INTEGER REFERENCES stores(id) ON DELETE SET NULL")
        cur.execute("UPDATE inventory_records SET store_id=%s WHERE store_id IS NULL", (default_store_id,))
        cur.execute("""
            DO $$
            BEGIN
                IF EXISTS (
                    SELECT 1 FROM pg_constraint
                    WHERE conname = 'inventory_records_recorded_date_key'
                      AND conrelid = 'inventory_records'::regclass
                ) THEN
                    ALTER TABLE inventory_records DROP CONSTRAINT inventory_records_recorded_date_key;
                END IF;
                IF NOT EXISTS (
                    SELECT 1 FROM pg_constraint
                    WHERE conname = 'inventory_records_product_id_recorded_date_key'
                      AND conrelid = 'inventory_records'::regclass
                ) THEN
                    ALTER TABLE inventory_records
                    ADD CONSTRAINT inventory_records_product_id_recorded_date_key
                    UNIQUE (product_id, recorded_date);
                END IF;
            END$$
        """)

        cur.execute("""
            CREATE TABLE IF NOT EXISTS inventory_batches (
                id          SERIAL PRIMARY KEY,
                record_id   INTEGER NOT NULL
                                REFERENCES inventory_records(id) ON DELETE CASCADE,
                expiry_date DATE NOT NULL,
                quantity    INTEGER NOT NULL DEFAULT 0
            )
        """)

        cur.execute("""
            CREATE TABLE IF NOT EXISTS staff_settings (
                id   SERIAL PRIMARY KEY,
                name TEXT NOT NULL
            )
        """)
        cur.execute("ALTER TABLE staff_settings ADD COLUMN IF NOT EXISTS store_id INTEGER REFERENCES stores(id) ON DELETE SET NULL")
        cur.execute("UPDATE staff_settings SET store_id=%s WHERE store_id IS NULL", (default_store_id,))
        cur.execute("""
            DO $$
            BEGIN
                IF EXISTS (
                    SELECT 1 FROM pg_constraint
                    WHERE conname = 'staff_settings_name_key' AND conrelid = 'staff_settings'::regclass
                ) THEN
                    ALTER TABLE staff_settings DROP CONSTRAINT staff_settings_name_key;
                END IF;
                IF NOT EXISTS (
                    SELECT 1 FROM pg_constraint
                    WHERE conname = 'staff_settings_store_id_name_key' AND conrelid = 'staff_settings'::regclass
                ) THEN
                    ALTER TABLE staff_settings ADD CONSTRAINT staff_settings_store_id_name_key UNIQUE (store_id, name);
                END IF;
            END$$
        """)

        cur.execute("""
            CREATE TABLE IF NOT EXISTS schedules (
                id               SERIAL PRIMARY KEY,
                schedule_time    TIME NOT NULL DEFAULT '21:00',
                active_receiver  TEXT,
                last_sent_date   DATE,
                updated_at       TIMESTAMPTZ DEFAULT NOW()
            )
        """)

        # ── line_groups（★ Phase 3：store_id 用來記錄綁定的店家）──
        cur.execute("""
            CREATE TABLE IF NOT EXISTS line_groups (
                id               SERIAL PRIMARY KEY,
                group_id         TEXT UNIQUE NOT NULL,
                group_name       TEXT,
                schedule_time    TIME NOT NULL DEFAULT '21:00',
                active           BOOLEAN NOT NULL DEFAULT TRUE,
                last_sent_date   DATE,
                weekly_time      TIME NOT NULL DEFAULT '09:00',
                weekly_active    BOOLEAN NOT NULL DEFAULT TRUE,
                last_weekly_date DATE,
                joined_at        TIMESTAMPTZ DEFAULT NOW(),
                updated_at       TIMESTAMPTZ DEFAULT NOW()
            )
        """)
        cur.execute("ALTER TABLE line_groups ADD COLUMN IF NOT EXISTS weekly_time      TIME    NOT NULL DEFAULT '09:00'")
        cur.execute("ALTER TABLE line_groups ADD COLUMN IF NOT EXISTS weekly_active    BOOLEAN NOT NULL DEFAULT TRUE")
        cur.execute("ALTER TABLE line_groups ADD COLUMN IF NOT EXISTS last_weekly_date DATE")
        cur.execute("ALTER TABLE line_groups ADD COLUMN IF NOT EXISTS store_id INTEGER REFERENCES stores(id) ON DELETE SET NULL")
        # ★ 注意：line_groups 的 store_id 不做「自動回填預設店」！
        # 這是「群組要收哪家店的通知」的綁定狀態，必須由使用者在群組裡輸入「綁定 店代碼」才設定，
        # 不能像其他表一樣自動假設是預設店，否則會把所有既有群組錯誤地綁到同一家店。
        # Phase 1~2 遷移時如果已經誤把舊群組 backfill 過 store_id，這裡不會再動它，維持現狀。

        cur.execute("""
            CREATE TABLE IF NOT EXISTS consumption_records (
                id            SERIAL PRIMARY KEY,
                product_id    INTEGER NOT NULL REFERENCES products(id) ON DELETE CASCADE,
                recorded_date DATE    NOT NULL,
                quantity      INTEGER NOT NULL DEFAULT 0,
                staff_name    TEXT,
                note          TEXT,
                created_at    TIMESTAMPTZ DEFAULT NOW(),
                UNIQUE (product_id, recorded_date)
            )
        """)
        cur.execute("ALTER TABLE consumption_records ADD COLUMN IF NOT EXISTS store_id INTEGER REFERENCES stores(id) ON DELETE SET NULL")
        cur.execute("UPDATE consumption_records SET store_id=%s WHERE store_id IS NULL", (default_store_id,))

        cur.execute("""
            CREATE TABLE IF NOT EXISTS purchase_orders (
                id               SERIAL PRIMARY KEY,
                product_id       INTEGER NOT NULL REFERENCES products(id) ON DELETE CASCADE,
                product_name     TEXT NOT NULL,
                unit             TEXT NOT NULL,
                quantity         INTEGER NOT NULL,
                ordered_by       TEXT NOT NULL,
                ordered_date     DATE NOT NULL,
                arrived          BOOLEAN NOT NULL DEFAULT FALSE,
                arrived_date     DATE,
                arrived_by       TEXT,
                arrived_quantity INTEGER,
                note             TEXT,
                created_at       TIMESTAMPTZ DEFAULT NOW()
            )
        """)
        cur.execute("ALTER TABLE purchase_orders ADD COLUMN IF NOT EXISTS arrived_quantity INTEGER")
        cur.execute("ALTER TABLE purchase_orders ADD COLUMN IF NOT EXISTS store_id INTEGER REFERENCES stores(id) ON DELETE SET NULL")
        cur.execute("UPDATE purchase_orders SET store_id=%s WHERE store_id IS NULL", (default_store_id,))

        # ── product_categories ──────────────────────────────────
        cur.execute("""
            CREATE TABLE IF NOT EXISTS product_categories (
                id         SERIAL PRIMARY KEY,
                name       TEXT NOT NULL,
                created_at TIMESTAMPTZ DEFAULT NOW()
            )
        """)
        cur.execute("ALTER TABLE product_categories ADD COLUMN IF NOT EXISTS store_id INTEGER REFERENCES stores(id) ON DELETE SET NULL")
        cur.execute("UPDATE product_categories SET store_id=%s WHERE store_id IS NULL", (default_store_id,))
        cur.execute("""
            DO $$
            BEGIN
                IF EXISTS (
                    SELECT 1 FROM pg_constraint
                    WHERE conname = 'product_categories_name_key' AND conrelid = 'product_categories'::regclass
                ) THEN
                    ALTER TABLE product_categories DROP CONSTRAINT product_categories_name_key;
                END IF;
                IF NOT EXISTS (
                    SELECT 1 FROM pg_constraint
                    WHERE conname = 'product_categories_store_id_name_key' AND conrelid = 'product_categories'::regclass
                ) THEN
                    ALTER TABLE product_categories ADD CONSTRAINT product_categories_store_id_name_key UNIQUE (store_id, name);
                END IF;
            END$$
        """)

        cur.execute("SELECT COUNT(*) AS n FROM schedules")
        if cur.fetchone()["n"] == 0:
            cur.execute("INSERT INTO schedules (schedule_time) VALUES ('21:00')")

        # ── 清理孤立記錄 ──────────────────────────────────────
        cur.execute("""
            DELETE FROM inventory_batches WHERE record_id IN (
                SELECT id FROM inventory_records WHERE product_id IS NULL
            )
        """)
        cur.execute("DELETE FROM inventory_records WHERE product_id IS NULL")

        conn.commit(); cur.close(); conn.close()
        log.info("資料表結構確認完成")

    except Exception as e:
        log.error("資料表初始化失敗：%s", e)
        raise


# ══════════════════════════════════════════════════════════════
# 主迴圈
# ══════════════════════════════════════════════════════════════
def main():
    log.info("StockFlow Scheduler 啟動，輪詢間隔 %d 秒", CHECK_INTERVAL)

    for attempt in range(1, 11):
        try:
            ensure_schema()
            break
        except Exception:
            log.warning("資料表初始化第 %d 次失敗，5 秒後重試", attempt)
            time.sleep(5)
    else:
        log.error("連續失敗 10 次，程序退出")
        return

    last_purge_date = None
    while True:
        TW    = timezone(timedelta(hours=8))
        today = datetime.now(TW).date()
        if last_purge_date != today:
            purge_old_records()
            last_purge_date = today
        check_and_send()
        time.sleep(CHECK_INTERVAL)


if __name__ == "__main__":
    main()