Автоматизация соцсетей — Instagram, Telegram, контент на автопилоте

# Для примеров ниже достаточно 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:
    """
    Публикует пост в Instagram через Meta Graph API.
    image_url должен быть публично доступным URL (не локальный файл!)
    Версию API (v20.0 в примере) замени на актуальную из документации Meta.
    """
    base_url = f"https://graph.facebook.com/v20.0/{INSTAGRAM_BUSINESS_ACCOUNT_ID}"
    
    # Шаг 1: Создаём медиа-контейнер
    container_response = requests.post(
        f"{base_url}/media",
        params={
            "image_url": image_url,  # Публичный URL картинки
            "caption": caption,
            "access_token": ACCESS_TOKEN,
        }
    )
    container_id = container_response.json().get("id")
    
    if not container_id:
        return {"error": container_response.json()}
    
    # Шаг 2: Публикуем контейнер
    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:
    """Получить статистику по посту"""
    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:
    """
    Запланировать пост (схема условная: проверь в документации Meta, поддерживает ли Instagram API
    отложенную публикацию. Надёжный вариант: хранить пост у себя и вызывать публикацию по расписанию).
    publish_time: ISO 8601 формат (например "2026-12-15T10:00:00+0000")
    """
    base_url = f"https://graph.facebook.com/v20.0/{INSTAGRAM_BUSINESS_ACCOUNT_ID}"
    
    # Конвертируем время в 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):
    """
    ВАЖНО: Это демонстрационный пример. Browser automation
    нарушает Terms of Service Instagram. Используй на свой риск
    и только для образовательных целей.
    """
    
    # Шаг 1: Генерируем контент через Claude
    response = claude.messages.create(
        model="claude-sonnet-5-5",  # название модели на октябрь 2026, актуальные: страница «Актуальное сейчас»
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""Напиши Instagram пост на тему: {topic}
            
            Требования:
            - 150-220 символов (до хэштегов)
            - Эмодзи для визуального разделения
            - 12-15 хэштегов в конце
            - Тон: дружелюбный, экспертный, не рекламный
            - Язык: русский"""
        }]
    )
    
    post_text = "".join(b.text for b in response.content if b.type == "text")
    print(f"Сгенерированный текст:\n{post_text}")
    
    # Шаг 2: (Опционально) Открываем браузер для публикации
    # 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")
    #     # ... дальнейшие действия
    
    return post_text

# Запуск
asyncio.run(generate_and_post_to_instagram("обзор рынка недвижимости в Эквадоре"))

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:
    """Создать запланированный пост через 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 — полный контент-конвейер

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]:
    """
    Генерирует план контента на неделю через Claude.
    Возвращает список постов с текстом, хэштегами, временем публикации.
    """
    response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=3000,
        messages=[{
            "role": "user",
            "content": f"""Создай контент-план на неделю для Instagram-аккаунта.

Контекст бренда: {brand_context}

Темы на эту неделю: {', '.join(topics)}

Для каждого из 7 дней создай:
1. Пост (150-200 слов + 15 хэштегов)
2. Stories идею (1-2 предложения)
3. Лучшее время публикации (учти что аудитория в СНГ, GMT+3)

Формат ответа: JSON список объектов {{
  "day": "Понедельник",
  "topic": "...",
  "post_text": "...",
  "hashtags": ["...", ...],
  "stories_idea": "...",
  "publish_time": "HH:MM"
}}

Верни только валидный JSON."""
        }]
    )
    
    # Парсим JSON ответ 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]:
    """Отправляем план постов в Postiz для планирования"""
    
    POSTIZ_API_KEY = os.environ["POSTIZ_API_KEY"]
    scheduled_posts = []
    
    # Определяем даты начиная с понедельника следующей недели
    today = datetime.now()
    days_ahead = 7 - today.weekday()  # до следующего понедельника
    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'])}"
        
        # Время публикации
        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']}: запланировано на {publish_at}")
    
    return scheduled_posts

def analyze_performance(platform_stats: Dict) -> str:
    """Анализируем статистику через Claude и получаем рекомендации"""
    
    response = claude.messages.create(
        model="claude-haiku-4-5",  # Haiku — быстрее и дешевле для анализа
        max_tokens=800,
        messages=[{
            "role": "user",
            "content": f"""Проанализируй статистику Instagram аккаунта за прошлую неделю:

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

Дай:
1. Топ-3 лучших поста (по engagement rate)
2. Что сработало хорошо (2-3 пункта)
3. Что улучшить на следующей неделе (2-3 пункта)
4. Рекомендуемое время публикации

Ответ кратко, в bullet points."""
        }]
    )
    
    return "".join(b.text for b in response.content if b.type == "text")

# Пример использования
if __name__ == "__main__":
    # Контекст бренда
    BRAND = """
    Acme Realty — агентство недвижимости в Эквадоре.
    Помогаем русскоязычным найти жильё и инвестировать.
    Тон: дружелюбный эксперт, не продажник.
    Аудитория: русскоязычные 35-55 лет, рассматривают переезд или инвестиции.
    """
    
    TOPICS = [
        "Жизнь в Куэнке: что нас удивило",
        "Сколько стоит снять квартиру: обзор цен",
        "Топ-5 районов для инвестиций",
        "Документы для покупки недвижимости иностранцем",
        "История клиента: переехали из Москвы",
        "Юридические тонкости сделки",
        "Еженедельный обзор рынка",
    ]
    
    print("🚀 Генерирую контент-план на неделю...")
    plan = generate_weekly_content_plan(BRAND, TOPICS)
    
    print("\n📅 Планирую публикации в Postiz...")
    scheduled = schedule_posts_to_postiz(plan)
    
    print("\n✅ Контент-план создан!")
    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"  # или числовой ID

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

async def post_daily_content():
    """Ежедневная автоматическая публикация в Telegram канал"""
    
    # Получаем тренды дня
    trending_response = claude.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1000,
        messages=[{
            "role": "user",
            "content": """Напиши пост для Telegram канала о недвижимости в Эквадоре.
            
            Формат:
            - Заголовок с эмодзи
            - 2-3 абзаца контента
            - Список "Что важно знать" (3-5 пунктов)
            - CTA (призыв к действию)
            - Максимум 800 символов
            
            Тема: актуальный совет по покупке или аренде недвижимости в Эквадоре"""
        }]
    )
    
    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"✅ Опубликовано: {time.strftime('%Y-%m-%d %H:%M')}")

# Расписание публикаций
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:
    """
    Классифицирует комментарий и готовит ответ.
    Возвращает: автоответ или флаг для ручной обработки.
    """
    
    response = claude.messages.create(
        model="claude-haiku-4-5",  # Haiku — быстро и дёшево для классификации
        max_tokens=200,
        messages=[{
            "role": "user",
            "content": f"""Комментарий к посту об Эквадорской недвижимости:
"{comment}"

Контекст поста: {post_context}

Задачи:
1. Классифицируй: positive/neutral/negative/spam/question
2. Если question или positive/neutral — дай короткий ответ (макс 100 символов)
3. Если negative или spam — ответь "MANUAL_REVIEW"

Формат: JSON {{
  "type": "...",
  "auto_reply": "..." или "MANUAL_REVIEW",
  "confidence": 0.0-1.0
}}"""
        }]
    )
    
    return json.loads("".join(b.text for b in response.content if b.type == "text"))

# Обработка комментариев
comments = [
    "Сколько стоит 2-комнатная квартира в Куэнке?",
    "Спасибо, очень полезно!",
    "Это мошенники! Не верьте!",
    "Купи мои услуги по ссылке в профиле",
]

for comment in comments:
    result = classify_and_respond_to_comment(
        comment, 
        "Пост о стоимости жилья в Куэнке"
    )
    
    if result["auto_reply"] == "MANUAL_REVIEW":
        print(f"⚠️ РУЧНАЯ ОБРАБОТКА: {comment[:50]}...")
    else:
        print(f"✅ Авто-ответ: {result['auto_reply']}")