ai-swap/backend/services/auto_post_service.py

import os
import time
import logging
from typing import List, Optional, Dict
from datetime import datetime
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.common.exceptions import WebDriverException

import os
import sys

# 获取当前文件的绝对路径
current_dir = os.path.dirname(os.path.abspath(__file__))
# 获取backend目录的路径
backend_dir = os.path.dirname(current_dir)
# 将backend目录添加到Python路径
if backend_dir not in sys.path:
    sys.path.insert(0, backend_dir)

from modules.auto_post.post_page import XiaohongshuPostPage
from modules.auto_post.post import kill_chrome_processes, clean_user_data, xiaohonshu_upload
from modules.auto_post.post_config import Config
from utils.logger_config import setup_logger

logger = setup_logger(__name__)

class AutoPostService:
    def __init__(self):
        pass
        
    def post_to_xiaohongshu(self, 
                           image_paths: List[str],
                           title: str,
                           description: str,
                           topics: List[str],
                           schedule_time: Optional[str] = None,
                           if_clean: bool = False) -> Dict:
        
        Config.set_post_config(
            image_paths=image_paths,
            title=title,
            description=description,
            topic=topics,
            upload_time=schedule_time
        )

        xiaohonshu_upload(if_clean)

auto_post_service = AutoPostService()

# import os
# import sys
# import time
# import threading
# from datetime import datetime
# from fastapi import FastAPI, HTTPException, BackgroundTasks
# from pydantic import BaseModel
# from typing import List, Optional, Dict



# # 创建FastAPI应用
# app = FastAPI(
#     title="小红书自动发布API",
#     description="用于自动发布内容到小红书的服务接口",
#     version="1.0.0"
# )

# # 创建线程锁防止并发冲突
# post_lock = threading.Lock()

# class XiaohongshuPostRequest(BaseModel):
#     """小红书发布请求数据结构"""
#     image_paths: List[str]
#     title: str
#     description: str
#     topics: List[str]
#     schedule_time: Optional[str] = None
#     if_clean: bool = False

# class PostResponse(BaseModel):
#     """API响应数据结构"""
#     success: bool
#     message: str
#     task_id: Optional[str] = None
#     timestamp: str

# @app.post("/xiaohongshu/post", response_model=PostResponse, tags=["发布管理"])
# async def post_to_xiaohongshu(request: XiaohongshuPostRequest):
#     """
#     发布内容到小红书
#     - **image_paths**: 图片路径列表
#     - **title**: 内容标题
#     - **description**: 详细描述
#     - **topics**: 话题标签列表
#     - **schedule_time**: 定时发布时间 (格式: YYYY-MM-DD HH:MM)
#     - **if_clean**: 是否清理浏览器缓存 (默认False)
#     """
#     task_id = f"task_{int(time.time())}"
#     logger.info(f"Received new post task {task_id}")
    
#     # 验证图片路径
#     for path in request.image_paths:
#         if not os.path.exists(path):
#             logger.error(f"Image not found: {path}")
#             raise HTTPException(
#                 status_code=400, 
#                 detail=f"图片文件不存在: {path}"
#             )
    
#     # 验证时间格式
#     if request.schedule_time:
#         try:
#             datetime.strptime(request.schedule_time, "%Y-%m-%d %H:%M")
#         except ValueError:
#             logger.error(f"Invalid time format: {request.schedule_time}")
#             raise HTTPException(
#                 status_code=400,
#                 detail="时间格式错误，请使用 YYYY-MM-DD HH:MM 格式"
#             )
    
#     # 使用线程锁防止并发操作
#     with post_lock:
#         try:
#             # 配置参数
#             Config.set_post_config(
#                 image_paths=request.image_paths,
#                 title=request.title,
#                 description=request.description,
#                 topic=request.topics,
#                 upload_time=request.schedule_time
#             )
            
#             # 执行发布
#             xiaohonshu_upload(request.if_clean)
            
#             return {
#                 "success": True,
#                 "message": "内容发布成功",
#                 "task_id": task_id,
#                 "timestamp": datetime.now().isoformat()
#             }
            
#         except Exception as e:
#             logger.exception(f"Post failed: {str(e)}")
#             raise HTTPException(
#                 status_code=500,
#                 detail=f"发布失败: {str(e)}"
#             ) from e

# @app.get("/health", response_model=Dict[str, str], tags=["系统状态"])
# def health_check():
#     """服务健康检查"""
#     return {
#         "status": "running",
#         "timestamp": datetime.now().isoformat()
#     }

# if __name__ == "__main__":
#     import uvicorn
#     uvicorn.run(
#         app, 
#         host="0.0.0.0", 
#         port=5000,
#         # 增加超时时间以适应较长的操作
#         timeout_keep_alive=120
#     )



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