小红书批量发布工具，批量发布笔记上传图文软件，python自动发布框架

import requests
import os
import time
import json
from PIL import Image
from io import BytesIO
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.options import Options
from selenium.common.exceptions import TimeoutException

class XiaohongshuBot:
    def __init__(self, username, password):
        self.username = username
        self.password = password
        self.session = requests.Session()
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
            'Referer': 'https://www.xiaohongshu.com/',
            'Origin': 'https://www.xiaohongshu.com'
        }
        self.driver = None
        self.login_cookies = None
        
    def init_driver(self):
        chrome_options = Options()
        chrome_options.add_argument('--headless')
        chrome_options.add_argument('--disable-gpu')
        chrome_options.add_argument('--no-sandbox')
        self.driver = webdriver.Chrome(options=chrome_options)
        
    def login(self):
        self.init_driver()
        self.driver.get('https://www.xiaohongshu.com/user/login')
        
        try:
            # 等待登录页面加载完成
            WebDriverWait(self.driver, 20).until(
                EC.presence_of_element_located((By.XPATH, '//input[@placeholder="请输入手机号"]'))
            )
            
            # 输入用户名和密码
            username_input = self.driver.find_element(By.XPATH, '//input[@placeholder="请输入手机号"]')
            password_input = self.driver.find_element(By.XPATH, '//input[@placeholder="请输入密码"]')
            
            username_input.send_keys(self.username)
            password_input.send_keys(self.password)
            
            # 点击登录按钮
            login_button = self.driver.find_element(By.XPATH, '//button[contains(@class, "login-btn")]')
            login_button.click()
            
            # 等待登录成功
            WebDriverWait(self.driver, 30).until(
                EC.presence_of_element_located((By.XPATH, '//div[contains(@class, "user-info")]'))
            )
            
            # 获取cookies
            self.login_cookies = {cookie['name']: cookie['value'] for cookie in self.driver.get_cookies()}
            self.session.cookies.update(self.login_cookies)
            
        except TimeoutException:
            print("登录超时，请检查网络或验证码")
            return False
        finally:
            self.driver.quit()
            
        return True
    
    def upload_image(self, image_path):
        if not os.path.exists(image_path):
            print(f"图片文件不存在: {image_path}")
            return None
            
        try:
            # 压缩图片
            img = Image.open(image_path)
            if img.size[0] > 2000 or img.size[1] > 2000:
                img.thumbnail((2000, 2000), Image.ANTIALIAS)
                
            buffer = BytesIO()
            img.save(buffer, format='JPEG', quality=85)
            buffer.seek(0)
            
            # 上传图片
            upload_url = 'https://www.xiaohongshu.com/api/upload/image'
            files = {'file': (os.path.basename(image_path), buffer, 'image/jpeg')}
            
            response = self.session.post(upload_url, files=files, headers=self.headers)
            if response.status_code == 200:
                result = response.json()
                if result.get('success'):
                    return result['data']['url']
                    
        except Exception as e:
            print(f"上传图片失败: {str(e)}")
            
        return None
    
    def publish_note(self, title, content, image_paths, tags=None, location=None):
        if not self.login_cookies:
            if not self.login():
                print("登录失败，无法发布笔记")
                return False
                
        # 上传所有图片
        image_urls = []
        for img_path in image_paths:
            img_url = self.upload_image(img_path)
            if img_url:
                image_urls.append(img_url)
                time.sleep(1)  # 避免频繁请求
                
        if not image_urls:
            print("没有图片上传成功")
            return False
            
        # 准备发布数据
        publish_data = {
            'title': title,
            'content': content,
            'image_urls': image_urls,
            'tags': tags or [],
            'location': location or '',
            'publish_time': int(time.time() * 1000)
        }
        
        # 发布笔记
        publish_url = 'https://www.xiaohongshu.com/api/note/publish'
        response = self.session.post(
            publish_url,
            data=json.dumps(publish_data),
            headers={**self.headers, 'Content-Type': 'application/json'}
        )
        
        if response.status_code == 200:
            result = response.json()
            if result.get('success'):
                print(f"笔记发布成功: {result['data']['note_id']}")
                return True
                
        print(f"笔记发布失败: {response.text}")
        return False
        
    def batch_publish(self, notes_data):
        success_count = 0
        for note in notes_data:
            print(f"正在发布: {note['title']}")
            if self.publish_note(
                title=note['title'],
                content=note['content'],
                image_paths=note['image_paths'],
                tags=note.get('tags'),
                location=note.get('location')
            ):
                success_count += 1
                time.sleep(5)  # 避免频繁发布
                
        print(f"批量发布完成，成功 {success_count}/{len(notes_data)} 篇")
        return success_count

if __name__ == '__main__':
    # 使用示例
    bot = XiaohongshuBot('your_username', 'your_password')
    
    # 准备批量发布的笔记数据
    notes_to_publish = [
        {
            'title': '夏日穿搭分享',
            'content': '今天分享一套适合夏天的清爽穿搭~',
            'image_paths': ['outfit1.jpg', 'outfit2.jpg'],
            'tags': ['穿搭', '夏日', '时尚'],
            'location': '上海'
        },
        {
            'title': '周末探店 | 小众咖啡馆',
            'content': '发现一家超有氛围的小众咖啡馆，咖啡和甜点都很棒！',
            'image_paths': ['cafe1.jpg', 'cafe2.jpg', 'cafe3.jpg'],
            'tags': ['探店', '咖啡', '周末好去处']
        }
    ]
    
    # 执行批量发布
    bot.batch_publish(notes_to_publish)