Automatizar redes sociales: Instagram, Telegram, contenido en piloto automático

# Para los ejemplos de abajo basta con requests
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:
    """
    Publica un post en Instagram con Meta Graph API.
    image_url debe ser una URL pública (¡no un archivo local!)
    Cambia la versión de la API (v20.0 en el ejemplo) por la vigente según la documentación de Meta.
    """
    base_url = f"https://graph.facebook.com/v20.0/{INSTAGRAM_BUSINESS_ACCOUNT_ID}"
    
    # Paso 1: creamos el contenedor de medios
    container_response = requests.post(
        f"{base_url}/media",
        params={
            "image_url": image_url,  # URL pública de la imagen
            "caption": caption,
            "access_token": ACCESS_TOKEN,
        }
    )
    container_id = container_response.json().get("id")
    
    if not container_id:
        return {"error": container_response.json()}
    
    # Paso 2: publicamos el contenedor
    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:
    """Obtener las estadísticas de un 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:
    """
    Programar un post (el esquema es ilustrativo: revisa en la documentación de Meta si la API de Instagram
    admite publicación programada. La opción segura: guardar el post de tu lado y llamar a la publicación según un horario).
    publish_time: formato ISO 8601 (por ejemplo "2026-12-15T10:00:00+0000")
    """
    base_url = f"https://graph.facebook.com/v20.0/{INSTAGRAM_BUSINESS_ACCOUNT_ID}"
    
    # Convertimos la hora 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):
    """
    IMPORTANTE: este es un ejemplo de demostración. La automatización del navegador
    viola los Terms of Service de Instagram. Úsala bajo tu propio riesgo
    y solo con fines educativos.
    """
    
    # Paso 1: generamos el contenido con Claude
    response = claude.messages.create(
        model="claude-sonnet-5-5",  # nombre del modelo a octubre de 2026; los vigentes: página «Lo vigente»
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""Escribe un post de Instagram sobre el tema: {topic}
            
            Requisitos:
            - 150-220 caracteres (sin contar hashtags)
            - Emojis para separar visualmente
            - 12-15 hashtags al final
            - Tono: amable, experto, no publicitario
            - Idioma: español"""
        }]
    )
    
    post_text = "".join(b.text for b in response.content if b.type == "text")
    print(f"Texto generado:\n{post_text}")
    
    # Paso 2: (opcional) abrimos el navegador para publicar
    # 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")
    #     # ... acciones siguientes
    
    return post_text

# Ejecución
asyncio.run(generate_and_post_to_instagram("panorama del mercado inmobiliario en 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:
    """Crear un post programado con la API de Postiz"""
    
    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: línea de producción de contenido completa

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]:
    """
    Genera el plan de contenido de la semana con Claude.
    Devuelve una lista de posts con texto, hashtags y hora de publicación.
    """
    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=3000,
        messages=[{
            "role": "user",
            "content": f"""Crea un plan de contenido semanal para una cuenta de Instagram.

Contexto de la marca: {brand_context}

Temas de esta semana: {', '.join(topics)}

Para cada uno de los 7 días crea:
1. Un post (150-200 palabras + 15 hashtags)
2. Una idea de Stories (1-2 oraciones)
3. La mejor hora para publicar (toma en cuenta que el público está en Latinoamérica, GMT-5)

Formato de respuesta: una lista JSON de objetos {{
  "day": "Lunes",
  "topic": "...",
  "post_text": "...",
  "hashtags": ["...", ...],
  "stories_idea": "...",
  "publish_time": "HH:MM"
}}

Devuelve solo JSON válido."""
        }]
    )
    
    # Leemos la respuesta JSON de Claude
    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]:
    """Mandamos el plan de posts a Postiz para programarlo"""
    
    POSTIZ_API_KEY = os.environ["POSTIZ_API_KEY"]
    scheduled_posts = []
    
    # Calculamos las fechas a partir del lunes de la próxima semana
    today = datetime.now()
    days_ahead = 7 - today.weekday()  # hasta el próximo lunes
    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'])}"
        
        # Hora de publicación
        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']}: programado para {publish_at}")
    
    return scheduled_posts

def analyze_performance(platform_stats: Dict) -> str:
    """Analizamos las estadísticas con Claude y obtenemos recomendaciones"""
    
    response = claude.messages.create(
        model="claude-haiku-4-5",  # Haiku: más rápido y barato para el análisis
        max_tokens=800,
        messages=[{
            "role": "user",
            "content": f"""Analiza las estadísticas de la cuenta de Instagram de la semana pasada:

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

Dame:
1. Los 3 mejores posts (por engagement rate)
2. Qué funcionó bien (2-3 puntos)
3. Qué mejorar la próxima semana (2-3 puntos)
4. La hora de publicación recomendada

Responde breve, en viñetas."""
        }]
    )
    
    return "".join(b.text for b in response.content if b.type == "text")

# Ejemplo de uso
if __name__ == "__main__":
    # Contexto de la marca
    BRAND = """
    Acme Realty es una agencia inmobiliaria en Ecuador.
    Ayudamos a personas de otros países a encontrar vivienda e invertir.
    Tono: experto amable, no vendedor.
    Público: personas de 35-55 años que piensan en mudarse o invertir.
    """
    
    TOPICS = [
        "La vida en Cuenca: lo que nos sorprendió",
        "Cuánto cuesta rentar un departamento: panorama de precios",
        "Las 5 mejores zonas para invertir",
        "Documentos para que un extranjero compre una propiedad",
        "Historia de un cliente: se mudaron desde Bogotá",
        "Detalles legales de la compraventa",
        "Resumen semanal del mercado",
    ]
    
    print("🚀 Generando el plan de contenido de la semana...")
    plan = generate_weekly_content_plan(BRAND, TOPICS)
    
    print("\n📅 Programando las publicaciones en Postiz...")
    scheduled = schedule_posts_to_postiz(plan)
    
    print("\n✅ ¡Plan de contenido creado!")
    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"  # o el ID numérico

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

async def post_daily_content():
    """Publicación automática diaria en un canal de Telegram"""
    
    # Generamos el post del día
    trending_response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1000,
        messages=[{
            "role": "user",
            "content": """Escribe un post para un canal de Telegram sobre bienes raíces en Ecuador.
            
            Formato:
            - Título con emoji
            - 2-3 párrafos de contenido
            - Una lista "Lo que conviene saber" (3-5 puntos)
            - CTA (llamado a la acción)
            - Máximo 800 caracteres
            
            Tema: un consejo actual sobre comprar o rentar una propiedad en 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"✅ Publicado: {time.strftime('%Y-%m-%d %H:%M')}")

# Horario de publicaciones
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:
    """
    Clasifica el comentario y prepara una respuesta.
    Devuelve: una respuesta automática o una marca para revisión manual.
    """
    
    response = claude.messages.create(
        model="claude-haiku-4-5",  # Haiku: rápido y barato para clasificar
        max_tokens=200,
        messages=[{
            "role": "user",
            "content": f"""Comentario en un post sobre bienes raíces en Ecuador:
"{comment}"

Contexto del post: {post_context}

Tareas:
1. Clasifica: positive/neutral/negative/spam/question
2. Si es question o positive/neutral, da una respuesta breve (máx. 100 caracteres)
3. Si es negative o spam, responde "MANUAL_REVIEW"

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

# Procesar comentarios
comments = [
    "¿Cuánto cuesta un departamento de 2 recámaras en Cuenca?",
    "¡Gracias, muy útil!",
    "¡Son estafadores! ¡No les crean!",
    "Compra mis servicios en el enlace de mi perfil",
]

for comment in comments:
    result = classify_and_respond_to_comment(
        comment, 
        "Post sobre el costo de la vivienda en Cuenca"
    )
    
    if result["auto_reply"] == "MANUAL_REVIEW":
        print(f"⚠️ REVISIÓN MANUAL: {comment[:50]}...")
    else:
        print(f"✅ Respuesta automática: {result['auto_reply']}")