Social Posting

SKILL.md

---
name: social-posting
description: Multi-platform social media posting service with automatic provider failover. Handles posting to 9 platforms (Twitter/X, LinkedIn, Instagram, Facebook, TikTok, Threads, Bluesky, YouTube, Pinterest) with per-user credential management, OAuth flow, media upload, scheduling, and post history tracking. Triggers on "post to social media", "publish to platforms", "schedule a post", "social post", or any cross-platform publishing request.
homepage: https://canlah.ai
metadata: {"category": "social-media", "platforms": ["twitter", "linkedin", "instagram", "facebook", "tiktok", "threads", "bluesky", "youtube", "pinterest"], "providers": ["PostForMe", "LATE"], "features": ["oauth", "scheduling", "media-upload", "failover", "post-history"]}
---

# 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

| Platform | Enum Value |
|----------|-----------|
| Twitter / X | `twitter` |
| LinkedIn | `linkedin` |
| Instagram | `instagram` |
| Facebook | `facebook` |
| TikTok | `tiktok` |
| Threads | `threads` |
| Bluesky | `bluesky` |
| YouTube | `youtube` |
| Pinterest | `pinterest` |

---

## Provider Architecture

Two providers with automatic failover:

| Provider | Role | API Base |
|----------|------|----------|
| **PostForMe** | Primary (cheaper) | `https://api.postforme.dev/v1` |
| **LATE** | Fallback (reliable) | `https://getlate.dev/api/v1` |

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

```python
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:

```python
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

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

### Get Connected Accounts

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

### Media Upload (Presigned URL Flow)

```python
# 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

```python
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

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

### Media Upload (Presigned URL Flow)

```python
# 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

```python
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

```bash
# 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

```python
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

```python
# 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

```python
# 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

```python
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

```python
# 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

```python
# 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

```sql
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

```sql
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)

```python
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.

| Error Condition | Result |
|-----------------|--------|
| No credentials configured | `PostResult(success=False, error="No social posting credentials configured...")` |
| No connected account for platform | `PostResult(success=False, error="No connected accounts for: [...]")` |
| Provider HTTP error | `PostResult(success=False, error="HTTP 4xx: ...")` |
| Network timeout | `PostResult(success=False, error="...")` |
| DB tracking failure | Logs error, still returns posting result |

---

## 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

---

## Author

**[Canlah AI](https://canlah.ai)** — Run performance marketing without breaking your brand.

- GitHub: [github.com/PHY041](https://github.com/PHY041)
- All Skills: [clawhub.ai/PHY041](https://clawhub.ai/PHY041)