Pipelines

from late import Late
from late.pipelines import CSVSchedulerPipeline

client = Late(api_key="your_api_key")
pipeline = CSVSchedulerPipeline(client)

# Schedule all posts from CSV
results = pipeline.schedule("posts.csv")

for result in results:
    if result.success:
        print(f"Row {result.row}: Created post {result.post_id}")
    else:
        print(f"Row {result.row}: Failed - {result.error}")

results = pipeline.schedule("posts.csv", dry_run=True)

if all(r.success for r in results):
    print("Validation passed! Ready to schedule.")
    # Actually create posts
    results = pipeline.schedule("posts.csv")
else:
    print("Validation errors:")
    for r in results:
        if not r.success:
            print(f"  Row {r.row}: {r.error}")

content,platform,account_id,scheduled_for,title,media_url,tags
"Hello world!",twitter,acc_123,2024-12-25T10:00:00Z,,,
"LinkedIn post",linkedin,acc_456,2024-12-25T11:00:00Z,My Title,,tag1,tag2
"With image",instagram,acc_789,2024-12-25T12:00:00Z,,https://example.com/image.jpg,

# Use custom date format
pipeline = CSVSchedulerPipeline(
    client,
    date_format="%Y-%m-%d %H:%M:%S",
    default_timezone="America/New_York"
)

results = pipeline.schedule("posts.csv")

import asyncio
from late import Late
from late.pipelines import CSVSchedulerPipeline

async def schedule_posts():
    async with Late(api_key="your_api_key") as client:
        pipeline = CSVSchedulerPipeline(client)
        results = await pipeline.aschedule("posts.csv")

        for r in results:
            print(f"Row {r.row}: {'OK' if r.success else r.error}")

asyncio.run(schedule_posts())

import asyncio
from late import Late, Platform
from late.pipelines import CrossPosterPipeline, PlatformConfig

async def cross_post():
    async with Late(api_key="your_api_key") as client:
        cross_poster = CrossPosterPipeline(client)

        results = await cross_poster.post(
            content="Exciting news! Our new feature is live.",
            platforms=[
                PlatformConfig(Platform.TWITTER, "tw_123"),
                PlatformConfig(Platform.LINKEDIN, "li_456"),
                PlatformConfig(Platform.INSTAGRAM, "ig_789"),
            ],
        )

        for r in results:
            if r.success:
                print(f"{r.platform}: Posted ({r.post_id})")
            else:
                print(f"{r.platform}: Failed - {r.error}")

asyncio.run(cross_post())

results = await cross_poster.post(
    content="Big announcement!",
    platforms=[
        PlatformConfig(Platform.TWITTER, "tw_123", delay_minutes=0),    # Post immediately
        PlatformConfig(Platform.LINKEDIN, "li_456", delay_minutes=5),   # 5 min later
        PlatformConfig(Platform.INSTAGRAM, "ig_789", delay_minutes=10), # 10 min later
    ],
)

results = await cross_poster.post(
    content="Default content for all platforms",
    platforms=[
        PlatformConfig(
            Platform.TWITTER,
            "tw_123",
            custom_content="Short version for Twitter! #announcement"
        ),
        PlatformConfig(
            Platform.LINKEDIN,
            "li_456",
            custom_content="Longer, more professional version for LinkedIn..."
        ),
        PlatformConfig(Platform.INSTAGRAM, "ig_789"),  # Uses default content
    ],
)

# Automatic content truncation (default: enabled)
results = await cross_poster.post(
    content="Very long content that exceeds some platform limits...",
    platforms=[...],
    adapt_content=True  # Truncates for each platform
)

# Disable adaptation
results = await cross_poster.post(
    content="Short content",
    platforms=[...],
    adapt_content=False  # Use exact content
)

from late import MediaType

results = await cross_poster.post(
    content="Check out our new product!",
    platforms=[
        PlatformConfig(Platform.TWITTER, "tw_123"),
        PlatformConfig(Platform.INSTAGRAM, "ig_456"),
    ],
    title="Product Launch",
    media_items=[{"type": MediaType.IMAGE, "url": "https://example.com/product.jpg"}],
    tags=["product", "launch", "tech"],
)

from late import Late, Platform
from late.pipelines import CrossPosterPipeline, PlatformConfig

client = Late(api_key="your_api_key")
cross_poster = CrossPosterPipeline(client)

# Synchronous cross-posting
results = cross_poster.post_sync(
    content="Announcement!",
    platforms=[
        PlatformConfig(Platform.TWITTER, "tw_123"),
        PlatformConfig(Platform.LINKEDIN, "li_456"),
    ],
)

CSVSchedulerPipeline(
    client: Late,
    date_format: str = "%Y-%m-%d %H:%M:%S",
    default_timezone: str = "UTC"
)

CrossPosterPipeline(
    client: Late,
    default_stagger: int = 5
)

PlatformConfig(
    platform: Platform | str,
    account_id: str,
    custom_content: str | None = None,
    delay_minutes: int = 0
)