Social Media Content Pipeline

pip install socaity

import os
import json
from socaity import deepseek_v3

ds = deepseek_v3(api_key=os.getenv("SOCAITY_API_KEY"))

def write_script(topic: str) -> dict:
    """Return a structured script with scene prompts."""
    prompt = (
        "You are a social media video scriptwriter. "
        "Return ONLY valid JSON with keys: 'title', 'scene_prompts' "
        "(list of 3 image prompts), 'script' (≤ 80 words for 30 s voice).

"
        f"Write a short social media video script about: {topic}"
    )
    response = ds.predictions(prompt=prompt).get_result()
    # deepseek_v3 streams tokens — join the chunks into the final string.
    script_text = "".join(response) if isinstance(response, list) else response
    return json.loads(script_text)

# Example
content = write_script("The future of AI in healthcare")
print(content["title"])
# → "AI is Rewriting Medicine — Here's What You Need to Know"

from socaity import flux_schnell
from socaity import speechcraft, face2face

flux = flux_schnell(api_key=os.getenv("SOCAITY_API_KEY"))
sc   = speechcraft(api_key=os.getenv("SOCAITY_API_KEY"))
f2f  = face2face(api_key=os.getenv("SOCAITY_API_KEY"))

def launch_parallel_stages(content: dict, avatar_img: str) -> tuple:
    """Submit all three GPU jobs simultaneously."""

    # Stage 2 — generate one hero image per scene prompt
    image_jobs = [
        flux(prompt=p, num_outputs=1)
        for p in content["scene_prompts"]
    ]

    # Stage 3 — synthesise the voiceover
    audio_job = sc.text2voice(
        text=content["script"],
        voice="en_male_calm",
    )

    # Stage 4 — swap presenter face into a pre-recorded template video clip
    video_job = f2f.swap_video(
        faces=avatar_img,
        target_video="./blank_avatar_30s.mp4",
    )

    return image_jobs, audio_job, video_job

import pathlib

def collect_and_save(topic: str, image_jobs, audio_job, video_job) -> str:
    slug = topic.lower().replace(" ", "_")[:30]
    out  = pathlib.Path(f"./output/{slug}")
    out.mkdir(parents=True, exist_ok=True)

    # Block and save images
    for i, job in enumerate(image_jobs):
        imgs = job.get_result()
        imgs[0].save(out / f"scene_{i}.png")

    # Block and save audio
    audio = audio_job.get_result()
    audio.save(out / "voiceover.mp3")

    # Block and save video
    video = video_job.get_result()
    video.save(out / "avatar_video.mp4")

    print(f"Pipeline complete → {out}")
    return str(out)

import os
import json
import pathlib
from socaity import deepseek_v3
from socaity import speechcraft, face2face
from socaity import flux_schnell

SOCAITY_KEY = os.getenv("SOCAITY_API_KEY")
AVATAR_IMG  = "./avatar.jpg"   # your presenter face

# --- Clients ---
# All four models authenticate with SOCAITY_API_KEY.
# Replicate-backed models (deepseek_v3, flux_schnell) are routed through the
# Socaity backend, which handles the upstream Replicate call for you.
ds   = deepseek_v3(api_key=SOCAITY_KEY)
flux = flux_schnell(api_key=SOCAITY_KEY)
sc   = speechcraft(api_key=SOCAITY_KEY)
f2f  = face2face(api_key=SOCAITY_KEY)

def run_pipeline(topic: str) -> str:
    print(f"[1/4] Writing script for: {topic}")
    resp = ds.predictions(prompt=(
        "You are a social media scriptwriter. Return ONLY valid JSON with keys: "
        "'title', 'scene_prompts' (list[str], 3 items), 'script' (str, ≤ 80 words).

"
        f"Topic: {topic}"
    )).get_result()
    # deepseek_v3 streams tokens — join the chunks into the final string.
    resp_text = "".join(resp) if isinstance(resp, list) else resp
    content = json.loads(resp_text)
    print(f"    Title: {content['title']}")

    print("[2-4/4] Submitting image / voice / video jobs in parallel...")
    image_jobs = [flux(prompt=p, num_outputs=1) for p in content["scene_prompts"]]
    audio_job  = sc.text2voice(text=content["script"], voice="en_male_calm")
    # face2face swaps the presenter face into an existing template video clip
    video_job  = f2f.swap_video(faces=AVATAR_IMG, target_video="./blank_30s.mp4")

    slug = topic.lower().replace(" ", "_")[:30]
    out  = pathlib.Path(f"./output/{slug}")
    out.mkdir(parents=True, exist_ok=True)

    for i, job in enumerate(image_jobs):
        job.get_result()[0].save(out / f"scene_{i}.png")
    audio_job.get_result().save(out / "voiceover.mp3")
    video_job.get_result().save(out / "avatar_video.mp4")

    print(f"Done! Output in {out}")
    return str(out)


if __name__ == "__main__":
    run_pipeline("The future of AI in healthcare")

import queue
import threading

topic_queue: queue.Queue = queue.Queue()

def worker():
    while True:
        topic = topic_queue.get()
        try:
            run_pipeline(topic)
        except Exception as e:
            print(f"Pipeline failed for '{topic}': {e}")
        finally:
            topic_queue.task_done()

# Spawn 4 parallel pipeline workers
for _ in range(4):
    threading.Thread(target=worker, daemon=True).start()

# Feed topics (e.g. from a database or Kafka)
topics = [
    "The future of AI in healthcare",
    "How to build a personal brand in 2026",
    "Top 5 productivity hacks for founders",
]
for t in topics:
    topic_queue.put(t)

topic_queue.join()
print("All topics processed.")