02ลองคุมลำดับเองด้วย Python ล้วน

import json
import tempfile
from pathlib import Path

WORK = Path(tempfile.mkdtemp())
STATE = WORK / "state.json"  # ← ความจำของตัวรัน อยู่บนดิสก์ ไม่ได้อยู่ในตัวแปร


def load():
    if not (WORK / "db-fixed").exists():
        raise ConnectionError("ฐานข้อมูลปลายทางตัดการเชื่อมต่อ")


STEPS = [  # ลำดับที่ต้องทำ — ขั้นละหนึ่งฟังก์ชัน
    ("receive_a", lambda: None),
    ("receive_b", lambda: None),
    ("receive_c", lambda: None),
    ("merge_check", lambda: None),
    ("transform", lambda: None),
    ("load", load),
    ("notify", lambda: None),
]


def run():
    state = json.loads(STATE.read_text()) if STATE.exists() else {}
    for name, step in STEPS:
        if state.get(name) == "success":
            print(f"  {name:<12} ข้าม — สำเร็จไปแล้ว")
            continue
        try:
            step()
        except Exception as e:
            state[name] = "failed"
            STATE.write_text(json.dumps(state))
            print(f"  {name:<12} พัง: {e} → หยุดรอบนี้")
            return
        state[name] = "success"
        STATE.write_text(json.dumps(state))  # จดทันทีที่แต่ละขั้นเสร็จ ไม่รอจบรอบ
        print(f"  {name:<12} สำเร็จ")


print("รอบที่ 1")
run()
(WORK / "db-fixed").touch()  # แก้ต้นเหตุแล้ว
print("รอบที่ 2 — เรียก run() ใหม่ ซึ่งรู้เรื่องรอบแรกจากไฟล์อย่างเดียว")
run()

รอบที่ 1
  receive_a    สำเร็จ
  receive_b    สำเร็จ
  receive_c    สำเร็จ
  merge_check  สำเร็จ
  transform    สำเร็จ
  load         พัง: ฐานข้อมูลปลายทางตัดการเชื่อมต่อ → หยุดรอบนี้
รอบที่ 2 — เรียก run() ใหม่ ซึ่งรู้เรื่องรอบแรกจากไฟล์อย่างเดียว
  receive_a    ข้าม — สำเร็จไปแล้ว
  receive_b    ข้าม — สำเร็จไปแล้ว
  receive_c    ข้าม — สำเร็จไปแล้ว
  merge_check  ข้าม — สำเร็จไปแล้ว
  transform    ข้าม — สำเร็จไปแล้ว
  load         สำเร็จ
  notify       สำเร็จ

import json
import tempfile
import time
from pathlib import Path

WORK = Path(tempfile.mkdtemp())
STATE = WORK / "state.json"  # สถานะ + จำนวนครั้งที่ลอง ของทุกขั้น
LOG = WORK / "runner.log"  # ← ชิ้นใหม่: บันทึกว่าเกิดอะไรขึ้นทีละครั้ง
RETRIES = 2  # ← ชิ้นใหม่: พังแล้วลองใหม่ได้อีกกี่ครั้ง
RETRY_DELAY = 0.2  # วินาที · ของจริงตั้งเป็นนาที


def receive_c():
    arrived = WORK / "stock_2026-05_c.xlsx"
    if not arrived.exists():
        arrived.touch()  # จำลองว่าสาขา ค อัปโหลดเสร็จระหว่างที่ตัวรันรอ
        raise FileNotFoundError("ไฟล์สาขา ค ยังมาไม่ถึง")


STEPS = [
    ("receive_a", lambda: None),
    ("receive_b", lambda: None),
    ("receive_c", receive_c),
    ("merge_check", lambda: None),
    ("transform", lambda: None),
    ("load", lambda: None),
    ("notify", lambda: None),
]


def log(line):
    with LOG.open("a", encoding="utf-8") as f:
        f.write(line + "\n")


def run():
    state = json.loads(STATE.read_text()) if STATE.exists() else {}
    for name, step in STEPS:
        cell = state.setdefault(name, {"state": "none", "tries": 0})
        while cell["state"] != "success":
            cell["tries"] += 1
            try:
                step()
                cell["state"] = "success"
                log(f"{name} ครั้งที่ {cell['tries']}: สำเร็จ")
            except Exception as e:
                cell["state"] = "failed"
                log(f"{name} ครั้งที่ {cell['tries']}: พัง — {e}")
                if cell["tries"] > RETRIES:
                    STATE.write_text(json.dumps(state))
                    return
                time.sleep(RETRY_DELAY)
            STATE.write_text(json.dumps(state))


run()
print(LOG.read_text(encoding="utf-8"), end="")
print("state.json → receive_c =", json.loads(STATE.read_text())["receive_c"])

receive_a ครั้งที่ 1: สำเร็จ
receive_b ครั้งที่ 1: สำเร็จ
receive_c ครั้งที่ 1: พัง — ไฟล์สาขา ค ยังมาไม่ถึง
receive_c ครั้งที่ 2: สำเร็จ
merge_check ครั้งที่ 1: สำเร็จ
transform ครั้งที่ 1: สำเร็จ
load ครั้งที่ 1: สำเร็จ
notify ครั้งที่ 1: สำเร็จ
state.json → receive_c = {'state': 'success', 'tries': 2}