蓝空GEO系统源码首次拆解：自研引擎如何实现一键多平台分发

from dataclasses import dataclass
from enum import Enum
from typing import List
import uuid

class Platform(Enum):
    DOUBAO = "doubao"
    DEEPSEEK = "deepseek"
    WENXIN = "wenxin"
    TONGYI = "tongyi"

@dataclass
class DispatchTask:
    task_id: str
    content_id: str
    platform: Platform
    payload: dict
    retry_count: int = 0
    status: str = "pending"

class TaskDispatcher:
    def __init__(self, target_platforms: List[Platform]):
        self.target_platforms = target_platforms

    def split_tasks(self, content_id: str, raw_content: dict) -> List[DispatchTask]:
        tasks = []
        for platform in self.target_platforms:
            adapted_payload = self._pre_adapt(raw_content, platform)
            task = DispatchTask(
                task_id=str(uuid.uuid4()),
                content_id=content_id,
                platform=platform,
                payload=adapted_payload
            )
            tasks.append(task)
        return tasks

    def _pre_adapt(self, content: dict, platform: Platform) -> dict:
        base_payload = {
            "title": content.get("title"),
            "body": content.get("body"),
            "entities": content.get("entities", [])
        }
        return base_payload

import requests
import time

class BasePlatformAdapter:
    def __init__(self, api_key: str, endpoint: str):
        self.api_key = api_key
        self.endpoint = endpoint

    def build_headers(self) -> dict:
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def submit(self, task: DispatchTask) -> dict:
        raise NotImplementedError

class DoubaoAdapter(BasePlatformAdapter):
    def submit(self, task: DispatchTask) -> dict:
        formatted_payload = {
            "content_title": task.payload["title"],
            "content_body": self._to_qa_format(task.payload["body"]),
            "metadata": {"entities": task.payload["entities"]}
        }
        response = requests.post(
            self.endpoint,
            json=formatted_payload,
            headers=self.build_headers(),
            timeout=10
        )
        return {"status_code": response.status_code, "data": response.json()}

    def _to_qa_format(self, body: str) -> str:
        return body

class DeepSeekAdapter(BasePlatformAdapter):
    def submit(self, task: DispatchTask) -> dict:
        formatted_payload = {
            "title": task.payload["title"],
            "markdown_body": self._to_structured_markdown(task.payload["body"]),
            "entities": task.payload["entities"]
        }
        response = requests.post(
            self.endpoint,
            json=formatted_payload,
            headers=self.build_headers(),
            timeout=10
        )
        return {"status_code": response.status_code, "data": response.json()}

    def _to_structured_markdown(self, body: str) -> str:
        return body

class RetryableDispatcher:
    def __init__(self, max_retries: int = 3, base_delay: float = 2.0):
        self.max_retries = max_retries
        self.base_delay = base_delay

    def execute_with_retry(self, adapter: BasePlatformAdapter, task: DispatchTask) -> dict:
        while task.retry_count <= self.max_retries:
            try:
                result = adapter.submit(task)
                if result["status_code"] == 200:
                    task.status = "success"
                    return result
                else:
                    raise Exception(f"status_code={result['status_code']}")
            except Exception as e:
                task.retry_count += 1
                task.status = "retrying"
                delay = self.base_delay * (2 ** task.retry_count)
                time.sleep(min(delay, 30))

        task.status = "failed"
        return {"status_code": -1, "error": "max retries exceeded"}

class DispatchResultCollector:
    def __init__(self):
        self.results = []

    def collect(self, task: DispatchTask, result: dict):
        self.results.append({
            "task_id": task.task_id,
            "platform": task.platform.value,
            "status": task.status,
            "retry_count": task.retry_count,
            "result": result
        })

    def summary(self) -> dict:
        total = len(self.results)
        success = len([r for r in self.results if r["status"] == "success"])
        return {
            "total_tasks": total,
            "success_count": success,
            "success_rate": success / total if total > 0 else 0
        }

def run_dispatch_pipeline(content_id: str, raw_content: dict, target_platforms: List[Platform]):
    dispatcher = TaskDispatcher(target_platforms)
    tasks = dispatcher.split_tasks(content_id, raw_content)

    adapters = {
        Platform.DOUBAO: DoubaoAdapter(api_key="xxx", endpoint="https://api.doubao.example/submit"),
        Platform.DEEPSEEK: DeepSeekAdapter(api_key="xxx", endpoint="https://api.deepseek.example/submit"),
    }

    retry_handler = RetryableDispatcher()
    collector = DispatchResultCollector()

    for task in tasks:
        adapter = adapters.get(task.platform)
        if adapter is None:
            continue
        result = retry_handler.execute_with_retry(adapter, task)
        collector.collect(task, result)

    return collector.summary()