Social Posting

Social Posting Skill

Multi-platform social media posting with automatic provider failover. Users connect their own social accounts via OAuth — credentials are encrypted at rest. Supports immediate posting, scheduling, and media attachments.

Supported Platforms

Provider Architecture

Two providers with automatic failover:

Failover order:

1. Try user's PostForMe credentials
2. Try user's LATE credentials
3. Fall back to global env-var credentials (PostForMe → LATE)

Core Data Structures

class Platform(Enum):
    TWITTER = "twitter"
    LINKEDIN = "linkedin"
    INSTAGRAM = "instagram"
    FACEBOOK = "facebook"
    TIKTOK = "tiktok"
    THREADS = "threads"
    BLUESKY = "bluesky"
    YOUTUBE = "youtube"
    PINTEREST = "pinterest"

@dataclass
class PostResult:
    success: bool
    post_id: Optional[str] = None
    platform_post_ids: Optional[Dict[str, str]] = None
    platform_post_urls: Optional[Dict[str, str]] = None
    error: Optional[str] = None
    provider: Optional[str] = None
    scheduled_for: Optional[datetime] = None

@dataclass
class AccountInfo:
    id: str
    platform: str
    username: str
    profile_id: Optional[str] = None

Provider Interface

Both providers implement the same abstract interface:

class SocialPostingProvider(ABC):
    @property
    @abstractmethod
    def name(self) -> str: ...

    @abstractmethod
    def get_accounts(self) -> List[AccountInfo]: ...

    @abstractmethod
    def upload_media(self, image_url: str) -> Optional[str]: ...

    @abstractmethod
    def post(
        self,
        content: str,
        platforms: List[str],
        media_urls: Optional[List[str]] = None,
        scheduled_for: Optional[datetime] = None
    ) -> PostResult: ...

PostForMe Provider (Primary)

Authentication: Authorization: Bearer {api_key}

OAuth URL Generation

POST /social-accounts/auth-url
{
  "platform": "twitter",
  "redirect_url": "https://your-app.com/callback"
}
# Returns: {"url": "..."} or {"data": {"auth_url": "..."}}

Get Connected Accounts

GET /social-accounts
# Returns: {"data": [{"id": "...", "platform": "twitter", "username": "..."}]}

Media Upload (Presigned URL Flow)

# 1. Get presigned URL
POST /media/create-upload-url
{"content_type": "image/jpeg"}
# Returns: {"upload_url": "...", "media_url": "..."}

# 2. PUT image bytes to upload_url
# 3. Use returned media_url in post payload

Create Post

POST /social-posts
{
  "caption": "Post content here",
  "social_accounts": ["account_id_1", "account_id_2"],
  "media": [{"url": "https://..."}],        # optional
  "scheduled_at": "2025-01-01T09:00:00"     # optional ISO datetime
}
# Returns: {"id": "post_id"} or {"data": {"id": "post_id"}}

Note: PostForMe normalizes twitter platform to both "twitter" and "x" internally.

LATE Provider (Fallback)

Authentication: Authorization: Bearer {api_key}

Get Connected Accounts

GET /accounts
# Returns: {"accounts": [{"_id": "...", "platform": "...", "username": "...", "profileId": "..."}]}

Media Upload (Presigned URL Flow)

# 1. Get presigned URL
POST /media/presign
{"filename": "media.jpg", "contentType": "image/jpeg"}
# Returns: {"uploadUrl": "...", "publicUrl": "..."}

# 2. PUT image bytes to uploadUrl (wait 1s for CDN propagation)
# 3. Use publicUrl in post payload

Create Post

POST /posts
{
  "content": "Post content here",
  "platforms": [
    {"platform": "twitter", "accountId": "...", "profileId": "..."}
  ],
  "mediaItems": [{"url": "https://...", "type": "image"}],  # optional
  "scheduledFor": "2025-01-01T09:00:00"                      # optional
}
# Returns: {"post": {"_id": "...", "platforms": [{"platform": "twitter", "platformPostId": "...", "platformPostUrl": "..."}]}}

Service Layer (SocialPostingService)

The service layer wraps both providers and adds database integration.

Environment Variables Required

# Provider API keys (for global fallback if user has no personal creds)
POSTFORME_API_KEY=your_postforme_key
LATE_API_KEY=your_late_key

# Encryption for stored credentials
ENCRYPTION_KEY=your_fernet_key  # Generate: Fernet.generate_key()

Credential Management

service = SocialPostingService()
service.init(supabase_client)

# Save user credentials
service.save_credentials(
    user_id="user-uuid",
    provider="postforme",  # or "late"
    api_key="sk-...",
    connected_platforms=["twitter", "linkedin"]
)

# Get credentials (auto-decrypted)
creds = service.get_credentials(user_id="user-uuid", provider="postforme")

# Delete credentials
service.delete_credentials(user_id="user-uuid", provider="postforme")

Credentials are encrypted using Fernet symmetric encryption before database storage. Set ENCRYPTION_KEY environment variable to a valid Fernet key.

OAuth Flow

# Generate OAuth URL for user to connect a platform
oauth_url = service.get_oauth_url(
    user_id="user-uuid",
    platform="twitter",
    redirect_url="https://your-app.com/oauth/callback"
)
# Returns: URL string or None

Posting

# Immediate post
result = service.create_post(
    user_id="user-uuid",
    content="Your post content",
    platforms=["twitter", "linkedin"],
    media_urls=["https://cdn.example.com/image.jpg"],  # optional
    scheduled_for=None,
    campaign_id="campaign-uuid",   # optional, for tracking
    batch_number=1                 # optional, for tracking
)

# Scheduled post
from datetime import datetime, timezone
result = service.create_post(
    user_id="user-uuid",
    content="Scheduled post content",
    platforms=["linkedin"],
    scheduled_for=datetime(2025, 6, 1, 9, 0, 0, tzinfo=timezone.utc)
)

# result.success → bool
# result.post_id → provider post ID
# result.platform_post_ids → {"twitter": "tweet_id", ...}
# result.platform_post_urls → {"twitter": "https://...", ...}
# result.error → error message if failed
# result.provider → "PostForMe" or "LATE"

Publish from Campaign Calendar

result = service.publish_batch(
    user_id="user-uuid",
    campaign_id="campaign-uuid",
    batch_number=3,
    platforms=["twitter", "instagram"],
    media_urls=["https://..."],    # selected images for this batch
    scheduled_for=None             # or datetime for scheduling
)

Looks up campaigns.creative_calendar.batches[n].caption from database and posts it.

Account Management

# Get connected platforms for a user
accounts = service.get_connected_accounts(user_id="user-uuid")
# Returns: [{"id": "...", "platform": "twitter", "username": "@handle"}]

# Refresh connected_platforms field in credentials table
service.refresh_connected_platforms(user_id="user-uuid")

Post History

# Get all posts
history = service.get_post_history(user_id="user-uuid", limit=50)

# Filter by status: "posted", "scheduled", "failed"
scheduled = service.get_post_history(user_id="user-uuid", status="scheduled")

# Get single post
post = service.get_post(post_id="post-uuid")

Database Schema

user_social_credentials

CREATE TABLE user_social_credentials (
    user_id         UUID NOT NULL,
    provider        TEXT NOT NULL,  -- 'postforme' | 'late'
    encrypted_api_key TEXT NOT NULL,
    connected_platforms TEXT[],
    updated_at      TIMESTAMPTZ,
    PRIMARY KEY (user_id, provider)
);

social_posts

CREATE TABLE social_posts (
    id                  UUID DEFAULT gen_random_uuid() PRIMARY KEY,
    user_id             UUID NOT NULL,
    provider            TEXT,
    provider_post_id    TEXT,
    platforms           TEXT[],
    platform_post_ids   JSONB,
    platform_post_urls  JSONB,
    content             TEXT,
    media_urls          TEXT[],
    scheduled_for       TIMESTAMPTZ,
    posted_at           TIMESTAMPTZ,
    status              TEXT,  -- 'posted' | 'scheduled' | 'failed'
    error_message       TEXT,
    campaign_id         UUID,
    batch_number        INTEGER,
    created_at          TIMESTAMPTZ DEFAULT NOW()
);

Failover Logic

User requests post to platforms X, Y, Z
         ↓
Check: user has PostForMe creds?
    YES → PostForMeProvider(user_key)
    NO  → Check: user has LATE creds?
              YES → LateProvider(user_key)
              NO  → Check: POSTFORME_API_KEY env var?
                        YES → PostForMeProvider(global_key)
                        NO  → Check: LATE_API_KEY env var?
                                  YES → LateProvider(global_key)
                                  NO  → Return error: no credentials configured
         ↓
Call provider.post(content, platforms, media_urls, scheduled_for)
         ↓
Track result in social_posts table
         ↓
Return PostResult

Usage Example (Standalone)

import os
from datetime import datetime
from social_posting_service import (
    SocialPostingService,
    PostForMeProvider,
    LateProvider
)

# Option A: Use PostForMe directly
provider = PostForMeProvider(api_key=os.getenv("POSTFORME_API_KEY"))
result = provider.post(
    content="Hello from the API!",
    platforms=["twitter", "linkedin"],
    media_urls=["https://example.com/image.jpg"]
)
print(result.success, result.post_id)

# Option B: Use service with database
from supabase import create_client
supabase = create_client(os.getenv("SUPABASE_URL"), os.getenv("SUPABASE_KEY"))

service = SocialPostingService()
service.init(supabase)
result = service.create_post(
    user_id="user-uuid",
    content="Post via service",
    platforms=["twitter"]
)

Error Handling

All methods return structured results — they do not raise exceptions to the caller.

Integration Checklist

• Install requests, cryptography packages
• Set ENCRYPTION_KEY env var (generate with Fernet.generate_key())
• Set at least one provider key: POSTFORME_API_KEY or LATE_API_KEY
• Create user_social_credentials table in database
• Create social_posts table in database
• Initialize service: service.init(supabase_client)
• Guide users through OAuth flow before first post