08ท่อรับไฟล์รายเดือน ฉบับ Airflow

# ~/airflow-lab/make_fake_files.py
"""สร้างไฟล์นับสต็อกรายเดือนของร้านเครื่องเขียนสมมติ 3 สาขา — ข้อมูลปลอมทั้งหมด หน่วยเป็นชิ้น

รันเปล่า ๆ = เขียนลงโฟลเดอร์ชั่วคราวแล้วพิมพ์สรุป · ใส่ path = เขียนลงโฟลเดอร์นั้น
"""

import random
import sys
import tempfile
from datetime import date
from pathlib import Path

from openpyxl import Workbook

SEED = 2569
BRANCHES = {"a": ("สาขา ก", 120), "b": ("สาขา ข", 340), "c": ("สาขา ค", 560)}  # ยอดเดือนแรก
MONTHS = [date(2026, 5, 31), date(2026, 6, 30), date(2026, 7, 31), date(2026, 8, 31), date(2026, 9, 30)]
CATALOG = {
    "ปากกา": ["ปากกาลูกลื่น", "ปากกาเจล", "ปากกาเน้นข้อความ"],
    "ดินสอ": ["ดินสอไม้ 2B", "ดินสอกด", "ยางลบ"],
    "กระดาษ": ["สมุดปกอ่อน", "กระดาษ A4", "กระดาษโน้ต"],
    "แฟ้ม": ["แฟ้มสอด", "แฟ้มห่วง"],
}
PACKS = ["แพ็ก 1", "แพ็ก 6", "แพ็ก 12"]
COLUMNS = ["รหัส", "ชื่อสินค้า", "หมวด", "จำนวน"]

# อาการที่ไฟล์จริงเป็น — หนึ่งไฟล์หนึ่งอาการ ให้แต่ละบทลองทีละเรื่อง
QUIRKS = {
    ("2026-06", "b"): "columns_moved",  # คอลัมน์สลับที่ มีคอลัมน์เพิ่ม และชื่อชีตเปลี่ยน
    ("2026-07", "a"): "autofilter_stuck",  # ตัวกรอง "จำนวน = 0" ค้าง ทุกแถวข้อมูลติดธงซ่อน
    ("2026-07", "b"): "second_sheet",  # ข้อมูลอยู่ชีตที่สอง และมีแถวคำอธิบายใต้หัวตาราง
    ("2026-08", "a"): "bad_date",  # ช่องวันที่นับเป็นข้อความ อ่านเป็นวันที่ไม่ได้
    ("2026-08", "b"): "new_category",  # หมวดใหม่โผล่มาเดือนนี้
    ("2026-08", "c"): "nbsp_blank",  # ช่องจำนวนที่ดูว่าง แต่มี non-breaking space
    ("2026-09", "a"): "missing_column",  # คอลัมน์ "จำนวน" หาย ด่านต้องหยุดท่อ
    ("2026-09", "c"): "late",  # ไฟล์มาช้า — แยกไว้ในโฟลเดอร์ late/
}


def catalog(quirk):
    items = [(cat, f"{name} {pack}") for cat, names in CATALOG.items() for name in names for pack in PACKS]
    if quirk == "new_category":
        items += [("อุปกรณ์ศิลปะ", f"สีไม้ {pack}") for pack in PACKS]
    return [(f"S{i:03d}", cat, name) for i, (cat, name) in enumerate(items, start=1)]


def split(total, n, rng):
    """แบ่ง total ชิ้นให้ n รายการ ทุกรายการได้อย่างน้อย 1 ชิ้น"""
    counts = [1] * n
    for _ in range(total - n):
        counts[rng.randrange(n)] += 1
    return counts


def make_rows(total, quirk, rng):
    items = catalog(quirk)
    counts = split(total, len(items), rng)
    rows = [
        {"รหัส": code, "ชื่อสินค้า": name, "หมวด": cat, "จำนวน": n}
        for (code, cat, name), n in zip(items, counts)
    ]
    if quirk == "nbsp_blank":
        for row in rng.sample(rows, 3):
            row["จำนวน"] = "\u00a0"
    if quirk == "missing_column":
        for row in rows:
            del row["จำนวน"]
    return rows


def write_book(path, title, month_end, rows, quirk):
    wb = Workbook()
    ws = wb.active
    ws.title = "นับสต็อก" if quirk == "columns_moved" else "สต็อก"
    if quirk == "second_sheet":
        ws.title = "หน้าปก"
        ws["A1"] = "ข้อมูลอยู่ชีตถัดไป"
        ws = wb.create_sheet("ข้อมูล")
    columns = list(COLUMNS)
    if quirk == "columns_moved":
        columns = ["ชื่อสินค้า", "รหัส", "จำนวน", "หมวด", "หมายเหตุ"]
    if quirk == "missing_column":
        columns.remove("จำนวน")
    ws["A1"] = f"นับสต็อก {title}"
    ws["A2"] = "วันที่นับ"
    ws["B2"] = "สิ้นเดือน ส.ค." if quirk == "bad_date" else month_end
    for letter, width in zip("ABCDE", [12, 24, 14, 10, 12]):
        ws.column_dimensions[letter].width = width
    header_row = 4  # แถว 3 เว้นว่าง
    for col, name in enumerate(columns, start=1):
        ws.cell(header_row, col, name)
    first = header_row + 1
    if quirk == "second_sheet":
        ws.cell(first, 1, "(หน่วย: ชิ้น · นับ ณ สิ้นเดือน)")
        first += 1
    for r, row in enumerate(rows, start=first):
        for col, name in enumerate(columns, start=1):
            ws.cell(r, col, row.get(name))
    last = first + len(rows) - 1
    if quirk == "autofilter_stuck":
        ws.auto_filter.ref = f"A{header_row}:{ws.cell(header_row, len(columns)).column_letter}{last}"
        ws.auto_filter.add_filter_column(columns.index("จำนวน"), ["0"])
        for r in range(first, last + 1):
            ws.row_dimensions[r].hidden = True
    wb.save(path)
    hidden = sum(1 for r in range(first, last + 1) if ws.row_dimensions[r].hidden)
    return wb.sheetnames, hidden


def main(out_dir):
    rng = random.Random(SEED)
    lines = []
    for i, month_end in enumerate(MONTHS):
        month = f"{month_end:%Y-%m}"
        for code, (title, base) in BRANCHES.items():
            quirk = QUIRKS.get((month, code), "-")
            total = base if i == 0 else round(base * rng.uniform(0.8, 1.2))
            rows = make_rows(total, quirk, rng)
            folder = out_dir / "late" if quirk == "late" else out_dir
            folder.mkdir(parents=True, exist_ok=True)
            name = f"stock_{month}_{code}.xlsx"
            sheets, hidden = write_book(folder / name, title, month_end, rows, quirk)
            quantities = [row.get("จำนวน") for row in rows]
            pieces = sum(n for n in quantities if isinstance(n, int))
            where = "late/" if quirk == "late" else ""
            lines.append(
                f"{where}{name} · ชีต {'/'.join(sheets)} · {len(rows)} แถว · ซ่อน {hidden} · "
                f"{pieces} ชิ้น · {quirk}"
            )
    print("\n".join(lines))
    first = [base for _, base in BRANCHES.values()]
    print(f"เดือนแรก: {' + '.join(map(str, first))} = {sum(first)} ชิ้น")


if __name__ == "__main__":
    if len(sys.argv) > 1:
        main(Path(sys.argv[1]))
    else:
        main(Path(tempfile.mkdtemp()))

stock_2026-05_a.xlsx · ชีต สต็อก · 33 แถว · ซ่อน 0 · 120 ชิ้น · -
stock_2026-05_b.xlsx · ชีต สต็อก · 33 แถว · ซ่อน 0 · 340 ชิ้น · -
stock_2026-05_c.xlsx · ชีต สต็อก · 33 แถว · ซ่อน 0 · 560 ชิ้น · -
stock_2026-06_a.xlsx · ชีต สต็อก · 33 แถว · ซ่อน 0 · 140 ชิ้น · -
stock_2026-06_b.xlsx · ชีต นับสต็อก · 33 แถว · ซ่อน 0 · 364 ชิ้น · columns_moved
stock_2026-06_c.xlsx · ชีต สต็อก · 33 แถว · ซ่อน 0 · 610 ชิ้น · -
stock_2026-07_a.xlsx · ชีต สต็อก · 33 แถว · ซ่อน 33 · 121 ชิ้น · autofilter_stuck
stock_2026-07_b.xlsx · ชีต หน้าปก/ข้อมูล · 33 แถว · ซ่อน 0 · 382 ชิ้น · second_sheet
stock_2026-07_c.xlsx · ชีต สต็อก · 33 แถว · ซ่อน 0 · 448 ชิ้น · -
stock_2026-08_a.xlsx · ชีต สต็อก · 33 แถว · ซ่อน 0 · 139 ชิ้น · bad_date
stock_2026-08_b.xlsx · ชีต สต็อก · 36 แถว · ซ่อน 0 · 406 ชิ้น · new_category
stock_2026-08_c.xlsx · ชีต สต็อก · 33 แถว · ซ่อน 0 · 550 ชิ้น · nbsp_blank
stock_2026-09_a.xlsx · ชีต สต็อก · 33 แถว · ซ่อน 0 · 0 ชิ้น · missing_column
stock_2026-09_b.xlsx · ชีต สต็อก · 33 แถว · ซ่อน 0 · 320 ชิ้น · -
late/stock_2026-09_c.xlsx · ชีต สต็อก · 33 แถว · ซ่อน 0 · 559 ชิ้น · late
เดือนแรก: 120 + 340 + 560 = 1020 ชิ้น

# ~/airflow-lab/dags/stock_steps.py — ตรรกะของทุกขั้นเป็นฟังก์ชันธรรมดา ไม่ import อะไรจาก Airflow
import json
import sqlite3
from pathlib import Path

from openpyxl import load_workbook

REQUIRED = ["รหัส", "ชื่อสินค้า", "หมวด", "จำนวน"]
NAMES = {"a": "ก", "b": "ข", "c": "ค"}


def find_header(book):
    """หาชีตกับแถวหัวตารางด้วยชื่อคอลัมน์ ไม่เชื่อตำแหน่ง → (ชีต, เลขแถว, {ชื่อคอลัมน์: เลขคอลัมน์})"""
    for ws in book.worksheets:
        for r in range(1, min(ws.max_row, 20) + 1):
            cols = {ws.cell(r, c).value: c for c in range(1, ws.max_column + 1)}
            if "รหัส" in cols and "ชื่อสินค้า" in cols:
                return ws, r, {name: c for name, c in cols.items() if name}
    raise ValueError("หาแถวหัวตารางไม่เจอในทุกชีต")


def read_branch(path):
    """อ่านทุกแถวใต้หัวตาราง แล้วติดป้ายว่าแถวไหนซ่อน — ไม่กรองแถวซ่อนทิ้งตอนอ่าน"""
    ws, header, cols = find_header(load_workbook(path))
    rows = []
    for r in range(header + 1, ws.max_row + 1):
        row = {name: ws.cell(r, c).value for name, c in cols.items()}
        if row["รหัส"] is None or row["ชื่อสินค้า"] is None:
            continue  # แถวคำอธิบายใต้หัวตาราง หรือแถวว่าง
        row["ซ่อน"] = bool(ws.row_dimensions[r].hidden)
        rows.append(row)
    return rows, [name for name in REQUIRED if name not in cols]


def stage(branch, rows, missing, path):
    """เขียนแถวทั้งหมดลงไฟล์ แล้วคืนแค่ใบสรุปเล็ก ๆ — ใบนี้คือสิ่งที่เดินทางผ่าน XCom"""
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(rows, ensure_ascii=False), encoding="utf-8")
    hidden = sum(row["ซ่อน"] for row in rows)
    return {"branch": branch, "path": str(path), "rows": len(rows), "hidden": hidden, "missing": missing}


def check_columns(staged):
    for s in staged:
        if s["missing"]:
            raise ValueError(f"สาขา {NAMES[s['branch']]} ไม่มีคอลัมน์ {', '.join(s['missing'])}")
    return staged


def transform(staged, path):
    rows = []
    for s in staged:
        for row in json.loads(Path(s["path"]).read_text(encoding="utf-8")):
            rows.append([s["branch"], row["รหัส"], int(row["จำนวน"]), row["ซ่อน"]])
    path.write_text(json.dumps(rows, ensure_ascii=False), encoding="utf-8")
    pieces = {s["branch"]: sum(row[2] for row in rows if row[0] == s["branch"]) for s in staged}
    return {"path": str(path), "rows": len(rows), "pieces": pieces}


def load(summary, db_path, month):
    """⚠️ ต่อท้ายตาราง — รันซ้ำแล้วจะเกิดอะไรขึ้น คือเรื่องของบทที่ 12"""
    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)")
    db.executemany("INSERT INTO stock VALUES (?, ?, ?, ?, ?)", [[month, *row] for row in rows])
    db.commit()
    return db.execute("SELECT COUNT(*) FROM stock WHERE month = ?", [month]).fetchone()[0]


def report(staged, summary, in_db):
    each = " · ".join(f"{NAMES[b]} {n:,}" for b, n in summary["pieces"].items())
    total = sum(summary["pieces"].values())
    lines = [f"สต็อกรวม {total:,} ชิ้น ({each}) จาก {summary['rows']} แถว · ในคลังเดือนนี้ {in_db} แถว"]
    lines += [f"⚠️ สาขา {NAMES[s['branch']]} ซ่อนอยู่ {s['hidden']} จาก {s['rows']} แถว — เปิดไฟล์ดูก่อนเชื่อยอด"
              for s in staged if s["hidden"]]
    return "\n".join(lines)


if __name__ == "__main__":  # ทดลองทุกขั้นโดยไม่มี Airflow — ไฟล์เล็กสองไฟล์ที่มีอาการแบบของจริง
    import tempfile

    from openpyxl import Workbook

    work = Path(tempfile.mkdtemp())
    book = Workbook()  # สาขา ก: ตัวกรองค้าง ทุกแถวข้อมูลติดธงซ่อน
    ws = book.active
    ws.append(["นับสต็อก สาขา ก"])
    ws.append([])
    ws.append([])
    ws.append(["รหัส", "ชื่อสินค้า", "หมวด", "จำนวน"])
    for i in range(1, 4):
        ws.append([f"S{i:03d}", f"ปากกา {i}", "ปากกา", 10 * i])
        ws.row_dimensions[ws.max_row].hidden = True
    book.save(work / "a.xlsx")
    book = Workbook()  # สาขา ข: ข้อมูลอยู่ชีตที่สอง คอลัมน์สลับที่ มีแถวคำอธิบายใต้หัวตาราง
    book.active.title = "หน้าปก"
    ws = book.create_sheet("ข้อมูล")
    ws.append(["ชื่อสินค้า", "รหัส", "จำนวน", "หมวด"])
    ws.append(["(หน่วย: ชิ้น)"])
    ws.append(["ดินสอ 1", "S004", 7, "ดินสอ"])
    ws.append(["ดินสอ 2", "S005", 5, "ดินสอ"])
    book.save(work / "b.xlsx")

    staged = [stage(b, *read_branch(work / f"{b}.xlsx"), work / f"{b}.json") for b in "ab"]
    for s in staged:
        print({k: v for k, v in s.items() if k != "path"})
    summary = transform(check_columns(staged), work / "all.json")
    print(report(staged, summary, load(summary, work / "warehouse.db", "2026-07")))

{'branch': 'a', 'rows': 3, 'hidden': 3, 'missing': []}
{'branch': 'b', 'rows': 2, 'hidden': 0, 'missing': []}
สต็อกรวม 72 ชิ้น (ก 60 · ข 12) จาก 5 แถว · ในคลังเดือนนี้ 5 แถว
⚠️ สาขา ก ซ่อนอยู่ 3 จาก 3 แถว — เปิดไฟล์ดูก่อนเชื่อยอด

# ~/airflow-lab/dags/stock_monthly.py — ต่อสายอย่างเดียว ตรรกะทั้งหมดอยู่ใน stock_steps.py ข้าง ๆ
from datetime import datetime
from pathlib import Path

import stock_steps as steps
from airflow.sdk import dag, task

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 receive(branch, **context):
        month = context["params"]["month"]
        rows, missing = steps.read_branch(LAB / "inbox" / f"stock_{month}_{branch}.xlsx")
        return steps.stage(branch, rows, missing, LAB / "staging" / month / f"{branch}.json")

    @task
    def merge_check(staged):
        for s in staged:  # สิ่งที่มาทาง XCom มีแค่ใบสรุป ไม่มีแถวข้อมูลสักแถว
            print(f"merge_check ← receive_{s['branch']}: {Path(s['path']).name} · {s['rows']} แถว · ซ่อน {s['hidden']}")
        return steps.check_columns(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))

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


stock_monthly()

uv pip install openpyxl --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-3.3.2/constraints-3.12.txt"
python ~/airflow-lab/make_fake_files.py ~/airflow-lab/inbox   # ผลเดียวกับหัวข้อ 8.2
# วาง stock_steps.py กับ stock_monthly.py ไว้ใน ~/airflow-lab/dags/

airflow dags test stock_monthly 2>&1 | grep -E "ชิ้น|แถว"

merge_check ← receive_a: a.json · 33 แถว · ซ่อน 33
merge_check ← receive_b: b.json · 33 แถว · ซ่อน 0
merge_check ← receive_c: c.json · 33 แถว · ซ่อน 0
สต็อกรวม 951 ชิ้น (ก 121 · ข 382 · ค 448) จาก 99 แถว · ในคลังเดือนนี้ 99 แถว
⚠️ สาขา ก ซ่อนอยู่ 33 จาก 33 แถว — เปิดไฟล์ดูก่อนเชื่อยอด