Telegram群发Bot数据库报错：UNIQUE约束冲突问题排查

基于aiogram框架开发了带数据库的Telegram私域群发Bot，需实现以下功能：

• 用户申请加入私密频道时自动通过并推送消息；
• 用户发送/start指令时将其ID存入数据库；
• 管理员通过/sendall指令向数据库中所有用户群发消息。

运行Bot时触发错误：sqlite3.IntegrityError: UNIQUE constraint failed: users.user_id，尝试在INSERT语句后添加OR REPLACE但无效，需排查修复。

from aiogram.contrib.fsm_storage.memory import MemoryStorage
import logging
import sqlite3
from aiogram import Bot, Dispatcher,executor,types
from aiogram.types.message import ContentTypes
from aiogram.types.message import ContentType
import markups as nav
import aiogram.utils.markdown as fmt
from db import Database

TOKEN = '--'

logging.basicConfig(level=logging.INFO)

bot = Bot(token=TOKEN, parse_mode=types.ParseMode.HTML)
dp = Dispatcher(bot)
db = Database('database.db')

storage = MemoryStorage()

# 当用户申请加入私密频道时，Bot自动通过并发送消息
@dp.chat_join_request_handler()
async def start1(update: types.ChatJoinRequest):
    await update.approve()
    await bot.send_message(chat_id=update.from_user.id, text=f"{fmt.hide_link('https://huntflow.ru/blog/wp-content/uploads/2017/12/9.png')}"
                           'Это официальный телеграм бот который рекомендует инетесный тебе контент\n'
                             '\n'
                             'Подписывайся на эти каналы👇\n'
                             '\n'
                             '\n'
                             '▲Pussy Tight▲ - https://t.me/+Pdfh8tg4CRNiMmJh\n'
                             '\n'
                             '▲Pussy Tight Hentai▲ - https://t.me/+UHyaBhvEj8Y5ZDQx', reply_markup=nav.MainMenu
                           )

@dp.message_handler(commands=['start'])
async def start(message: types.Message):
    if message.chat.type == 'private':
        if not db.user_exists(message.from_user.id):
            db.add_user(message.from_user.id)
        await bot.send_message(message.from_user.id, text=f"{fmt.hide_link('https://huntflow.ru/blog/wp-content/uploads/2017/12/9.png')}"
                           'Это официальный телеграм бот который рекомендует инетесный тебе контент\n'
                             '\n'
                             'Подписывайся на эти каналы👇\n'
                             '\n'
                             '\n'
                             '▲Pussy Tight▲ - https://t.me/+Pdfh8tg4CRNiMmJh\n'
                             '\n'
                             '▲Pussy Tight Hentai▲ - https://t.me/+UHyaBhvEj8Y5ZDQx', reply_markup=nav.MainMenu
                           )

# 管理员通过该指令向数据库中所有用户群发消息
@dp.message_handler(commands=['sendall'])
async def sendall(message: types.Message):
    if message.chat.type == 'private':
        if message.from_user.id == id_admin:
         text = message.text[9:]
         users = db.get_users()
         for row in users:
             try:
                await bot.send_message(row[0], text)
                if int(row[1]) != 1:
                    db.set_active(row[0], 1)
             except:
                 db.set_active(row[0], 0)
         await bot.send_message(message.from_user.id, 'Успешная рассылка')

if __name__ == '__main__':
    executor.start_polling(dp, skip_updates=True)

用户尝试在INSERT后添加OR REPLACE，但代码执行无效果：

import sqlite3

class Database:
    def __init__(self, db_file):
        self.connection = sqlite3.connect(db_file)
        self.cursor = self.connection.cursor()

    def user_exists(self, user_id):
        with self.connection:
            result = self.cursor.execute("SELECT * FROM 'users' WHERE 'user_id' = ?", (user_id,)).fetchmany(1)
            return bool(len(result))

    def add_user(self, user_id):
        with self.connection:
            return self.cursor.execute("INSERT INTO 'users' ('user_id') VALUES (?)", (user_id,))

    def set_active(self, user_id, active):
        with self.connection:
            return self.cursor.execute("UPDATE 'users' SET 'active' = ? WHERE 'user_id' = ?", (active, user_id,))

    def get_users(self):
        with self.connection:
            return self.cursor.execute("SELECT 'user_id', 'active' FROM 'users'").fetchall()

调整括号后问题仍存在：

File "D:\spambot\venv\db.py", line 15, in add_user
    return self.cursor.execute("INSERT INTO 'users' ('user_id') VALUES (?)", (user_id,))
sqlite3.IntegrityError: UNIQUE constraint failed: users.user_id

核心原因 #

报错是因为重复执行用户插入操作，而user_id作为唯一约束字段不允许重复值。虽然/start指令处理逻辑中做了用户存在性判断，但仍存在两个潜在问题：

1. 频道申请通过的用户未被存入数据库，若这类用户后续触发/start，可能因并发或判断逻辑疏漏导致重复插入；
2. 原add_user方法未处理重复插入的情况，即使添加OR REPLACE也可能因语法或表结构问题无效。

修复步骤 #

1. 修改add_user方法，处理重复插入 #

使用INSERT OR IGNORE语句，当用户已存在时直接忽略插入操作（适合仅需确保用户存在的场景）：

def add_user(self, user_id):
    with self.connection:
        return self.cursor.execute("INSERT OR IGNORE INTO 'users' ('user_id') VALUES (?)", (user_id,))

2. 补充频道申请的用户入库逻辑 #

在chat_join_request_handler中添加用户存在性检查和入库操作，确保通过频道申请的用户也进入群发列表：

@dp.chat_join_request_handler()
async def start1(update: types.ChatJoinRequest):
    await update.approve()
    # 添加用户到数据库
    if not db.user_exists(update.from_user.id):
        db.add_user(update.from_user.id)
    await bot.send_message(chat_id=update.from_user.id, text=f"{fmt.hide_link('https://huntflow.ru/blog/wp-content/uploads/2017/12/9.png')}"
                           'Это официальный телеграм бот который рекомендует инетесный тебе контент\n'
                             '\n'
                             'Подписывайся на эти каналы👇\n'
                             '\n'
                             '\n'
                             '▲Pussy Tight▲ - https://t.me/+Pdfh8tg4CRNiMmJh\n'
                             '\n'
                             '▲Pussy Tight Hentai▲ - https://t.me/+UHyaBhvEj8Y5ZDQx', reply_markup=nav.MainMenu
                           )

3. 优化用户存在性判断逻辑 #

用fetchone()替代fetchmany(1)，简化判断逻辑：

def user_exists(self, user_id):
    with self.connection:
        result = self.cursor.execute("SELECT 1 FROM 'users' WHERE 'user_id' = ?", (user_id,)).fetchone()
        return result is not None

4. 确保数据库表结构正确 #

在Database类的初始化方法中添加建表语句，保证user_id为主键（自带唯一约束），active字段有默认值：

def __init__(self, db_file):
    self.connection = sqlite3.connect(db_file)
    self.cursor = self.connection.cursor()
    # 创建users表（如果不存在）
    self.cursor.execute('''CREATE TABLE IF NOT EXISTS users
                          (user_id INTEGER PRIMARY KEY, active INTEGER DEFAULT 1)''')
    self.connection.commit()

完整修复后的数据库代码 #

import sqlite3

class Database:
    def __init__(self, db_file):
        self.connection = sqlite3.connect(db_file)
        self.cursor = self.connection.cursor()
        # 初始化用户表
        self.cursor.execute('''CREATE TABLE IF NOT EXISTS users
                              (user_id INTEGER PRIMARY KEY, active INTEGER DEFAULT 1)''')
        self.connection.commit()

    def user_exists(self, user_id):
        with self.connection:
            result = self.cursor.execute("SELECT 1 FROM 'users' WHERE 'user_id' = ?", (user_id,)).fetchone()
            return result is not None

    def add_user(self, user_id):
        with self.connection:
            return self.cursor.execute("INSERT OR IGNORE INTO 'users' ('user_id') VALUES (?)", (user_id,))

    def set_active(self, user_id, active):
        with self.connection:
            return self.cursor.execute("UPDATE 'users' SET 'active' = ? WHERE 'user_id' = ?", (active, user_id,))

    def get_users(self):
        with self.connection:
            return self.cursor.execute("SELECT 'user_id', 'active' FROM 'users'").fetchall()

内容的提问来源于stack exchange，提问作者Karaoki