Загрузка...

Script
Please check the code

Thread in Python created by dftyhfturtugyh Jan 13, 2026. 345 views

  1. dftyhfturtugyh
    dftyhfturtugyh Topic starter Jan 13, 2026 И я такой: «Пау-пау-пау», а ты типо: «Пум-пум-пум»
    Python
    import os
    import sqlite3
    import logging
    from datetime import datetime
    from typing import Optional, Tuple, List, Dict, Any

    from telegram import (
    Update, InlineKeyboardButton, InlineKeyboardMarkup, Message
    )
    from telegram.constants import ParseMode, ChatType
    from telegram.ext import (
    Application, CommandHandler, MessageHandler, CallbackQueryHandler,
    ContextTypes, filters
    )

    logging.basicConfig(
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    level=logging.INFO
    )
    log = logging.getLogger("support-bot")

    DB_PATH = os.getenv("DB_PATH", "support.db")
    BOT_TOKEN = os.getenv("BOT_TOKEN", "").strip()

    # ADMIN_IDS="123,456"
    ADMIN_IDS = {int(x.strip()) for x in os.getenv("ADMIN_IDS", "").split(",") if x.strip()}


    def db() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.execute("PRAGMA journal_mode=WAL;")
    return conn


    def now_iso() -> str:
    return datetime.utcnow().isoformat(timespec="seconds")


    def init_db() -> None:
    with db() as conn:
    conn.execute("""
    CREATE TABLE IF NOT EXISTS tickets (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id INTEGER,
    username TEXT,
    user_chat_id INTEGER, -- where to reply (usually user's private chat id)
    source_chat_id INTEGER, -- where ticket was created (group/private)
    created_at TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'open', -- open/closed
    escalated INTEGER NOT NULL DEFAULT 0,
    last_user_text TEXT
    )
    """)
    conn.execute("""
    CREATE TABLE IF NOT EXISTS messages (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    ticket_id INTEGER NOT NULL,
    sender TEXT NOT NULL, -- user/admin/system
    text TEXT,
    tg_message_id INTEGER,
    tg_chat_id INTEGER,
    created_at TEXT NOT NULL,
    FOREIGN KEY(ticket_id) REFERENCES tickets(id)
    )
    """)
    conn.execute("""
    CREATE TABLE IF NOT EXISTS attachments (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    ticket_id INTEGER NOT NULL,
    kind TEXT NOT NULL, -- photo/video/document/voice/audio/sticker/etc
    file_id TEXT NOT NULL,
    file_unique_id TEXT,
    caption TEXT,
    created_at TEXT NOT NULL,
    FOREIGN KEY(ticket_id) REFERENCES tickets(id)
    )
    """)


    def is_admin(user_id: int) -> bool:
    return user_id in ADMIN_IDS


    def get_open_ticket_for_user(user_id: int) -> Optional[int]:
    with db() as conn:
    row = conn.execute(
    "SELECT id FROM tickets WHERE user_id=? AND status='open' ORDER BY id DESC LIMIT 1",
    (user_id,)
    ).fetchone()
    return int(row[0]) if row else None


    def create_ticket(
    user_id: int,
    username: Optional[str],
    user_chat_id: int,
    source_chat_id: int,
    first_text: Optional[str]
    ) -> int:
    with db() as conn:
    cur = conn.execute(
    "INSERT INTO tickets(user_id, username, user_chat_id, source_chat_id, created_at, status, escalated, last_user_text) "
    "VALUES(?,?,?,?,?,'open',0,?)",
    (user_id, username or "", user_chat_id, source_chat_id, now_iso(), first_text or "")
    )
    ticket_id = int(cur.lastrowid)
    return ticket_id


    def add_message(ticket_id: int, sender: str, text: Optional[str], tg_chat_id: Optional[int], tg_message_id: Optional[int]) -> None:
    with db() as conn:
    conn.execute(
    "INSERT INTO messages(ticket_id, sender, text, tg_chat_id, tg_message_id, created_at) VALUES(?,?,?,?,?,?)",
    (ticket_id, sender, text or "", tg_chat_id, tg_message_id, now_iso())
    )
    if sender == "user" and text:
    conn.execute("UPDATE tickets SET last_user_text=? WHERE id=?", (text, ticket_id))


    def add_attachment(
    ticket_id: int,
    kind: str,
    file_id: str,
    file_unique_id: Optional[str],
    caption: Optional[str]
    ) -> None:
    with db() as conn:
    conn.execute(
    "INSERT INTO attachments(ticket_id, kind, file_id, file_unique_id, caption, created_at) VALUES(?,?,?,?,?,?)",
    (ticket_id, kind, file_id, file_unique_id or "", caption or "", now_iso())
    )


    def get_ticket(ticket_id: int) -> Optional[Tuple[int, int, int, str, int]]:
    # returns (id, user_id, user_chat_id, status, escalated)
    with db() as conn:
    row = conn.execute(
    "SELECT id, user_id, user_chat_id, status, escalated FROM tickets WHERE id=?",
    (ticket_id,)
    ).fetchone()
    return row


    def close_ticket(ticket_id: int) -> bool:
    with db() as conn:
    cur = conn.execute("UPDATE tickets SET status='closed' WHERE id=? AND status='open'", (ticket_id,))
    return cur.rowcount > 0


    def set_escalated(ticket_id: int, value: int) -> None:
    with db() as conn:
    conn.execute("UPDATE tickets SET escalated=? WHERE id=?", (value, ticket_id))


    def list_open_tickets(limit: int = 20) -> List[Tuple[int, int, str, int]]:
    # returns (id, user_id, username, escalated)
    with db() as conn:
    rows = conn.execute(
    "SELECT id, user_id, username, escalated FROM tickets WHERE status='open' ORDER BY escalated DESC, id DESC LIMIT ?",
    (limit,)
    ).fetchall()
    return rows


    def ticket_keyboard(ticket_id: int) -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup([
    [
    InlineKeyboardButton(" Закрыть", callback_data=f"close:{ticket_id}"),
    InlineKeyboardButton(" Эскалация", callback_data=f"esc:{ticket_id}")
    ],
    [
    InlineKeyboardButton(" Как прикрепить", callback_data=f"attach:{ticket_id}")
    ]
    ])


    async def notify_admins(context: ContextTypes.DEFAULT_TYPE, text: str, ticket_id: int) -> None:
    if not ADMIN_IDS:
    return
    kb = ticket_keyboard(ticket_id)
    for admin_id in ADMIN_IDS:
    try:
    await context.bot.send_message(
    chat_id=admin_id,
    text=text,
    parse_mode=ParseMode.HTML,
    reply_markup=kb
    )
    except Exception as e:
    log.exception("Failed to notify admin %s: %s", admin_id, e)


    async def forward_to_admins(context: ContextTypes.DEFAULT_TYPE, msg: Message) -> None:
    """Copy any message to all admins (keeps media)."""
    if not ADMIN_IDS:
    return
    for admin_id in ADMIN_IDS:
    try:
    await msg.copy(chat_id=admin_id)
    except Exception as e:
    log.exception("Failed to copy to admin %s: %s", admin_id, e)


    async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text(
    "Привет! Это бот поддержки.\n\n"
    "Напиши вопрос — я создам тикет (или добавлю к текущему) и передам оператору.\n"
    "Можно отправлять фото/видео/файлы/голосовые."
    )


    async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    if is_admin(update.effective_user.id):
    await update.message.reply_text(
    "Админ-команды:\n"
    "/tickets — список открытых\n"
    "/reply <ticket_id> <текст> — ответить\n"
    "/close <ticket_id> — закрыть\n"
    "\n"
    "Также можно нажимать кнопки под уведомлением."
    )
    else:
    await update.message.reply_text(
    "Пользователь:\n"
    "— просто напиши сообщение (можно с файлами)\n"
    "— в группе используй /support, чтобы создать обращение"
    )


    async def cmd_support(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Command for group chats: open a ticket and ask user to continue in DM."""
    if not update.message:
    return

    chat = update.effective_chat
    user = update.effective_user

    if chat.type not in (ChatType.GROUP, ChatType.SUPERGROUP):
    await update.message.reply_text("Эта команда нужна в группе. В личке — просто напиши сообщение.")
    return

    # Create/open ticket but replies go to user's private chat id only if they message bot in DM.
    # Here we create ticket with user_chat_id = user.id (works for DM send; if user never started bot, sending may fail).
    existing = get_open_ticket_for_user(user.id)
    if existing:
    ticket_id = existing
    else:
    ticket_id = create_ticket(
    user_id=user.id,
    username=user.username,
    user_chat_id=user.id,
    source_chat_id=chat.id,
    first_text=f"(Создано из группы {chat.title or ''})"
    )
    add_message(ticket_id, "system", f"Ticket created from group {chat.id}", chat.id, update.message.message_id)

    await update.message.reply_text(
    f" Тикет #{ticket_id} создан.\n"
    "Напиши мне в личку детали (и файлы), чтобы оператор мог ответить тебе напрямую."
    )

    admin_text = (
    f" <b>Тикет #{ticket_id} (из группы)</b>\n"
    f"Пользователь: {user.mention_html()} (id: <code>{user.id}</code>)\n"
    f"Username: @{user.username if user.username else '—'}\n"
    f"Группа: <code>{chat.id}</code>\n"
    f"Сообщение: /support"
    )
    await notify_admins(context, admin_text, ticket_id)


    def extract_attachment_info(msg: Message) -> Optional[Dict[str, Any]]:
    if msg.photo:
    p = msg.photo[-1]
    return {"kind": "photo", "file_id": p.file_id, "file_unique_id": p.file_unique_id, "caption": msg.caption}
    if msg.video:
    v = msg.video
    return {"kind": "video", "file_id": v.file_id, "file_unique_id": v.file_unique_id, "caption": msg.caption}
    if msg.document:
    d = msg.document
    return {"kind": "document", "file_id": d.file_id, "file_unique_id": d.file_unique_id, "caption": msg.caption}
    if msg.voice:
    v = msg.voice
    return {"kind": "voice", "file_id": v.file_id, "file_unique_id": v.file_unique_id, "caption": msg.caption}
    if msg.audio:
    a = msg.audio
    return {"kind": "audio", "file_id": a.file_id, "file_unique_id": a.file_unique_id, "caption": msg.caption}
    if msg.sticker:
    s = msg.sticker
    return {"kind": "sticker", "file_id": s.file_id, "file_unique_id": s.file_unique_id, "caption": None}
    if msg.video_note:
    vn = msg.video_note
    return {"kind": "video_note", "file_id": vn.file_id, "file_unique_id": vn.file_unique_id, "caption": None}
    return None


    async def handle_user_any(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Handle user messages in private chats: text + attachments -> one active ticket."""
    if not update.message:
    return

    chat = update.effective_chat
    user = update.effective_user
    msg = update.message

    # In groups we only react to /support (handled elsewhere).
    if chat.type in (ChatType.GROUP, ChatType.SUPERGROUP):
    return

    ticket_id = get_open_ticket_for_user(user.id)
    if not ticket_id:
    # First contact creates ticket
    ticket_id = create_ticket(
    user_id=user.id,
    username=user.username,
    user_chat_id=chat.id,
    source_chat_id=chat.id,
    first_text=msg.text or msg.caption or ""
    )
    await msg.reply_text(f" Принято. Создан тикет #{ticket_id}. Оператор ответит здесь.")
    admin_text = (
    f" <b>Новый тикет #{ticket_id}</b>\n"
    f"Пользователь: {user.mention_html()} (id: <code>{user.id}</code>)\n"
    f"Username: @{user.username if user.username else '—'}\n\n"
    f"<b>Текст:</b>\n{(msg.text or msg.caption or '—')}"
    )
    await notify_admins(context, admin_text, ticket_id)
    else:
    # Existing ticket
    await msg.reply_text(f" Добавил к тикету #{ticket_id}.")

    # Log message
    add_message(ticket_id, "user", msg.text or msg.caption or "", chat.id, msg.message_id)

    # Handle attachments
    att = extract_attachment_info(msg)
    if att:
    add_attachment(ticket_id, att["kind"], att["file_id"], att.get("file_unique_id"), att.get("caption"))

    # Send/copy user message to admins (keeps media)
    # plus a small header
    header = (
    f" <b>Апдейт тикета #{ticket_id}</b>\n"
    f"От: {user.mention_html()} (id: <code>{user.id}</code>)"
    )
    await notify_admins(context, header, ticket_id)
    await forward_to_admins(context, msg)


    async def cmd_tickets(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    if not update.message:
    return
    if not is_admin(update.effective_user.id):
    await update.message.reply_text(" Команда только для админов.")
    return

    rows = list_open_tickets(limit=30)
    if not rows:
    await update.message.reply_text("Открытых тикетов нет.")
    return

    lines = ["<b>Открытые тикеты:</b>"]
    for tid, uid, uname, esc in rows:
    flag = "" if esc else "•"
    lines.append(f"{flag} <code>{tid}</code> — id <code>{uid}</code> @{uname if uname else '—'}")
    await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.HTML)


    async def cmd_reply(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    if not update.message:
    return
    if not is_admin(update.effective_user.id):
    await update.message.reply_text(" Команда только для админов.")
    return

    if len(context.args) < 2:
    await update.message.reply_text("Формат: /reply <ticket_id> <текст>")
    return

    try:
    ticket_id = int(context.args[0])
    except ValueError:
    await update.message.reply_text("ticket_id должен быть числом.")
    return

    reply_text = " ".join(context.args[1:]).strip()
    t = get_ticket(ticket_id)
    if not t:
    await update.message.reply_text("Тикет не найден.")
    return

    _tid, _uid, user_chat_id, status, _esc = t
    if status != "open":
    await update.message.reply_text("Тикет закрыт.")
    return

    try:
    await context.bot.send_message(
    chat_id=user_chat_id,
    text=f" Ответ по тикету #{ticket_id}:\n{reply_text}"
    )
    add_message(ticket_id, "admin", reply_text, update.effective_chat.id, update.message.message_id)
    await update.message.reply_text(f" Отправлено по тикету #{ticket_id}.", reply_markup=ticket_keyboard(ticket_id))
    except Exception as e:
    log.exception("Failed to reply to user: %s", e)
    await update.message.reply_text("Не удалось отправить пользователю (возможно, он не запускал бота или заблокировал).")


    async def cmd_close(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    if not update.message:
    return
    if not is_admin(update.effective_user.id):
    await update.message.reply_text(" Команда только для админов.")
    return
    if len(context.args) != 1:
    await update.message.reply_text("Формат: /close <ticket_id>")
    return
    try:
    ticket_id = int(context.args[0])
    except ValueError:
    await update.message.reply_text("ticket_id должен быть числом.")
    return

    t = get_ticket(ticket_id)
    if not t:
    await update.message.reply_text("Тикет не найден.")
    return

    if close_ticket(ticket_id):
    await update.message.reply_text(f" Тикет #{ticket_id} закрыт.")
    _tid, _uid, user_chat_id, _status, _esc = t
    try:
    await context.bot.send_message(chat_id=user_chat_id, text=f" Тикет #{ticket_id} закрыт. Если нужно — напиши снова.")
    except Exception:
    pass
    else:
    await update.message.reply_text("Тикет уже закрыт или не найден.")


    async def on_callback(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    q = update.callback_query
    if not q:
    return
    await q.answer()

    user_id = q.from_user.id
    if not is_admin(user_id):
    await q.edit_message_reply_markup(reply_markup=None)
    await q.message.reply_text(" Кнопки доступны только админам.")
    return

    data = q.data or ""
    try:
    action, tid_s = data.split(":", 1)
    ticket_id = int(tid_s)
    except Exception:
    return

    t = get_ticket(ticket_id)
    if not t:
    await q.message.reply_text("Тикет не найден.")
    return

    if action == "close":
    if close_ticket(ticket_id):
    await q.message.reply_text(f" Закрыто: тикет #{ticket_id}")
    _tid, _uid, user_chat_id, _status, _esc = t
    try:
    await context.bot.send_message(chat_id=user_chat_id, text=f" Тикет #{ticket_id} закрыт. Если нужно — напиши снова.")
    except Exception:
    pass
    else:
    await q.message.reply_text("Тикет уже закрыт.")
    return

    if action == "esc":
    # toggle escalate
    _tid, _uid, user_chat_id, status, esc = t
    new_val = 0 if esc else 1
    set_escalated(ticket_id, new_val)
    state = " ЭСКАЛИРОВАН" if new_val else "Эскалация снята"
    await q.message.reply_text(f"{state}: тикет #{ticket_id}")
    return

    if action == "attach":
    await q.message.reply_text(
    " Пользователь может прикрепить файлы просто отправив их в чат с ботом.\n"
    "Поддерживаются: фото/видео/*********/голосовые/аудио/стикеры.\n"
    "Они автоматически попадут в тикет и будут пересланы админам."
    )
    return


    def main() -> None:
    if not BOT_TOKEN:
    raise SystemExit("Укажи BOT_TOKEN в переменной окружения.")
    if not ADMIN_IDS:
    log.warning("ADMIN_IDS не указаны — админы не будут получать тикеты.")

    init_db()

    app = Application.builder().token(BOT_TOKEN).build()

    app.add_handler(CommandHandler("start", cmd_start))
    app.add_handler(CommandHandler("help", cmd_help))

    # Group entry point
    app.add_handler(CommandHandler("support", cmd_support))

    # Admin
    app.add_handler(CommandHandler("tickets", cmd_tickets))
    app.add_handler(CommandHandler("reply", cmd_reply))
    app.add_handler(CommandHandler("close", cmd_close))

    # Buttons
    app.add_handler(CallbackQueryHandler(on_callback))

    # Private messages: accept everything except commands (text + media)
    app.add_handler(MessageHandler(
    (filters.ChatType.PRIVATE) & (~filters.COMMAND) &
    (filters.TEXT | filters.PHOTO | filters.VIDEO | filters.Document.ALL | filters.VOICE |
    filters.AUDIO | filters.Sticker.ALL | filters.VIDEO_NOTE),
    handle_user_any
    ))

    log.info("Support bot started.")
    app.run_polling(allowed_updates=Update.ALL_TYPES)


    if __name__ == "__main__":
    main()
     
  2. Eblocrut
    вроде не новокек а сидит с ии хуйню какую то скинул в 600 строк
     
    1. View previous comments (4)
    2. dftyhfturtugyh Topic starter
      avatarEblocrut, но он когда скинул мне 600 строк я даже не смотрел нихуя, я просто понял что это хуета на 600 строк
    3. dftyhfturtugyh Topic starter
      avatarEblocrut, нужно было мнение человека, который знает больше и может посмотреть со всех сторон.
    4. атака
      avatardftyhfturtugyh, если реально нужен бот, лучше купи. без знаний нельзя ИИ юзать, хуйня получится
  3. Aisan
    Чекнул, красивый
     
  4. Апатия
    [IMG]
    Синхронная либа, дальше второй строчки можно даже не продолжать
     
    1. атака
      avatarАпатия, меня больше смутило то что кто-то пишет ботов на фреймворке telegram. а тут на синхронный sql похуй, ботом не будет пользоваться <5 юзеров
    2. Апатия
      avatarатака, ну вот тем более, я дальше sqlite даже смотреть не стал
  5. argoars
    argoars Jan 13, 2026 5 Jun 24, 2023
  6. noonegrata
    noonegrata Jan 13, 2026 да я бьюти да я в тренде меня волнуют только деньги 36 Feb 14, 2019
    Проблема в том, что примерно 95% задач неэффективно решать через чат джбт. GPT это языковая модель, и применять её стоит исходя из её природы. Сильнее всего она работает с текстом и информацией: может сокращать и расширять, пересказывать, менять структуру и все в этом духе, посыл я думаю вы поняли)
     
    1. атака
      avatarnoonegrata, повседневные задачи она решает на ура, но код за ней нужно дорабатывать. поэтому человек который ничего не понимает, не сможет юзать ИИ для программирования. эта тема тому пример. итог: юзать ИИ для кода--НОРМ
    2. g2void
      avatarатака, Сам чат гпт да, но например codex будет хорош в коде. А вообще как говорят клауду равных пока нету для кода, но гемини тоже хорош
    3. kekch127
      avatarnoonegrata, как по мне ии программисту стоит юзать только для типовых задач либо для структуризации какой-то конкретной задачи. Поступающая новая задача не будет ей решена верно в любом случае, поэтому легче просто запросить у нее план реализации и сделать самому. ИМХО так будет легче в любом случае, если в голове тяжело быстро представить решение проблемы
Loading...