11พังกลางทาง: ล้างสถานะเฉพาะขั้นแล้วรันต่อ

# ~/airflow-lab/dags/stock_monthly.py — บทที่ 11: เพิ่มข้อความเช้า (all_done) กับ watcher ที่ทำให้รอบแดงจริง
from datetime import datetime, timedelta
from pathlib import Path

import stock_gate
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 steps.load(summary, LAB / "warehouse.db", context["params"]["month"])

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

cd ~/airflow-lab && python -c "
import make_fake_files as m
from pathlib import Path
del m.QUIRKS[('2026-09', 'a')]
m.main(Path('fixed'))" && cp fixed/stock_2026-09_a.xlsx inbox/

ก่อน Clear                      ทันทีหลัง Confirm               จบรอบ
รอบ → failed                    รอบ → running                    รอบ → success
  arrived       success 1         arrived       success 1          arrived       success 1
  receive_a     success 1         receive_a     running 2          receive_a     success 2
  receive_b     success 1         receive_b     success 1          receive_b     success 1
  receive_c     success 1         receive_c     success 1          receive_c     success 1
  gate          failed 1          gate          none 1             gate          success 2
  transform     upstream_failed 0 transform     none 0             transform     success 1
  load          upstream_failed 0 load          none 0             load          success 1
  notify        upstream_failed 0 notify        none 0             notify        success 1
  morning       success 1         morning       none 1             morning       success 2
  watcher       failed 1          watcher       none 1             watcher       skipped 1