ยิง API สองหมื่นกว่าครั้ง ให้หยุดกลางทางแล้วต่อได้ และไม่ได้ไฟล์เสีย

พอดีผมมีงานต้องสร้างไฟล์เสียงจาก API ของผู้ให้บริการรายหนึ่ง จำนวนรวม 26,850 ไฟล์ คิดจากคำศัพท์ห้าพันกว่าคำคูณด้วยเสียงผู้พูดห้าแบบ บัญชีที่ใช้เป็นระดับฟรี โควตาเขียนไว้ชัดว่า 20 ครั้งต่อ 60 วินาที ผมเขียนตัวยิงรอบแรกแบบตรงไปตรงมาคือวนลูปแล้วยิง พอปล่อยจริงได้ไม่กี่นาทีก็เจอทั้ง 429 และข้อความว่ายิงเกินโควตา ระหว่างที่นั่งแก้ผมเลยได้ข้อสรุปว่าตัวยิงแบบนี้ต้องมีสามชิ้นเสมอ ไม่ใช่ชิ้นเดียว คือตัวหน่วง ตัวต่องาน และตัวตรวจ ขาดชิ้นไหนไปก็เจ็บคนละแบบ

บทความนี้ผมเขียนเป็นสูตรพร้อมโค้ดที่ใช้จริง ตัดชื่อผู้ให้บริการกับค่าเชื่อมต่อออกหมดแล้ว ใครเอาไปใช้กับ API อะไรก็ได้ที่มีเพดานจำนวนครั้ง

เพดานเป็นของทั้งบัญชี ไม่ใช่ของแต่ละเส้น

ความผิดพลาดแรกของผมคือไปใส่ time.sleep() ไว้ในตัว worker แต่ละตัว คิดว่าถ้าหน่วงทุกเส้นเท่ากันแล้วจะไม่เกิน ซึ่งผิด เพราะเพดานที่ผู้ให้บริการนับคือจำนวนครั้งต่อบัญชี ไม่ใช่ต่อเส้น ถ้าเปิด 10 เส้น แล้วแต่ละเส้นหน่วง 3 วินาที ของจริงที่ออกไปคือ 10 ครั้งต่อ 3 วินาที เท่ากับ 200 ครั้งต่อนาที เกินเพดานไปสิบเท่า

ตัวหน่วงที่ถูกต้องต้องเป็นของกลาง มีตัวแปรตัวเดียวจำเวลาที่ยิงครั้งล่าสุด แล้วทุกเส้นต้องมาเข้าคิวขอผ่านตัวเดียวกันนี้

import threading, time

_rate_lock = threading.Lock()
_last_call = 0.0
_min_interval = 3.0          # 20 ครั้ง/60 วิ = 1 ครั้งต่อ 3 วินาที

def set_rps(rps):
    global _min_interval
    _min_interval = 1.0 / max(rps, 0.01)

def throttle_global():
    """ทุกเส้นต้องผ่านประตูนี้ก่อนยิง ไม่ว่าจะเปิดกี่เส้น"""
    global _last_call
    while True:
        with _rate_lock:
            now = time.time()
            since = now - _last_call
            if since >= _min_interval:
                _last_call = now
                return
            wait = _min_interval - since
        time.sleep(wait)      # นอนนอก lock เสมอ

ค่าที่ผมเลือกใช้คือ 0.3 ครั้งต่อวินาที ซึ่งเท่ากับ 18 ครั้งต่อนาที ไม่ได้ตั้งเต็ม 20 เพราะเวลาฝั่งเรากับฝั่งเขาไม่ตรงกันเป๊ะ และโควตาแบบหน้าต่างเลื่อนจะนับครั้งสุดท้ายของนาทีก่อนมาปนกับนาทีนี้เสมอ เผื่อไว้สิบเปอร์เซ็นต์ถูกกว่ามานั่งไล่ 429 ทีหลัง

เปิดกี่เส้นก็ไม่เร็วขึ้น ถ้าเพดานเป็นของกลาง

พอมีตัวหน่วงของกลางแล้ว จำนวน worker ก็เปลี่ยนบทบาทไปเลย มันไม่ได้ทำให้เร็วขึ้นแม้แต่นิดเดียว เพราะความเร็วถูกล็อกไว้ที่ตัวหน่วงแล้ว สิ่งที่ worker ทำคือซ่อนเวลารอเน็ตของแต่ละครั้งเท่านั้น

ผมเริ่มที่ 10 เส้นแล้วเจอ max retries exceeded ประปราย เพราะเส้นที่เข้าคิวรอนานเกินไปจนหมดเวลาเชื่อมต่อก่อนได้ยิง ลดลงมาเหลือ 5 เส้นแล้วอาการหาย ความเร็วรวมเท่าเดิมเป๊ะ

ตัวเลข 25 ชั่วโมงนี่แหละคือเหตุผลที่ต้องมีชิ้นที่สอง งานที่กินเวลาข้ามคืนจะโดนขัดจังหวะแน่นอน ไม่เน็ตหลุดก็ไฟกระพริบ ไม่งั้นก็คนสั่งปิดเครื่องเอง

ลองซ้ำเป็นคนละเรื่องกับหน่วง อย่าเอามารวมกัน

ตัวหน่วงคุมความถี่ขาออก ส่วนตัวลองซ้ำคุมความผิดพลาดรายครั้ง สองอันนี้ต้องแยกกันชัด ๆ และที่สำคัญคือ ทุกครั้งที่ลองใหม่ต้องผ่านตัวหน่วงอีกรอบ ไม่ใช่ยิงซ้ำทันที

def generate_one(session, endpoint, task, retries=5):
    key, payload, out_path = task
    last_error = "unknown"
    for attempt in range(retries):
        try:
            throttle_global()                       # เข้าคิวใหม่ทุกครั้ง
            r = session.post(endpoint, data=payload, timeout=20)
            if r.status_code == 200 and len(r.content) >= 100:
                out_path.parent.mkdir(parents=True, exist_ok=True)
                tmp = out_path.with_suffix(out_path.suffix + ".part")
                tmp.write_bytes(r.content)
                tmp.replace(out_path)               # เปลี่ยนชื่อทีเดียวจบ
                return True, len(r.content)
            if r.status_code == 429:
                time.sleep(2 ** attempt + 0.5)      # ถอยแบบทวีคูณ
                last_error = "rate limited"
                continue
            body = (r.text or "")[:150].replace("\n", " ")
            return False, f"HTTP {r.status_code}: {body}"   # ผิดถาวร ไม่ต้องลองซ้ำ
        except requests.RequestException as e:
            last_error = str(e)[:100]
            if attempt < retries - 1:
                time.sleep(1 + attempt)
                continue
            return False, last_error
    return False, f"max retries ({last_error})"

จุดที่ผมแก้เพิ่มทีหลังคือสองบรรทัด tmp กับ replace เดิมผมเขียนทับไฟล์ปลายทางตรง ๆ ซึ่งเป็นต้นตอของไฟล์เสียทั้งหมดที่เจอตอนท้าย ถ้าเน็ตหลุดตอนเขียนยังไม่จบ สิ่งที่ค้างอยู่บนดิสก์คือไฟล์นามสกุลถูกที่มีเนื้อไม่ครบ การเขียนลงชื่อชั่วคราวแล้วค่อยเปลี่ยนชื่อทำให้ไฟล์ปลายทางมีสองสถานะเท่านั้นคือไม่มีเลยกับครบ

ต่องานให้ดูจากของจริงบนดิสก์ ไม่ใช่จากไฟล์สถานะ

ตอนออกแบบใหม่ ๆ ผมเคยคิดจะเก็บไฟล์สถานะไว้ว่าทำถึงลำดับไหนแล้ว สุดท้ายทิ้งความคิดนั้นไป เพราะไฟล์สถานะกับของจริงบนดิสก์หลุดจากกันได้ตลอดเวลา ถ้าโปรแกรมถูกฆ่ากลางคัน ไฟล์สถานะจะค้างอยู่ที่ค่าก่อนหน้า และถ้ามีคนเข้าไปลบไฟล์ทิ้งเองก็ยิ่งเละ

วิธีที่ทนที่สุดคือสร้างรายการงานทั้งหมดขึ้นมาใหม่ทุกครั้งที่รัน แล้วคัดของที่มีอยู่แล้วออก ดิสก์คือแหล่งความจริงเพียงที่เดียว

tasks, cached = [], 0
for item_id, text, level in items:
    for alias in chosen_voices:
        out = AUDIO_DIR / alias / f"lv{level}" / f"item_{item_id}.mp3"
        if out.exists() and out.stat().st_size > 100:
            cached += 1
            continue
        tasks.append((item_id, text, level, alias, out))

total_target = len(items) * len(chosen_voices)
pct = cached / total_target * 100 if total_target else 0
print(f"Target: {total_target} · cached: {cached} · to generate: {len(tasks)}")
print(f"Resuming from {cached}/{total_target} ({pct:.1f}% already done)")
if not tasks:
    print("Nothing to do · all cached.")

เงื่อนไข st_size > 100 ในบรรทัดนั้นคือด่านหยาบ ๆ กันไฟล์ศูนย์ไบต์ที่เกิดตอนดิสก์เต็มหรือถูกฆ่าตอนเพิ่งเปิดไฟล์ มันกันได้แค่กรณีหยาบที่สุด ไฟล์ที่เนื้อขาดครึ่งมันไม่มีทางรู้ ตรงนี้เป็นงานของชิ้นที่สามที่จะพูดถึงต่อไป

ผลตอนรันรอบที่สองออกมาแบบนี้ อ่านแล้วรู้ทันทีว่าของเดิมไม่ได้ถูกทำซ้ำ

$ python tts_batch.py --workers 5 --rps 0.3
Rate limit: 0.3 req/sec (18 req/min)
Loaded 5370 items from DB
Target: 26850 files · cached(already exist): 13936 · to generate: 12914
Resuming from 13936/26850 (51.9% already done)
 52%|#####     | 13936/26850 [00:00<20:01:12, 0.18file/s, ok=0, fail=0]

แถบความคืบหน้าต้องเริ่มนับจากของเก่า ไม่ใช่จากศูนย์

เรื่องเล็กแต่มีผลกับคนเฝ้างานมาก ถ้าแถบเริ่มที่ 0 จาก 12,914 คนดูจะอ่านไม่ออกว่าตกลงงานทั้งกองเหลืออีกเท่าไร ผมเลยตั้งยอดรวมเป็นเป้าหมายจริงแล้วให้แถบเริ่มที่จำนวนที่ทำไว้แล้ว

with ThreadPoolExecutor(max_workers=args.workers) as ex:
    futures = {ex.submit(generate_one, session, endpoint, t): t for t in tasks}
    with tqdm(total=total_target, initial=cached, unit="file", smoothing=0.1) as pbar:
        for fut in as_completed(futures):
            ok_flag, info = fut.result()
            if ok_flag:
                ok += 1
            else:
                fail += 1
                if fail <= 20:
                    tqdm.write(f"FAIL {futures[fut][1]}: {info}")
            pbar.update(1)
            pbar.set_postfix({"ok": ok, "fail": fail})

initial=cached บรรทัดเดียวเปลี่ยนเลขบนจอจาก 0 จาก 12,914 เป็น 13,936 จาก 26,850 ซึ่งเป็นเลขที่เอาไปตอบคนถามได้ ส่วน if fail <= 20 มีไว้กันล็อกท่วมจอ ถ้าฝั่งปลายทางล่มทั้งระบบ เราไม่ต้องการบรรทัดผิดพลาดหมื่นสองพันบรรทัด เห็นยี่สิบบรรทัดแรกก็พอรู้แล้วว่าเป็นเรื่องเดียวกันหมด

ตัวตรวจต้องเป็นคนละโปรแกรม และต้องดูไบต์แรกของไฟล์

ชิ้นที่สามคือชิ้นที่ผมเพิ่มทีหลังสุด และเป็นชิ้นที่ทำให้กองไฟล์เชื่อถือได้จริง เหตุผลที่มันต้องแยกเป็นคนละโปรแกรมคือ ตัวยิงมีหน้าที่ทำให้ครบ ส่วนตัวตรวจมีหน้าที่ไม่เชื่อตัวยิง ถ้ารวมสองอย่างไว้ในโปรแกรมเดียว ตัวตรวจจะเชื่อผลของตัวเองเสมอ

หลักการตรวจมีแค่สองชั้น ชั้นแรกดูขนาดขั้นต่ำ ชั้นที่สองดูไบต์แรกว่าเป็นหัวไฟล์ของชนิดนั้นจริงไหม

def is_mp3_valid(path, min_size):
    try:
        size = path.stat().st_size
    except OSError as e:
        return False, f"stat error: {e}"
    if size < min_size:
        return False, f"size={size}<min({min_size})"
    with open(path, "rb") as f:
        head = f.read(10)
    if len(head) < 4:
        return False, "header<4 bytes"
    if head[0] == 0xFF and (head[1] & 0xE0) == 0xE0:   # เฟรม MPEG audio
        return True, ""
    if head[:3] == b"ID3":                              # มีแท็ก ID3v2 นำหน้า
        return True, ""
    return False, f"bad magic: {head[:4].hex()}"

ค่าขนาดขั้นต่ำอย่าตั้งมั่ว ผมคิดจากของจริงคือ 24 กิโลเฮิรตซ์ที่ 96 กิโลบิตต่อวินาที เท่ากับ 12 กิโลไบต์ต่อวินาที คำที่สั้นที่สุดยาวราวครึ่งวินาที ตั้งที่ 2,000 ไบต์เลยเป็นเส้นที่ของดีไม่มีทางต่ำกว่า ถ้าตั้งสูงไปจะไล่ลบของดีทิ้ง ถ้าตั้งต่ำไปก็ปล่อยของเสียผ่าน

ให้ตัวตรวจลบเอง แล้วปล่อยตัวยิงวนกลับมาอีกรอบ

ผมไม่ให้ตัวตรวจสั่งสร้างไฟล์ใหม่เอง เพราะจะกลายเป็นสองโปรแกรมที่ยิง API ได้ทั้งคู่ และตัวหน่วงของกลางจะคุมไม่ถึงกัน ทางที่สะอาดกว่าคือให้ตัวตรวจทำได้แค่ลบ แล้วปล่อยให้ตัวยิงเดิมมองเห็นเป็นช่องว่างแล้วเติมให้เองในรอบถัดไป

# 1) ตรวจอย่างเดียว ยังไม่แตะอะไร
python verify_files.py --verbose

# 2) พอใจแล้วค่อยลบตัวเสีย
python verify_files.py --fix

# 3) วนกลับไปยิงรอบใหม่ เดี๋ยวมันเห็นช่องว่างแล้วเติมเอง
python tts_batch.py --workers 5 --rps 0.3

ค่าปริยายของตัวตรวจต้องเป็นโหมดรายงานเสมอ ส่วนการลบต้องพิมพ์ธงเพิ่มเอง ผมโดนมาแล้วกับสคริปต์ที่ลบเป็นค่าปริยาย ตั้งค่าขนาดขั้นต่ำผิดตัวเดียวแล้วของดีหายไปหลายพันไฟล์ ต้องนั่งยิงใหม่ข้ามคืน

รายงานที่ออกมาผมให้แยกตามกลุ่มด้วย เพราะถ้าของเสียกระจุกอยู่กลุ่มเดียวแปลว่าเป็นปัญหาของอินพุตกลุ่มนั้น ไม่ใช่ปัญหาเน็ต

ตรวจสอบ 26850 ไฟล์ · OK: 26761 · BAD: 89

Per group:
  group-a       5361     9
  group-b       5352    18
  group-c       5310    60
  group-d       5369     1
  group-e       5369     1

BAD breakdown:
  group-c/lv4: 47
  group-c/lv5: 13

กรณีข้างบนนี่เห็นแล้วรู้เลยว่ากลุ่ม c ช่วง lv4 มีอะไรบางอย่างต่างจากชาวบ้าน ไม่ใช่เน็ตหลุดแบบสุ่ม เพราะเน็ตหลุดมันไม่เลือกกลุ่ม

ลำดับที่ผมใช้จริง เอาไปวางกับงานอื่นได้เลย

1. หาเพดานจริงก่อนเขียนโค้ด ว่าเป็นกี่ครั้งต่อกี่วินาที และนับต่อบัญชีหรือต่อเส้น แล้วตั้งค่าที่ใช้จริงต่ำกว่านั้นสักสิบเปอร์เซ็นต์
2. ทำตัวหน่วงของกลาง ตัวเดียวที่ทุกเส้นต้องผ่าน และเรียก sleep นอกล็อกเสมอ
3. แยกผิดถาวรกับผิดชั่วคราว กลุ่มแรกคืนค่าทันที กลุ่มหลังถอยแบบทวีคูณไม่เกิน 5 ครั้ง
4. เขียนลงชื่อชั่วคราวแล้วค่อยเปลี่ยนชื่อ ไฟล์ปลายทางจะมีแค่สองสถานะคือไม่มีกับครบ
5. สร้างรายการงานใหม่ทุกครั้งที่รัน แล้วคัดของที่มีบนดิสก์ออก ห้ามพึ่งไฟล์สถานะ
6. ให้แถบความคืบหน้าเริ่มจากของเก่า คนเฝ้างานจะได้อ่านเลขรวมออก
7. เขียนตัวตรวจเป็นคนละโปรแกรม ดูขนาดขั้นต่ำกับไบต์แรกของไฟล์ และตั้งโหมดรายงานเป็นค่าปริยาย
8. ให้ตัวตรวจทำได้แค่ลบ แล้ววนกลับไปให้ตัวยิงเติมช่องว่าง ไม่ให้สองโปรแกรมยิง API พร้อมกัน

เรื่องค่าเชื่อมต่อกับรหัสผ่าน ผมไม่เขียนลงในสคริปต์เด็ดขาด อ่านจากตัวแปรสภาพแวดล้อมอย่างเดียว เวลาเอาสคริปต์ไปวางเครื่องอื่นหรือส่งให้คนอื่นดูจะได้ไม่มีอะไรติดไปด้วย

import os, sys, requests

def make_session():
    key = os.environ.get("TTS_API_KEY")
    if not key:
        sys.exit("ยังไม่ได้ตั้ง TTS_API_KEY")
    s = requests.Session()
    s.headers.update({"Authorization": f"Bearer {key}",
                      "Content-Type": "application/ssml+xml"})
    s.mount("https://", requests.adapters.HTTPAdapter(
        pool_connections=32, pool_maxsize=64, max_retries=2))
    return s

ส่วนตัวผมคิดว่าสามชิ้นนี้ใช้ได้กับงานยิงเป็นกองทุกแบบ ไม่ใช่แค่งานสร้างเสียง จะเป็นดึงรูป แปลงเอกสาร หรือเรียกโมเดลอะไรก็ตาม ตราบใดที่ปลายทางมีเพดานและงานกินเวลาข้ามคืน สูตรเดิมนี้ใช้ซ้ำได้หมด ใครจะปรับตัวเลขหรือย้ายไปเขียนด้วยภาษาอื่นก็ตามสะดวกครับ ตรงไหนผมอธิบายตกหล่นหรือพิมพ์ผิด เดี๋ยวกลับมาแก้เพิ่มให้ภายหลังครับ