Social Media Content Pipeline

pip install socaity
export SOCAITY_API_KEY=sk-...

import os
import json
from socaity.sdk.replicate.deepseek_ai 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' (str, max 80 words for 30 s voice).\n\n"
        f"Write a short social media video script about: {topic}"
    )
    # deepseek_v3 defaults: max_tokens=2048, temperature=0.1. Override only if needed.
    response = ds.predictions(prompt=prompt).get_result()
    # Some Replicate-backed text models return a list of token chunks; coerce to str.
    script_text = "".join(response) if isinstance(response, list) else response
    return json.loads(script_text)

content = write_script("The future of AI in healthcare")
print(content["title"])

from socaity.sdk.replicate.black_forest_labs import flux_schnell
from socaity.sdk.replicate.jaaari import kokoro_82m
# face2face is self-hosted via APIPod, not a catalog import. Deploy it yourself
# and import the generated client. See /apipod/host-existing-model.
from my_apipod_clients import face2face

flux = flux_schnell(api_key=os.getenv("SOCAITY_API_KEY"))
tts  = kokoro_82m(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 at once. Each call returns immediately."""

    # Stage 2: one hero image per scene prompt.
    # flux_schnell defaults: num_inference_steps=4, output_format='webp', num_outputs=1.
    image_jobs = [
        flux(prompt=p) for p in content["scene_prompts"]
    ]

    # Stage 3: synthesise the voiceover.
    # kokoro_82m takes text plus an optional speed and selectable voice id.
    audio_job = tts(text=content["script"])

    # Stage 4: swap the presenter face into a pre-recorded template clip.
    # face2face defaults to enhance_face_model='gpen_bfr_512'.
    video_job = f2f.swap_video(
        faces=avatar_img,
        target_video="./blank_avatar_30s.mp4",
    )

    return image_jobs, audio_job, video_job

import pathlib
from fastsdk import gather_results

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)

    # gather_results waits on every job in parallel. results_only=True returns a
    # list in submission order (default returns a dict keyed by job name). Flatten
    # the image jobs into the list so the whole batch blocks once instead of
    # stage-by-stage.
    all_jobs = [*image_jobs, audio_job, video_job]
    results  = gather_results(all_jobs, results_only=True)

    images = results[: len(image_jobs)]
    audio  = results[len(image_jobs)]
    video  = results[len(image_jobs) + 1]

    # flux_schnell returns one image per submission (num_outputs=1 default).
    for i, img in enumerate(images):
        img.save(out / f"scene_{i}.webp")

    audio.save(out / "voiceover.wav")
    video.save(out / "avatar_video.mp4")

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

import os
import json
import pathlib
from fastsdk import gather_results
from socaity.sdk.replicate.deepseek_ai import deepseek_v3
from socaity.sdk.replicate.black_forest_labs import flux_schnell
from socaity.sdk.replicate.jaaari import kokoro_82m
# face2face is self-hosted via APIPod, not a catalog import. Deploy it yourself
# and import the generated client. See /apipod/host-existing-model.
from my_apipod_clients import face2face

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

# All four clients authenticate with SOCAITY_API_KEY. The Replicate-backed
# models (deepseek_v3, flux_schnell, kokoro_82m) are routed through the Socaity
# backend, which proxies the upstream Replicate call.
ds   = deepseek_v3(api_key=SOCAITY_KEY)
flux = flux_schnell(api_key=SOCAITY_KEY)
tts  = kokoro_82m(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, max 80 words).\n\n"
        f"Topic: {topic}"
    )).get_result()
    # Coerce list-of-chunks into a single string before JSON parsing.
    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, and video jobs in parallel...")
    # flux_schnell defaults: num_inference_steps=4, output_format='webp'.
    image_jobs = [flux(prompt=p) for p in content["scene_prompts"]]
    # kokoro_82m takes text plus an optional speed and selectable voice id.
    audio_job  = tts(text=content["script"])
    # face2face replaces the face in an existing template clip; it does not
    # animate from audio. Default enhance_face_model='gpen_bfr_512'.
    video_job  = f2f.swap_video(faces=AVATAR_IMG, target_video="./blank_30s.mp4")

    # Wait on every outstanding job in one call. results_only=True returns a list
    # in submission order (default returns a dict keyed by job name).
    all_jobs = [*image_jobs, audio_job, video_job]
    results  = gather_results(all_jobs, results_only=True)
    images, audio, video = results[:-2], results[-2], results[-1]

    slug = topic.lower().replace(" ", "_")[:30]
    out  = pathlib.Path(f"./output/{slug}")
    out.mkdir(parents=True, exist_ok=True)
    for i, img in enumerate(images):
        img.save(out / f"scene_{i}.webp")
    audio.save(out / "voiceover.wav")
    video.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")

from fastsdk import gather_results

# A submit-only version of the pipeline. It builds the per-topic job graph,
# then returns the handles without blocking. The caller waits on all of them
# together with gather_results, so SocAIty keeps every topic's GPU work in
# flight at the same time.
def submit_pipeline(topic: str):
    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, max 80 words).\n\n"
        f"Topic: {topic}"
    )).get_result()
    resp_text = "".join(resp) if isinstance(resp, list) else resp
    content = json.loads(resp_text)

    image_jobs = [flux(prompt=p) for p in content["scene_prompts"]]
    audio_job  = tts(text=content["script"])
    video_job  = f2f.swap_video(faces=AVATAR_IMG, target_video="./blank_30s.mp4")
    return topic, content, image_jobs, audio_job, video_job

topics = [
    "The future of AI in healthcare",
    "How to build a personal brand in 2026",
    "Top 5 productivity hacks for founders",
]

# Submit every topic's GPU jobs, then wait on the entire batch in one call.
batches = [submit_pipeline(t) for t in topics]
flat    = [j for _, _, imgs, a, v in batches for j in (*imgs, a, v)]
results = gather_results(flat, raise_on_error=False, results_only=True)

print(f"All {len(topics)} topics processed.")