12รันซ้ำแล้วต้องได้ผลเท่าเดิม

# ~/airflow-lab/dags/stock_load.py — โหลดที่รันซ้ำได้: เขียนทับทั้งเดือนในธุรกรรมเดียว (ฟังก์ชันธรรมดา ไม่ import Airflow)
import json
import sqlite3
from pathlib import Path


def append(db, month, rows):
    """แบบบทที่ 8: ต่อท้ายทีละแถว — ทุกแถวที่เขียนแล้วอยู่ถาวรทันที"""
    for row in rows:
        db.execute("INSERT INTO stock VALUES (?, ?, ?, ?, ?)", [month, *row])
        db.commit()


def replace_month(db, month, rows):
    """ลบของเดือนนี้ทั้งหมดแล้วเขียนใหม่ทั้งเดือน ในธุรกรรมเดียว: สำเร็จทั้งก้อน หรือเหมือนไม่เคยเกิด"""
    with db:  # ← ออกจาก with ปกติ = commit · มี error กลางทาง = rollback ทั้งก้อน
        db.execute("DELETE FROM stock WHERE month = ?", [month])
        db.executemany("INSERT INTO stock VALUES (?, ?, ?, ?, ?)", ([month, *row] for row in rows))


def load_month(summary, db_path, month):
    """ขั้น load ของท่อ — ผลขึ้นกับเดือนกับไฟล์ของเดือนนั้นเท่านั้น รันกี่ครั้งก็ได้ตารางเดียวกัน"""
    rows = json.loads(Path(summary["path"]).read_text(encoding="utf-8"))
    db = sqlite3.connect(db_path)
    db.execute("CREATE TABLE IF NOT EXISTS stock (month, branch, code, qty, hidden)")
    replace_month(db, month, rows)
    return db.execute("SELECT COUNT(*) FROM stock WHERE month = ?", [month]).fetchone()[0]


if __name__ == "__main__":  # เทียบสองแบบโดยไม่มี Airflow: พังกลางการเขียน แล้วรันใหม่ รันซ้ำ
    rows = [["a", f"S{i:03d}", 3, False] for i in range(1, 100)]  # 99 แถว เท่าเดือนกรกฎาคม

    def cut_at(n):  # ฐานข้อมูลตัดการเชื่อมต่อหลังเขียนไปได้ n แถว
        for i, row in enumerate(rows):
            if i == n:
                raise ConnectionError("ฐานข้อมูลตัดการเชื่อมต่อ")
            yield row

    for write in (append, replace_month):
        db = sqlite3.connect(":memory:")
        db.execute("CREATE TABLE stock (month, branch, code, qty, hidden)")
        seen = []
        for label, data in [("พังกลางทาง", cut_at(50)), ("รันใหม่", rows), ("รันซ้ำ", rows), ("พังกลางทางอีก", cut_at(50))]:
            try:
                write(db, "2026-07", data)
            except ConnectionError:
                pass
            seen.append(f"{label} {db.execute('SELECT COUNT(*) FROM stock').fetchone()[0]} แถว")
        print(f"{write.__name__:<14}" + " → ".join(seen))

append        พังกลางทาง 50 แถว → รันใหม่ 149 แถว → รันซ้ำ 248 แถว → พังกลางทางอีก 298 แถว
replace_month พังกลางทาง 0 แถว → รันใหม่ 99 แถว → รันซ้ำ 99 แถว → พังกลางทางอีก 99 แถว

# ~/airflow-lab/dags/stock_monthly.py — บทที่ 12: load เขียนทับทั้งเดือน รันซ้ำแล้วได้ผลเท่าเดิม
from datetime import datetime, timedelta
from pathlib import Path

import stock_gate
import stock_load
import stock_steps as steps
from airflow.sdk import dag, task
from airflow.sdk.exceptions import AirflowFailException, AirflowSkipException

LAB = Path.home() / "airflow-lab"


@dag(
    schedule=None,
    start_date=datetime(2026, 5, 1),
    catchup=False,
    params={"month": "2026-07"},
    default_args={"retries": 2, "retry_delay": timedelta(seconds=20)},  # แล็บรอ 20 วินาที ของจริงหลายนาที
)
def stock_monthly():
    @task
    def arrived(**context):
        month = context["params"]["month"]
        files = sorted((LAB / "inbox").glob(f"stock_{month}_*.xlsx"))
        if not files:  # ยังไม่มีไฟล์ของเดือนนี้สักสาขา — ไม่มีอะไรให้ทำ ไม่ใช่ความผิดของใคร
            raise AirflowSkipException(f"ยังไม่มีไฟล์ของเดือน {month}")
        return len(files)

    @task
    def receive(branch, **context):
        month = context["params"]["month"]
        path = LAB / "inbox" / f"stock_{month}_{branch}.xlsx"
        if not path.exists():  # ไฟล์มาช้า — ชั่วคราว ปล่อยให้ retries ลองใหม่
            raise FileNotFoundError(f"ไฟล์สาขา {stock_gate.NAMES[branch]} เดือน {month} ยังมาไม่ถึง")
        rows, missing = steps.read_branch(path)
        staged = steps.stage(branch, rows, missing, LAB / "staging" / month / f"{branch}.json")
        return {**staged, "problems": stock_gate.inspect(path, rows, missing, month)}

    @task
    def gate(staged):
        for s in staged:
            print(f"ด่าน ← receive_{s['branch']}: {s['rows']} แถว · ปัญหา {len(s['problems'])} ข้อ")
        try:
            return stock_gate.check(staged)
        except stock_gate.BadFile as e:  # ไฟล์ผิดรูป ลองกี่ครั้งก็ไม่หาย — หยุดเลย ไม่ใช้ retries ที่เหลือ
            raise AirflowFailException(str(e)) from e

    @task
    def transform(staged, **context):
        return steps.transform(staged, LAB / "staging" / context["params"]["month"] / "all.json")

    @task
    def load(summary, **context):
        return stock_load.load_month(summary, LAB / "warehouse.db", context["params"]["month"])  # ← เดิม steps.load

    @task
    def notify(staged, summary, in_db):
        print(steps.report(staged, summary, in_db))

    @task(trigger_rule="all_done")  # ส่งทุกเช้า ไม่ว่าท่อจะผ่านหรือพัง — แบบ notify ของบทที่ 7
    def morning():
        print("morning: ส่งข้อความเช้าแล้ว")

    @task(trigger_rule="one_failed")  # watcher: มีขั้นไหนพัง ขั้นนี้ได้รันแล้วพังเสมอ
    def watcher():
        raise AirflowFailException("มีขั้นที่พังในรอบนี้")

    found = arrived()
    received = [receive.override(task_id=f"receive_{b}")(b) for b in "abc"]
    found >> received
    staged = gate(received)
    summary = transform(staged)
    loaded = load(summary)
    done = notify(staged, summary, loaded)
    note = morning()
    done >> note
    [found, *received, staged, summary, loaded, done, note] >> watcher()  # ต่อจากทุกขั้น


stock_monthly()

for i in 1 2; do airflow dags test stock_monthly 2>&1 | grep "ในคลัง"; done

สต็อกรวม 951 ชิ้น (ก 121 · ข 382 · ค 448) จาก 99 แถว · ในคลังเดือนนี้ 99 แถว
สต็อกรวม 951 ชิ้น (ก 121 · ข 382 · ค 448) จาก 99 แถว · ในคลังเดือนนี้ 99 แถว

import sqlite3

# รหัสสินค้าในไฟล์ของสาขา ข: กรกฎาคม 33 รายการ · สิงหาคมเพิ่มหมวดใหม่อีก 3 (S034–S036) แบบไฟล์ปลอมของบทที่ 8
FILES = {
    "2026-07": [f"S{i:03d}" for i in range(1, 34)],
    "2026-08": [f"S{i:03d}" for i in range(1, 37)],
}


def import_latest(db, month):
    """ตัวนำเข้าแบบ "สถานะล่าสุด": ของที่อยู่ในไฟล์ = ขายอยู่ · ของที่ไม่อยู่ในไฟล์ = เลิกขาย"""
    codes = FILES[month]
    with db:
        db.executemany("INSERT OR REPLACE INTO items VALUES (?, 1)", [[c] for c in codes])
        db.execute(f"UPDATE items SET selling = 0 WHERE code NOT IN ({','.join('?' * len(codes))})", codes)


def import_period(db, month):
    """ตัวนำเข้าแบบ "ฟังก์ชันของงวด": เขียนทับเฉพาะแถวของเดือนนั้น · "ขายอยู่" คิดจากเดือนล่าสุดตอนอ่าน"""
    with db:
        db.execute("DELETE FROM listed WHERE month = ?", [month])
        db.executemany("INSERT INTO listed VALUES (?, ?)", [[month, c] for c in FILES[month]])


def selling_latest(db):
    return db.execute("SELECT COUNT(*) FROM items WHERE selling = 1").fetchone()[0]


def selling_period(db):
    return db.execute("SELECT COUNT(*) FROM listed WHERE month = (SELECT MAX(month) FROM listed)").fetchone()[0]


db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE items (code PRIMARY KEY, selling)")
db.execute("CREATE TABLE listed (month, code)")
for label, month in [("นำเข้ากรกฎาคม", "2026-07"), ("นำเข้าสิงหาคม", "2026-08"), ("รันกรกฎาคมซ้ำ", "2026-07")]:
    import_latest(db, month)
    import_period(db, month)
    print(f"{label:<14} → ขายอยู่ตามแบบสถานะล่าสุด {selling_latest(db)} · ตามแบบงวด {selling_period(db)}")

นำเข้ากรกฎาคม  → ขายอยู่ตามแบบสถานะล่าสุด 33 · ตามแบบงวด 33
นำเข้าสิงหาคม  → ขายอยู่ตามแบบสถานะล่าสุด 36 · ตามแบบงวด 36
รันกรกฎาคมซ้ำ  → ขายอยู่ตามแบบสถานะล่าสุด 33 · ตามแบบงวด 36