soul/scripts/send_poster_to_feishu.py

#!/usr/bin/env python3
"""
Soul 文章海报发飞书：生成海报图片（含小程序码）并发送，不发链接

流程：解析文章 → 调 Soul API 获取小程序码 → 合成海报图 → 上传飞书 → 以图片形式发送

用法:
  python3 scripts/send_poster_to_feishu.py <文章md路径>
  python3 scripts/send_poster_to_feishu.py --id 9.15

环境: pip install Pillow requests
飞书图片上传需配置: FEISHU_APP_ID, FEISHU_APP_SECRET（与 webhook 同租户的飞书应用）
"""

import argparse
import base64
import io
import json
import os
from pathlib import Path

# 可选：从脚本同目录 .env.feishu 加载 FEISHU_APP_ID, FEISHU_APP_SECRET
_env = Path(__file__).resolve().parent / ".env.feishu"
if _env.exists():
    for line in _env.read_text().strip().split("\n"):
        if "=" in line and not line.strip().startswith("#"):
            k, v = line.split("=", 1)
            os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))
import os
import re
import sys
import tempfile
from pathlib import Path
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError

FEISHU_WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/8b7f996e-2892-4075-989f-aa5593ea4fbc"
SOUL_API = "https://soul.quwanzhi.com/api/miniprogram/qrcode"
POSTER_W, POSTER_H = 600, 900  # 2x 小程序 300x450


def parse_article(filepath: Path) -> dict:
    """从 md 解析：标题、金句、日期、section id"""
    text = filepath.read_text(encoding="utf-8")
    lines = [l.strip() for l in text.split("\n")]

    title = ""
    date_line = ""
    quote = ""
    section_id = ""

    for line in lines:
        if line.startswith("# "):
            title = line.lstrip("# ").strip()
            m = re.match(r"^(\d+\.\d+)\s", title)
            if m:
                section_id = m.group(1)
        elif re.match(r"^\d{4}年\d{1,2}月\d{1,2}日", line):
            date_line = line
        elif title and not quote and line and not line.startswith("-") and "---" not in line:
            if 10 <= len(line) <= 120:
                quote = line

    if not section_id and filepath.stem:
        m = re.match(r"^(\d+\.\d+)", filepath.stem)
        if m:
            section_id = m.group(1)

    return {
        "title": title or filepath.stem,
        "quote": quote or title or "来自 Soul 创业派对的真实故事",
        "date": date_line,
        "section_id": section_id or "9.1",
    }


def fetch_qrcode(section_id: str) -> bytes | None:
    """从 Soul 后端获取小程序码图片"""
    scene = f"id={section_id}"
    body = json.dumps({"scene": scene, "page": "pages/read/read", "width": 280}).encode()
    req = Request(SOUL_API, data=body, headers={"Content-Type": "application/json"}, method="POST")
    try:
        with urlopen(req, timeout=15) as resp:
            data = json.loads(resp.read().decode())
            if not data.get("success") or not data.get("image"):
                return None
            b64 = data["image"].split(",", 1)[-1] if "," in data["image"] else data["image"]
            return base64.b64decode(b64)
    except Exception as e:
        print("获取小程序码失败:", e)
        return None


def draw_poster(data: dict, qr_bytes: bytes | None) -> bytes:
    """合成海报图，返回 PNG bytes"""
    try:
        from PIL import Image, ImageDraw, ImageFont
    except ImportError:
        print("需要安装 Pillow: pip install Pillow")
        sys.exit(1)

    img = Image.new("RGB", (POSTER_W, POSTER_H), color=(26, 26, 46))
    draw = ImageDraw.Draw(img)

    # 顶部装饰条
    draw.rectangle([0, 0, POSTER_W, 8], fill=(0, 206, 209))

    # 字体（系统 fallback）
    try:
        font_sm = ImageFont.truetype("/System/Library/Fonts/PingFang.ttc", 24)
        font_md = ImageFont.truetype("/System/Library/Fonts/PingFang.ttc", 32)
        font_lg = ImageFont.truetype("/System/Library/Fonts/PingFang.ttc", 40)
    except OSError:
        font_sm = font_md = font_lg = ImageFont.load_default()

    y = 50
    draw.text((40, y), "Soul创业派对", fill=(255, 255, 255), font=font_sm)
    y += 60

    # 标题（多行）
    title = data["title"]
    for i in range(0, len(title), 18):
        draw.text((40, y), title[i : i + 18], fill=(255, 255, 255), font=font_lg)
        y += 50
    y += 20

    draw.line([(40, y), (POSTER_W - 40, y)], fill=(255, 255, 255, 40), width=1)
    y += 40

    # 金句
    quote = data["quote"]
    for i in range(0, min(len(quote), 80), 20):
        draw.text((40, y), quote[i : i + 20], fill=(255, 255, 255, 200), font=font_md)
        y += 40
    if data["date"]:
        draw.text((40, y), data["date"], fill=(255, 255, 255, 180), font=font_sm)
        y += 40
    y += 20

    # 底部提示 + 小程序码
    draw.text((40, POSTER_H - 120), "长按识别小程序码", fill=(255, 255, 255), font=font_sm)
    draw.text((40, POSTER_H - 90), "阅读全文", fill=(255, 255, 255, 180), font=font_sm)

    if qr_bytes:
        qr_img = Image.open(io.BytesIO(qr_bytes))
        qr_img = qr_img.resize((160, 160))
        img.paste(qr_img, (POSTER_W - 200, POSTER_H - 180))

    buf = io.BytesIO()
    img.save(buf, format="PNG")
    return buf.getvalue()


def get_feishu_token() -> str | None:
    app_id = os.environ.get("FEISHU_APP_ID")
    app_secret = os.environ.get("FEISHU_APP_SECRET")
    if not app_id or not app_secret:
        return None
    req = Request(
        "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
        data=json.dumps({"app_id": app_id, "app_secret": app_secret}).encode(),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(req, timeout=10) as resp:
            data = json.loads(resp.read().decode())
            return data.get("tenant_access_token")
    except Exception:
        return None


def upload_image_to_feishu(png_bytes: bytes, token: str) -> str | None:
    """上传图片到飞书，返回 image_key"""
    try:
        import requests
    except ImportError:
        print("需要安装: pip install requests")
        return None

    headers = {"Authorization": f"Bearer {token}"}
    files = {"image": ("poster.png", png_bytes, "image/png")}
    data = {"image_type": "message"}
    try:
        r = requests.post(
            "https://open.feishu.cn/open-apis/im/v1/images",
            headers=headers,
            data=data,
            files=files,
            timeout=15,
        )
        j = r.json()
        if j.get("code") == 0 and j.get("data", {}).get("image_key"):
            return j["data"]["image_key"]
    except Exception as e:
        print("上传飞书失败:", e)
    return None


def send_image_to_feishu(image_key: str) -> bool:
    payload = {"msg_type": "image", "content": {"image_key": image_key}}
    req = Request(
        FEISHU_WEBHOOK,
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urlopen(req, timeout=10) as resp:
            r = json.loads(resp.read().decode())
            return r.get("code") == 0
    except Exception as e:
        print("发送失败:", e)
        return False


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("file", nargs="?", help="文章 md 路径")
    parser.add_argument("--id", help="章节 id")
    parser.add_argument("--save", help="仅保存海报到本地路径，不发飞书")
    args = parser.parse_args()

    base = Path("/Users/karuo/Documents/个人/2、我写的书/《一场soul的创业实验》/第四篇｜真实的赚钱/第9章｜我在Soul上亲访的赚钱案例")

    if args.id:
        candidates = list(base.glob(f"{args.id}*.md"))
        if not candidates:
            print(f"未找到 id={args.id} 的文章")
            sys.exit(1)
        filepath = candidates[0]
    elif args.file:
        filepath = Path(args.file)
        if not filepath.exists():
            print("文件不存在:", filepath)
            sys.exit(1)
    else:
        parser.print_help()
        sys.exit(1)

    data = parse_article(filepath)
    print("生成海报:", data["title"])

    qr_bytes = fetch_qrcode(data["section_id"])
    png_bytes = draw_poster(data, qr_bytes)

    if args.save:
        Path(args.save).write_bytes(png_bytes)
        print("已保存:", args.save)
        return

    token = get_feishu_token()
    if not token:
        out = Path(tempfile.gettempdir()) / f"soul_poster_{data['section_id']}.png"
        out.write_bytes(png_bytes)
        print("未配置 FEISHU_APP_ID / FEISHU_APP_SECRET，无法上传。海报已保存到:", out)
        sys.exit(1)

    image_key = upload_image_to_feishu(png_bytes, token)
    if not image_key:
        print("上传图片失败")
        sys.exit(1)

    if send_image_to_feishu(image_key):
        print("已发送到飞书（图片）")
    else:
        sys.exit(1)


if __name__ == "__main__":
    main()