Xiaohongshu

# NanoBanana API 配置
# 获取API Key: https://nanobananaapi.ai/api-key

NANOBANANA_API_KEY=your_api_key_here

# 可选：自定义API地址（默认使用官方地址）
# NANOBANANA_API_URL=https://api.nanobananaapi.ai/api/v1/nanobanana

{
  "feeds": [],
  "meta": {
    "created_at": "2026-02-04T17:59:12.166756",
    "version": 1
  }
}

requests

#!/usr/bin/env python3
"""
小红书高赞笔记本地数据库
支持：添加、查询、分析、导出
"""

import json
import sys
import os
from pathlib import Path
from datetime import datetime
from collections import Counter

DB_DIR = Path.cwd() / "database"
DB_FILE = DB_DIR / "feeds.json"
SUMMARY_FILE = DB_DIR / "summary.md"


def _load_db() -> dict:
    if DB_FILE.exists():
        with open(DB_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {"feeds": [], "meta": {"created_at": datetime.now().isoformat(), "version": 1}}


def _save_db(db: dict):
    DB_DIR.mkdir(parents=True, exist_ok=True)
    with open(DB_FILE, "w", encoding="utf-8") as f:
        json.dump(db, f, ensure_ascii=False, indent=2)


def add(feed_json: str):
    """添加一条笔记到数据库，输入为 JSON 字符串"""
    feed = json.loads(feed_json)

    required = ["feed_id", "title"]
    for key in required:
        if key not in feed:
            print(json.dumps({"success": False, "error": f"缺少必填字段: {key}"}))
            sys.exit(1)

    db = _load_db()

    # 去重：同 feed_id 则更新
    existing_ids = {f["feed_id"]: i for i, f in enumerate(db["feeds"])}
    feed["updated_at"] = datetime.now().isoformat()

    if feed["feed_id"] in existing_ids:
        idx = existing_ids[feed["feed_id"]]
        feed["created_at"] = db["feeds"][idx].get("created_at", feed["updated_at"])
        db["feeds"][idx] = feed
        action = "updated"
    else:
        feed["created_at"] = feed["updated_at"]
        db["feeds"].append(feed)
        action = "added"

    _save_db(db)
    print(json.dumps({"success": True, "action": action, "feed_id": feed["feed_id"], "total": len(db["feeds"])}))


def list_feeds(domain: str = None, min_likes: int = 0, limit: int = 50):
    """列出数据库中的笔记"""
    db = _load_db()
    feeds = db["feeds"]

    if domain:
        feeds = [f for f in feeds if f.get("domain") == domain]
    if min_likes > 0:
        feeds = [f for f in feeds if f.get("likes", 0) >= min_likes]

    # 按赞数降序
    feeds.sort(key=lambda x: x.get("likes", 0), reverse=True)
    feeds = feeds[:limit]

    result = []
    for f in feeds:
        result.append({
            "feed_id": f["feed_id"],
            "title": f.get("title", ""),
            "author": f.get("author", ""),
            "likes": f.get("likes", 0),
            "favorites": f.get("favorites", 0),
            "comments": f.get("comments", 0),
            "domain": f.get("domain", ""),
            "created_at": f.get("created_at", ""),
        })

    print(json.dumps({"total": len(db["feeds"]), "filtered": len(result), "feeds": result}, ensure_ascii=False, indent=2))


def get(feed_id: str):
    """获取单条笔记完整数据"""
    db = _load_db()
    for f in db["feeds"]:
        if f["feed_id"] == feed_id:
            print(json.dumps(f, ensure_ascii=False, indent=2))
            return
    print(json.dumps({"error": f"未找到 feed_id={feed_id}"}))
    sys.exit(1)


def analyze():
    """分析数据库中所有笔记，生成洞察摘要"""
    db = _load_db()
    feeds = db["feeds"]

    if not feeds:
        print(json.dumps({"error": "数据库为空"}))
        return

    total = len(feeds)
    likes_list = [f.get("likes", 0) for f in feeds]
    favs_list = [f.get("favorites", 0) for f in feeds]
    comments_list = [f.get("comments", 0) for f in feeds]

    # 互动数据统计
    avg_likes = sum(likes_list) / total
    avg_favs = sum(favs_list) / total
    avg_comments = sum(comments_list) / total

    # 收藏/赞比
    fav_like_ratios = []
    for f in feeds:
        likes = f.get("likes", 0)
        if likes > 0:
            fav_like_ratios.append(f.get("favorites", 0) / likes)

    avg_fav_ratio = sum(fav_like_ratios) / len(fav_like_ratios) if fav_like_ratios else 0

    # 领域分布
    domain_counter = Counter(f.get("domain", "未分类") for f in feeds)

    # 标题类型分布
    title_type_counter = Counter()
    for f in feeds:
        analysis = f.get("analysis", {})
        if analysis.get("title_type"):
            title_type_counter[analysis["title_type"]] += 1

    # 配图风格分布
    image_style_counter = Counter()
    for f in feeds:
        analysis = f.get("analysis", {})
        if analysis.get("image_style"):
            image_style_counter[analysis["image_style"]] += 1

    # 内容类型分布
    content_type_counter = Counter()
    for f in feeds:
        analysis = f.get("analysis", {})
        if analysis.get("content_type"):
            content_type_counter[analysis["content_type"]] += 1

    # 提取高频关键要素
    all_elements = []
    for f in feeds:
        analysis = f.get("analysis", {})
        all_elements.extend(analysis.get("key_elements", []))
    element_counter = Counter(all_elements)

    # 提取高频标签
    all_tags = []
    for f in feeds:
        analysis = f.get("analysis", {})
        all_tags.extend(analysis.get("tags_used", []))
    tag_counter = Counter(all_tags)

    # TOP 5 高赞
    top5 = sorted(feeds, key=lambda x: x.get("likes", 0), reverse=True)[:5]

    insights = {
        "total_feeds": total,
        "engagement": {
            "avg_likes": round(avg_likes, 1),
            "avg_favorites": round(avg_favs, 1),
            "avg_comments": round(avg_comments, 1),
            "avg_fav_like_ratio": round(avg_fav_ratio, 2),
        },
        "domains": dict(domain_counter.most_common()),
        "title_types": dict(title_type_counter.most_common()),
        "image_styles": dict(image_style_counter.most_common()),
        "content_types": dict(content_type_counter.most_common()),
        "top_elements": dict(element_counter.most_common(10)),
        "top_tags": dict(tag_counter.most_common(15)),
        "top5_feeds": [{"title": f.get("title"), "likes": f.get("likes", 0), "author": f.get("author", "")} for f in top5],
        "analyzed_at": datetime.now().isoformat(),
    }

    print(json.dumps(insights, ensure_ascii=False, indent=2))

    # 同时生成 markdown 摘要
    _generate_summary(insights, feeds)


def _generate_summary(insights: dict, feeds: list):
    """生成 markdown 格式的分析摘要"""
    DB_DIR.mkdir(parents=True, exist_ok=True)

    lines = [
        "# 小红书高赞笔记分析摘要",
        "",
        f"> 数据库共 {insights['total_feeds']} 篇笔记 | 更新时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}",
        "",
        "---",
        "",
        "## 互动数据均值",
        "",
        f"| 指标 | 均值 |",
        f"|------|------|",
        f"| 赞 | {insights['engagement']['avg_likes']} |",
        f"| 收藏 | {insights['engagement']['avg_favorites']} |",
        f"| 评论 | {insights['engagement']['avg_comments']} |",
        f"| 收藏/赞比 | {insights['engagement']['avg_fav_like_ratio']} |",
        "",
    ]

    if insights["domains"]:
        lines.extend(["## 领域分布", ""])
        for domain, count in insights["domains"].items():
            lines.append(f"- {domain}: {count}篇")
        lines.append("")

    if insights["title_types"]:
        lines.extend(["## 标题类型分布", ""])
        for ttype, count in insights["title_types"].items():
            lines.append(f"- {ttype}: {count}篇")
        lines.append("")

    if insights["image_styles"]:
        lines.extend(["## 配图风格分布", ""])
        for style, count in insights["image_styles"].items():
            lines.append(f"- {style}: {count}篇")
        lines.append("")

    if insights["content_types"]:
        lines.extend(["## 内容类型分布", ""])
        for ctype, count in insights["content_types"].items():
            lines.append(f"- {ctype}: {count}篇")
        lines.append("")

    if insights["top_elements"]:
        lines.extend(["## 高频成功要素 TOP 10", ""])
        for elem, count in insights["top_elements"].items():
            lines.append(f"- {elem} ({count}次)")
        lines.append("")

    if insights["top_tags"]:
        lines.extend(["## 高频标签 TOP 15", ""])
        for tag, count in insights["top_tags"].items():
            lines.append(f"- {tag} ({count}次)")
        lines.append("")

    if insights["top5_feeds"]:
        lines.extend(["## TOP 5 高赞笔记", ""])
        lines.append("| 排名 | 标题 | 作者 | 赞数 |")
        lines.append("|------|------|------|------|")
        for i, f in enumerate(insights["top5_feeds"], 1):
            lines.append(f"| {i} | {f['title']} | {f['author']} | {f['likes']} |")
        lines.append("")

    lines.extend([
        "---",
        "",
        "## 创作建议",
        "",
        "基于以上数据，创作时优先参考：",
    ])

    if insights["title_types"]:
        top_title = list(insights["title_types"].keys())[0]
        lines.append(f"- 标题采用 **{top_title}** 效果最佳")
    if insights["image_styles"]:
        top_style = list(insights["image_styles"].keys())[0]
        lines.append(f"- 配图优先用 **{top_style}** 风格")
    if insights["top_elements"]:
        top3_elems = list(insights["top_elements"].keys())[:3]
        lines.append(f"- 必备要素：{', '.join(top3_elems)}")

    lines.append("")

    with open(SUMMARY_FILE, "w", encoding="utf-8") as f:
        f.write("\n".join(lines))

    print(f"\n摘要已保存到: {SUMMARY_FILE}", file=sys.stderr)


def delete(feed_id: str):
    """删除一条笔记"""
    db = _load_db()
    before = len(db["feeds"])
    db["feeds"] = [f for f in db["feeds"] if f["feed_id"] != feed_id]
    after = len(db["feeds"])
    _save_db(db)
    print(json.dumps({"success": True, "deleted": before - after, "remaining": after}))


def stats():
    """输出数据库基本统计"""
    db = _load_db()
    feeds = db["feeds"]
    domains = Counter(f.get("domain", "未分类") for f in feeds)
    print(json.dumps({
        "total": len(feeds),
        "domains": dict(domains.most_common()),
        "db_path": str(DB_FILE),
        "db_size": DB_FILE.stat().st_size if DB_FILE.exists() else 0,
    }, ensure_ascii=False, indent=2))


def main():
    if len(sys.argv) < 2:
        print("用法: feed_database.py <command> [args]")
        print("命令:")
        print("  add '<json>'      添加/更新笔记")
        print("  list [--domain X] [--min-likes N] [--limit N]")
        print("  get <feed_id>     获取单条详情")
        print("  delete <feed_id>  删除")
        print("  analyze           生成分析摘要")
        print("  stats             数据库统计")
        sys.exit(0)

    cmd = sys.argv[1]

    if cmd == "add":
        if len(sys.argv) < 3:
            # 从 stdin 读取
            feed_json = sys.stdin.read()
        else:
            feed_json = sys.argv[2]
        add(feed_json)

    elif cmd == "list":
        kwargs = {}
        args = sys.argv[2:]
        i = 0
        while i < len(args):
            if args[i] == "--domain" and i + 1 < len(args):
                kwargs["domain"] = args[i + 1]
                i += 2
            elif args[i] == "--min-likes" and i + 1 < len(args):
                kwargs["min_likes"] = int(args[i + 1])
                i += 2
            elif args[i] == "--limit" and i + 1 < len(args):
                kwargs["limit"] = int(args[i + 1])
                i += 2
            else:
                i += 1
        list_feeds(**kwargs)

    elif cmd == "get":
        get(sys.argv[2])

    elif cmd == "delete":
        delete(sys.argv[2])

    elif cmd == "analyze":
        analyze()

    elif cmd == "stats":
        stats()

    else:
        print(f"未知命令: {cmd}")
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
小红书配图生成脚本
使用 Atlas Cloud Nano Banana API 生成图片
"""

import os
import sys
import json
import time
import argparse
import requests
from pathlib import Path
from typing import Dict, Optional, Tuple

DEFAULT_API_BASE = "https://api.atlascloud.ai/v1"
ENV_FILE_ENV_VAR = "XHS_ENV_FILE"


class ConfigError(RuntimeError):
    """Raised when required API configuration is missing or invalid."""


def load_env_file(env_path: str) -> Dict[str, str]:
    """从.env文件加载配置"""
    config_path = Path(env_path).expanduser()
    if not config_path.is_file():
        raise ConfigError(f"配置文件不存在: {config_path}")

    config = {}
    with open(config_path) as f:
        for line in f:
            line = line.strip()
            if line and not line.startswith("#") and "=" in line:
                key, value = line.split("=", 1)
                # 去除行内注释 (# 后面的内容)
                if " #" in value:
                    value = value.split(" #")[0]
                # 去除可能的引号和空白
                value = value.strip().strip('"').strip("'")
                config[key.strip()] = value
    return config


def resolve_config(env_file: Optional[str] = None) -> Tuple[str, str]:
    """获取API配置"""
    # 优先从环境变量获取
    api_key = os.environ.get("ATLAS_API_KEY") or os.environ.get("LLM_API_KEY")
    api_base = os.environ.get("ATLAS_API_BASE") or os.environ.get("LLM_API_BASE")

    # 只在显式指定时读取.env文件，避免从作者本机路径或隐式文件读取凭据
    explicit_env_file = env_file or os.environ.get(ENV_FILE_ENV_VAR)
    env_config = load_env_file(explicit_env_file) if explicit_env_file else {}
    if env_config and (not api_key or not api_base):
        api_key = api_key or env_config.get("ATLAS_API_KEY") or env_config.get("LLM_API_KEY")
        api_base = api_base or env_config.get("ATLAS_API_BASE") or env_config.get("LLM_API_BASE")

    if not api_key:
        raise ConfigError(
            "未找到 API Key，请设置 ATLAS_API_KEY/LLM_API_KEY，"
            "或通过 --env-file/XHS_ENV_FILE 显式指定配置文件"
        )

    if not api_base:
        api_base = DEFAULT_API_BASE

    return api_key, api_base


def get_config(env_file: Optional[str] = None) -> Tuple[str, str]:
    """获取API配置，CLI入口使用清晰错误信息退出。"""
    try:
        return resolve_config(env_file)
    except ConfigError as exc:
        sys.stderr.write(f"错误: {exc}\n")
        sys.exit(1)


def generate_image(
    prompt: str,
    aspect_ratio: str = "3:4",
    num_images: int = 1,
    output_dir: str = None,
    model: str = "google/nano-banana/text-to-image",
    env_file: Optional[str] = None
):
    """
    生成图片

    Args:
        prompt: 图片描述prompt
        aspect_ratio: 图片比例 (3:4, 1:1, 4:3, 16:9, 9:16)
        num_images: 生成数量
        output_dir: 输出目录
        model: 模型名称
        env_file: 显式指定的.env配置文件路径
    """
    api_key, api_base = get_config(env_file)

    # Atlas Cloud 图片生成 endpoint
    # 将 /v1 替换为完整的图片生成路径
    base_url = api_base.rstrip("/")
    if base_url.endswith("/v1"):
        base_url = base_url[:-3]  # 移除 /v1
    endpoint = f"{base_url}/api/v1/model/generateImage"

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }

    print(f"正在生成图片...")
    print(f"模型: {model}")
    print(f"比例: {aspect_ratio}")
    print(f"Prompt: {prompt[:100]}...")

    saved_files = []

    for i in range(num_images):
        payload = {
            "model": model,
            "prompt": prompt,
            "aspect_ratio": aspect_ratio,
            "output_format": "png",
            "enable_sync_mode": True  # 同步模式，等待完成
        }

        try:
            response = requests.post(endpoint, headers=headers, json=payload, timeout=120)

            if response.status_code != 200:
                print(f"API错误 ({response.status_code}): {response.text}", file=sys.stderr)
                continue

            result = response.json()

            # Atlas Cloud 响应格式: {"code": 200, "data": {"outputs": [...]}}
            data = result.get("data", result)  # 兼容两种格式

            # 获取图片URL
            outputs = data.get("outputs", [])
            if not outputs:
                # 尝试从其他字段获取
                if data.get("urls"):
                    outputs = list(data["urls"].values())
                elif result.get("outputs"):
                    outputs = result["outputs"]

            if not outputs:
                print(f"警告: 第{i+1}张图片未返回结果", file=sys.stderr)
                continue

            # 保存图片
            output_path = Path(output_dir) if output_dir else Path.cwd() / "images"
            output_path.mkdir(parents=True, exist_ok=True)

            for j, img_url in enumerate(outputs):
                if img_url.startswith("data:"):
                    # Base64 编码的图片
                    import base64
                    # 格式: data:image/png;base64,xxxxx
                    header, data = img_url.split(",", 1)
                    img_data = base64.b64decode(data)
                else:
                    # URL 图片
                    img_response = requests.get(img_url, timeout=60)
                    img_data = img_response.content

                filename = f"xhs_image_{int(time.time())}_{i+1}_{j+1}.png"
                filepath = output_path / filename

                with open(filepath, "wb") as f:
                    f.write(img_data)

                saved_files.append(str(filepath))
                print(f"已保存: {filepath}")

        except requests.exceptions.Timeout:
            print(f"超时: 第{i+1}张图片生成超时", file=sys.stderr)
            continue
        except requests.exceptions.RequestException as e:
            print(f"网络错误: {e}", file=sys.stderr)
            continue
        except Exception as e:
            print(f"错误: {e}", file=sys.stderr)
            continue

    if saved_files:
        # 输出JSON结果供Claude读取
        print(json.dumps({
            "success": True,
            "files": saved_files,
            "count": len(saved_files)
        }))
        return saved_files
    else:
        print(json.dumps({
            "success": False,
            "error": "未能生成任何图片"
        }))
        sys.exit(1)


def main():
    parser = argparse.ArgumentParser(description="小红书配图生成器 (Atlas Cloud)")
    parser.add_argument("prompt", help="图片描述prompt")
    parser.add_argument("--ratio", "-r", default="3:4",
                        choices=["3:4", "4:3", "1:1", "16:9", "9:16", "2:3", "3:2", "4:5", "5:4"],
                        help="图片比例 (默认: 3:4 竖版)")
    parser.add_argument("--num", "-n", type=int, default=1,
                        help="生成数量 (默认: 1)")
    parser.add_argument("--output", "-o", help="输出目录")
    parser.add_argument("--model", "-m", default="google/nano-banana/text-to-image",
                        choices=[
                            "google/nano-banana/text-to-image",
                            "google/nano-banana-pro/text-to-image"
                        ],
                        help="模型 (默认: nano-banana)")
    parser.add_argument("--env-file", help="显式指定 .env 配置文件；也可设置 XHS_ENV_FILE")

    args = parser.parse_args()
    generate_image(args.prompt, args.ratio, args.num, args.output, args.model, args.env_file)


if __name__ == "__main__":
    main()