# 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()