File size: 2,476 Bytes
59215bb 345855e 59215bb 345855e 59215bb 345855e 59215bb 345855e 59215bb 345855e 59215bb 345855e 59215bb 345855e 59215bb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | from publisher.platforms.facebook import publish_facebook
from publisher.platforms.reels import publish_reels
from publisher.platforms.shorts import publish_shorts
from publisher.platforms.tiktok import publish_tiktok
PUBLISHERS = {
"tiktok": publish_tiktok,
"reels": publish_reels,
"shorts": publish_shorts,
"facebook": publish_facebook,
}
async def publish_all(
source,
platforms,
metadata=None,
hashtags=None,
thumbnail=None,
schedule_time=None,
):
results = {}
metadata = metadata or {}
hashtags = hashtags or {}
caption = metadata.get("metadata") if isinstance(metadata, dict) else str(metadata)
tag_value = hashtags.get("hashtags", hashtags) if isinstance(hashtags, dict) else hashtags
if isinstance(tag_value, str):
tag_value = [tag.strip() for tag in tag_value.split(",") if tag.strip()]
for platform in platforms:
publisher = PUBLISHERS.get(platform)
if not publisher:
results[platform] = {"status": "skipped", "error": "Unsupported platform"}
continue
platform_payload = {
"video_path": source,
"caption": caption or "",
"hashtags": tag_value or [],
"thumbnail": thumbnail,
"schedule_time": schedule_time,
}
try:
results[platform] = await publisher(platform_payload)
except Exception as exc:
results[platform] = {"status": "failed", "error": str(exc)}
return results
async def publish(payload):
if isinstance(payload, dict):
source = payload.get("source") or payload.get("video_path")
platforms = payload.get("platforms", [])
metadata = payload.get("metadata")
hashtags = payload.get("hashtags")
thumbnail = payload.get("thumbnail")
schedule_time = payload.get("schedule_time")
else:
source = getattr(payload, "source", None) or getattr(payload, "video_path", None)
platforms = getattr(payload, "platforms", [])
metadata = getattr(payload, "metadata", None)
hashtags = getattr(payload, "hashtags", None)
thumbnail = getattr(payload, "thumbnail", None)
schedule_time = getattr(payload, "schedule_time", None)
return await publish_all(
source=source,
platforms=platforms,
metadata=metadata,
hashtags=hashtags,
thumbnail=thumbnail,
schedule_time=schedule_time,
)
|