cfc/运营文案/xhs_feishu.py

"""
飞书企业机器人推送工具
======================
通过飞书开放平台（open.feishu.cn API）发送富文本卡片消息 + 图片附件。

用法：
    from xhs_feishu import send_daily_report
    send_daily_report(config_path, date, overview, notes, screenshot_path)

依赖：
    requests（第三方库）
    配置 JSON 文件（含 feishu_app_id / feishu_app_secret / feishu_chat_id）
"""
import io
import json
import os
import sys
import time

import requests

FEISHU_TOKEN_URL = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
FEISHU_MSG_URL = "https://open.feishu.cn/open-apis/im/v1/messages"
FEISHU_IMG_URL = "https://open.feishu.cn/open-apis/im/v1/images"

# 进程内 token 缓存：{key: (token, expire_ts)}
_TOKEN_CACHE = {}


def _get_token(app_id: str, app_secret: str) -> str:
    """获取 tenant_access_token（带进程内缓存，过期的自动刷新）。

    飞书 token 默认有效期 7200s，这里提前 60s 视为过期。
    """
    cache_key = f"{app_id}:{app_secret}"
    now = time.time()
    if cache_key in _TOKEN_CACHE:
        token, expire_ts = _TOKEN_CACHE[cache_key]
        if expire_ts > now:
            return token
    resp = requests.post(
        FEISHU_TOKEN_URL,
        json={"app_id": app_id, "app_secret": app_secret},
        timeout=15,
    )
    resp.raise_for_status()
    data = resp.json()
    if data.get("code") != 0:
        raise RuntimeError(f"获取飞书 token 失败: {data.get('msg')}")
    token = data["tenant_access_token"]
    expire = data.get("expire", 7200)
    _TOKEN_CACHE[cache_key] = (token, now + expire - 60)
    return token


def _upload_image(token: str, image_path: str) -> str:
    """上传图片到飞书，返回 image_key。

    注意：`image_type` 是 multipart form-data 的字段（不是 URL query 参数）。
    此前误写成 `?image_type=message` 会导致 400 Bad Request（code=234001）。
    """
    url = FEISHU_IMG_URL
    with open(image_path, "rb") as f:
        resp = requests.post(
            url,
            headers={"Authorization": f"Bearer {token}"},
            files={"image": (os.path.basename(image_path), f, "image/jpeg")},
            data={"image_type": "message"},
            timeout=30,
        )
    resp.raise_for_status()
    data = resp.json()
    if data.get("code") != 0:
        raise RuntimeError(f"上传图片失败: {data.get('msg')}")
    return data["data"]["image_key"]


def _build_card_json(date: str, overview: dict, notes: list) -> str:
    """构建飞书富文本卡片 JSON 字符串。

    overview 字段: fans_count, interact_count, follow_count,
                   period_7d: {曝光数/观看数/点赞数/评论数/收藏数/分享数/净涨粉...},
                   period_30d: 同上
    """
    ov = overview or {}

    def _period_lines(period, label):
        """格式化一个周期段"""
        if not period:
            return [f"⚠️ {label}数据暂不可用"]
        return [
            f"{label} 曝光 {period.get('曝光数', '-')}",
            f"     观看 {period.get('观看数', '-')} | 点赞 {period.get('点赞数', '-')} | "
            f"评论 {period.get('评论数', '-')} | 收藏 {period.get('收藏数', '-')} | "
            f"分享 {period.get('分享数', '-')}",
            f"     净涨粉 {period.get('净涨粉', '-')}",
        ]

    overview_lines = [
        f"粉丝数：{ov.get('fans_count', 0)}  关注：{ov.get('follow_count', 0)}  "
        f"获赞与收藏：{ov.get('interact_count', 0)}",
    ]
    # 近7日
    overview_lines += _period_lines(ov.get("period_7d", {}), "近7日")
    # 近30日（如果有）
    if ov.get("period_30d"):
        overview_lines += _period_lines(ov.get("period_30d", {}), "近30日")
    note_lines = []
    for i, n in enumerate(notes, 1):
        note_lines.append(
            f"{i}. 《{n.get('title', '')}》\n"
            f"   📅{n.get('publish_date', '')}  "
            f"👀{n.get('read_count', 0):,}  "
            f"❤️{n.get('like_count', 0):,}  "
            f"⭐{n.get('collect_count', 0):,}  "
            f"💬{n.get('comment_count', 0):,}  "
            f"🔄{n.get('share_count', 0):,}\n"
            f"   📮未回复评论：{n.get('unreplied_comment_count', 0)}"
        )

    content = "\n\n".join([
        "**📈 账号概览**\n" + "\n".join(overview_lines),
        f"**📝 笔记列表（{len(notes)}篇）**\n" + "\n\n".join(note_lines),
        "🖼️ 附：创作者中心页面截图",
    ])

    card = {
        "config": {"wide_screen_mode": True},
        "header": {
            "template": "blue",
            "title": {"tag": "plain_text", "content": f"小红书创作者中心日报 | {date}"},
        },
        "elements": [
            {"tag": "div", "text": {"tag": "lark_md", "content": content}},
        ],
    }
    return json.dumps(card, ensure_ascii=False)


def _send_message(token: str, cfg: dict, msg_type: str, content: str) -> dict:
    """发送一条飞书消息，返回响应数据。

    重要：飞书 im/v1/messages 接口的 receive_id_type 是 URL 查询参数，
    不是 JSON body 字段（否则报 99992402 receive_id_type is required）。
    """
    rtype = cfg.get("receive_id_type", "chat_id")
    url = f"{FEISHU_MSG_URL}?receive_id_type={rtype}"
    body = {
        "receive_id": cfg["feishu_chat_id"],
        "msg_type": msg_type,
        "content": content,
    }
    resp = requests.post(
        url,
        headers={"Authorization": f"Bearer {token}"},
        json=body,
        timeout=30,
    )
    resp.raise_for_status()
    data = resp.json()
    if data.get("code") != 0:
        raise RuntimeError(f"发送飞书{msg_type}消息失败: {data.get('msg')}")
    return data


def send_daily_report(
    config_path: str,
    date: str,
    overview: dict,
    notes: list,
    screenshot_path: str = None,
    screenshot_paths: list = None,
):
    """发送小红书创作者中心日报到飞书群。

    Args:
        config_path: xhs_daily_config.json 路径
        date: 报告日期 YYYY-MM-DD
        overview: 账号总览 dict
        notes: 笔记列表 list[dict]
        screenshot_path: 单张截图文件路径（可选，向后兼容）
        screenshot_paths: 多张截图文件路径列表（可选，优先使用）
    """
    with open(config_path, "r", encoding="utf-8") as f:
        cfg = json.load(f)
    required = ["feishu_app_id", "feishu_app_secret", "feishu_chat_id"]
    missing = [k for k in required if not cfg.get(k)]
    if missing:
        raise ValueError(f"配置缺少必填字段: {', '.join(missing)}")

    token = _get_token(cfg["feishu_app_id"], cfg["feishu_app_secret"])
    card_json = _build_card_json(date, overview, notes)
    data = _send_message(token, cfg, "interactive", card_json)

    # 收集要附发的截图
    shot_list = []
    if screenshot_paths:
        shot_list = [p for p in screenshot_paths if p and os.path.exists(p)]
    elif screenshot_path and os.path.exists(screenshot_path):
        shot_list = [screenshot_path]

    for shot in shot_list:
        try:
            image_key = _upload_image(token, shot)
            _send_message(
                token, cfg, "image",
                json.dumps({"image_key": image_key}),
            )
        except Exception as e:
            print(f"[WARN] 发送截图失败，但卡片已推送: {e}", flush=True)

    print(f"[OK] 飞书日报已推送，message_id={data['data']['message_id']}", flush=True)
    return data["data"]["message_id"]


if __name__ == "__main__":

    if not sys.stdout.isatty():
        try:
            sys.stdout.reconfigure(encoding='utf-8', errors='replace')
        except Exception:
            pass
    print("飞书推送工具模块加载成功")
    print(f"token URL: {FEISHU_TOKEN_URL}")