он загружает и загружает их с помощью Google Диска, что означает скрытые ссылки в соответствующем типе данных.
он работает, но
загрузка и выгрузка кода логики не работает
вот мой код -
Код: Выделить всё
import os
import logging
import io
import requests
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.http import MediaIoBaseDownload, MediaFileUpload
from telegram import Update, InputFile
from telegram.ext import (
Application,
CommandHandler,
MessageHandler,
filters,
CallbackContext,
ConversationHandler,
)
# Enable logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Bot states
FILE, BATCH_CHOICE, MANUAL_BATCH, NAME, THUMBNAIL = range(5)
# Hardcoded token (Replace with your Telegram bot token)
TOKEN = '6*******************'
# Hardcoded Google Drive API credentials
client_id = '**************************'
client_secret = '***********************'
redirect_uri = '**********'
refresh_token = '***************'
# Google Drive API scopes
SCOPES = ['https://www.googleapis.com/auth/drive']
# Function to initialize Google Drive API
def initialize_drive_api():
creds = Credentials.from_authorized_user_info({
'client_id': client_id,
'client_secret': client_secret,
'refresh_token': refresh_token,
'token_uri': refresh_token # Updated token_uri here
}, SCOPES)
return build('drive', 'v3', credentials=creds)
# Initialize Google Drive API service
drive_service = initialize_drive_api()
# Start command handler
async def start(update: Update, context: CallbackContext) -> int:
await update.message.reply_text('Hello! I\'m Popeye bot. Please send me a .txt file containing Google Drive file IDs to download.')
return FILE
# Handle received file
async def handle_file(update: Update, context: CallbackContext) -> int:
file_id = update.message.document.file_id
file_info = await context.bot.get_file(file_id)
file_url = file_info.file_path
context.user_data['file_url'] = file_url
await update.message.reply_text('File received! Do you want to grab batch from file or add it manually? (type "grab" or "manual")')
return BATCH_CHOICE
# Handle batch choice
async def handle_batch_choice(update: Update, context: CallbackContext) -> int:
choice = update.message.text.lower()
if choice == 'grab':
file_url = context.user_data.get('file_url')
if not file_url:
await update.message.reply_text('No file URL found. Please start over.')
return ConversationHandler.END
# Download file from Telegram
try:
response = requests.get(file_url, stream=True)
response.raise_for_status()
# Save file locally
with open('temp.txt', 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
# Read file IDs from temp.txt
with open('temp.txt', 'r', encoding='utf-8') as f:
lines = [line.strip() for line in f.readlines()]
batch = []
for line in lines:
parts = line.split(' ', 1)
if len(parts) == 2:
title, file_id = parts
batch.append((title, file_id))
# Store batch in user data
context.user_data['batch'] = batch
# Remove local file
os.remove('temp.txt')
await update.message.reply_text('Batch grabbed! Please enter your name:')
return NAME
except Exception as e:
logger.error(f"Error downloading file: {e}")
await update.message.reply_text('Error downloading file. Please try again.')
return ConversationHandler.END
elif choice == 'manual':
await update.message.reply_text('Please enter the batch manually (separated by commas with title and file ID):')
return MANUAL_BATCH
else:
await update.message.reply_text('Invalid input. Please try again.')
return BATCH_CHOICE
# Handle manual batch input
async def handle_manual_batch(update: Update, context: CallbackContext) -> int:
batch_input = update.message.text.strip()
batch = [part.split(' ', 1) for part in batch_input.split(',')]
context.user_data['batch'] = batch
await update.message.reply_text('Batch added! Please enter your name:')
return NAME
# Handle name input
async def handle_name(update: Update, context: CallbackContext) -> int:
name = update.message.text.strip()
context.user_data['name'] = name
await update.message.reply_text('Please send an image as video thumbnail:')
return THUMBNAIL
# Handle thumbnail input
async def handle_thumbnail(update: Update, context: CallbackContext) -> int:
file_id = update.message.photo[-1].file_id
file_info = await context.bot.get_file(file_id)
file_url = file_info.file_path
context.user_data['thumbnail_url'] = file_url
await download_and_upload(update, context)
return ConversationHandler.END
# Download and upload batch
async def download_and_upload(update: Update, context: CallbackContext):
try:
batch = context.user_data.get('batch')
name = context.user_data.get('name')
thumbnail_url = context.user_data.get('thumbnail_url')
chat_id = update.effective_chat.id
if not batch or not name or not thumbnail_url:
await update.message.reply_text('Error: Missing required data. Please start over.')
return
# Download each file from Google Drive and upload to Telegram
for title, drive_file_id in batch:
request = drive_service.files().get_media(fileId=drive_file_id)
fh = io.BytesIO()
downloader = MediaIoBaseDownload(fh, request)
done = False
while done is False:
status, done = downloader.next_chunk()
fh.seek(0)
# Send video to Telegram chat
await context.bot.send_video(chat_id=chat_id, video=fh, caption=f'{title}\nExtracted by -- {name}', thumb=InputFile(thumbnail_url))
await context.bot.send_message(chat_id=chat_id, text=f'Batch processed: {batch[0][0]}')
except Exception as e:
logger.error(f"Error processing batch: {e}")
await update.message.reply_text('Error processing batch. Please try again later.')
finally:
# Clean up user data
context.user_data.clear()
# Main function
def main():
application = Application.builder().token(TOKEN).build()
conv_handler = ConversationHandler(
entry_points=[CommandHandler('Popeye', start)],
states={
FILE: [MessageHandler(filters.Document.MimeType('text/plain'), handle_file)],
BATCH_CHOICE: [MessageHandler(filters.TEXT & ~filters.COMMAND, handle_batch_choice)],
MANUAL_BATCH: [MessageHandler(filters.TEXT & ~filters.COMMAND, handle_manual_batch)],
NAME: [MessageHandler(filters.TEXT & ~filters.COMMAND, handle_name)],
THUMBNAIL: [MessageHandler(filters.PHOTO, handle_thumbnail)]
},
fallbacks=[]
)
application.add_handler(conv_handler)
application.add_error_handler(error_handler)
application.run_polling()
def error_handler(update, context):
logger.warning('Update "%s" caused error "%s"', update, context.error)
# Optionally, you can restart the bot here after logging the error
# Example: main()
if __name__ == '__main__':
main()
Код: Выделить всё
INFO:telegram.ext.Application:Application started
Код: Выделить всё
INFO:httpx:HTTP Request: POST https://api.telegram.org/bot**** "HTTP/1.1 200 OK"
Подробнее здесь: https://stackoverflow.com/questions/786 ... am-drm-bot
