09ด่านตรวจก่อนโหลด

# ~/airflow-lab/dags/stock_gate.py — ด่านตรวจโครงก่อนโหลด: ฟังก์ชันธรรมดา ไม่ import อะไรจาก Airflow
from datetime import datetime

from openpyxl import load_workbook

NAMES = {"a": "ก", "b": "ข", "c": "ค"}
ROWS = range(10, 201)  # กว้างโดยตั้งใจ: จับไฟล์ผิดตัวหรือถูกตัดหาย ไม่ได้ล็อกขนาดแคตตาล็อกที่สาขาเป็นเจ้าของ


class BadFile(ValueError):
    """ไฟล์ผิดรูป — ต้องให้สาขาส่งใหม่ถึงจะหาย"""


def count_date(path):
    """ค่าในช่องข้าง ๆ คำว่า "วันที่นับ" — หาด้วยคำ ไม่เชื่อตำแหน่ง แบบเดียวกับหัวตาราง"""
    for ws in load_workbook(path).worksheets:
        for row in ws.iter_rows(max_row=10):
            for cell in row:
                if cell.value == "วันที่นับ":
                    return ws.cell(cell.row, cell.column + 1).value
    return None


def inspect(path, rows, missing, month):
    """ตรวจโครงของไฟล์หนึ่งสาขา → รายการปัญหา (ว่าง = ผ่าน) · ไม่ตรวจค่าที่สาขาเป็นเจ้าของ เช่นชื่อหมวด"""
    problems = []
    if missing:
        problems.append(f"ไม่มีคอลัมน์ {', '.join(missing)}")
    if len(rows) not in ROWS:
        problems.append(f"มี {len(rows)} แถว นอกช่วง {ROWS.start}–{ROWS.stop - 1}")
    if "จำนวน" not in missing:
        # .strip() เปล่า ๆ ตัด non-breaking space ได้ · == "" กับ .strip(" ") ตัดไม่ได้
        blank = [row["รหัส"] for row in rows if row["จำนวน"] is None or str(row["จำนวน"]).strip() == ""]
        if blank:
            problems.append(f"ช่องจำนวนว่าง {len(blank)} แถว (รหัส {', '.join(blank)})")
    when = count_date(path)
    if not isinstance(when, datetime):
        problems.append(f"วันที่นับอ่านไม่ได้: {when!r}")
    elif f"{when:%Y-%m}" != month:
        problems.append(f"วันที่นับ {when:%Y-%m-%d} ไม่ใช่เดือน {month}")
    return problems


def check(staged):
    """รวมปัญหาของทุกสาขาแล้วหยุดครั้งเดียว — หยุดที่ปัญหาแรก = แก้ทีละเรื่อง รันซ้ำทีละรอบ"""
    bad = [f"สาขา {NAMES[s['branch']]}: {p}" for s in staged for p in s["problems"]]
    if bad:
        raise BadFile("ด่านหยุดท่อ — " + " · ".join(bad))
    return staged


if __name__ == "__main__":  # ทดลองด่านโดยไม่มี Airflow — สามสาขา สาขาละอาการ (แถวแบบที่ read_branch คืน)
    import tempfile
    from pathlib import Path

    from openpyxl import Workbook

    nbsp = "\u00a0"
    print(f'== "" → {nbsp == ""} · .strip(" ") → {nbsp.strip(" ") == ""} · .strip() → {nbsp.strip() == ""}')

    work = Path(tempfile.mkdtemp())
    cases = {  # สาขา: (วันที่นับ, แถว, คอลัมน์ที่ขาด)
        "a": (datetime(2026, 8, 31), [], []),  # มีแต่หัวตาราง
        "b": (datetime(2026, 8, 31), [{"รหัส": f"S{i:03d}", "หมวด": "สีไม้" if i > 10 else "ปากกา",
                                        "จำนวน": nbsp if i in (3, 7) else 5} for i in range(1, 13)], []),
        "c": ("สิ้นเดือน", [{"รหัส": f"S{i:03d}", "หมวด": "ปากกา"} for i in range(1, 13)], ["จำนวน"]),
    }
    staged = []
    for b, (date, rows, missing) in cases.items():
        wb = Workbook()
        wb.active.append(["วันที่นับ", date])
        wb.save(work / f"{b}.xlsx")
        staged.append({"branch": b, "problems": inspect(work / f"{b}.xlsx", rows, missing, "2026-08")})
    try:
        check(staged)
    except BadFile as e:
        print(type(e).__name__, "·", str(e).replace(" · สาขา", "\n  สาขา"))

== "" → False · .strip(" ") → False · .strip() → True
BadFile · ด่านหยุดท่อ — สาขา ก: มี 0 แถว นอกช่วง 10–200
  สาขา ข: ช่องจำนวนว่าง 2 แถว (รหัส S003, S007)
  สาขา ค: ไม่มีคอลัมน์ จำนวน
  สาขา ค: วันที่นับอ่านไม่ได้: 'สิ้นเดือน'

# ~/airflow-lab/dags/stock_monthly.py — บทที่ 9: ขั้น arrived (ยังไม่มีไฟล์ = ข้าม) + ด่านตรวจโครงแทน merge_check
from datetime import datetime
from pathlib import Path

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

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


@dag(schedule=None, start_date=datetime(2026, 5, 1), catchup=False, params={"month": "2026-07"})
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"
        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'])} ข้อ")
        return stock_gate.check(staged)

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

    received = [receive.override(task_id=f"receive_{b}")(b) for b in "abc"]
    arrived() >> received
    staged = gate(received)
    summary = transform(staged)
    notify(staged, summary, load(summary))


stock_monthly()

# ~/airflow-lab/states.py — พิมพ์คอลัมน์ของรอบล่าสุดจากฐานข้อมูลของแล็บ ในรูปตารางของบทที่ 4
import importlib
import sqlite3
import sys
from pathlib import Path

import airflow  # noqa: F401 — ต้องรันด้วย python ของแล็บ เพราะต้อง import ไฟล์ DAG เพื่ออ่านกราฟ

LAB = Path.home() / "airflow-lab"
dag_id = sys.argv[1]
sys.path.insert(0, str(LAB / "dags"))
dag = getattr(importlib.import_module(dag_id), dag_id)()  # ประกาศกราฟอย่างเดียว ไม่รันงาน (บทที่ 6)
ups = {t.task_id: t.upstream_task_ids for t in dag.tasks}


def beat(task):  # จังหวะแบบบทที่ 3: ขั้นที่ไม่รอใคร = 1 · ขั้นอื่น = 1 + จังหวะที่มากที่สุดของต้นน้ำ
    return 1 + max((beat(u) for u in ups[task]), default=0)


db = sqlite3.connect(f"file:{LAB / 'airflow.db'}?mode=ro", uri=True)  # อ่านอย่างเดียว ไม่แก้ตาราง
run_id, run_state, conf = db.execute(
    "SELECT run_id, state, conf FROM dag_run WHERE dag_id = ? ORDER BY id DESC LIMIT 1", [dag_id]
).fetchone()
cells = {
    task: (state, tries)
    for task, state, tries in db.execute(
        "SELECT task_id, state, try_number FROM task_instance WHERE dag_id = ? AND run_id = ?", [dag_id, run_id]
    )
}
print(f"รอบล่าสุดของ {dag_id} {conf} → {run_state}")
for task in sorted(ups, key=lambda t: (beat(t), t)):
    state, tries = cells.get(task, (None, 0))
    print(f"  {task:<14}{state or 'none'} {tries}")  # ช่องว่างในฐานข้อมูล = none ของบทที่ 4

airflow dags test stock_monthly -c '{"month": "2026-08"}' 2>&1 | grep -E "ด่าน ←|BadFile:"

ด่าน ← receive_a: 33 แถว · ปัญหา 1 ข้อ
ด่าน ← receive_b: 36 แถว · ปัญหา 0 ข้อ
ด่าน ← receive_c: 33 แถว · ปัญหา 1 ข้อ
stock_gate.BadFile: ด่านหยุดท่อ — สาขา ก: วันที่นับอ่านไม่ได้: 'สิ้นเดือน ส.ค.' · สาขา ค: ช่องจำนวนว่าง 3 แถว (รหัส S007, S008, S023)

python ~/airflow-lab/states.py stock_monthly

รอบล่าสุดของ stock_monthly {"month": "2026-08"} → failed
  arrived       success 1
  receive_a     success 1
  receive_b     success 1
  receive_c     success 1
  gate          failed 1
  transform     upstream_failed 0
  load          upstream_failed 0
  notify        upstream_failed 0

airflow dags test stock_monthly -c '{"month": "2026-10"}' 2>&1 | grep -o "reason=.*"

reason='ยังไม่มีไฟล์ของเดือน 2026-10'

python ~/airflow-lab/states.py stock_monthly

รอบล่าสุดของ stock_monthly {"month": "2026-10"} → success
  arrived       skipped 1
  receive_a     skipped 0
  receive_b     skipped 0
  receive_c     skipped 0
  gate          skipped 0
  transform     skipped 0
  load          skipped 0
  notify        skipped 0