Social media automation: Instagram, Telegram, content on autopilot

# requests is enough for the examples below
pip install requests

import requests
import os

INSTAGRAM_BUSINESS_ACCOUNT_ID = os.environ["INSTAGRAM_ACCOUNT_ID"]
ACCESS_TOKEN = os.environ["META_ACCESS_TOKEN"]

def publish_instagram_post(image_url: str, caption: str) -> dict:
    """
    Publishes a post to Instagram through the Meta Graph API.
    image_url must be a publicly accessible URL (not a local file!)
    Replace the API version (v20.0 in the example) with the current one from Meta's documentation.
    """
    base_url = f"https://graph.facebook.com/v20.0/{INSTAGRAM_BUSINESS_ACCOUNT_ID}"
    
    # Step 1: Create a media container
    container_response = requests.post(
        f"{base_url}/media",
        params={
            "image_url": image_url,  # Public URL of the image
            "caption": caption,
            "access_token": ACCESS_TOKEN,
        }
    )
    container_id = container_response.json().get("id")
    
    if not container_id:
        return {"error": container_response.json()}
    
    # Step 2: Publish the container
    publish_response = requests.post(
        f"{base_url}/media_publish",
        params={
            "creation_id": container_id,
            "access_token": ACCESS_TOKEN,
        }
    )
    
    return publish_response.json()

def get_instagram_insights(post_id: str) -> dict:
    """Get the stats for a post"""
    response = requests.get(
        f"https://graph.facebook.com/v20.0/{post_id}/insights",
        params={
            "metric": "reach,impressions,likes,comments,shares,saves",
            "access_token": ACCESS_TOKEN,
        }
    )
    return response.json()

def schedule_post(image_url: str, caption: str, publish_time: str) -> dict:
    """
    Schedule a post (a rough sketch: check Meta's documentation to see whether the Instagram API
    supports scheduled publishing. The reliable option: store the post yourself and call publish on a schedule).
    publish_time: ISO 8601 format (for example "2026-12-15T10:00:00+0000")
    """
    base_url = f"https://graph.facebook.com/v20.0/{INSTAGRAM_BUSINESS_ACCOUNT_ID}"
    
    # Convert the time to a Unix timestamp
    from datetime import datetime
    scheduled_ts = int(datetime.fromisoformat(publish_time).timestamp())
    
    response = requests.post(
        f"{base_url}/media",
        params={
            "image_url": image_url,
            "caption": caption,
            "scheduled_publish_time": scheduled_ts,
            "is_carousel_item": False,
            "access_token": ACCESS_TOKEN,
        }
    )
    return response.json()

from playwright.async_api import async_playwright
import asyncio
import anthropic
import os

claude = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

async def generate_and_post_to_instagram(topic: str):
    """
    IMPORTANT: This is a demo example. Browser automation
    violates Instagram's Terms of Service. Use at your own risk
    and only for educational purposes.
    """
    
    # Step 1: Generate the content with Claude
    response = claude.messages.create(
        model="claude-sonnet-5-5",  # model name as of October 2026; current ones: the What's current page
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""Write an Instagram post on the topic: {topic}
            
            Requirements:
            - 150-220 characters (before hashtags)
            - Emoji for visual separation
            - 12-15 hashtags at the end
            - Tone: friendly, expert, not salesy
            - Language: English"""
        }]
    )
    
    post_text = "".join(b.text for b in response.content if b.type == "text")
    print(f"Generated text:\n{post_text}")
    
    # Step 2: (Optional) Open a browser to publish
    # async with async_playwright() as p:
    #     browser = await p.chromium.launch(headless=False)
    #     page = await browser.new_page()
    #     await page.goto("https://www.instagram.com")
    #     # ... further actions
    
    return post_text

# Run it
asyncio.run(generate_and_post_to_instagram("an overview of the real estate market in Ecuador"))

import requests
import os

POSTIZ_API_KEY = os.environ["POSTIZ_API_KEY"]
POSTIZ_BASE_URL = "https://your-postiz-instance.com/api"

def create_scheduled_post(content: str, media_urls: list, 
                          publish_at: str, platforms: list) -> dict:
    """Create a scheduled post through the Postiz API"""
    
    response = requests.post(
        f"{POSTIZ_BASE_URL}/posts",
        headers={
            "Authorization": f"Bearer {POSTIZ_API_KEY}",
            "Content-Type": "application/json",
        },
        json={
            "content": content,
            "media": media_urls,
            "publishAt": publish_at,
            "platforms": platforms,  # ["instagram", "telegram", "linkedin"]
        }
    )
    
    return response.json()

#!/usr/bin/env python3
# content-pipeline.py: the full content pipeline

import anthropic
import requests
import json
import os
from datetime import datetime, timedelta
from typing import List, Dict

claude = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

def generate_weekly_content_plan(brand_context: str, topics: List[str]) -> List[Dict]:
    """
    Generates a weekly content plan with Claude.
    Returns a list of posts with text, hashtags and publishing time.
    """
    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=3000,
        messages=[{
            "role": "user",
            "content": f"""Create a weekly content plan for an Instagram account.

Brand context: {brand_context}

Topics for this week: {', '.join(topics)}

For each of the 7 days, create:
1. A post (150-200 words + 15 hashtags)
2. A Stories idea (1-2 sentences)
3. The best time to publish (keep in mind the audience is in the US, Eastern Time)

Response format: a JSON list of objects {{
  "day": "Monday",
  "topic": "...",
  "post_text": "...",
  "hashtags": ["...", ...],
  "stories_idea": "...",
  "publish_time": "HH:MM"
}}

Return only valid JSON."""
        }]
    )
    
    # Parse Claude's JSON response
    content_plan = json.loads("".join(b.text for b in response.content if b.type == "text"))
    return content_plan

def schedule_posts_to_postiz(content_plan: List[Dict], 
                              platform: str = "instagram") -> List[Dict]:
    """Send the post plan to Postiz for scheduling"""
    
    POSTIZ_API_KEY = os.environ["POSTIZ_API_KEY"]
    scheduled_posts = []
    
    # Work out dates starting from next Monday
    today = datetime.now()
    days_ahead = 7 - today.weekday()  # until next Monday
    start_date = today + timedelta(days=days_ahead)
    
    for i, post in enumerate(content_plan):
        post_date = start_date + timedelta(days=i)
        post_text = f"{post['post_text']}\n\n{' '.join(post['hashtags'])}"
        
        # Publishing time
        hour, minute = post['publish_time'].split(':')
        publish_at = post_date.replace(
            hour=int(hour), minute=int(minute), second=0
        ).isoformat()
        
        result = requests.post(
            "https://your-postiz.com/api/posts",
            headers={"Authorization": f"Bearer {POSTIZ_API_KEY}"},
            json={
                "content": post_text,
                "publishAt": publish_at,
                "platforms": [platform],
            }
        )
        
        scheduled_posts.append({
            "day": post["day"],
            "status": "scheduled" if result.status_code == 200 else "error",
            "publish_at": publish_at,
        })
        
        print(f"✅ {post['day']}: scheduled for {publish_at}")
    
    return scheduled_posts

def analyze_performance(platform_stats: Dict) -> str:
    """Analyze the stats with Claude and get recommendations"""
    
    response = claude.messages.create(
        model="claude-haiku-4-5",  # Haiku is faster and cheaper for analysis
        max_tokens=800,
        messages=[{
            "role": "user",
            "content": f"""Analyze the Instagram account's stats for last week:

{json.dumps(platform_stats, ensure_ascii=False, indent=2)}

Give me:
1. The top 3 posts (by engagement rate)
2. What worked well (2-3 points)
3. What to improve next week (2-3 points)
4. The recommended publishing time

Keep it short, in bullet points."""
        }]
    )
    
    return "".join(b.text for b in response.content if b.type == "text")

# Example usage
if __name__ == "__main__":
    # Brand context
    BRAND = """
    Acme Realty is a real estate agency in Ecuador.
    We help Americans find a home and invest.
    Tone: a friendly expert, not a salesperson.
    Audience: Americans aged 35-55 considering a move or an investment.
    """
    
    TOPICS = [
        "Life in Cuenca: what surprised us",
        "What it costs to rent an apartment: a price overview",
        "Top 5 neighborhoods for investment",
        "Documents a foreigner needs to buy property",
        "A client's story: they moved here from Chicago",
        "The legal fine print of a deal",
        "Weekly market overview",
    ]
    
    print("🚀 Generating the weekly content plan...")
    plan = generate_weekly_content_plan(BRAND, TOPICS)
    
    print("\n📅 Scheduling posts in Postiz...")
    scheduled = schedule_posts_to_postiz(plan)
    
    print("\n✅ Content plan created!")
    for post in scheduled:
        print(f"  {post['day']}: {post['publish_at']} — {post['status']}")

import asyncio
import os
from telegram import Bot
import anthropic
import schedule
import time

BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
CHANNEL_ID = "@acme_realty"  # or a numeric ID

bot = Bot(token=BOT_TOKEN)
claude = anthropic.Anthropic()

async def post_daily_content():
    """Automatic daily posting to a Telegram channel"""
    
    # Get the day's post
    trending_response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1000,
        messages=[{
            "role": "user",
            "content": """Write a post for a Telegram channel about real estate in Ecuador.
            
            Format:
            - A headline with an emoji
            - 2-3 paragraphs of content
            - A "What you need to know" list (3-5 points)
            - A CTA (call to action)
            - 800 characters max
            
            Topic: a timely tip on buying or renting property in Ecuador"""
        }]
    )
    
    post_text = "".join(b.text for b in trending_response.content if b.type == "text")
    
    await bot.send_message(
        chat_id=CHANNEL_ID,
        text=post_text,
        parse_mode="Markdown"
    )
    
    print(f"✅ Published: {time.strftime('%Y-%m-%d %H:%M')}")

# Publishing schedule
def run_scheduler():
    schedule.every().day.at("09:00").do(
        lambda: asyncio.run(post_daily_content())
    )
    schedule.every().day.at("18:00").do(
        lambda: asyncio.run(post_daily_content())
    )
    
    while True:
        schedule.run_pending()
        time.sleep(60)

if __name__ == "__main__":
    run_scheduler()

def classify_and_respond_to_comment(comment: str, post_context: str) -> dict:
    """
    Classifies a comment and drafts a reply.
    Returns: an auto-reply or a flag for manual handling.
    """
    
    response = claude.messages.create(
        model="claude-haiku-4-5",  # Haiku is fast and cheap for classification
        max_tokens=200,
        messages=[{
            "role": "user",
            "content": f"""A comment on a post about real estate in Ecuador:
"{comment}"

Post context: {post_context}

Tasks:
1. Classify it: positive/neutral/negative/spam/question
2. If question or positive/neutral, give a short reply (100 characters max)
3. If negative or spam, answer "MANUAL_REVIEW"

Format: JSON {{
  "type": "...",
  "auto_reply": "..." or "MANUAL_REVIEW",
  "confidence": 0.0-1.0
}}"""
        }]
    )
    
    return json.loads("".join(b.text for b in response.content if b.type == "text"))

# Process comments
comments = [
    "How much is a 2-bedroom apartment in Cuenca?",
    "Thanks, really helpful!",
    "These people are scammers! Don't trust them!",
    "Buy my services at the link in my bio",
]

for comment in comments:
    result = classify_and_respond_to_comment(
        comment, 
        "A post about housing costs in Cuenca"
    )
    
    if result["auto_reply"] == "MANUAL_REVIEW":
        print(f"⚠️ MANUAL REVIEW: {comment[:50]}...")
    else:
        print(f"✅ Auto-reply: {result['auto_reply']}")