ai-swap/backend/modules/auto_post/post.py

import os
import time
import shutil
import logging
import psutil
import subprocess
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from .post_page import XiaohongshuPostPage
from .post_config import Config


def kill_chrome_processes():
    """结束所有Chrome相关进程"""
    try:
        if os.name == 'nt':  # Windows
            subprocess.run(['taskkill', '/F', '/IM', 'chrome.exe'], capture_output=True)
            subprocess.run(['taskkill', '/F', '/IM', 'chromedriver.exe'], capture_output=True)
        else:  # Linux/Mac
            os.system("pkill -f chrome")
            os.system("pkill -f chromedriver")
        logging.info("已清理Chrome相关进程")
    except Exception as e:
        logging.warning(f"清理Chrome进程失败: {str(e)}")

def clean_user_data(if_clean=True):
    """清理用户数据目录"""
    user_data_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), 'user_data'))
    
    # 检查并清理锁文件
    lock_files = ['SingletonLock', 'SingletonCookie', 'SingletonSocket', 'Singleton*']
    for lock_file in lock_files:
        try:
            lock_path = os.path.join(user_data_dir, 'Default', lock_file)
            if os.path.exists(lock_path):
                os.remove(lock_path)
                logging.info(f"已删除锁文件: {lock_path}")
        except Exception as e:
            logging.warning(f"删除锁文件失败: {str(e)}")
    
    if os.path.exists(user_data_dir) and if_clean:
        logging.info(f"清理并创建新用户数据目录: {user_data_dir}")
        try:
            # 先尝试删除可能导致问题的文件
            problem_files = [
                os.path.join(user_data_dir, 'Default', 'Preferences'),
                os.path.join(user_data_dir, 'Default', 'Network Persistent State'),
                os.path.join(user_data_dir, 'Default', 'Network Action Predictor'),
                os.path.join(user_data_dir, 'Default', 'History'),
                os.path.join(user_data_dir, 'Default', 'History-journal')
            ]
            for file in problem_files:
                if os.path.exists(file):
                    os.remove(file)
                    logging.info(f"已删除文件: {file}")
            
            # 如果需要完全清理
            if if_clean:
                shutil.rmtree(user_data_dir)
                os.makedirs(user_data_dir)
                logging.info(f"已完全清理用户数据目录: {user_data_dir}")
        except Exception as e:
            logging.warning(f"清理用户数据目录失败: {str(e)}")
    else:
        logging.info(f"沿用旧用户数据目录")
        # 确保目录存在
        os.makedirs(user_data_dir, exist_ok=True)
    
    return user_data_dir

def xiaohonshu_upload(if_clean=False):
    """上传图文到小红书"""
    driver = None
    max_attempts = 3
    
    for attempt in range(max_attempts):
        try:
            # 清理Chrome进程和用户数据
            kill_chrome_processes()
            user_data_dir = clean_user_data(if_clean=False)
            time.sleep(2)  # 等待进程完全结束
            
            logging.info(f"第{attempt + 1}次尝试启动浏览器，用户数据目录: {user_data_dir}")
            
            # 初始化WebDriver
            options = webdriver.ChromeOptions()
            options.add_argument(f'--user-data-dir={user_data_dir}')
            options.add_argument('--profile-directory=Default')
            options.add_experimental_option('excludeSwitches', ['enable-automation'])
            options.add_experimental_option('useAutomationExtension', False)
            options.add_experimental_option("detach", True)

            
            # 添加必要的配置
            # options.add_argument('--headless')
            options.add_argument('--disable-gpu')
            options.add_argument('--no-sandbox')
            options.add_argument('--disable-dev-shm-usage')
            options.add_argument('--window-size=1920,1080')
            options.add_argument('--start-maximized')
            options.add_argument('--disable-blink-features=AutomationControlled')
            options.add_argument('--disable-extensions')
            options.add_argument('--disable-popup-blocking')
            options.add_argument('--disable-notifications')
            options.add_argument('--remote-debugging-port=0')  # 使用随机调试端口
            
            service = Service(Config.CHROME_DRIVER_PATH)
            driver = webdriver.Chrome(service=service, options=options)
            
            # 如果成功创建driver，跳出重试循环
            break
            
        except Exception as e:
            logging.error(f"第{attempt + 1}次启动浏览器失败: {str(e)}")
            if driver:
                try:
                    driver.quit()
                except:
                    pass
            if attempt == max_attempts - 1:
                raise
            time.sleep(5)  # 等待一段时间后重试
    
    try:
        post_page = XiaohongshuPostPage(driver)
        post_page.open()
        post_page.click_upload_button()
        post_page.upload_images(Config._POST_CONFIG['image_paths'])
        post_page.input_title(Config._POST_CONFIG['title'])
        post_page.input_description(Config._POST_CONFIG['description'], Config._POST_CONFIG['topic'])
        post_page.schedule_post(Config._POST_CONFIG['upload_time'])
        
    except Exception as e:
        print(f"发生错误: {str(e)}")
        driver.save_screenshot("error.png")
        raise
    finally:
        time.sleep(10)
        driver.quit()

if __name__ == "__main__":
    Config.set_post_config(
        image_paths=[r"D:\桌面\研究生\组会\rednote\data\flower.jpg", r"D:\桌面\研究生\组会\rednote\data\test_img.png"],
        title="💙被问爆的蓝色仙女裙，美到犯规！",
        description="👗宝子们，挖到一条超绝的蓝色长裙！温柔的浅蓝色，仿佛把天空穿在了身上~V领设计巧妙修饰颈部线条，增添了一丝小性感。泡泡袖又带着点复古甜美感，谁穿谁是在逃公主！腰部的设计很贴心，能很好地勾勒出腰线，显得腰细细的~裙摆大大的，走路都带风，氛围感直接拉满！无论是日常出街还是度假穿都超合适，真的会被美到失语，闭眼入不亏~",
        topic=["每日穿搭", "今日分享", "蓝色长裙", "仙女裙", "复古甜美", "小性感", "氛围感", "闭眼入不亏"],
        upload_time="2025-07-09 10:26"  
    )
    xiaohonshu_upload(if_clean=False)