soul-yongping/scripts/send_feishu_text_and_images.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
向开发群 webhook 发送一条长文本 + 若干本地 PNG（先上传飞书再发 image_key）。
依赖：与 send_chapter_poster_to_feishu.py 相同，需 scripts/.env.feishu 内 FEISHU_APP_ID / FEISHU_APP_SECRET。

用法:
  python3 send_feishu_text_and_images.py --text-file recap.txt \\
    --images a.png b.png
  python3 send_feishu_text_and_images.py -t "单行文本" --images x.png
"""
from __future__ import annotations

import argparse
import os
import sys
from pathlib import Path

SCRIPT_DIR = Path(__file__).resolve().parent

try:
    import requests
except ImportError:
    print("pip install requests", file=sys.stderr)
    sys.exit(1)

DEFAULT_WEBHOOK = os.environ.get(
    "FEISHU_DEV_GROUP_WEBHOOK",
    "https://open.feishu.cn/open-apis/bot/v2/hook/c558df98-e13a-419f-a3c0-7e428d15f494",
)


def load_env_feishu():
    p = SCRIPT_DIR / ".env.feishu"
    if not p.is_file():
        return
    for line in p.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if line and not line.startswith("#") and "=" in line:
            k, v = line.split("=", 1)
            os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))


def tenant_token() -> str | None:
    load_env_feishu()
    app_id = os.environ.get("FEISHU_APP_ID", "")
    sec = os.environ.get("FEISHU_APP_SECRET", "")
    if not app_id or not sec:
        print("缺少 FEISHU_APP_ID / FEISHU_APP_SECRET（.env.feishu）", file=sys.stderr)
        return None
    r = requests.post(
        "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
        json={"app_id": app_id, "app_secret": sec},
        timeout=15,
    )
    data = r.json() or {}
    if data.get("code") != 0:
        print("token 失败:", data, file=sys.stderr)
        return None
    return data.get("tenant_access_token")


def send_text(webhook: str, text: str) -> bool:
    r = requests.post(webhook, json={"msg_type": "text", "content": {"text": text}}, timeout=15)
    d = r.json() or {}
    if d.get("code") != 0:
        print("文本发送失败:", d, file=sys.stderr)
        return False
    return True


def upload_png(token: str, path: Path) -> str | None:
    url = "https://open.feishu.cn/open-apis/im/v1/images"
    headers = {"Authorization": f"Bearer {token}"}
    with path.open("rb") as f:
        r = requests.post(
            url,
            headers=headers,
            files={"image": (path.name, f, "image/png")},
            data={"image_type": "message"},
            timeout=60,
        )
    out = r.json() or {}
    if out.get("code") != 0:
        print("上传失败", path, out, file=sys.stderr)
        return None
    return (out.get("data") or {}).get("image_key")


def send_image(webhook: str, image_key: str) -> bool:
    r = requests.post(
        webhook,
        json={"msg_type": "image", "content": {"image_key": image_key}},
        timeout=15,
    )
    d = r.json() or {}
    if d.get("code") != 0:
        print("图片消息失败:", d, file=sys.stderr)
        return False
    return True


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("-t", "--text", default="", help="直接传入文本")
    ap.add_argument("--text-file", type=Path, help="从文件读文本（utf-8）")
    ap.add_argument("--webhook", "-w", default=DEFAULT_WEBHOOK)
    ap.add_argument("--images", "-i", nargs="*", default=[], help="PNG 路径列表")
    args = ap.parse_args()

    body = args.text.strip()
    if args.text_file:
        body = args.text_file.read_text(encoding="utf-8").strip()
    if not body:
        ap.error("需要 -t 或 --text-file")

    if not send_text(args.webhook, body[:20000]):
        sys.exit(1)
    print("已发文本")

    if not args.images:
        return

    tok = tenant_token()
    if not tok:
        sys.exit(1)
    for p in args.images:
        path = Path(p).expanduser().resolve()
        if not path.is_file():
            print("跳过（不存在）:", path, file=sys.stderr)
            continue
        key = upload_png(tok, path)
        if key and send_image(args.webhook, key):
            print("已发图:", path.name)


if __name__ == "__main__":
    main()