forked from LiveCarta/ContentAutomation
77 lines
2.8 KiB
Python
77 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
from content_automation.adapters.storage.base import StorageAdapterBase
|
|
from content_automation.interfaces import PublishedContentStore, SocialNetworkAdapter
|
|
from content_automation.models import PublishedContentRecord
|
|
from content_automation.settings import AppSettings
|
|
|
|
|
|
class PublishController:
|
|
"""Coordinates storage lookup and cross-network publishing."""
|
|
|
|
def __init__(
|
|
self,
|
|
settings: AppSettings,
|
|
storage: StorageAdapterBase,
|
|
social_adapters: dict[str, SocialNetworkAdapter],
|
|
published_content_store: PublishedContentStore | None = None,
|
|
) -> None:
|
|
self._settings = settings
|
|
self._storage = storage
|
|
self._social_adapters = social_adapters
|
|
self._published_content_store = published_content_store
|
|
|
|
def publish(self, relative_path: str, caption: str) -> dict[str, Any]:
|
|
if not self._storage.exists(relative_path):
|
|
raise FileNotFoundError(
|
|
f"Media file is not available in storage: {relative_path}"
|
|
)
|
|
|
|
media_url = self._storage.get_public_url(relative_path)
|
|
result: dict[str, Any] = {}
|
|
for network in self._settings.target_social_networks:
|
|
adapter = self._social_adapters.get(network)
|
|
if adapter is None:
|
|
raise ValueError(f"No adapter configured for network: {network}")
|
|
result[network] = adapter.post_media(media_url=media_url, caption=caption)
|
|
|
|
if self._published_content_store is not None:
|
|
record = PublishedContentRecord(
|
|
relative_path=relative_path,
|
|
media_url=media_url,
|
|
caption=caption,
|
|
platform_responses=self._serialize_platform_responses(result),
|
|
)
|
|
self._published_content_store.save(record)
|
|
|
|
return result
|
|
|
|
def _serialize_platform_responses(
|
|
self, responses: Mapping[str, Any]
|
|
) -> dict[str, Any]:
|
|
serialized: dict[str, Any] = {}
|
|
for network, response in responses.items():
|
|
if network == "youtube":
|
|
serialized[network] = self._extract_youtube_video_id(response)
|
|
continue
|
|
serialized[network] = response
|
|
return serialized
|
|
|
|
@staticmethod
|
|
def _extract_youtube_video_id(response: Any) -> str:
|
|
if isinstance(response, str):
|
|
return response
|
|
|
|
if isinstance(response, Mapping):
|
|
video_id = response.get("id")
|
|
if isinstance(video_id, str) and video_id:
|
|
return video_id
|
|
|
|
raise ValueError(
|
|
"Unsupported YouTube publish response: expected a string ID "
|
|
"or payload containing an 'id' field."
|
|
)
|