mirror of
https://gitverse.ru/kpa39l/telepost.git
synced 2026-09-29 01:50:08 +00:00
Implement phase 3 and 4: Telegram and Twitter integration
This commit is contained in:
@@ -121,7 +121,7 @@ telepost-clone/
|
||||
|
||||
## Дальнейшие этапы
|
||||
|
||||
2. **Этап 2**: Создание и редактирование постов
|
||||
3. **Этап 3**: Публикация в Telegram
|
||||
4. **Этап 4**: Добавление интеграции с внешней платформой
|
||||
2. **Этап 2**: Создание и редактирование постов [x]
|
||||
3. **Этап 3**: Публикация в Telegram [x]
|
||||
4. **Этап 4**: Добавление интеграции с внешней платформой [x]
|
||||
5. **Этап 5**: Расширенные функции (планирование, история, черновики)
|
||||
|
||||
+2
-2
@@ -83,8 +83,8 @@ telepost-clone/
|
||||
- `media_files`: Загруженные медиафайлы
|
||||
|
||||
## 4. Потоки работы
|
||||
|
||||
### 4.1. Аутентификация
|
||||
|
||||
```
|
||||
1. Пользователь переходит на /login
|
||||
2. Нажимает кнопку "Войти через Telegram"
|
||||
@@ -96,7 +96,7 @@ telepost-clone/
|
||||
```
|
||||
|
||||
### 4.2. Создание и публикация поста
|
||||
```
|
||||
```text
|
||||
1. Пользователь создает новый пост
|
||||
2. Пользователь выбирает каналы для публикации
|
||||
3. Пользователь может назначить время публикации
|
||||
|
||||
@@ -98,4 +98,85 @@ def delete_post(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Post not found"
|
||||
)
|
||||
return {"message": "Post deleted successfully"}
|
||||
return {"message": "Post deleted successfully"}
|
||||
|
||||
|
||||
@router.post("/{post_id}/publish_telegram")
|
||||
def publish_post_to_telegram(
|
||||
post_id: int,
|
||||
chat_id: str,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
"""
|
||||
Опубликовать пост в Telegram
|
||||
"""
|
||||
post_service = PostService(db)
|
||||
|
||||
# Проверяем, что пост принадлежит пользователю
|
||||
post = post_service.get_post(post_id=post_id, owner_id=current_user.id)
|
||||
if not post:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Post not found"
|
||||
)
|
||||
|
||||
try:
|
||||
success = post_service.publish_post_to_telegram(post_id=post_id, chat_id=chat_id)
|
||||
if success:
|
||||
return {"message": "Post published to Telegram successfully", "success": True}
|
||||
else:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail="Failed to publish post to Telegram"
|
||||
)
|
||||
except ValueError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=str(e)
|
||||
)
|
||||
except Exception as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Error publishing post to Telegram: {str(e)}"
|
||||
)
|
||||
|
||||
|
||||
@router.post("/{post_id}/publish_twitter")
|
||||
def publish_post_to_twitter(
|
||||
post_id: int,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
"""
|
||||
Опубликовать пост в Twitter
|
||||
"""
|
||||
post_service = PostService(db)
|
||||
|
||||
# Проверяем, что пост принадлежит пользователю
|
||||
post = post_service.get_post(post_id=post_id, owner_id=current_user.id)
|
||||
if not post:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Post not found"
|
||||
)
|
||||
|
||||
try:
|
||||
success = post_service.publish_post_to_twitter(post_id=post_id)
|
||||
if success:
|
||||
return {"message": "Post published to Twitter successfully", "success": True}
|
||||
else:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail="Failed to publish post to Twitter"
|
||||
)
|
||||
except ValueError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=str(e)
|
||||
)
|
||||
except Exception as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail=f"Error publishing post to Twitter: {str(e)}"
|
||||
)
|
||||
@@ -0,0 +1,4 @@
|
||||
from .telegram_bot import TelegramBot
|
||||
from .twitter_api import TwitterAPI
|
||||
|
||||
__all__ = ["TelegramBot", "TwitterAPI"]
|
||||
@@ -0,0 +1,127 @@
|
||||
import requests
|
||||
from typing import Dict, Optional
|
||||
from fastapi import HTTPException
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TelegramBot:
|
||||
def __init__(self, bot_token: str):
|
||||
self.bot_token = bot_token
|
||||
self.base_url = f"https://api.telegram.org/bot{bot_token}"
|
||||
|
||||
def send_message(self, chat_id: str, text: str, parse_mode: str = "HTML") -> Dict:
|
||||
"""
|
||||
Отправить текстовое сообщение в Telegram
|
||||
|
||||
Args:
|
||||
chat_id: ID чата или канала
|
||||
text: Текст сообщения
|
||||
parse_mode: Режим парсинга ('HTML' или 'MarkdownV2')
|
||||
|
||||
Returns:
|
||||
Ответ от Telegram API
|
||||
"""
|
||||
url = f"{self.base_url}/sendMessage"
|
||||
payload = {
|
||||
"chat_id": chat_id,
|
||||
"text": text,
|
||||
"parse_mode": parse_mode,
|
||||
"disable_web_page_preview": True
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.post(url, json=payload)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
|
||||
if not result.get("ok"):
|
||||
raise HTTPException(status_code=400, detail=f"Telegram API error: {result.get('description')}")
|
||||
|
||||
return result
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.error(f"Error sending message to Telegram: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"Failed to send message to Telegram: {str(e)}")
|
||||
|
||||
def send_photo(self, chat_id: str, photo_url: str, caption: str = "", parse_mode: str = "HTML") -> Dict:
|
||||
"""
|
||||
Отправить фото в Telegram
|
||||
|
||||
Args:
|
||||
chat_id: ID чата или канала
|
||||
photo_url: URL фото
|
||||
caption: Подпись к фото
|
||||
parse_mode: Режим парсинга ('HTML' или 'MarkdownV2')
|
||||
|
||||
Returns:
|
||||
Ответ от Telegram API
|
||||
"""
|
||||
url = f"{self.base_url}/sendPhoto"
|
||||
payload = {
|
||||
"chat_id": chat_id,
|
||||
"photo": photo_url,
|
||||
"caption": caption,
|
||||
"parse_mode": parse_mode
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.post(url, json=payload)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
|
||||
if not result.get("ok"):
|
||||
raise HTTPException(status_code=400, detail=f"Telegram API error: {result.get('description')}")
|
||||
|
||||
return result
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.error(f"Error sending photo to Telegram: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"Failed to send photo to Telegram: {str(e)}")
|
||||
|
||||
def get_me(self) -> Dict:
|
||||
"""
|
||||
Получить информацию о боте
|
||||
|
||||
Returns:
|
||||
Информация о боте от Telegram API
|
||||
"""
|
||||
url = f"{self.base_url}/getMe"
|
||||
|
||||
try:
|
||||
response = requests.get(url)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
|
||||
if not result.get("ok"):
|
||||
raise HTTPException(status_code=400, detail=f"Telegram API error: {result.get('description')}")
|
||||
|
||||
return result
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.error(f"Error getting bot info from Telegram: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"Failed to get bot info from Telegram: {str(e)}")
|
||||
|
||||
def get_chat(self, chat_id: str) -> Dict:
|
||||
"""
|
||||
Получить информацию о чате
|
||||
|
||||
Args:
|
||||
chat_id: ID чата или канала
|
||||
|
||||
Returns:
|
||||
Информация о чате от Telegram API
|
||||
"""
|
||||
url = f"{self.base_url}/getChat"
|
||||
payload = {"chat_id": chat_id}
|
||||
|
||||
try:
|
||||
response = requests.post(url, json=payload)
|
||||
response.raise_for_status()
|
||||
result = response.json()
|
||||
|
||||
if not result.get("ok"):
|
||||
raise HTTPException(status_code=400, detail=f"Telegram API error: {result.get('description')}")
|
||||
|
||||
return result
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.error(f"Error getting chat info from Telegram: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"Failed to get chat info from Telegram: {str(e)}")
|
||||
@@ -0,0 +1,126 @@
|
||||
import tweepy
|
||||
from typing import Dict, Optional
|
||||
from fastapi import HTTPException
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TwitterAPI:
|
||||
def __init__(self, consumer_key: str, consumer_secret: str, access_token: str, access_token_secret: str):
|
||||
"""
|
||||
Инициализация Twitter API клиента
|
||||
|
||||
Args:
|
||||
consumer_key: Ключ приложения Twitter
|
||||
consumer_secret: Секрет приложения Twitter
|
||||
access_token: Токен доступа
|
||||
access_token_secret: Секрет токена доступа
|
||||
"""
|
||||
try:
|
||||
# Создаем объект аутентификации
|
||||
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
|
||||
auth.set_access_token(access_token, access_token_secret)
|
||||
|
||||
# Создаем API клиент
|
||||
self.api = tweepy.API(auth, wait_on_rate_limit=True)
|
||||
|
||||
# Проверяем валидность аутентификации
|
||||
self.api.verify_credentials()
|
||||
except Exception as e:
|
||||
logger.error(f"Error initializing Twitter API: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"Failed to initialize Twitter API: {str(e)}")
|
||||
|
||||
def post_tweet(self, text: str) -> Dict:
|
||||
"""
|
||||
Опубликовать твит
|
||||
|
||||
Args:
|
||||
text: Текст твита (до 280 символов)
|
||||
|
||||
Returns:
|
||||
Информация о созданном твите
|
||||
"""
|
||||
try:
|
||||
# Публикуем твит
|
||||
tweet = self.api.update_status(text)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"tweet_id": tweet.id,
|
||||
"tweet_text": tweet.text,
|
||||
"created_at": tweet.created_at.isoformat() if tweet.created_at else None
|
||||
}
|
||||
except tweepy.TweepyException as e:
|
||||
logger.error(f"Error posting tweet: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"Failed to post tweet: {str(e)}")
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error posting tweet: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"Failed to post tweet: {str(e)}")
|
||||
|
||||
def post_thread(self, tweets: list) -> Dict:
|
||||
"""
|
||||
Опубликовать тред (цепочку твитов)
|
||||
|
||||
Args:
|
||||
tweets: Список текстов твитов в порядке публикации
|
||||
|
||||
Returns:
|
||||
Информация о созданном треде
|
||||
"""
|
||||
try:
|
||||
previous_tweet_id = None
|
||||
tweet_ids = []
|
||||
|
||||
for i, text in enumerate(tweets):
|
||||
if previous_tweet_id:
|
||||
# Отвечаем на предыдущий твит, чтобы создать тред
|
||||
tweet = self.api.update_status(
|
||||
status=text,
|
||||
in_reply_to_status_id=previous_tweet_id,
|
||||
auto_populate_reply_metadata=True
|
||||
)
|
||||
else:
|
||||
# Первый твит в треде
|
||||
tweet = self.api.update_status(text)
|
||||
|
||||
tweet_ids.append(tweet.id)
|
||||
previous_tweet_id = tweet.id
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"tweet_ids": tweet_ids,
|
||||
"thread_length": len(tweets)
|
||||
}
|
||||
except tweepy.TweepyException as e:
|
||||
logger.error(f"Error posting thread: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"Failed to post thread: {str(e)}")
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error posting thread: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"Failed to post thread: {str(e)}")
|
||||
|
||||
def get_user_info(self) -> Dict:
|
||||
"""
|
||||
Получить информацию о текущем пользователе
|
||||
|
||||
Returns:
|
||||
Информация о пользователе
|
||||
"""
|
||||
try:
|
||||
user = self.api.verify_credentials()
|
||||
|
||||
return {
|
||||
"id": user.id,
|
||||
"screen_name": user.screen_name,
|
||||
"name": user.name,
|
||||
"followers_count": user.followers_count,
|
||||
"friends_count": user.friends_count,
|
||||
"statuses_count": user.statuses_count,
|
||||
"description": user.description
|
||||
}
|
||||
except tweepy.TweepyException as e:
|
||||
logger.error(f"Error getting user info: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"Failed to get user info: {str(e)}")
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error getting user info: {str(e)}")
|
||||
raise HTTPException(status_code=500, detail=f"Failed to get user info: {str(e)}")
|
||||
@@ -4,6 +4,9 @@ from sqlalchemy import and_, or_
|
||||
from ..models.post import Post
|
||||
from ..schemas.post import PostCreate, PostUpdate
|
||||
from datetime import datetime
|
||||
from ..integrations.telegram_bot import TelegramBot
|
||||
from ..integrations.twitter_api import TwitterAPI
|
||||
import os
|
||||
|
||||
|
||||
class PostService:
|
||||
@@ -85,4 +88,146 @@ class PostService:
|
||||
"""Получить посты по статусу"""
|
||||
return self.db.query(Post).filter(
|
||||
and_(Post.owner_id == owner_id, Post.status == status)
|
||||
).all()
|
||||
).all()
|
||||
|
||||
def publish_post_to_telegram(self, post_id: int, chat_id: str) -> bool:
|
||||
"""
|
||||
Опубликовать пост в Telegram
|
||||
|
||||
Args:
|
||||
post_id: ID поста в базе данных
|
||||
chat_id: ID чата или канала в Telegram
|
||||
|
||||
Returns:
|
||||
Успешность публикации
|
||||
"""
|
||||
# Получаем пост из базы данных
|
||||
post = self.db.query(Post).filter(Post.id == post_id).first()
|
||||
|
||||
if not post:
|
||||
return False
|
||||
|
||||
# Получаем токен бота из переменных окружения
|
||||
bot_token = os.getenv("TELEGRAM_BOT_TOKEN")
|
||||
if not bot_token:
|
||||
raise ValueError("TELEGRAM_BOT_TOKEN environment variable is not set")
|
||||
|
||||
# Создаем экземпляр TelegramBot
|
||||
bot = TelegramBot(bot_token)
|
||||
|
||||
try:
|
||||
# Формируем сообщение из заголовка и содержимого поста
|
||||
message = f"<b>{post.title}</b>\n\n{post.content}"
|
||||
|
||||
# Отправляем сообщение в Telegram
|
||||
result = bot.send_message(chat_id=chat_id, text=message)
|
||||
|
||||
# Если публикация успешна, обновляем статус поста и время публикации
|
||||
if result.get("ok"):
|
||||
post.status = "published"
|
||||
post.published_at = datetime.utcnow()
|
||||
self.db.commit()
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"Error publishing post to Telegram: {str(e)}")
|
||||
return False
|
||||
|
||||
def publish_scheduled_posts(self, chat_id: str) -> int:
|
||||
"""
|
||||
Опубликовать все запланированные посты, время которых наступило
|
||||
|
||||
Args:
|
||||
chat_id: ID чата или канала в Telegram
|
||||
|
||||
Returns:
|
||||
Количество опубликованных постов
|
||||
"""
|
||||
scheduled_posts = self.get_scheduled_posts()
|
||||
published_count = 0
|
||||
|
||||
for post in scheduled_posts:
|
||||
try:
|
||||
# Пытаемся опубликовать пост в Telegram
|
||||
success = self.publish_post_to_telegram(post.id, chat_id)
|
||||
|
||||
if success:
|
||||
published_count += 1
|
||||
# Обновляем статус поста на published
|
||||
post.status = "published"
|
||||
post.published_at = datetime.utcnow()
|
||||
self.db.commit()
|
||||
else:
|
||||
# Если публикация не удалась, помечаем пост как cancelled
|
||||
post.status = "cancelled"
|
||||
self.db.commit()
|
||||
except Exception as e:
|
||||
print(f"Error publishing scheduled post {post.id}: {str(e)}")
|
||||
# Если возникла ошибка, помечаем пост как cancelled
|
||||
post.status = "cancelled"
|
||||
self.db.commit()
|
||||
|
||||
return published_count
|
||||
|
||||
def publish_post_to_twitter(self, post_id: int) -> bool:
|
||||
"""
|
||||
Опубликовать пост в Twitter
|
||||
|
||||
Args:
|
||||
post_id: ID поста в базе данных
|
||||
|
||||
Returns:
|
||||
Успешность публикации
|
||||
"""
|
||||
# Получаем пост из базы данных
|
||||
post = self.db.query(Post).filter(Post.id == post_id).first()
|
||||
|
||||
if not post:
|
||||
return False
|
||||
|
||||
# Получаем параметры для Twitter API из переменных окружения
|
||||
twitter_consumer_key = os.getenv("TWITTER_CONSUMER_KEY")
|
||||
twitter_consumer_secret = os.getenv("TWITTER_CONSUMER_SECRET")
|
||||
twitter_access_token = os.getenv("TWITTER_ACCESS_TOKEN")
|
||||
twitter_access_token_secret = os.getenv("TWITTER_ACCESS_TOKEN_SECRET")
|
||||
|
||||
if not all([twitter_consumer_key, twitter_consumer_secret, twitter_access_token, twitter_access_token_secret]):
|
||||
raise ValueError("Twitter API credentials are not fully configured in environment variables")
|
||||
|
||||
# Создаем экземпляр TwitterAPI
|
||||
twitter_api = TwitterAPI(
|
||||
consumer_key=twitter_consumer_key,
|
||||
consumer_secret=twitter_consumer_secret,
|
||||
access_token=twitter_access_token,
|
||||
access_token_secret=twitter_access_token_secret
|
||||
)
|
||||
|
||||
try:
|
||||
# Формируем текст твита из заголовка и содержимого поста
|
||||
# Учитываем ограничение на 280 символов
|
||||
combined_text = f"{post.title}\n\n{post.content}"
|
||||
if len(combined_text) > 280:
|
||||
# Обрезаем текст до 280 символов, возможно сокращая содержимое
|
||||
truncated_title = post.title[:int(280*0.3)] # используем 30% на заголовок
|
||||
remaining_chars = 280 - len(truncated_title) - 3 # -3 для "..."
|
||||
|
||||
if len(post.content) > remaining_chars:
|
||||
combined_text = f"{truncated_title}...\n\n{post.content[:remaining_chars-4]}..."
|
||||
else:
|
||||
combined_text = f"{truncated_title}...\n\n{post.content}"
|
||||
|
||||
# Отправляем твит
|
||||
result = twitter_api.post_tweet(combined_text)
|
||||
|
||||
# Если публикация успешна, обновляем статус поста и время публикации
|
||||
if result.get("success"):
|
||||
post.status = "published"
|
||||
post.published_at = datetime.utcnow()
|
||||
self.db.commit()
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"Error publishing post to Twitter: {str(e)}")
|
||||
return False
|
||||
@@ -11,3 +11,5 @@ pydantic==2.5.0
|
||||
pydantic-settings==2.1.0
|
||||
redis==5.0.1
|
||||
cachetools==5.3.2
|
||||
python-telegram-bot==20.7
|
||||
tweepy==4.14.0
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from app.main import app
|
||||
from app.database.database import get_db
|
||||
from app.models.post import Post
|
||||
from app.models.user import User
|
||||
from datetime import datetime
|
||||
|
||||
# Тестовый клиент
|
||||
client = TestClient(app)
|
||||
|
||||
# Настройка тестовой базы данных
|
||||
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
|
||||
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
|
||||
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
|
||||
def override_get_db():
|
||||
try:
|
||||
db = TestingSessionLocal()
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
|
||||
|
||||
def test_create_post():
|
||||
"""Тест создания поста"""
|
||||
user_data = {
|
||||
"username": "testuser",
|
||||
"email": "test@example.com",
|
||||
"full_name": "Test User"
|
||||
}
|
||||
|
||||
# Создаем пользователя
|
||||
response = client.post("/users/", json=user_data)
|
||||
assert response.status_code == 200
|
||||
|
||||
# Создаем пост
|
||||
post_data = {
|
||||
"title": "Test Post",
|
||||
"content": "This is a test post content",
|
||||
"status": "draft"
|
||||
}
|
||||
|
||||
response = client.post("/posts/", json=post_data)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["title"] == "Test Post"
|
||||
assert data["content"] == "This is a test post content"
|
||||
|
||||
return data["id"]
|
||||
|
||||
|
||||
def test_publish_to_telegram():
|
||||
"""Тест публикации в Telegram (без фактической отправки)"""
|
||||
# Создаем пост для тестирования
|
||||
post_id = test_create_post()
|
||||
|
||||
# Тестируем публикацию в Telegram
|
||||
# Так как у нас нет реального токена в тестовой среде,
|
||||
# мы ожидаем ошибку связанную с отсутствием настроек
|
||||
response = client.post(f"/posts/{post_id}/publish_telegram?chat_id=test_chat")
|
||||
# Ожидаем ошибку из-за отсутствия TELEGRAM_BOT_TOKEN
|
||||
assert response.status_code in [422, 500]
|
||||
|
||||
|
||||
def test_publish_to_twitter():
|
||||
"""Тест публикации в Twitter (без фактической отправки)"""
|
||||
# Создаем пост для тестирования
|
||||
post_id = test_create_post()
|
||||
|
||||
# Тестируем публикацию в Twitter
|
||||
# Так как у нас нет реальных ключей в тестовой среде,
|
||||
# мы ожидаем ошибку связанную с отсутствием настроек
|
||||
response = client.post(f"/posts/{post_id}/publish_twitter")
|
||||
# Ожидаем ошибку из-за отсутствия TWITTER_* переменных окружения
|
||||
assert response.status_code in [422, 500]
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main()
|
||||
@@ -11,6 +11,9 @@
|
||||
let status = 'draft';
|
||||
let loading = false;
|
||||
let error = null;
|
||||
let telegramPublishLoading = false;
|
||||
let telegramChatId = '';
|
||||
let twitterPublishLoading = false;
|
||||
|
||||
// Load post data if editing
|
||||
onMount(async () => {
|
||||
@@ -131,6 +134,97 @@
|
||||
e.preventDefault();
|
||||
await savePost();
|
||||
}
|
||||
|
||||
async function publishToTelegram() {
|
||||
if (!postId) {
|
||||
error = 'Post must be saved before publishing to Telegram';
|
||||
return;
|
||||
}
|
||||
|
||||
if (!telegramChatId.trim()) {
|
||||
error = 'Please enter a Telegram Chat ID';
|
||||
return;
|
||||
}
|
||||
|
||||
telegramPublishLoading = true;
|
||||
error = null;
|
||||
|
||||
try {
|
||||
const token = localStorage.getItem('token');
|
||||
if (!token) {
|
||||
throw new Error('No authentication token found');
|
||||
}
|
||||
|
||||
const response = await fetch(`${import.meta.env.VITE_API_URL || '/api/v1'}/posts/${postId}/publish_telegram?chat_id=${encodeURIComponent(telegramChatId)}`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Authorization': `Bearer ${token}`,
|
||||
'Content-Type': 'application/json'
|
||||
}
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
if (response.status === 401) {
|
||||
localStorage.removeItem('token');
|
||||
goto('/login');
|
||||
return;
|
||||
}
|
||||
const errorData = await response.json();
|
||||
throw new Error(errorData.detail || 'Failed to publish post to Telegram');
|
||||
}
|
||||
|
||||
const result = await response.json();
|
||||
alert('Post published to Telegram successfully!');
|
||||
} catch (err) {
|
||||
console.error('Error publishing to Telegram:', err);
|
||||
error = err.message;
|
||||
} finally {
|
||||
telegramPublishLoading = false;
|
||||
}
|
||||
}
|
||||
|
||||
async function publishToTwitter() {
|
||||
if (!postId) {
|
||||
error = 'Post must be saved before publishing to Twitter';
|
||||
return;
|
||||
}
|
||||
|
||||
twitterPublishLoading = true;
|
||||
error = null;
|
||||
|
||||
try {
|
||||
const token = localStorage.getItem('token');
|
||||
if (!token) {
|
||||
throw new Error('No authentication token found');
|
||||
}
|
||||
|
||||
const response = await fetch(`${import.meta.env.VITE_API_URL || '/api/v1'}/posts/${postId}/publish_twitter`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Authorization': `Bearer ${token}`,
|
||||
'Content-Type': 'application/json'
|
||||
}
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
if (response.status === 401) {
|
||||
localStorage.removeItem('token');
|
||||
goto('/login');
|
||||
return;
|
||||
}
|
||||
const errorData = await response.json();
|
||||
throw new Error(errorData.detail || 'Failed to publish post to Twitter');
|
||||
}
|
||||
|
||||
const result = await response.json();
|
||||
alert('Post published to Twitter successfully!');
|
||||
} catch (err) {
|
||||
console.error('Error publishing to Twitter:', err);
|
||||
error = err.message;
|
||||
} finally {
|
||||
twitterPublishLoading = false;
|
||||
}
|
||||
}
|
||||
</script>
|
||||
|
||||
<div class="post-editor">
|
||||
@@ -193,6 +287,32 @@
|
||||
Cancel
|
||||
</button>
|
||||
</div>
|
||||
|
||||
{#if postId}
|
||||
<div class="telegram-publish-section">
|
||||
<h3>Publish to Telegram</h3>
|
||||
<div class="form-group">
|
||||
<label for="telegram-chat-id">Telegram Chat ID:</label>
|
||||
<input
|
||||
id="telegram-chat-id"
|
||||
type="text"
|
||||
bind:value={telegramChatId}
|
||||
placeholder="Enter Telegram Chat ID or @channel_name"
|
||||
disabled={telegramPublishLoading}
|
||||
/>
|
||||
</div>
|
||||
<button type="button" on:click={publishToTelegram} disabled={telegramPublishLoading || loading}>
|
||||
{telegramPublishLoading ? 'Publishing...' : 'Publish to Telegram'}
|
||||
</button>
|
||||
</div>
|
||||
|
||||
<div class="twitter-publish-section">
|
||||
<h3>Publish to Twitter</h3>
|
||||
<button type="button" on:click={publishToTwitter} disabled={twitterPublishLoading || loading}>
|
||||
{twitterPublishLoading ? 'Publishing...' : 'Publish to Twitter'}
|
||||
</button>
|
||||
</div>
|
||||
{/if}
|
||||
</form>
|
||||
</div>
|
||||
|
||||
@@ -269,4 +389,26 @@
|
||||
border-radius: 4px;
|
||||
margin-bottom: 1rem;
|
||||
}
|
||||
|
||||
.telegram-publish-section {
|
||||
margin-top: 2rem;
|
||||
padding-top: 1.5rem;
|
||||
border-top: 1px solid #eee;
|
||||
}
|
||||
|
||||
.telegram-publish-section h3 {
|
||||
margin-bottom: 1rem;
|
||||
color: #333;
|
||||
}
|
||||
|
||||
.twitter-publish-section {
|
||||
margin-top: 2rem;
|
||||
padding-top: 1.5rem;
|
||||
border-top: 1px solid #eee;
|
||||
}
|
||||
|
||||
.twitter-publish-section h3 {
|
||||
margin-bottom: 1rem;
|
||||
color: #333;
|
||||
}
|
||||
</style>
|
||||
Generated
+6
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"name": "telepost",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {}
|
||||
}
|
||||
Reference in New Issue
Block a user