164 lines
4.9 KiB
Python
164 lines
4.9 KiB
Python
"""
|
||
CLI для управления ботом и БД
|
||
Позволяет выполнять операции без запуска самого бота
|
||
"""
|
||
|
||
import asyncio
|
||
import click
|
||
from app.database import AsyncSessionLocal, init_db
|
||
from app.database.repository import (
|
||
GroupRepository, MessageRepository, MessageGroupRepository
|
||
)
|
||
|
||
|
||
@click.group()
|
||
def cli():
|
||
"""TG Autoposter - CLI для управления ботом"""
|
||
pass
|
||
|
||
|
||
@cli.group()
|
||
def message():
|
||
"""Команды для управления сообщениями"""
|
||
pass
|
||
|
||
|
||
@message.command()
|
||
@click.option('--title', prompt='Название сообщения', help='Короткое описание')
|
||
@click.option('--text', prompt='Текст сообщения', help='Текст для отправки')
|
||
@click.option('--parse-mode', default='HTML', help='Режим парсинга (HTML/Markdown)')
|
||
def create(title, text, parse_mode):
|
||
"""Создать новое сообщение"""
|
||
async def do_create():
|
||
await init_db()
|
||
async with AsyncSessionLocal() as session:
|
||
repo = MessageRepository(session)
|
||
msg = await repo.add_message(text, title, parse_mode)
|
||
click.echo(f"✅ Сообщение создано (ID: {msg.id})")
|
||
|
||
asyncio.run(do_create())
|
||
|
||
|
||
@message.command()
|
||
def list():
|
||
"""Список всех сообщений"""
|
||
async def do_list():
|
||
await init_db()
|
||
async with AsyncSessionLocal() as session:
|
||
repo = MessageRepository(session)
|
||
messages = await repo.get_all_messages()
|
||
|
||
if not messages:
|
||
click.echo("Сообщений не найдено")
|
||
return
|
||
|
||
click.echo(f"\nВсего сообщений: {len(messages)}\n")
|
||
for msg in messages:
|
||
status = "✅" if msg.is_active else "❌"
|
||
click.echo(f"{status} ID: {msg.id}")
|
||
click.echo(f" Название: {msg.title}")
|
||
click.echo(f" Текст: {msg.text[:50]}...")
|
||
click.echo()
|
||
|
||
asyncio.run(do_list())
|
||
|
||
|
||
@message.command()
|
||
@click.option('--id', type=int, prompt='ID сообщения', help='ID для удаления')
|
||
def delete(id):
|
||
"""Удалить сообщение"""
|
||
async def do_delete():
|
||
await init_db()
|
||
async with AsyncSessionLocal() as session:
|
||
repo = MessageRepository(session)
|
||
msg = await repo.get_message(id)
|
||
|
||
if not msg:
|
||
click.echo(f"❌ Сообщение {id} не найдено")
|
||
return
|
||
|
||
await repo.delete_message(id)
|
||
click.echo(f"✅ Сообщение {id} удалено")
|
||
|
||
asyncio.run(do_delete())
|
||
|
||
|
||
@cli.group()
|
||
def group():
|
||
"""Команды для управления группами"""
|
||
pass
|
||
|
||
|
||
@group.command()
|
||
def list():
|
||
"""Список всех групп"""
|
||
async def do_list():
|
||
await init_db()
|
||
async with AsyncSessionLocal() as session:
|
||
repo = GroupRepository(session)
|
||
groups = await repo.get_all_active_groups()
|
||
|
||
if not groups:
|
||
click.echo("Групп не найдено")
|
||
return
|
||
|
||
click.echo(f"\nВсего групп: {len(groups)}\n")
|
||
for g in groups:
|
||
click.echo(f"✅ ID: {g.id}")
|
||
click.echo(f" Название: {g.title}")
|
||
click.echo(f" Chat ID: {g.chat_id}")
|
||
click.echo(f" Slow Mode: {g.slow_mode_delay}s")
|
||
click.echo()
|
||
|
||
asyncio.run(do_list())
|
||
|
||
|
||
@cli.group()
|
||
def db():
|
||
"""Команды для управления БД"""
|
||
pass
|
||
|
||
|
||
@db.command()
|
||
def init():
|
||
"""Инициализировать БД"""
|
||
async def do_init():
|
||
click.echo("Инициализация БД...")
|
||
await init_db()
|
||
click.echo("✅ БД инициализирована")
|
||
|
||
asyncio.run(do_init())
|
||
|
||
|
||
@db.command()
|
||
def reset():
|
||
"""Сбросить БД (удалить все данные)"""
|
||
if not click.confirm("⚠️ Вы уверены? Все данные будут удалены"):
|
||
click.echo("Отменено")
|
||
return
|
||
|
||
async def do_reset():
|
||
from app.database import engine
|
||
from app.models import Base
|
||
|
||
click.echo("Удаление всех таблиц...")
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.drop_all)
|
||
|
||
click.echo("Создание таблиц...")
|
||
await init_db()
|
||
click.echo("✅ БД сброшена")
|
||
|
||
asyncio.run(do_reset())
|
||
|
||
|
||
@cli.command()
|
||
def run():
|
||
"""Запустить бота"""
|
||
from app import main
|
||
asyncio.run(main())
|
||
|
||
|
||
if __name__ == '__main__':
|
||
cli()
|