13รอบนี้ประมวล⁠ผลข้อมูลช่วงไหน

cron สตริง                 logical_date= 09:26  start= 09:26  end= 09:26
CronDataIntervalTimetable  logical_date= 09:24  start= 09:24  end= 09:26

# ~/airflow-lab/dags/stock_monthly.py — บทที่ 13: รอบตามตารางรู้เดือนจากช่วงข้อมูลของตัวเอง
from datetime import datetime, timedelta
from pathlib import Path

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

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


def month_of(context):
    """เดือนที่รอบนี้ประมวลผล: บอกมาทาง -c ก่อน · ไม่บอกก็ใช้ต้นช่วงข้อมูลของรอบ · ไม่มีทั้งคู่ = หยุด"""
    if context["params"]["month"]:
        return context["params"]["month"]
    start = context.get("data_interval_start")  # รอบที่กดรันเองไม่มีคีย์นี้เลย — ["..."] จะยก KeyError
    if start is None:
        raise AirflowFailException("รอบนี้ไม่มีช่วงข้อมูล — บอกเดือนด้วย -c หรือย้อนรันด้วย backfill")
    return f"{start:%Y-%m}"


@dag(
    schedule=CronDataIntervalTimetable("0 0 1 * *", timezone="UTC"),  # ทุกต้นเดือน รอบละหนึ่งเดือนที่จบแล้ว
    start_date=datetime(2026, 5, 1),
    catchup=False,
    params={"month": ""},
    default_args={"retries": 2, "retry_delay": timedelta(seconds=20)},  # แล็บรอ 20 วินาที ของจริงหลายนาที
)
def stock_monthly():
    @task
    def arrived(**context):
        month = month_of(context)
        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 = month_of(context)
        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" / month_of(context) / "all.json")

    @task
    def load(summary, **context):
        return stock_load.load_month(summary, LAB / "warehouse.db", month_of(context))  # ← เดิม 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()

airflow dags test stock_monthly 2026-07-01 2>&1 | grep "ในคลัง"

สต็อกรวม 1,114 ชิ้น (ก 140 · ข 364 · ค 610) จาก 99 แถว · ในคลังเดือนนี้ 99 แถว

$ airflow dags unpause stock_monthly
scheduled  2026-09-01 → 2026-10-01  failed      ← รอบเดียว: งวดล่าสุดที่จบแล้ว

$ airflow backfill create --dag-id stock_monthly --from-date 2026-05-01 --to-date 2026-07-01
backfill   2026-05-01 → 2026-06-01  success
backfill   2026-06-01 → 2026-07-01  success
backfill   2026-07-01 → 2026-08-01  success
คลัง: 2026-05 99 แถว · 2026-06 99 แถว · 2026-07 99 แถว

$ airflow dags trigger stock_monthly                    ← กดรันเองโดยไม่บอกเดือน
manual     None       → None        failed      arrived failed 1
AirflowFailException: รอบนี้ไม่มีช่วงข้อมูล — บอกเดือนด้วย -c หรือย้อนรันด้วย backfill

$ airflow backfill create --dag-id stock_monthly --from-date 2026-05-01 --to-date 2026-07-01
ไม่มีรอบใหม่ — ทั้งสามเดือนถูกข้ามด้วยเหตุผล already exists