social_media_auto_publisher/docs/PROJECT_PLAN.md

社交媒体自动发布器 - 项目方案

1. 项目概述

1.1 项目目标

构建一个统一的社交媒体自动化发布平台，支持以下功能：

• 登录管理：小红书、抖音 账号登录与Cookie管理
• 内容发布：小红书笔记/视频、抖音视频的自动化发布
• 反检测机制：模仿人类操作，降低自动化风险
• 批量操作：支持多账号、多内容批量发布

1.2 支持平台与功能

平台	登录	图文笔记	视频发布	批量发布
小红书	✅	✅	✅	✅
抖音	✅	❌	✅	✅

2. 项目架构设计

2.1 整体架构

social_media_auto_publisher/
├── core/                   # 核心引擎
│   ├── publisher.py        # 统一发布管理器
│   ├── session_manager.py  # 会话管理
│   └── task_scheduler.py   # 任务调度器
├── platforms/              # 平台适配层
│   ├── xiaohongshu/        # 小红书适配器
│   └── douyin/             # 抖音适配器
├── auth/                   # 认证模块
│   ├── base_auth.py        # 基础认证类
│   ├── xiaohongshu_auth.py # 小红书认证
│   └── douyin_auth.py      # 抖音认证
├── utils/                  # 工具模块
│   ├── browser.py          # 浏览器管理
│   ├── human_behavior.py   # 人类行为模拟
│   ├── media_handler.py    # 媒体文件处理
│   └── logger.py           # 日志管理
├── config/                 # 配置管理
│   ├── settings.py         # 系统配置
│   └── platform_config.py  # 平台配置
├── tests/                  # 测试用例
├── examples/               # 使用示例
├── docs/                   # 文档
└── media/                  # 媒体文件存储
    ├── images/
    └── videos/

2.2 核心设计模式

2.2.1 适配器模式 (Adapter Pattern)

# 统一的平台接口
class PlatformAdapter:
    def login(self) -> bool
    def publish_content(self, content: Content) -> bool
    def get_upload_status(self) -> UploadStatus

# 各平台适配器
class XiaoHongShuAdapter(PlatformAdapter)
class DouyinAdapter(PlatformAdapter)

2.2.2 工厂模式 (Factory Pattern)

class PlatformAdapterFactory:
    @staticmethod
    def create_adapter(platform: PlatformType) -> PlatformAdapter

2.2.3 策略模式 (Strategy Pattern)

class HumanBehaviorStrategy:
    def typing_behavior(self, text: str) -> None
    def clicking_behavior(self, element) -> None
    def waiting_behavior(self, duration: float) -> None

2.3 数据流设计

用户输入 → 内容解析 → 平台选择 → 认证检查 → 内容发布 → 状态反馈
    ↓           ↓           ↓           ↓           ↓           ↓
   API     ContentModel  Factory      Auth      Publisher   Logger

3. 模块详细设计

3.1 核心引擎 (core/)

3.1.1 统一发布管理器

class Publisher:
    """统一发布管理器"""
    def __init__(self):
        self.adapters = {}
        self.session_manager = SessionManager()

    async def publish(self,
                     platform: PlatformType,
                     content: Content,
                     account: AccountInfo) -> PublishResult

    async def batch_publish(self,
                           tasks: List[PublishTask]) -> List[PublishResult]

3.1.2 会话管理器

class SessionManager:
    """浏览器会话管理"""
    def __init__(self):
        self.active_sessions = {}
        self.cookie_store = CookieStore()

    async def get_session(self, platform: PlatformType, account: str) -> Browser
    async def save_session(self, platform: PlatformType, account: str) -> None

3.1.3 任务调度器

class TaskScheduler:
    """异步任务调度"""
    def __init__(self, max_concurrent: int = 3):
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def schedule_tasks(self, tasks: List[Task]) -> List[TaskResult]

3.2 平台适配层 (platforms/)

3.2.1 小红书适配器

class XiaoHongShuAdapter(PlatformAdapter):
    """小红书平台适配器"""

    async def login(self, account: AccountInfo) -> bool:
        """扫码登录"""

    async def publish_image_note(self, note: ImageNote) -> bool:
        """发布图文笔记"""

    async def publish_video_note(self, note: VideoNote) -> bool:
        """发布视频笔记"""

3.2.2 抖音适配器

class DouyinAdapter(PlatformAdapter):
    """抖音平台适配器"""

    async def login(self, account: AccountInfo) -> bool:
        """扫码登录"""

    async def publish_video(self, video: VideoContent) -> bool:
        """发布视频内容"""

3.3 认证模块 (auth/)

3.3.1 基础认证类

class BaseAuth:
    """基础认证类"""

    def __init__(self, platform: PlatformType):
        self.platform = platform
        self.cookie_path = f"auth/cookies/{platform.value}/"

    async def check_login_status(self, page: Page) -> bool:
        """检查登录状态"""

    async def save_cookies(self, page: Page, account: str) -> None:
        """保存Cookie"""

    async def load_cookies(self, page: Page, account: str) -> bool:
        """加载Cookie"""

3.3.2 平台认证实现

class XiaoHongShuAuth(BaseAuth):
    """小红书认证实现"""

    # 登录检测元素
    LOGIN_INDICATORS = [
        "[data-testid='avatar']",
        ".creator-center-avatar",
        ".user-info"
    ]

class DouyinAuth(BaseAuth):
    """抖音认证实现"""

    # 登录检测元素
    LOGIN_INDICATORS = [
        ".user-avatar",
        ".creator-workspace",
        ".publish-btn"
    ]

3.4 工具模块 (utils/)

3.4.1 浏览器管理器

class BrowserManager:
    """浏览器实例管理"""

    def __init__(self, headless: bool = False):
        self.headless = headless
        self.playwright = None
        self.browsers = {}

    async def get_browser(self,
                         platform: PlatformType,
                         user_data_dir: str = None) -> Browser

    async def inject_stealth_script(self, page: Page) -> None:
        """注入反检测脚本"""

3.4.2 人类行为模拟器

class HumanBehaviorSimulator:
    """人类行为模拟"""

    def __init__(self, config: HumanBehaviorConfig):
        self.typing_speed = config.typing_speed  # 80-150ms/字符
        self.pause_probability = config.pause_probability  # 15%
        self.random_delay_range = config.random_delay_range  # 0.3-0.8s

    async def human_type(self, page: Page, selector: str, text: str) -> None:
        """模拟人类输入"""

    async def human_click(self, page: Page, selector: str) -> None:
        """模拟人类点击"""

    async def random_scroll(self, page: Page, distance: int) -> None:
        """随机滚动"""

3.4.3 媒体处理器

class MediaHandler:
    """媒体文件处理"""

    @staticmethod
    def validate_image(file_path: str) -> bool:
        """验证图片格式和大小"""

    @staticmethod
    def validate_video(file_path: str) -> bool:
        """验证视频格式和大小"""

    @staticmethod
    def compress_image(file_path: str, target_size: int) -> str:
        """压缩图片"""

    @staticmethod
    def compress_video(file_path: str, target_size: int) -> str:
        """压缩视频"""

3.5 配置管理 (config/)

3.5.1 系统配置

class Settings:
    """系统配置"""

    # 浏览器配置
    BROWSER_HEADLESS = False
    BROWSER_TIMEOUT = 30000  # 30秒

    # 并发配置
    MAX_CONCURRENT_TASKS = 3
    TASK_RETRY_COUNT = 3

    # 反检测配置
    HUMAN_TYPING_SPEED = (80, 150)  # ms/字符
    RANDOM_PAUSE_PROBABILITY = 0.15
    RANDOM_DELAY_RANGE = (0.3, 0.8)  # 秒

    # 文件大小限制
    MAX_IMAGE_SIZE = 10 * 1024 * 1024  # 10MB
    MAX_VIDEO_SIZE = 500 * 1024 * 1024  # 500MB

3.5.2 平台配置

class PlatformConfig:
    """平台特定配置"""

    XIAOHONGSHU = {
        'login_url': 'https://creator.xiaohongshu.com/',
        'image_note_url': 'https://creator.xiaohongshu.com/publish/publish',
        'video_note_url': 'https://creator.xiaohongshu.com/publish/video',
        'max_images': 9,
        'max_video_duration': 300,  # 5分钟
    }

    DOUYIN = {
        'login_url': 'https://creator.douyin.com/creator-micro/home',
        'upload_url': 'https://creator.douyin.com/creator-micro/content/upload',
        'max_video_duration': 600,  # 10分钟
        'supported_formats': ['mp4', 'mov', 'avi'],
    }

4. 数据模型设计

4.1 内容模型

@dataclass
class Content:
    """内容基类"""
    title: str
    description: str
    tags: List[str]
    visibility: str = "public"  # public, private, friends

@dataclass
class ImageNote(Content):
    """图文笔记"""
    images: List[str]  # 图片路径列表
    cover_image: str = None

@dataclass
class VideoContent(Content):
    """视频内容"""
    video_path: str
    cover_image: str = None
    duration: int = None

4.2 账号模型

@dataclass
class AccountInfo:
    """账号信息"""
    platform: PlatformType
    username: str
    cookie_file: str
    is_active: bool = True

@dataclass
class PublishTask:
    """发布任务"""
    id: str
    platform: PlatformType
    account: AccountInfo
    content: Content
    scheduled_time: datetime = None

@dataclass
class PublishResult:
    """发布结果"""
    task_id: str
    success: bool
    message: str
    published_url: str = None
    timestamp: datetime = None

5. API 设计

5.1 简单 API 接口

# 初始化发布器
publisher = Publisher(headless=False)

# 登录小红书
await publisher.login_platform("xiaohongshu", "user1")

# 发布图文笔记
note = ImageNote(
    title="我的旅行分享",
    description="今天去了很美的地方...",
    images=["path/to/image1.jpg", "path/to/image2.jpg"],
    tags=["旅行", "美景", "分享"]
)

result = await publisher.publish("xiaohongshu", note, "user1")

# 批量发布
tasks = [
    PublishTask("xiaohongshu", note, "user1"),
    PublishTask("douyin", video, "user2"),
]
results = await publisher.batch_publish(tasks)

5.2 高级 API 接口

# 配置化发布
config = {
    "platforms": ["xiaohongshu", "douyin"],
    "accounts": {
        "xiaohongshu": ["user1", "user2"],
        "douyin": ["user1"]
    },
    "schedule": {
        "type": "interval",  # interval, fixed_time
        "interval": 3600,    # 每小时发布一次
        "start_time": "09:00",
        "end_time": "21:00"
    }
}

scheduler = TaskScheduler(config)
await scheduler.start_auto_publish(content_queue)

6. 反自动化检测策略

6.1 多层检测规避

1. 浏览器指纹伪装

   • 使用 stealth.min.js 脚本
   • 修改 webdriver、navigator 等属性
   • 随机化浏览器版本和插件信息

2. 行为模拟

   • 随机打字速度 (80-150ms/字符)
   • 模拟真实点击和滚动
   • 随机暂停和操作间隔

3. 环境伪装

   • 使用真实的 User-Agent
   • 随机化视口大小
   • 模拟真实网络环境

6.2 智能重试机制

class RetryStrategy:
    """智能重试策略"""

    def __init__(self):
        self.max_retries = 3
        self.backoff_factor = 2
        self.retry_conditions = [
            NetworkError,
            TimeoutError,
            ElementNotFoundError,
            LoginRequiredError
        ]

    async def execute_with_retry(self, func, *args, **kwargs):
        """带重试的执行"""

7. 错误处理与日志

7.1 异常类型定义

class SocialMediaError(Exception):
    """基础异常类"""

class LoginFailedError(SocialMediaError):
    """登录失败异常"""

class UploadFailedError(SocialMediaError):
    """上传失败异常"""

class ContentRejectedError(SocialMediaError):
    """内容被拒绝异常"""

class RateLimitError(SocialMediaError):
    """频率限制异常"""

7.2 结构化日志

# 日志级别和格式
LOG_FORMAT = (
    "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
    "<level>{level: <8}</level> | "
    "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
    "<level>{message}</level>"
)

# 日志文件配置
logger.add(
    "logs/{time:YYYY-MM-DD}.log",
    rotation="1 day",
    retention="30 days",
    compression="zip",
    level="INFO"
)

8. 安全考虑

8.1 数据安全

• Cookie 文件加密存储
• 敏感信息不记录到日志
• 定期清理临时文件

8.2 访问控制

• API 访问频率限制
• 账号操作间隔控制
• 异常访问检测

9. 部署与使用

9.1 环境要求

Python >= 3.8
Playwright >= 1.40.0
Chrome/Chromium 浏览器

9.2 安装步骤

# 克隆项目
git clone https://github.com/your-repo/social_media_auto_publisher.git

# 安装依赖
pip install -r requirements.txt
playwright install chromium

# 配置环境
cp config/settings.example.py config/settings.py

9.3 快速开始

from social_media_auto_publisher import Publisher

async def main():
    # 初始化发布器
    publisher = Publisher(headless=False)

    # 登录平台
    await publisher.setup_platform("xiaohongshu")

    # 发布内容
    result = await publisher.publish_image_note(
        title="测试笔记",
        description="这是一个测试笔记",
        images=["test.jpg"],
        tags=["测试", "自动化"]
    )

    print(f"发布结果: {result}")

if __name__ == "__main__":
    asyncio.run(main())

10. 开发计划

10.1 第一阶段 (2周)

• 基础架构搭建
• 小红书登录模块
• 小红书图文笔记发布
• 基础反检测机制

10.2 第二阶段 (2周)

• 抖音登录模块
• 抖音视频发布
• 批量发布功能
• 错误处理优化

10.3 第三阶段 (1周)

• 任务调度系统
• 配置管理完善
• 文档和示例
• 测试用例

10.4 第四阶段 (1周)

• 性能优化
• 安全加固
• 部署脚本
• 用户手册

11. 技术难点与解决方案

11.1 平台反爬虫机制

难点: 各平台频繁更新反自动化策略 解决方案:

• 建立快速响应机制，及时更新选择器和脚本
• 多策略备选方案，提高容错性
• 与社区保持同步，共享最新的反检测方案

11.2 跨平台兼容性

难点: 不同平台的页面结构和API差异 解决方案:

• 使用适配器模式隔离平台差异
• 抽象统一的操作接口
• 灵活的配置系统支持平台特性

11.3 稳定性和可靠性

难点: 网络波动、平台变更等导致的失败 解决方案:

• 完善的重试机制和错误恢复
• 详细的日志记录和监控
• 优雅的降级策略

12. 扩展性设计

12.1 新平台接入

通过实现 PlatformAdapter 接口，可以轻松添加新的社交平台支持：

class BilibiliAdapter(PlatformAdapter):
    """B站适配器"""
    async def login(self, account: AccountInfo) -> bool: ...
    async def publish_video(self, video: VideoContent) -> bool: ...

12.2 功能扩展

• 内容智能生成 (AI文案、图片处理)
• 数据分析和报表
• 定时发布和内容规划
• 多账号统一管理

12.3 API扩展

• RESTful API 服务
• Web 管理界面
• 移动端支持
• 第三方集成

---

这个项目方案基于现有的 xiaohongshu_note_publisher 项目经验，设计了一个更加完善、可扩展的多平台社交媒体自动发布系统。通过模块化设计和现代化的架构，既能满足当前需求，又具备良好的扩展性和维护性。