ส่งไฟล์จากงานอัตโนมัติเข้า Telegram กับเพดาน 50 MB ของ Bot API

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

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

ขอบเขต: ออกทางเดียว ไม่มีขาเข้า

ข้อนี้ต้องพูดก่อนเลยเพราะเป็นหัวใจของการออกแบบ ผมเคยเขียนเรื่อง วาง Telegram Bot เฝ้า DA Server แบบ blast radius เล็ก ไว้แล้ว ตัวนั้นเป็นบอทที่รับคำสั่ง จากแชทมาสั่งงานเซิร์ฟเวอร์ ซึ่งเป็นคนละเรื่องกับตัวนี้โดยสิ้นเชิง

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

เพดานจริงของ Bot API ที่ต้องรู้ก่อนเขียน

เรื่องที่พลาดกันบ่อยที่สุดคือรูป ถ้าส่งรูปด้วย sendPhoto ฝั่ง Telegram จะบีบให้เล็กลงเสมอ ภาพหน้าจอที่แคปมาเพื่อดูตัวหนังสือเล็ก ๆ จะอ่านไม่ออกทันที และถ้ารูปใหญ่เกิน 10 MB มันจะปฏิเสธไปเลย ทางแก้คือส่งเป็น sendDocument แทน ซึ่งไม่แตะไฟล์เลยแม้แต่ไบต์เดียว

โค้ดส่วนที่เลือกวิธีส่งสั้นมาก แต่เป็นส่วนที่ใช้บ่อยที่สุด:

PHOTO_EXT = {".jpg", ".jpeg", ".png", ".webp"}
VIDEO_EXT = {".mp4", ".mov", ".m4v"}
MAX_BYTES = 50 * 1024 * 1024
PHOTO_MAX = 10 * 1024 * 1024

def send_file(path: Path, caption: str = "", as_doc: bool = False) -> dict:
    if not path.is_file():
        raise FileNotFoundError(str(path))
    size = path.stat().st_size
    if size > MAX_BYTES:
        raise ValueError(f"{path.name} is {size/1048576:.1f} MB, over the 50 MB bot limit")
    ext = path.suffix.lower()
    if not as_doc and ext in PHOTO_EXT and size <= PHOTO_MAX:
        method, field = "sendPhoto", "photo"
    elif not as_doc and ext in VIDEO_EXT:
        method, field = "sendVideo", "video"
    else:
        method, field = "sendDocument", "document"
    mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    with path.open("rb") as fh:
        res = tg(method, {"chat_id": CHAT_ID, "caption": caption[:1024]},
                 {field: (path.name, fh, mime)})
    log(f"{method} {path.name} ({size/1024:.0f} KB)")
    return res

จุดที่ต้องระวังคือ caption[:1024] กับ text[:4096] ต้องตัดเองตั้งแต่ฝั่งเรา ถ้าปล่อยให้เกินแล้วส่งไป API จะตอบ 400 กลับมาแล้วไฟล์ไม่ถูกส่งเลยทั้งที่ตัวไฟล์ไม่มีปัญหาอะไร เสียเที่ยวเพราะข้อความยาวเกินอย่างเดียว

ตัวรับในเครื่อง: ผูกกับ 127.0.0.1 เท่านั้น

ตัวบริการเป็น HTTP server เล็ก ๆ จาก http.server ใน stdlib ล้วน ฟังอยู่ที่ 127.0.0.1 พอร์ตเดียว มีสามปลายทางเท่านั้นคือ /health, /message และ /file

class Handler(BaseHTTPRequestHandler):
    def _authed(self) -> bool:
        return self.headers.get("X-Token") == API_TOKEN

    def do_POST(self):
        route = self.path.split("?")[0]
        if not self._authed():
            return self._reply(401, {"ok": False, "error": "bad or missing X-Token"})
        body = self._body()
        if route == "/message":
            text = body.get("text") or ""
            if not text.strip():
                return self._reply(400, {"ok": False, "error": "text is empty"})
            res = send_message(text)
            return self._reply(200, {"ok": True, "message_id": res.get("message_id")})
        if route == "/file":
            p = body.get("path")
            if not p:
                return self._reply(400, {"ok": False, "error": "path is required"})
            res = send_file(Path(p), body.get("caption", ""), bool(body.get("as_doc")))
            return self._reply(200, {"ok": True, "message_id": res.get("message_id")})
        return self._reply(404, {"ok": False, "error": "unknown endpoint"})

def main() -> None:
    srv = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)      # ห้ามเป็น 0.0.0.0
    srv.serve_forever()

ความลับทั้งหมดอยู่ในไฟล์ตั้งค่าไฟล์เดียวข้างนอกโค้ด:

{
  "bot_token": "<bot_token ของคุณ>",
  "chat_id":   "<หมายเลขแชทของคุณ>",
  "api_token": "<สตริงสุ่มยาว ๆ สำหรับ X-Token>",
  "port": 8787
}

CLI ที่ยังทำงานได้ตอนบริการล่ม

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

def service_up(c: dict) -> bool:
    try:
        return requests.get(service_url(c) + "/health", timeout=3).json().get("ok") is True
    except Exception:
        return False

up = False if args.direct else service_up(c)
for f in args.files:
    p = Path(f)
    if not p.is_file():
        sys.exit(f"not a file: {p}")
    if up:
        svc_post(c, "/file", {"path": str(p.resolve()),
                              "caption": args.caption, "as_doc": args.as_doc})
    else:
        direct_file(c, p, args.caption, args.as_doc)      # ทางถอย ยิงตรง
    print(f"[OK] {p.name} ({p.stat().st_size/1024:.0f} KB)")

เรียกใช้จริงหน้าตาแบบนี้ ซึ่งเป็นสิ่งเดียวที่ผมต้องจำ:

PY="$USERPROFILE/anaconda3/envs/ml313/python.exe"
"$PY" ~/.claude/tools/tg_send.py --check                      # เช็กบริการ + บอท
"$PY" ~/.claude/tools/tg_send.py -m "งานเมื่อคืนเสร็จแล้ว"
"$PY" ~/.claude/tools/tg_send.py out.pdf shot.png -c "รายงาน"
"$PY" ~/.claude/tools/tg_send.py big.png --as-doc             # ไม่ให้บีบรูป

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

@echo off
set PY=%USERPROFILE%\anaconda3\envs\ml313\python.exe
"%PY%" C:\jobs\build_report.py > C:\jobs\out\report.log 2>&1
if errorlevel 1 (
  "%PY%" %USERPROFILE%\.claude\tools\tg_send.py C:\jobs\out\report.log -c "งานล้ม ดู log"
) else (
  "%PY%" %USERPROFILE%\.claude\tools\tg_send.py C:\jobs\out\report.pdf -c "รายงานประจำวัน"
)

กับดักภาษาไทยตอนเรียกผ่าน curl บน Windows

อันนี้เสียเวลาผมไปพักใหญ่กว่าจะรู้ว่าไม่ใช่ความผิดของโค้ด ถ้าเรียกปลายทาง /message ด้วย curl แล้ววางข้อความไทยลงใน --data ตรง ๆ จากหน้าต่างคำสั่งบนเครื่องนี้ ข้อความที่ไปถึงแชทจะเป็นตัวยึกยือทั้งหมด

# ผิด - ข้อความไทยผ่านบรรทัดคำสั่งแล้วเสียตั้งแต่ต้นทาง
curl -X POST http://127.0.0.1:8787/message \
     -H "X-Token: $TOKEN" --data '{"text":"งานเสร็จแล้ว"}'

# ถูก - เขียนเป็นไฟล์ UTF-8 ก่อน แล้วส่งไบต์ดิบ
printf '{"text":"งานเสร็จแล้ว"}' > /tmp/msg.json
curl -X POST http://127.0.0.1:8787/message \
     -H "X-Token: $TOKEN" -H "Content-Type: application/json; charset=utf-8" \
     --data-binary @/tmp/msg.json

# เช็กสุขภาพ ไม่ต้องใช้ token
curl -s http://127.0.0.1:8787/health

ให้มันเปิดเองตอนล็อกอิน และเห็นได้ว่ายังอยู่

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

@echo off
title Telegram Local API :8787
taskkill /f /fi "WINDOWTITLE eq Telegram Local API :8787" >nul 2>&1
"%USERPROFILE%\anaconda3\envs\ml313\python.exe" "%USERPROFILE%\.claude\tools\tg_local_api.py"

เอา shortcut ของ bat ตัวนี้ไปวางในโฟลเดอร์ Startup แล้วจบ ไม่ต้องลงทะเบียนเป็น service ให้ยุ่งยาก

1. สร้างบอทกับ BotFather เก็บ token ไว้ในไฟล์ตั้งค่านอกโค้ด
2. ทักหาบอทของตัวเองหนึ่งข้อความก่อน บอทเริ่มแชทกับเราเองไม่ได้
3. เขียนบริการที่ผูก 127.0.0.1 อย่างเดียว และตรวจ X-Token ทุกคำขอ
4. ให้ CLI มีทางถอยยิงตรงเมื่อบริการไม่ได้เปิด งานจะได้ไม่ตายตาม
5. ตัด caption ที่ 1024 และข้อความที่ 4096 ตั้งแต่ฝั่งเรา อย่ารอให้ API ปฏิเสธ

ข้อที่สองในรายการข้างบนสำคัญกว่าที่คิด บอทของ Telegram เริ่มบทสนทนากับใครเองไม่ได้ ต้องให้ปลายทางทักมาก่อนอย่างน้อยหนึ่งครั้ง ตอนตั้งครั้งแรกผมนั่งงงกับ chat not found อยู่พักหนึ่งเพราะยังไม่เคยกดคุยกับบอทตัวเองเลย

สรุป

โครงที่ผมใช้อยู่มีแค่สามชิ้นคือ บริการเล็ก ๆ ในเครื่องที่ฟัง localhost, CLI ที่มีทางถอย และไฟล์ตั้งค่าที่เก็บความลับไว้ที่เดียว งานอัตโนมัติทุกตัวบนเครื่องเรียกใช้ผ่าน CLI ตัวเดียวกันหมด เลยได้ log รวมที่เดียวด้วย

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

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