mirror of
https://github.com/comeonzhj/Auto-Redbook-Skills.git
synced 2026-03-27 12:49:27 +08:00
refactor md2Redbook skill with themes and paging
This commit is contained in:
ZhangJia
@@ -1,72 +1,107 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
小红书笔记发布脚本
|
||||
将生成的图片卡片发布到小红书
|
||||
小红书笔记发布脚本 - 增强版
|
||||
支持直接发布（本地签名）和通过 API 服务发布两种方式
|
||||
|
||||
使用方法:
|
||||
python publish_xhs.py --title "标题" --desc "描述" --images cover.png card_1.png card_2.png
|
||||
# 直接发布（使用本地签名）
|
||||
python publish_xhs.py --title "标题" --desc "描述" --images cover.png card_1.png
|
||||
|
||||
# 通过 API 服务发布
|
||||
python publish_xhs.py --title "标题" --desc "描述" --images cover.png card_1.png --api-mode
|
||||
|
||||
环境变量:
|
||||
在同目录下创建 .env 文件，配置 XHS_COOKIE：
|
||||
在同目录或项目根目录下创建 .env 文件，配置：
|
||||
|
||||
# 必需：小红书 Cookie
|
||||
XHS_COOKIE=your_cookie_string_here
|
||||
|
||||
# 可选：API 服务地址（使用 --api-mode 时需要）
|
||||
XHS_API_URL=http://localhost:5005
|
||||
|
||||
依赖安装:
|
||||
pip install xhs python-dotenv
|
||||
pip install xhs python-dotenv requests
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import List, Optional, Dict, Any
|
||||
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
from xhs import XhsClient
|
||||
import requests
|
||||
except ImportError as e:
|
||||
print(f"缺少依赖: {e}")
|
||||
print("请运行: pip install xhs python-dotenv")
|
||||
print("请运行: pip install python-dotenv requests")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def load_cookie():
|
||||
def load_cookie() -> str:
|
||||
"""从 .env 文件加载 Cookie"""
|
||||
# 尝试从当前目录加载 .env
|
||||
env_path = Path.cwd() / '.env'
|
||||
if env_path.exists():
|
||||
load_dotenv(env_path)
|
||||
# 尝试从多个位置加载 .env
|
||||
env_paths = [
|
||||
Path.cwd() / '.env',
|
||||
Path(__file__).parent.parent / '.env',
|
||||
Path(__file__).parent.parent.parent / '.env',
|
||||
]
|
||||
|
||||
# 也尝试从脚本目录加载
|
||||
script_env = Path(__file__).parent.parent / '.env'
|
||||
if script_env.exists():
|
||||
load_dotenv(script_env)
|
||||
for env_path in env_paths:
|
||||
if env_path.exists():
|
||||
load_dotenv(env_path)
|
||||
break
|
||||
|
||||
cookie = os.getenv('XHS_COOKIE')
|
||||
if not cookie:
|
||||
print("❌ 错误: 未找到 XHS_COOKIE 环境变量")
|
||||
print("请在当前目录创建 .env 文件，添加以下内容：")
|
||||
print("请创建 .env 文件，添加以下内容：")
|
||||
print("XHS_COOKIE=your_cookie_string_here")
|
||||
print("\nCookie 获取方式：")
|
||||
print("1. 在浏览器中登录小红书（https://www.xiaohongshu.com）")
|
||||
print("2. 打开开发者工具（F12）")
|
||||
print("3. 在 Network 标签中查看任意请求的 Cookie 头")
|
||||
print("4. 复制完整的 cookie 字符串")
|
||||
sys.exit(1)
|
||||
|
||||
return cookie
|
||||
|
||||
|
||||
def create_client(cookie: str) -> XhsClient:
|
||||
"""创建小红书客户端"""
|
||||
try:
|
||||
# 使用本地签名
|
||||
from xhs.help import sign as local_sign
|
||||
|
||||
def sign_func(uri, data=None, a1="", web_session=""):
|
||||
return local_sign(uri, data, a1=a1)
|
||||
|
||||
client = XhsClient(cookie=cookie, sign=sign_func)
|
||||
return client
|
||||
except Exception as e:
|
||||
print(f"❌ 创建客户端失败: {e}")
|
||||
sys.exit(1)
|
||||
def parse_cookie(cookie_string: str) -> Dict[str, str]:
|
||||
"""解析 Cookie 字符串为字典"""
|
||||
cookies = {}
|
||||
for item in cookie_string.split(';'):
|
||||
item = item.strip()
|
||||
if '=' in item:
|
||||
key, value = item.split('=', 1)
|
||||
cookies[key.strip()] = value.strip()
|
||||
return cookies
|
||||
|
||||
|
||||
def validate_images(image_paths: list) -> list:
|
||||
def validate_cookie(cookie_string: str) -> bool:
|
||||
"""验证 Cookie 是否包含必要的字段"""
|
||||
cookies = parse_cookie(cookie_string)
|
||||
|
||||
# 检查必需的 cookie 字段
|
||||
required_fields = ['a1', 'web_session']
|
||||
missing = [f for f in required_fields if f not in cookies]
|
||||
|
||||
if missing:
|
||||
print(f"⚠️ Cookie 可能不完整，缺少字段: {', '.join(missing)}")
|
||||
print("这可能导致签名失败，请确保 Cookie 包含 a1 和 web_session 字段")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def get_api_url() -> str:
|
||||
"""获取 API 服务地址"""
|
||||
return os.getenv('XHS_API_URL', 'http://localhost:5005')
|
||||
|
||||
|
||||
def validate_images(image_paths: List[str]) -> List[str]:
|
||||
"""验证图片文件是否存在"""
|
||||
valid_images = []
|
||||
for path in image_paths:
|
||||
@@ -82,51 +117,218 @@ def validate_images(image_paths: list) -> list:
|
||||
return valid_images
|
||||
|
||||
|
||||
def publish_note(client: XhsClient, title: str, desc: str, images: list,
|
||||
is_private: bool = False, post_time: str = None):
|
||||
"""发布图文笔记"""
|
||||
try:
|
||||
print(f"\n🚀 准备发布笔记...")
|
||||
class LocalPublisher:
|
||||
"""本地发布模式：直接使用 xhs 库"""
|
||||
|
||||
def __init__(self, cookie: str):
|
||||
self.cookie = cookie
|
||||
self.client = None
|
||||
|
||||
def init_client(self):
|
||||
"""初始化 xhs 客户端"""
|
||||
try:
|
||||
from xhs import XhsClient
|
||||
from xhs.help import sign as local_sign
|
||||
except ImportError:
|
||||
print("❌ 错误: 缺少 xhs 库")
|
||||
print("请运行: pip install xhs")
|
||||
sys.exit(1)
|
||||
|
||||
# 解析 a1 值
|
||||
cookies = parse_cookie(self.cookie)
|
||||
a1 = cookies.get('a1', '')
|
||||
|
||||
def sign_func(uri, data=None, a1_param="", web_session=""):
|
||||
# 使用 cookie 中的 a1 值
|
||||
return local_sign(uri, data, a1=a1 or a1_param)
|
||||
|
||||
self.client = XhsClient(cookie=self.cookie, sign=sign_func)
|
||||
|
||||
def get_user_info(self) -> Optional[Dict[str, Any]]:
|
||||
"""获取当前登录用户信息"""
|
||||
try:
|
||||
info = self.client.get_self_info()
|
||||
print(f"👤 当前用户: {info.get('nickname', '未知')}")
|
||||
return info
|
||||
except Exception as e:
|
||||
print(f"⚠️ 无法获取用户信息: {e}")
|
||||
return None
|
||||
|
||||
def publish(self, title: str, desc: str, images: List[str],
|
||||
is_private: bool = False, post_time: str = None) -> Dict[str, Any]:
|
||||
"""发布图文笔记"""
|
||||
print(f"\n🚀 准备发布笔记（本地模式）...")
|
||||
print(f" 📌 标题: {title}")
|
||||
print(f" 📝 描述: {desc[:50]}..." if len(desc) > 50 else f" 📝 描述: {desc}")
|
||||
print(f" 🖼️ 图片数量: {len(images)}")
|
||||
|
||||
result = client.create_image_note(
|
||||
title=title,
|
||||
desc=desc,
|
||||
files=images,
|
||||
is_private=is_private,
|
||||
post_time=post_time
|
||||
)
|
||||
|
||||
print("\n✨ 笔记发布成功！")
|
||||
if isinstance(result, dict):
|
||||
note_id = result.get('note_id') or result.get('id')
|
||||
if note_id:
|
||||
print(f" 📎 笔记ID: {note_id}")
|
||||
print(f" 🔗 链接: https://www.xiaohongshu.com/explore/{note_id}")
|
||||
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ 发布失败: {e}")
|
||||
sys.exit(1)
|
||||
try:
|
||||
result = self.client.create_image_note(
|
||||
title=title,
|
||||
desc=desc,
|
||||
files=images,
|
||||
is_private=is_private,
|
||||
post_time=post_time
|
||||
)
|
||||
|
||||
print("\n✨ 笔记发布成功！")
|
||||
if isinstance(result, dict):
|
||||
note_id = result.get('note_id') or result.get('id')
|
||||
if note_id:
|
||||
print(f" 📎 笔记ID: {note_id}")
|
||||
print(f" 🔗 链接: https://www.xiaohongshu.com/explore/{note_id}")
|
||||
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
error_msg = str(e)
|
||||
print(f"\n❌ 发布失败: {error_msg}")
|
||||
|
||||
# 提供具体的错误排查建议
|
||||
if 'sign' in error_msg.lower() or 'signature' in error_msg.lower():
|
||||
print("\n💡 签名错误排查建议：")
|
||||
print("1. 确保 Cookie 包含有效的 a1 和 web_session 字段")
|
||||
print("2. Cookie 可能已过期，请重新获取")
|
||||
print("3. 尝试使用 --api-mode 通过 API 服务发布")
|
||||
elif 'cookie' in error_msg.lower():
|
||||
print("\n💡 Cookie 错误排查建议：")
|
||||
print("1. 确保 Cookie 格式正确")
|
||||
print("2. Cookie 可能已过期，请重新获取")
|
||||
print("3. 确保 Cookie 来自已登录的小红书网页版")
|
||||
|
||||
raise
|
||||
|
||||
|
||||
def get_user_info(client: XhsClient):
|
||||
"""获取当前登录用户信息"""
|
||||
try:
|
||||
info = client.get_self_info()
|
||||
print(f"\n👤 当前用户: {info.get('nickname', '未知')}")
|
||||
return info
|
||||
except Exception as e:
|
||||
print(f"⚠️ 无法获取用户信息: {e}")
|
||||
return None
|
||||
class ApiPublisher:
|
||||
"""API 发布模式：通过 xhs-api 服务发布"""
|
||||
|
||||
def __init__(self, cookie: str, api_url: str = None):
|
||||
self.cookie = cookie
|
||||
self.api_url = api_url or get_api_url()
|
||||
self.session_id = 'md2redbook_session'
|
||||
|
||||
def init_client(self):
|
||||
"""初始化 API 客户端"""
|
||||
print(f"📡 连接 API 服务: {self.api_url}")
|
||||
|
||||
# 健康检查
|
||||
try:
|
||||
resp = requests.get(f"{self.api_url}/health", timeout=5)
|
||||
if resp.status_code != 200:
|
||||
raise Exception("API 服务不可用")
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"❌ 无法连接到 API 服务: {e}")
|
||||
print(f"\n💡 请确保 xhs-api 服务已启动：")
|
||||
print(f" cd xhs-api && python app_full.py")
|
||||
sys.exit(1)
|
||||
|
||||
# 初始化 session
|
||||
try:
|
||||
resp = requests.post(
|
||||
f"{self.api_url}/init",
|
||||
json={
|
||||
"session_id": self.session_id,
|
||||
"cookie": self.cookie
|
||||
},
|
||||
timeout=30
|
||||
)
|
||||
result = resp.json()
|
||||
|
||||
if resp.status_code == 200 and result.get('status') == 'success':
|
||||
print(f"✅ API 初始化成功")
|
||||
user_info = result.get('user_info', {})
|
||||
if user_info:
|
||||
print(f"👤 当前用户: {user_info.get('nickname', '未知')}")
|
||||
elif result.get('status') == 'warning':
|
||||
print(f"⚠️ {result.get('message')}")
|
||||
else:
|
||||
raise Exception(result.get('error', '初始化失败'))
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ API 初始化失败: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
def get_user_info(self) -> Optional[Dict[str, Any]]:
|
||||
"""获取当前登录用户信息"""
|
||||
try:
|
||||
resp = requests.post(
|
||||
f"{self.api_url}/user/info",
|
||||
json={"session_id": self.session_id},
|
||||
timeout=10
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
result = resp.json()
|
||||
if result.get('status') == 'success':
|
||||
info = result.get('user_info', {})
|
||||
print(f"👤 当前用户: {info.get('nickname', '未知')}")
|
||||
return info
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f"⚠️ 无法获取用户信息: {e}")
|
||||
return None
|
||||
|
||||
def publish(self, title: str, desc: str, images: List[str],
|
||||
is_private: bool = False, post_time: str = None) -> Dict[str, Any]:
|
||||
"""发布图文笔记"""
|
||||
print(f"\n🚀 准备发布笔记（API 模式）...")
|
||||
print(f" 📌 标题: {title}")
|
||||
print(f" 📝 描述: {desc[:50]}..." if len(desc) > 50 else f" 📝 描述: {desc}")
|
||||
print(f" 🖼️ 图片数量: {len(images)}")
|
||||
|
||||
try:
|
||||
payload = {
|
||||
"session_id": self.session_id,
|
||||
"title": title,
|
||||
"desc": desc,
|
||||
"files": images,
|
||||
"is_private": is_private
|
||||
}
|
||||
if post_time:
|
||||
payload["post_time"] = post_time
|
||||
|
||||
resp = requests.post(
|
||||
f"{self.api_url}/publish/image",
|
||||
json=payload,
|
||||
timeout=120
|
||||
)
|
||||
result = resp.json()
|
||||
|
||||
if resp.status_code == 200 and result.get('status') == 'success':
|
||||
print("\n✨ 笔记发布成功！")
|
||||
publish_result = result.get('result', {})
|
||||
if isinstance(publish_result, dict):
|
||||
note_id = publish_result.get('note_id') or publish_result.get('id')
|
||||
if note_id:
|
||||
print(f" 📎 笔记ID: {note_id}")
|
||||
print(f" 🔗 链接: https://www.xiaohongshu.com/explore/{note_id}")
|
||||
return publish_result
|
||||
else:
|
||||
raise Exception(result.get('error', '发布失败'))
|
||||
|
||||
except Exception as e:
|
||||
error_msg = str(e)
|
||||
print(f"\n❌ 发布失败: {error_msg}")
|
||||
raise
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description='将图片发布为小红书笔记'
|
||||
description='将图片发布为小红书笔记',
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog='''
|
||||
示例:
|
||||
# 基本用法
|
||||
python publish_xhs.py -t "我的标题" -d "正文内容" -i cover.png card_1.png card_2.png
|
||||
|
||||
# 使用 API 模式
|
||||
python publish_xhs.py -t "我的标题" -d "正文内容" -i *.png --api-mode
|
||||
|
||||
# 设为私密笔记
|
||||
python publish_xhs.py -t "我的标题" -d "正文内容" -i *.png --private
|
||||
|
||||
# 定时发布
|
||||
python publish_xhs.py -t "我的标题" -d "正文内容" -i *.png --post-time "2024-12-01 10:00:00"
|
||||
'''
|
||||
)
|
||||
parser.add_argument(
|
||||
'--title', '-t',
|
||||
@@ -154,6 +356,16 @@ def main():
|
||||
default=None,
|
||||
help='定时发布时间（格式：2024-01-01 12:00:00）'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--api-mode',
|
||||
action='store_true',
|
||||
help='使用 API 模式发布（需要 xhs-api 服务运行）'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--api-url',
|
||||
default=None,
|
||||
help='API 服务地址（默认: http://localhost:5005）'
|
||||
)
|
||||
parser.add_argument(
|
||||
'--dry-run',
|
||||
action='store_true',
|
||||
@@ -170,32 +382,43 @@ def main():
|
||||
# 加载 Cookie
|
||||
cookie = load_cookie()
|
||||
|
||||
# 验证 Cookie 格式
|
||||
validate_cookie(cookie)
|
||||
|
||||
# 验证图片
|
||||
valid_images = validate_images(args.images)
|
||||
|
||||
# 创建客户端
|
||||
client = create_client(cookie)
|
||||
|
||||
# 获取用户信息（验证 Cookie 有效性）
|
||||
get_user_info(client)
|
||||
|
||||
if args.dry_run:
|
||||
print("\n🔍 验证模式 - 不会实际发布")
|
||||
print(f" 📌 标题: {args.title}")
|
||||
print(f" 📝 描述: {args.desc}")
|
||||
print(f" 🖼️ 图片: {valid_images}")
|
||||
print(f" 🔒 私密: {args.private}")
|
||||
print(f" ⏰ 定时: {args.post_time or '立即发布'}")
|
||||
print(f" 📡 模式: {'API' if args.api_mode else '本地'}")
|
||||
print("\n✅ 验证通过，可以发布")
|
||||
return
|
||||
|
||||
# 选择发布方式
|
||||
if args.api_mode:
|
||||
publisher = ApiPublisher(cookie, args.api_url)
|
||||
else:
|
||||
publisher = LocalPublisher(cookie)
|
||||
|
||||
# 初始化客户端
|
||||
publisher.init_client()
|
||||
|
||||
# 发布笔记
|
||||
publish_note(
|
||||
client=client,
|
||||
title=args.title,
|
||||
desc=args.desc,
|
||||
images=valid_images,
|
||||
is_private=args.private,
|
||||
post_time=args.post_time
|
||||
)
|
||||
try:
|
||||
publisher.publish(
|
||||
title=args.title,
|
||||
desc=args.desc,
|
||||
images=valid_images,
|
||||
is_private=args.private,
|
||||
post_time=args.post_time
|
||||
)
|
||||
except Exception as e:
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
Reference in New Issue
Block a user