feat: AI Press Secretary - content generation and publishing platform

Features:
- AI-powered post generation from brand profiles
- Image generation with templates
- Content planning
- Telegram publishing
- Social profile analyzer
- Community organization support
- Markdown to HTML conversion for Telegram
- Inline AI chat for post editing

Deploy:
- Add deployment scripts for Linux servers
- Add PowerShell deploy script for Windows
- Add Nginx and SSL setup scripts
- Add backup and restore scripts
This commit is contained in:
apuc committed 2026-08-27 15:24:15 +03:00
1 parent 81625b2e6f
commit c6048aeea0
37 files changed
+7037 -225

No files matched your search

+281 -14
View File
@@ -7,13 +7,32 @@ from config import PIAPI_API_KEY, PIAPI_IMAGE_MODEL, get_proxies
PIAPI_IMAGE_URL = "https://api.piapi.ai/v1/images/generations"
# Доступные размеры (gpt-image-1)
IMAGE_SIZES = {
"square": ("1024x1024", "Квадрат 1024x1024"),
"wide": ("1536x1024", "Горизонтальный 1536x1024"),
"tall": ("1024x1536", "Вертикальный 1024x1536"),
# Размеры по моделям
# gpt-image-1: 1024x1024, 1536x1024, 1024x1536
# dall-e-3: 1024x1024, 1792x1024, 1024x1792
_MODEL_SIZES = {
"gpt-image-1": {
"square": ("1024x1024", "Квадрат 1024×1024"),
"wide": ("1536x1024", "Горизонтальный 1536×1024"),
"tall": ("1024x1536", "Вертикальный 1024×1536"),
},
"dall-e-3": {
"square": ("1024x1024", "Квадрат 1024×1024"),
"wide": ("1792x1024", "Горизонтальный 1792×1024"),
"tall": ("1024x1792", "Вертикальный 1024×1792"),
},
}
# Размеры по умолчанию (для неизвестных моделей — как у gpt-image-1)
DEFAULT_SIZES = {
"square": ("1024x1024", "Квадрат 1024×1024"),
"wide": ("1536x1024", "Горизонтальный 1536×1024"),
"tall": ("1024x1536", "Вертикальный 1024×1536"),
}
# Обратная совместимость: IMAGE_SIZES для шаблонов
IMAGE_SIZES = DEFAULT_SIZES
# Пресеты стилей - дописываются к промпту
STYLE_PRESETS = {
"none": ("Без стиля", ""),
@@ -24,6 +43,132 @@ STYLE_PRESETS = {
"flat": ("Плоская графика", ", flat vector illustration, bold shapes, social media style"),
}
# Шаблоны изображений — готовые пресеты для быстрой генерации
# prompt — для свободной генерации (подставляется в textarea)
# prompt_template — для генерации к посту ({post_text} заменяется на текст поста)
IMAGE_TEMPLATES = [
{
"key": "product",
"name": "Товар",
"icon": "🛍️",
"prompt": "Professional product photo on a clean studio background, soft gradient lighting, high-end commercial photography",
"prompt_template": "Professional product photo on a clean studio background, soft gradient lighting, high-end commercial photography. Product related to: {post_text}",
"style": "photo",
"size": "square",
"hint": "Фото товара на чистом фоне — для карточек и каталогов",
},
{
"key": "lifestyle",
"name": "Лайфстайл",
"icon": "📸",
"prompt": "Natural lifestyle scene, warm tones, candid feel, natural light, everyday setting",
"prompt_template": "Natural lifestyle scene related to: {post_text}. A person in a cozy everyday setting, warm tones, candid feel, natural light",
"style": "photo",
"size": "wide",
"hint": "Живая сцена из жизни — для постов и сторис",
},
{
"key": "food",
"name": "Еда",
"icon": "🍕",
"prompt": "Beautiful food photography, overhead flat lay or 45-degree angle, garnished plate, rustic wooden table, soft natural light, appetizing colors",
"prompt_template": "Beautiful food photography of {post_text}, overhead flat lay or 45-degree angle, garnished plate, rustic wooden table, soft natural light, appetizing colors",
"style": "photo",
"size": "square",
"hint": "Вкусная еда — для кафе, ресторанов, рецептов",
},
{
"key": "business",
"name": "Бизнес",
"icon": "🏢",
"prompt": "Professional business setting, modern office or meeting room, people in smart casual clothing, clean and bright atmosphere",
"prompt_template": "Professional business setting illustrating: {post_text}. Modern office, people in smart casual, clean bright atmosphere",
"style": "photo",
"size": "wide",
"hint": "Деловая атмосфера — для LinkedIn и презентаций",
},
{
"key": "abstract",
"name": "Абстрактный фон",
"icon": "🎨",
"prompt": "Abstract gradient background with flowing organic shapes, vibrant modern color palette, smooth transitions, depth effect",
"prompt_template": "Abstract gradient background with flowing organic shapes and stylized text overlay reading: {post_text}. Vibrant modern color palette, smooth transitions, depth effect",
"style": "none",
"size": "square",
"hint": "Красивый фон — для цитат, анонсов, обложек",
},
{
"key": "portrait",
"name": "Портрет",
"icon": "👤",
"prompt": "Professional portrait photo, soft studio lighting, blurred background, sharp focus on face, natural skin tones, confident expression",
"prompt_template": "Professional portrait photo of a person who {post_text}. Soft studio lighting, blurred background, sharp focus, natural skin tones, confident expression",
"style": "photo",
"size": "square",
"hint": "Портрет или аватар — для профилей и «О нас»",
},
{
"key": "infographic",
"name": "Инфографика",
"icon": "📊",
"prompt": "Clean infographic layout with icons, charts, and data visualization on white background, modern flat design, bold readable typography",
"prompt_template": "Clean infographic design with the following text and data clearly rendered on it: {post_text}. White background, modern flat design, icons and charts, bold readable typography — the text from the post MUST be visible and legible on the image",
"style": "flat",
"size": "tall",
"hint": "Визуализация данных — для статистик и гайдов",
},
{
"key": "banner",
"name": "Рекламный баннер",
"icon": "🎯",
"prompt": "Eye-catching advertising banner with bold typography, vibrant colors, modern promotional design, call to action",
"prompt_template": "Eye-catching advertising banner with the following text prominently displayed: {post_text}. Vibrant colors, modern promotional design, the text must be the main focus of the image, bold readable typography",
"style": "none",
"size": "wide",
"hint": "Баннер для рекламы — с текстом поста",
},
{
"key": "nature",
"name": "Природа",
"icon": "🌄",
"prompt": "Breathtaking nature landscape, golden hour lighting, vivid colors, dramatic sky, professional landscape photography",
"prompt_template": "Breathtaking nature landscape illustrating the theme: {post_text}. Golden hour lighting, vivid colors, dramatic sky, professional landscape photography",
"style": "photo",
"size": "wide",
"hint": "Пейзажи и природа — для мотивации и фона",
},
{
"key": "minimal",
"name": "Минимализм",
"icon": "✨",
"prompt": "Minimalist composition, single focal point, lots of white space, clean geometric shapes, soft muted colors, elegant simplicity",
"prompt_template": "Minimalist design with the following text as the centerpiece: {post_text}. Clean white background, elegant typography, lots of breathing space, soft muted accent color",
"style": "minimal",
"size": "square",
"hint": "Чистый минимализм — для премиальных брендов",
},
{
"key": "story",
"name": "Сторис",
"icon": "📱",
"prompt": "Vertical social media story, gradient background, centered composition, modern aesthetic, bold colors",
"prompt_template": "Vertical social media story with the following text as the main element: {post_text}. Gradient background, centered composition, modern aesthetic, bold colors, the text must be clearly readable",
"style": "none",
"size": "tall",
"hint": "Вертикальный формат — для Instagram/TikTok сторис",
},
{
"key": "before_after",
"name": "До / После",
"icon": "🔄",
"prompt": "Split composition before/after comparison, left side 'before', right side 'after', clean dividing line, matching lighting",
"prompt_template": "Split composition before/after comparison related to: {post_text}. Left side shows 'before' state, right side shows 'after' transformation, clean dividing line, matching lighting",
"style": "photo",
"size": "wide",
"hint": "Сравнение до/после — для кейсов и результатов",
},
]
def detect_ext(data: bytes) -> str:
"""Определяет расширение файла по magic-байтам."""
@@ -38,24 +183,56 @@ def detect_ext(data: bytes) -> str:
return ".png"
class ImageGenerator:
"""Генерация изображений через PiAPI (OpenAI-совместимый /images/generations)."""
def get_sizes_for_model(model: str) -> dict:
"""Возвращает доступные размеры для конкретной модели."""
if not model:
return DEFAULT_SIZES
# Точное совпадение
if model in _MODEL_SIZES:
return _MODEL_SIZES[model]
# Частичное совпадение (напр. openai/dall-e-3)
for key, sizes in _MODEL_SIZES.items():
if key in model:
return sizes
return DEFAULT_SIZES
def __init__(self, model: str | None = None):
self.model = model or PIAPI_IMAGE_MODEL
self.proxies = get_proxies()
class ImageGenerator:
"""Генерация изображений через OpenAI-совместимый /images/generations."""
def __init__(self, model=None, provider_settings=None):
"""provider_settings — опциональный dict:
api_key, base_url, image_model
Если не передан — используются глобальные настройки PiAPI."""
if provider_settings and provider_settings.get("api_key"):
self.api_key = provider_settings["api_key"]
base = provider_settings.get("base_url") or "https://api.piapi.ai/v1"
self.image_url = base.rstrip("/") + "/images/generations"
self.model = model or provider_settings.get("image_model") or PIAPI_IMAGE_MODEL
self.provider_type = provider_settings.get("provider_type", "custom")
# Прокси для всех провайдеров (доступ из России к зарубежным API)
self.proxies = get_proxies()
else:
self.api_key = PIAPI_API_KEY
self.image_url = PIAPI_IMAGE_URL
self.model = model or PIAPI_IMAGE_MODEL
self.provider_type = "piapi"
self.proxies = get_proxies()
def generate(self, prompt: str, size: str = "square") -> bytes:
"""Генерирует картинку и возвращает её байты. Поднимает исключение при ошибке."""
if size not in IMAGE_SIZES:
sizes = get_sizes_for_model(self.model)
if size not in sizes:
size = "square"
size_value = IMAGE_SIZES[size][0]
size_value = sizes[size][0]
# DALL-E 3 больше не поддерживается OpenAI — используем gpt-image-1
# response_format и quality не передаём (gpt-image-1 возвращает b64_json по умолчанию)
response = requests.post(
PIAPI_IMAGE_URL,
self.image_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {PIAPI_API_KEY}",
"Authorization": f"Bearer {self.api_key}",
},
json={
"model": self.model,
@@ -87,3 +264,93 @@ class ImageGenerator:
if style in STYLE_PRESETS:
return prompt.strip() + STYLE_PRESETS[style][1]
return prompt.strip()
def generate_with_reference(self, prompt: str, reference_image_bytes: bytes, size: str = "square", chat_settings=None) -> bytes:
"""Генерирует изображение на основе промпта + исходного изображения.
Пробует: 1) text-провайдер chat/completions, 2) image-провайдер chat/completions, 3) fallback без картинки.
"""
sizes = get_sizes_for_model(self.model)
if size not in sizes:
size = "square"
size_value = sizes[size][0]
# Кодируем исходное изображение в base64
img_b64 = base64.b64encode(reference_image_bytes).decode("utf-8")
mime = "image/png"
if reference_image_bytes[:2] == b"\xff\xd8":
mime = "image/jpeg"
elif reference_image_bytes[:4] == b"RIFF":
mime = "image/webp"
# Список провайдеров для попытки chat/completions с image input
candidates = []
if chat_settings and chat_settings.get("api_key"):
candidates.append((chat_settings["api_key"], chat_settings.get("base_url", "https://api.openai.com/v1"), "text"))
# image-провайдер тоже может уметь chat/completions (PiAPI/OpenAI)
if self.api_key and (not chat_settings or self.api_key != chat_settings.get("api_key")):
candidates.append((self.api_key, self.image_url.rsplit("/images/generations", 1)[0], "image"))
# Fallback: default OpenAI
if not candidates:
candidates.append((self.api_key, "https://api.openai.com/v1", "default"))
last_error = None
for api_key, base_url, label in candidates:
chat_url = base_url.rstrip("/") + "/chat/completions"
print(f"[ImageGenerator] reference image -> {label} provider: {chat_url}")
try:
response = requests.post(
chat_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
},
json={
"model": self.model,
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {"url": f"data:{mime};base64,{img_b64}"}
}
]
}
],
"max_tokens": 4096,
},
proxies=self.proxies,
timeout=240,
)
if response.status_code >= 400:
last_error = f"{response.status_code} {response.reason}"
print(f"[ImageGenerator] {label} провайдер вернул {last_error}, пробуем следующий")
continue
resp_json = response.json()
msg = resp_json.get("choices", [{}])[0].get("message", {})
content = msg.get("content", "")
# Парсим картинку из ответа
if isinstance(content, list):
for part in content:
if isinstance(part, dict) and part.get("type") == "image_url":
url = part.get("image_url", {}).get("url", "")
if url.startswith("data:"):
return base64.b64decode(url.split(",", 1)[1])
if isinstance(content, str) and len(content) > 1000:
try:
return base64.b64decode(content)
except Exception:
pass
print(f"[ImageGenerator] {label} провайдер не вернул картинку")
last_error = "no image in response"
except Exception as e:
last_error = str(e)
print(f"[ImageGenerator] {label} провайдер ошибка: {last_error}")
# Все провайдеры не сработали — fallback без картинки
print(f"[ImageGenerator] ни один провайдер не поддержал image input, генерируем без референса")
return self.generate(prompt, size)
+599 -16
View File
@@ -6,20 +6,45 @@ from config import PIAPI_API_KEY, PIAPI_MODEL, get_proxies
PIAPI_URL = "https://api.piapi.ai/v1/chat/completions"
class ContentGenerator:
def __init__(self):
"""Инициализация генератора контента"""
def __init__(self, provider_settings=None):
"""Инициализация генератора контента.
provider_settings — опциональный dict с ключами:
provider_type, api_key, base_url, text_model
Если не передан — используются глобальные настройки PiAPI из config."""
print("🔄 Инициализация ContentGenerator...")
# Проверяем, есть ли ключ PiAPI
if PIAPI_API_KEY:
if provider_settings and provider_settings.get("api_key"):
# Per-user настройки
self.api_key = provider_settings["api_key"]
self.base_url = provider_settings.get("base_url") or PIAPI_URL.replace("/chat/completions", "")
self.model = provider_settings.get("text_model") or PIAPI_MODEL
self.use_piapi = True
self.chat_url = self.base_url.rstrip("/") + "/chat/completions"
self.provider_type = provider_settings.get("provider_type", "custom")
# Прокси используется для всех провайдеров (если настроен в .env)
# Это нужно для доступа из России к зарубежным API
self.proxies = get_proxies()
print(f"✅ Провайдер: {self.provider_type} | модель: {self.model} | прокси: {'да' if self.proxies else 'нет'}")
elif PIAPI_API_KEY:
# Глобальные настройки из config
self.api_key = PIAPI_API_KEY
self.base_url = PIAPI_URL.replace("/chat/completions", "")
self.model = PIAPI_MODEL
self.use_piapi = True
print("✅ PiAPI подключен!")
self.chat_url = PIAPI_URL
self.provider_type = "piapi"
self.proxies = get_proxies()
print("✅ PiAPI подключен (глобальные настройки)!")
else:
print("⚠️ PiAPI ключ не найден, использую шаблоны")
print("⚠️ API ключ не найден, использую шаблоны")
self.use_piapi = False
self.api_key = None
self.chat_url = PIAPI_URL
self.model = PIAPI_MODEL
self.provider_type = "piapi"
self.proxies = get_proxies()
# Шаблоны для запасного варианта (если PiAPI не работает)
self.templates = {
'instagram': [
@@ -159,10 +184,10 @@ class ContentGenerator:
"""
response = requests.post(
PIAPI_URL,
self.chat_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {PIAPI_API_KEY}"
"Authorization": f"Bearer {self.api_key}"
},
json={
"model": self.model,
@@ -173,7 +198,7 @@ class ContentGenerator:
"temperature": 0.9,
"max_tokens": 500
},
proxies=get_proxies(),
proxies=self.proxies,
timeout=30
)
response.raise_for_status()
@@ -215,10 +240,10 @@ class ContentGenerator:
"""
response = requests.post(
PIAPI_URL,
self.chat_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {PIAPI_API_KEY}"
"Authorization": f"Bearer {self.api_key}"
},
json={
"model": self.model,
@@ -229,7 +254,7 @@ class ContentGenerator:
"temperature": 0.9,
"max_tokens": 500
},
proxies=get_proxies(),
proxies=self.proxies,
timeout=30
)
response.raise_for_status()
@@ -244,6 +269,470 @@ class ContentGenerator:
post = self._generate_with_templates(business_type, business_description, platform)
return f"{post}\n\n💭 Пожелания учтены частично (без ИИ): {wishes}"
def generate_post_from_brand_profile(self, brand_data, platform, style, wishes):
"""Генерирует пост на основе медиа-образа (данных опроса brand_survey).
brand_data — dict с полями опроса."""
print(f"🔄 Генерация поста из медиа-образа для {platform}...")
brand_type = brand_data.get("brand_type", "personal")
is_org = (brand_type == "organization")
is_community = (brand_type == "community")
# Собираем ключевые поля медиа-образа
client_name = brand_data.get("client_name", "")
client_role = brand_data.get("client_role", "")
brand_essence = brand_data.get("brand_essence", "")
positioning_who = brand_data.get("positioning_who", "")
positioning_problem = brand_data.get("positioning_problem", "")
positioning_result = brand_data.get("positioning_result", "")
usp = brand_data.get("usp", "")
tov_description = brand_data.get("tov_description", "")
tov_references = brand_data.get("tov_references", "")
tov_axes = brand_data.get("tov_axes", "")
prompt_context = brand_data.get("prompt_context", "")
prompt_tov = brand_data.get("prompt_tov", "")
prompt_structure = brand_data.get("prompt_structure", "")
prompt_directions = brand_data.get("prompt_directions", "")
prompt_formats = brand_data.get("prompt_formats", "")
prompt_technical = brand_data.get("prompt_technical", "")
# Поля организации
org_name = brand_data.get("org_name", "")
org_industry = brand_data.get("org_industry", "")
org_mission = brand_data.get("org_mission", "")
org_products = brand_data.get("org_products", "")
org_team_size = brand_data.get("org_team_size", "")
positioning_difference = brand_data.get("positioning_difference", "")
community_formats = brand_data.get("community_formats", "")
platform_names = {
'instagram': 'Instagram',
'linkedin': 'LinkedIn',
'telegram': 'Telegram',
'twitter': 'Twitter/X',
'ad': 'рекламный пост'
}
if is_community:
identity_name = org_name
identity_role = org_industry
identity_block = f"""=== ОБЩЕСТВЕННАЯ ОРГАНИЗАЦИЯ ===
Название: {org_name}
Сфера: {org_industry}
Миссия: {org_mission}
Направления деятельности: {org_products}
Участников: {org_team_size}
Формат взаимодействия: {community_formats}
Сущность бренда: {brand_essence}
УТП: {usp}"""
positioning_block = f"""=== ПОЗИЦИОНИРОВАНИЕ ===
Для кого создано: {positioning_who}
Какую проблему решает: {positioning_problem}
Ценность для участника: {positioning_result}
Чем отличаемся от других: {positioning_difference}"""
tone_intro = "Ты — профессиональный копирайтер, создающий контент для общественной организации / бизнес-сообщества."
post_intro = f"Напиши пост для {platform_names.get(platform, 'соцсетей')} от имени организации «{org_name}»."
system_msg = f"Ты профессиональный копирайтер. Пишешь контент для бизнес-сообщества «{org_name}» в соответствии с его миссией, ценностями и Tone of Voice."
elif is_org:
identity_name = org_name
identity_role = org_industry
identity_block = f"""=== ОРГАНИЗАЦИЯ ===
Название: {org_name}
Сфера: {org_industry}
Миссия: {org_mission}
Продукты/услуги: {org_products}
Сотрудников: {org_team_size}
Сущность бренда: {brand_essence}
УТП: {usp}"""
positioning_block = f"""=== ПОЗИЦИОНИРОВАНИЕ ===
Кому помогаем: {positioning_who}
Какую проблему решаем: {positioning_problem}
Какой результат получает клиент: {positioning_result}
Почему выбирают нас: {positioning_difference}"""
tone_intro = "Ты — профессиональный копирайтер, создающий контент для организации."
post_intro = f"Напиши пост для {platform_names.get(platform, 'соцсетей')} от имени организации «{org_name}»."
system_msg = f"Ты профессиональный копирайтер. Пишешь контент для организации «{org_name}» в соответствии с её Tone of Voice и позиционированием."
else:
identity_name = client_name
identity_role = client_role
identity_block = f"""=== ЛИЧНОСТЬ И КОНТЕКСТ ===
Имя: {client_name}
Роль: {client_role}
Сущность бренда: {brand_essence}
УТП: {usp}"""
positioning_block = f"""=== ПОЗИЦИОНИРОВАНИЕ ===
Кому помогаю: {positioning_who}
Какую проблему решаю: {positioning_problem}
Какой результат получает клиент: {positioning_result}"""
tone_intro = "Ты — профессиональный копирайтер-психолог, который глубоко понимает образ и стратегию личного бренда."
post_intro = f"Напиши пост для {platform_names.get(platform, 'соцсетей')} от лица автора медиа-образа."
system_msg = "Ты профессиональный копирайтер-психолог. Пишешь в точном соответствии с образом и Tone of Voice автора."
if self.use_piapi:
try:
prompt = f"""
{tone_intro}
{post_intro}
{identity_block}
{positioning_block}
=== TONE OF VOICE ===
Описание тона: {tov_description}
Референсы: {tov_references}
Оси ToV: {tov_axes}
=== СТРУКТУРА МЫСЛИ ===
{prompt_structure}
=== КОНТЕНТ-НАПРАВЛЕНИЯ ===
{prompt_directions}
=== ФОРМАТЫ И ШАБЛОНЫ ===
{prompt_formats}
=== ТЕХНИЧЕСКИЕ ТРЕБОВАНИЯ ===
{prompt_technical}
=== ПАРАМЕТРЫ ЭТОГО ПОСТА ===
Платформа: {platform_names.get(platform, 'соцсети')}
Стиль: {style}
Пожелания: {wishes}
Напиши только текст поста, без лишних пояснений. Следуй структуре мысли и Tone of Voice.
"""
response = requests.post(
self.chat_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
},
json={
"model": self.model,
"messages": [
{"role": "system", "content": system_msg},
{"role": "user", "content": prompt}
],
"temperature": 0.9,
"max_tokens": 4000
},
proxies=self.proxies,
timeout=90
)
response.raise_for_status()
msg = response.json()["choices"][0]["message"]
post = (msg.get("content") or "").strip()
if post and len(post) > 20:
print("✅ Пост из медиа-образа сгенерирован через API!")
return post
# Если content пустой — возможно reasoning-модель не успела завершить
raise RuntimeError("API вернул пустой ответ (content пуст). Увеличьте max_tokens или попробуйте другую модель.")
except requests.exceptions.HTTPError as e:
status = e.response.status_code if e.response is not None else "?"
body = e.response.text[:200] if e.response is not None else ""
raise RuntimeError(f"API вернул ошибку {status}: {body}")
except requests.exceptions.ConnectionError as e:
raise RuntimeError(f"Не удалось подключиться к API: {e}")
except requests.exceptions.Timeout:
raise RuntimeError("Превышен таймаут запроса к API (30 сек)")
except Exception as e:
raise RuntimeError(f"Ошибка API: {e}")
# Fallback — шаблонный пост с данными медиа-образа
print("🔄 Использую шаблонный пост из медиа-образа...")
return self._generate_template_brand_post(brand_data, platform, wishes)
def modify_post(self, original_post: str, instruction: str) -> str:
"""Модифицирует существующий пост по инструкции пользователя.
original_post — текущий текст поста
instruction — что нужно сделать («сократи вдвое», «сделай официальнее» и т.д.)
Возвращает новый текст поста.
"""
print(f"🔄 Модификация поста: «{instruction[:60]}»...")
prompt = f"""Ты — профессиональный копирайтер. Пользователь просит изменить пост.
=== ТЕКУЩИЙ ПОСТ ===
{original_post}
=== ИНСТРУКЦИЯ ===
{instruction}
=== ЗАДАЧА ===
Перепиши пост, выполнив инструкцию. Сохрани общий смысл и стиль. Верни ТОЛЬКО готовый пост без пояснений и markdown."""
if self.use_piapi:
try:
response = requests.post(
self.chat_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
},
json={
"model": self.model,
"messages": [
{"role": "system", "content": "Ты профессиональный копирайтер. Ты редактируешь посты строго по инструкции."},
{"role": "user", "content": prompt}
],
"temperature": 0.7,
"max_tokens": 4000
},
proxies=self.proxies,
timeout=90
)
response.raise_for_status()
post = response.json()["choices"][0]["message"]["content"].strip()
if post and len(post) > 20:
print("✅ Пост модифицирован!")
return post
except Exception as e:
print(f"⚠️ Ошибка API при модификации: {e}")
# Если ИИ не сработал — возвращаем оригинал с пометкой
return f"{original_post}\n\n💬 [ИИ не смог изменить пост: {instruction}]"
def _generate_template_brand_post(self, brand_data, platform, wishes):
"""Простой пост из медиа-образа без обращения к ИИ (запасной вариант)."""
client_name = brand_data.get("client_name", "Автор")
brand_essence = brand_data.get("brand_essence", "")
usp = brand_data.get("usp", "")
positioning_result = brand_data.get("positioning_result", "")
platform_emoji = {
"instagram": "📸", "linkedin": "💼", "telegram": "📱",
"twitter": "🐦", "ad": "📢"
}
emoji = platform_emoji.get(platform, "📝")
post = (
f"{emoji} {brand_essence}\n\n"
f"Меня зовут {client_name}.\n"
f"{usp}\n\n"
f"Я помогаю получить: {positioning_result}\n\n"
)
if wishes:
post += f"💭 {wishes}\n\n"
post += "Напишите в комментариях, какая у вас сейчас задача 👇\n\n#личныйбренд #бизнес"
return post
def suggest_image_concepts(self, post_text, brand_data=None):
"""Анализирует текст поста и предлагает 4 варианта концепции изображения.
Возвращает список dict: [{title, description, style, size}, ...]"""
print(f"🔄 Анализ поста для подбора концепции изображения ({len(post_text)} символов)...")
if not self.use_piapi:
# Fallback — базовые концепции без ИИ
return [
{"title": "Тематическая фотография", "description": "Профессиональная фотография, иллюстрирующая ключевую идею поста", "style": "photo", "size": "wide"},
{"title": "Минималистичный баннер", "description": "Чистый баннер с текстом поста на минималистичном фоне", "style": "minimal", "size": "wide"},
{"title": "Абстрактная иллюстрация", "description": "Цветная абстрактная иллюстрация в тон поста", "style": "illustration", "size": "square"},
{"title": "Инфографика", "description": "Визуализация данных и ключевых тезисов поста", "style": "flat", "size": "tall"},
]
brand_context = ""
if brand_data:
brand_context = f"""
Контекст бренда:
- Имя: {brand_data.get('client_name', '')}
- Роль: {brand_data.get('client_role', '')}
- Сущность: {brand_data.get('brand_essence', '')}
- УТП: {brand_data.get('usp', '')}
"""
prompt = f"""Ты — арт-директор и визуальный стратег. Проанализируй текст поста и предложи 4 разных варианта визуального оформления.
=== ТЕКСТ ПОСТА ===
{post_text}
{brand_context}
=== ЗАДАЧА ===
Предложи 4 контрастных варианта изображения. Каждый вариант — это отдельная идея с уникальным настроением и стилем.
Для каждого варианта укажи:
- title: краткое название (2-4 слова на русском)
- description: что изображено (1 предложение на русском)
- style: один из: photo, illustration, minimal, 3d, flat
- size: один из: square, wide, tall
Верни ТОЛЬКО JSON-массив без markdown и пояснений:
[{{"title": "...", "description": "...", "style": "...", "size": "..."}}, ...]"""
try:
response = requests.post(
self.chat_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
},
json={
"model": self.model,
"messages": [
{"role": "system", "content": "Ты — арт-директор. Отвечай только валидным JSON без markdown-обёрток."},
{"role": "user", "content": prompt}
],
"temperature": 0.8,
"max_tokens": 1000
},
proxies=self.proxies,
timeout=60
)
response.raise_for_status()
raw = response.json()["choices"][0]["message"]["content"].strip()
# Убираем markdown-обёртку если есть
if raw.startswith("```"):
raw = raw.split("\n", 1)[1]
if raw.endswith("```"):
raw = raw[:-3]
raw = raw.strip()
concepts = json.loads(raw)
if isinstance(concepts, list) and len(concepts) > 0:
print(f"✅ Получено {len(concepts)} концепций от ИИ")
return concepts[:5] # макс 5
except Exception as e:
print(f"⚠️ Ошибка при получении концепций: {e}")
# Fallback
return [
{"title": "Тематическая фотография", "description": "Профессиональная фотография, иллюстрирующая ключевую идею поста", "style": "photo", "size": "wide"},
{"title": "Минималистичный баннер", "description": "Чистый баннер с текстом поста на минималистичном фоне", "style": "minimal", "size": "wide"},
{"title": "Абстрактная иллюстрация", "description": "Цветная абстрактная иллюстрация в тон поста", "style": "illustration", "size": "square"},
{"title": "Инфографика", "description": "Визуализация данных и ключевых тезисов поста", "style": "flat", "size": "tall"},
]
def parse_brand_doc_to_survey(self, doc_text):
"""Разбирает текст .docx документа стратегии личного бренда
и извлекает структурированные поля опроса через PiAPI.
Возвращает dict с полями brand_survey.data или None при ошибке."""
print(f"🔄 Разбор документа медиа-образа ({len(doc_text)} символов)...")
if not self.use_piapi:
print("⚠️ PiAPI недоступен — разбор документа невозможен")
return None
# Ограничиваем длину, чтобы влезть в контекст
text = doc_text[:12000]
prompt = f"""
Ты — аналитик. Извлеки структурированные данные из документа стратегии личного бренда и верни их в виде JSON.
Документ:
---
{text}
---
Извлеки следующие поля (если поле отсутствует в документе — оставь пустую строку ""):
{{
"client_name": "Имя клиента",
"client_role": "Профессия/роль",
"client_location": "Город/регион",
"current_situation": "Описание текущей ситуации",
"goal_audience_number": "число подписчиков",
"goal_audience_platform": "основная платформа",
"goal_audience_description": "описание аудитории",
"goal_expert_field": "область экспертизы",
"goal_expert_speaking": "число выступлений в год",
"goal_commercial_leads": "число заявок в месяц",
"goal_commercial_services": "услуги",
"goal_content_system": "описание контент-системы",
"brand_essence": "сущность бренда одной строкой",
"positioning_who": "кому помогаю",
"positioning_problem": "какую проблему решаю",
"positioning_result": "какой результат получает клиент",
"usp": "УТП",
"ideal_audiences": "идеальные подписчики, по строкам",
"visual_concept": "концепция визуала",
"color_main_name": "название основного цвета",
"color_main_hex": "HEX основного цвета",
"color_accent1_name": "название акцентного цвета 1",
"color_accent1_hex": "HEX акцентного цвета 1",
"color_accent2_name": "название акцентного цвета 2",
"color_accent2_hex": "HEX акцентного цвета 2",
"color_neutral": "нейтральная база",
"typography_headings": "шрифт заголовков",
"typography_body": "шрифт текста",
"typography_accents": "шрифт акцентов",
"tov_description": "описание тона",
"tov_references": "референсы",
"tov_axes": "оси ToV по строкам",
"pillar1_name": "название столпа 1",
"pillar1_goal": "цель столпа 1",
"pillar1_formats": "форматы столпа 1 по строкам",
"pillar1_frequency": "частота столпа 1",
"pillar2_name": "название столпа 2",
"pillar2_goal": "цель столпа 2",
"pillar2_formats": "форматы столпа 2 по строкам",
"pillar2_frequency": "частота столпа 2",
"pillar3_name": "название столпа 3",
"pillar3_goal": "цель столпа 3",
"pillar3_formats": "форматы столпа 3 по строкам",
"pillar3_frequency": "частота столпа 3",
"special_formats": "особые форматы",
"product_lead_magnet": "лид-магнит",
"product_simple": "простая услуга",
"product_converter": "услуга-конвертер",
"product_complex": "комплексные решения",
"product_expert": "экспертные продукты",
"metrics_quantitative": "количественные метрики, формат: Название | Значение | Измерение, по строкам",
"metrics_qualitative": "качественные метрики, тот же формат",
"metrics_commercial": "коммерческие метрики, тот же формат",
"risks": "риски, формат: Риск | Решение, по строкам",
"recommendations": "рекомендации по строкам",
"archetype_key_name": "название ключевого архетипа",
"archetype_key_essence": "суть ключевого архетипа",
"archetype_key_manifestations": "проявления по строкам",
"archetype_supporting": "вспомогательные архетипы, формат: Название | Суть | Проявления, по строкам",
"archetype_synthesis": "синтез архетипов",
"archetype_table": "таблица проявлений, формат: Ситуация | Архетип | Описание, по строкам",
"prompt_context": "контекст и личность для промпта",
"prompt_tov": "Tone of Voice для промпта",
"prompt_structure": "структура мысли",
"prompt_directions": "контент-направления по строкам",
"prompt_formats": "форматы и шаблоны",
"prompt_technical": "технические требования к тексту"
}}
Верни ТОЛЬКО валидный JSON без пояснений и markdown.
"""
try:
response = requests.post(
self.chat_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
},
json={
"model": self.model,
"messages": [
{"role": "system", "content": "Ты отвечаешь только валидным JSON, без markdown и пояснений."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 3000,
"response_format": {"type": "json_object"}
},
proxies=self.proxies,
timeout=60
)
response.raise_for_status()
content = response.json()["choices"][0]["message"]["content"].strip()
parsed = json.loads(content)
print("✅ Документ медиа-образа разобран через PiAPI!")
# Гарантируем наличие channels и funnel
if "channels" not in parsed:
parsed["channels"] = []
if "funnel" not in parsed:
parsed["funnel"] = []
return parsed
except Exception as e:
print(f"⚠️ Ошибка разбора документа: {e}")
return None
def generate_content_plan(self, business_type, business_description, days):
"""Генерирует контент-план на несколько дней вперёд"""
print(f"🔄 Генерация контент-плана на {days} дн...")
@@ -263,10 +752,10 @@ class ContentGenerator:
"""
response = requests.post(
PIAPI_URL,
self.chat_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {PIAPI_API_KEY}"
"Authorization": f"Bearer {self.api_key}"
},
json={
"model": self.model,
@@ -278,7 +767,7 @@ class ContentGenerator:
"max_tokens": 1000,
"response_format": {"type": "json_object"}
},
proxies=get_proxies(),
proxies=self.proxies,
timeout=30
)
response.raise_for_status()
@@ -294,6 +783,100 @@ class ContentGenerator:
print("🔄 Использую шаблонный план...")
return self._generate_template_plan(business_type, days)
def generate_content_plan_from_brand_profile(self, brand_data, days):
"""Генерирует контент-план на основе медиа-образа."""
print(f"🔄 Генерация контент-плана из медиа-образа на {days} дн...")
brand_type = brand_data.get("brand_type", "personal")
is_org = (brand_type == "organization")
is_community = (brand_type == "community")
client_name = brand_data.get("client_name", "")
org_name = brand_data.get("org_name", "")
brand_essence = brand_data.get("brand_essence", "")
positioning_who = brand_data.get("positioning_who", "")
usp = brand_data.get("usp", "")
tov_description = brand_data.get("tov_description", "")
prompt_directions = brand_data.get("prompt_directions", "")
pillar1_formats = brand_data.get("pillar1_formats", "")
pillar2_formats = brand_data.get("pillar2_formats", "")
pillar3_formats = brand_data.get("pillar3_formats", "")
if is_community:
entity_name = org_name
pillar1_label = "О миссии"
pillar2_label = "Мероприятия"
pillar3_label = "Участники"
entity_type = "Бизнес-сообщество / общественная организация"
elif is_org:
entity_name = org_name
pillar1_label = "О продукте"
pillar2_label = "Экспертиза"
pillar3_label = "Вовлечение"
entity_type = "Организация"
else:
entity_name = client_name
pillar1_label = "Кто я"
pillar2_label = "Решаю боль"
pillar3_label = "Втягиваю"
entity_type = "Автор"
if self.use_piapi:
try:
prompt = f"""
Ты — SMM-стратег. Составь контент-план на {days} дней.
{entity_type}: {entity_name}
Сущность бренда: {brand_essence}
УТП: {usp}
Аудитория: {positioning_who}
Tone of Voice: {tov_description}
Контент-направления:
{prompt_directions}
Три столпа контента:
1. «{pillar1_label}» (форматы): {pillar1_formats}
2. «{pillar2_label}» (форматы): {pillar2_formats}
3. «{pillar3_label}» (форматы): {pillar3_formats}
Для каждого дня укажи платформу (instagram, linkedin, telegram, twitter или ad) и короткую идею поста (1 предложение), учитывая три столпа и Tone of Voice.
Верни ТОЛЬКО валидный JSON без пояснений в формате:
{{"plan": [{{"day": 1, "platform": "instagram", "idea": "..."}}, ...]}}
"""
response = requests.post(
self.chat_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
},
json={
"model": self.model,
"messages": [
{"role": "system", "content": "Ты отвечаешь только валидным JSON, без markdown и пояснений."},
{"role": "user", "content": prompt}
],
"temperature": 0.7,
"max_tokens": 1200,
"response_format": {"type": "json_object"}
},
proxies=self.proxies,
timeout=30
)
response.raise_for_status()
content = response.json()["choices"][0]["message"]["content"].strip()
parsed = json.loads(content)
plan = parsed.get("plan")
if plan:
print("✅ Контент-план из медиа-образа сгенерирован через PiAPI!")
return plan
except Exception as e:
print(f"⚠️ Ошибка PiAPI: {e}, использую шаблонный план")
print("🔄 Использую шаблонный план...")
return self._generate_template_plan(brand_essence or client_name or "бизнес", days)
def _generate_template_plan(self, business_type, days):
"""Простой контент-план без обращения к ИИ (запасной вариант)"""
platforms = ['instagram', 'linkedin', 'telegram', 'twitter', 'ad']
+187
View File
@@ -0,0 +1,187 @@
"""
Анализатор профилей соцсетей.
Определяет платформу по URL, извлекает посты, отправляет на AI-анализ
для заполнения полей медиа-образа.
"""
import re
import json
import requests
from config import get_proxies
# === Определение платформы ===
def detect_platform(url: str) -> dict:
"""Определяет платформу по URL.
Возвращает {"platform": str, "channel": str, "fetchable": bool}
"""
url = url.strip()
# Telegram: t.me/channel_name или t.me/s/channel_name
m = re.match(r'https?://t\.me/(?:s/)?([a-zA-Z0-9_]+)', url)
if m:
return {"platform": "telegram", "channel": m.group(1), "fetchable": True}
# VK: vk.com/username or vk.ru/username
m = re.match(r'https?://(?:vk\.com|vk\.ru)/([a-zA-Z0-9_.]+)', url)
if m:
return {"platform": "vk", "channel": m.group(1), "fetchable": False}
# Tenchat: tenchat.ru/username
m = re.match(r'https?://tenchat\.ru/([a-zA-Z0-9_]+)', url)
if m:
return {"platform": "tenchat", "channel": m.group(1), "fetchable": False}
# Instagram
m = re.match(r'https?://(?:www\.)?instagram\.com/([a-zA-Z0-9_.]+)/?', url)
if m:
return {"platform": "instagram", "channel": m.group(1), "fetchable": False}
return {"platform": "unknown", "channel": url, "fetchable": False}
# === Парсинг Telegram ===
def fetch_telegram_posts(channel_name: str, limit: int = 20) -> list:
"""Загружает посты из публичного Telegram-канала через t.me/s/.
Возвращает список {"text": str, "date": str}.
"""
url = f"https://t.me/s/{channel_name}"
proxies = get_proxies()
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
try:
r = requests.get(url, proxies=proxies, timeout=15, headers=headers)
r.raise_for_status()
except Exception as e:
print(f"[SocialAnalyzer] Ошибка загрузки Telegram: {e}")
return []
html = r.text
# Извлекаем тексты постов
texts = re.findall(
r'class="tgme_widget_message_text[^"]*"[^>]*>(.*?)</div>',
html, re.DOTALL
)
# Извлекаем даты
dates = re.findall(
r'class="tgme_widget_message_date"[^>]*>.*?<time[^>]*datetime="([^"]*)"',
html, re.DOTALL
)
posts = []
for i, raw_text in enumerate(texts[:limit]):
clean = re.sub(r'<[^>]+>', ' ', raw_text).strip()
clean = re.sub(r'\s+', ' ', clean)
if len(clean) > 10: # Пропускаем служебные сообщения
date = dates[i] if i < len(dates) else ""
posts.append({"text": clean, "date": date})
print(f"[SocialAnalyzer] Telegram: загружено {len(posts)} постов из @{channel_name}")
return posts
# === AI-анализ постов ===
ANALYSIS_PROMPT_TEMPLATE = """
Ты — эксперт по брендингу и SMM. Проанализируй {count} постов из профиля/канала и извлеки параметры для медиа-образа.
=== ПОСТЫ ДЛЯ АНАЛИЗА ===
{posts_text}
=== ЗАДАЧА ===
Проанализируй стиль, тон, темы, подачу и структуру постов. Верни JSON со следующими полями:
{{
"client_name": "Имя или название канала/профиля (если можно определить)",
"brand_essence": "Сущность бренда одной строкой (что этот человек/бренд транслирует)",
"tov_description": "Описание Tone of Voice (тон общения, манера подачи)",
"tov_axes": "Оси ToV (по строкам, 3-5 шт)",
"tov_references": "На кого/что похож стиль (референсы)",
"prompt_directions": "Контент-направления (по строкам, какие темы затрагиваются)",
"prompt_structure": "Типичная структура постов (как строются тексты)",
"prompt_formats": "Типичные форматы постов (экспертный, личный, вовлекающий и т.д.)",
"current_situation": "Краткое описание: кто этот человек/бренд, чем занимается",
"client_role": "Роль / профессия (если определяется)",
"positioning_who": "Для кого контент (целевая аудитория)",
"positioning_problem": "Какую проблему решает",
"positioning_result": "Какой результат получает аудитория"
}}
Верни ТОЛЬКО валидный JSON без markdown и пояснений.
"""
def analyze_posts_with_ai(posts: list, provider_settings: dict = None) -> dict:
"""Отправляет посты на AI-анализ и возвращает структурированные данные
для заполнения медиа-образа.
"""
if not posts:
return {"error": "Нет постов для анализа"}
# Склеиваем посты
posts_text = ""
for i, p in enumerate(posts[:20], 1):
date_str = f" ({p['date']})" if p.get("date") else ""
posts_text += f"\n--- Пост {i}{date_str} ---\n{p['text']}\n"
prompt = ANALYSIS_PROMPT_TEMPLATE.format(
count=len(posts[:20]),
posts_text=posts_text
)
# Определяем AI-провайдер
if provider_settings and provider_settings.get("api_key"):
api_key = provider_settings["api_key"]
base_url = provider_settings.get("base_url", "https://api.piapi.ai/v1")
model = provider_settings.get("text_model", "gpt-4o")
else:
from config import PIAPI_API_KEY, PIAPI_MODEL
api_key = PIAPI_API_KEY
base_url = "https://api.piapi.ai/v1"
model = PIAPI_MODEL
if not api_key:
return {"error": "AI-провайдер не настроен. Настройте ИИ в разделе «ИИ»."}
chat_url = base_url.rstrip("/") + "/chat/completions"
proxies = get_proxies()
try:
response = requests.post(
chat_url,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}"
},
json={
"model": model,
"messages": [
{"role": "system", "content": "Ты эксперт по брендингу. Отвечаешь только валидным JSON."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 2000,
"response_format": {"type": "json_object"}
},
proxies=proxies,
timeout=90
)
response.raise_for_status()
content = response.json()["choices"][0]["message"]["content"].strip()
# Парсим JSON
result = json.loads(content)
print(f"[SocialAnalyzer] AI-анализ завершён, получено {len(result)} полей")
return result
except json.JSONDecodeError as e:
print(f"[SocialAnalyzer] Ошибка парсинга JSON от AI: {e}")
return {"error": "AI вернул некорректный JSON. Попробуйте ещё раз."}
except Exception as e:
print(f"[SocialAnalyzer] Ошибка AI: {e}")
return {"error": f"Ошибка AI: {e}"}
+186
View File
@@ -0,0 +1,186 @@
"""Публикация постов в Telegram-каналы и чаты через Bot API.
Использует TOKEN из config — тот же токен, что и у бота.
Бот должен быть админом в канале/чате, куда публикуем."""
import re
import requests
from config import TOKEN, get_proxies
TELEGRAM_API = f"https://api.telegram.org/bot{TOKEN}"
def _md_to_html(text: str) -> str:
"""Конвертирует простой Markdown в HTML для Telegram.
Поддерживает: **жирный**, *курсив*, __подчёркнутый~~зачёркнутый~~,
`моноширинный`, ```блок кода```, заголовки ### в <b>,
нумерованные/маркированные списки.
Telegram HTML: <b>, <i>, <u>, <s>, <code>, <pre>, <a>.
"""
# Сначала экранируем HTML-символы в тексте (но не наши теги)
text = text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
# Блоки кода ```...```
text = re.sub(r'```(\w*)\n(.*?)```', r'<pre>\2</pre>', text, flags=re.DOTALL)
text = re.sub(r'```(.*?)```', r'<pre>\1</pre>', text, flags=re.DOTALL)
# Inline code `...`
text = re.sub(r'`([^`]+)`', r'<code>\1</code>', text)
# Заголовки ### **Текст** → <b>Текст</b>
text = re.sub(r'^#{1,6}\s+\*\*(.+?)\*\*\s*$', r'<b>\1</b>', text, flags=re.MULTILINE)
text = re.sub(r'^#{1,6}\s+(.+?)\s*$', r'<b>\1</b>', text, flags=re.MULTILINE)
# Жирный **...**
text = re.sub(r'\*\*(.+?)\*\*', r'<b>\1</b>', text)
# Курсив *...* (но не внутри <b> тегов)
text = re.sub(r'(?<!\*)\*([^*\n]+?)\*(?!\*)', r'<i>\1</i>', text)
# Подчёркнутый __...__
text = re.sub(r'__(.+?)__', r'<u>\1</u>', text)
# Зачёркнутый ~~...~~
text = re.sub(r'~~(.+?)~~', r'<s>\1</s>', text)
return text
def _api_call(method, params):
"""Вызывает Telegram Bot API метод. Возвращает (ok, result_or_error)."""
try:
resp = requests.post(
f"{TELEGRAM_API}/{method}",
data=params,
proxies=get_proxies(),
timeout=30,
)
data = resp.json()
if data.get("ok"):
return True, data.get("result", {})
return False, data.get("description", "Неизвестная ошибка")
except Exception as e:
return False, str(e)
def resolve_chat(username_or_id):
"""Разрешает @username или числовой ID в информацию о чате.
Возвращает dict: {chat_id, title, username, type} или {error}."""
# Нормализуем
identifier = username_or_id.strip()
if identifier.startswith("@"):
identifier = identifier # оставляем как есть для API
elif identifier.lstrip("-").isdigit():
identifier = int(identifier)
ok, result = _api_call("getChat", {"chat_id": identifier})
if not ok:
return {"error": result}
chat_id = result.get("id")
chat_type = result.get("type", "unknown")
title = result.get("title") or result.get("first_name", "") or result.get("username", "")
username = result.get("username", "")
return {
"chat_id": chat_id,
"title": title,
"username": username,
"type": chat_type,
}
def check_bot_is_admin(chat_id):
"""Проверяет, является ли бот админом в чате/канале.
Возвращает (is_admin, error_message)."""
ok, result = _api_call("getChatMember", {
"chat_id": chat_id,
"user_id": _get_bot_id(),
})
if not ok:
return False, f"Не удалось проверить статус: {result}"
status = result.get("status", "")
# administrator — для чатов, owner — для каналов
if status in ("administrator", "creator", "owner"):
return True, None
return False, f"Бот не админ в этом чате (статус: {status}). Добавьте бота в админы."
def _get_bot_id():
"""Возвращает ID бота (кешируется)."""
global _cached_bot_id
if _cached_bot_id is not None:
return _cached_bot_id
ok, result = _api_call("getMe", {})
if ok:
_cached_bot_id = result.get("id")
return _cached_bot_id
return None
_cached_bot_id = None
def publish_post(chat_id, text, image_bytes=None):
"""Публикует пост в канал/чат.
- text: текст поста (Markdown конвертируется в HTML)
- image_bytes: опционально, байты картинки
Возвращает (ok, message_or_message_id)."""
text = _md_to_html(text)
if image_bytes:
ok, result = _api_call_send_photo(chat_id, image_bytes, text)
else:
ok, result = _api_call("sendMessage", {
"chat_id": chat_id,
"text": text,
"parse_mode": "HTML",
})
if ok:
msg_id = result.get("message_id")
return True, f"Опубликовано (message_id={msg_id})"
return False, result
def _api_call_send_photo(chat_id, image_bytes, caption):
"""Отправляет фото с подписью.
Если текст длиннее 1024 символов — фото без подписи + текст отдельным сообщением."""
long_text = len(caption) > 1024
try:
files = {"photo": ("image.png", image_bytes, "image/png")}
data = {
"chat_id": chat_id,
"caption": "" if long_text else caption,
"parse_mode": "HTML" if not long_text else None,
}
# Убираем None из data
data = {k: v for k, v in data.items() if v is not None}
resp = requests.post(
f"{TELEGRAM_API}/sendPhoto",
data=data,
files=files,
proxies=get_proxies(),
timeout=60,
)
data = resp.json()
if not data.get("ok"):
return False, data.get("description", "Ошибка отправки фото")
# Если текст длинный — отправляем его отдельным сообщением
if long_text:
ok2, result2 = _api_call("sendMessage", {
"chat_id": chat_id,
"text": caption,
"parse_mode": "HTML",
})
if not ok2:
# Фото ушло, но текст не отправился — всё равно считаем успехом,
# но возвращаем предупреждение
return True, data.get("result", {})
return True, data.get("result", {})
except Exception as e:
return False, str(e)