fix(orm): upsert concurrency-safe to avoid UNIQUE constraint on GroupSetting.group_id

.gitignore

@@ -203,3 +203,10 @@ alembic.ini
 
 #文件修改记录
 .history/
+
+# Claude Code 本地配置
+.claude/settings.local.json
+.claude/*.local.*
+.claude/local/
+
+.serena/

CLAUDE.md

@@ -0,0 +1,111 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Project Overview
+
+This is a QQ bot built on the Graia Ariadne framework, named "xiaomai-bot" (小麦机器人). It's a comprehensive chatbot with AI integration, gaming features (especially Battlefield 1), image processing, entertainment functions, and management capabilities.
+
+## Development Commands
+
+### Core Commands
+- **Run the bot**: `uv run main.py`
+- **Install dependencies**: `uv sync` (using uv package manager)
+- **Linting**: `ruff check` (configured in pyproject.toml)
+- **Formatting**: `ruff format` (configured in pyproject.toml)  
+- **Tests**: `pytest` (test files in tests/ directory)
+
+### Convenience Scripts
+- **Windows**: `run.bat` - Quick start script for Windows
+- **Linux**: `run.sh` - Quick start script for Linux
+
+### Version Management
+- **Bump version**: Uses `bump-my-version` tool (configured in pyproject.toml)
+- **Changelog generation**: Uses `git-cliff` (configured in pyproject.toml)
+
+## Architecture Overview
+
+### Core Components
+- **main.py**: Application entry point with message listeners and bot initialization
+- **core/**: Contains the bot's core functionality
+  - **bot.py**: Main bot class (Umaru) with module loading and initialization
+  - **config.py**: Global configuration access interface  
+  - **control.py**: Permission, frequency, and feature control components
+  - **orm/**: Database ORM layer with SQLAlchemy + Alembic migrations
+  - **models/**: Control component models (frequency, response, saya)
+
+### Module System
+The bot uses a plugin-based architecture organized into:
+- **modules/required/**: Essential plugins (auto_upgrade, saya_manager, perm_manager, helper, status, etc.)
+- **modules/self_contained/**: Built-in feature plugins (AI chat, BF1 features, image processing, entertainment)
+- **modules/third_party/**: External plugins
+
+Each module has a `metadata.json` file defining its configuration, permissions, and usage.
+
+### Database Architecture
+- Uses SQLAlchemy ORM with async support (AsyncORM)
+- SQLite database by default (`data.db`)
+- Alembic for database migrations
+- Tables defined in `core/orm/tables.py`
+
+### Key Features
+- **AI Chat**: Multi-provider support (OpenAI, DeepSeek) with plugin system for tools
+- **Battlefield 1 Integration**: Complete server management and player statistics
+- **Image Processing**: Search, generation, meme creation using Playwright
+- **Permission System**: Granular user and group permissions
+- **Multi-Account Support**: Supports multiple bot accounts with response management
+
+## Configuration
+
+### Primary Config
+- **config.yaml**: Main configuration file (copy from config_demo.yaml)
+- Environment variables supported for Docker deployment
+- Critical settings: bot_accounts, mirai_host, verify_key, Master (admin user)
+
+### Prerequisites
+- Python 3.10-3.12
+- Mirai Console with Mirai API HTTP plugin
+- UV package manager (recommended)
+
+## Important Patterns
+
+### Module Development
+- Each module should have proper metadata.json configuration
+- Use the control system for permissions and rate limiting
+- Follow the existing patterns for message handling and command parsing
+
+### Database Operations
+- Use the AsyncORM from `core.orm` for database operations  
+- All database models should inherit from Base in `core.orm`
+- Use Alembic migrations for schema changes
+
+### Error Handling
+- Comprehensive logging with loguru
+- Error logs stored in `log/` directory organized by date
+- Exception catcher module handles unhandled exceptions
+
+## Testing
+
+- Test files located in `tests/` directory
+- Uses pytest with async support
+- Includes specialized tests for components like md2img, Minecraft integration
+
+## Deployment
+
+Supports multiple deployment methods:
+- Direct Python execution
+- Docker containers  
+- Docker Compose
+- All methods support environment variable configuration
+
+## Dependencies
+
+Major dependencies include:
+- graia-ariadne: Core bot framework
+- SQLAlchemy + Alembic: Database ORM and migrations  
+- FastAPI: Web interface
+- Playwright: Browser automation for image generation
+- httpx: HTTP client
+- Various AI providers (openai, etc.)
+
+The project uses UV for dependency management with lock file support.

core/orm/__init__.py

@@ -3,6 +3,7 @@
 from creart import create
 from loguru import logger
 from sqlalchemy import MetaData, inspect, delete, update, select, insert, text, event
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.engine import Engine
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
 from sqlalchemy.orm import declarative_base, sessionmaker
@@ -205,23 +206,43 @@ async def update_batch(self, table, data_list, conditions_list):
         # 执行批量更新操作
         await self.execute_all(update_stmts)
 
+    def _build_update_stmt(self, table, data, condition):
+        """
+        构造 UPDATE 语句的辅助方法
+        :param table: 表
+        :param data: 数据
+        :param condition: 条件
+        :return: 配置好的 UPDATE 语句
+        """
+        stmt = update(table).where(*condition).values(**data)
+        stmt = stmt.execution_options(synchronize_session="fetch")
+        return stmt
+
     async def insert_or_update(self, table, data, condition):
         """
-        如果满足条件则更新，否则插入
+        如果满足条件则更新，否则插入（并发安全）
+        - 先尝试 UPDATE，如果受影响行数为 0，则尝试 INSERT
+        - 若 INSERT 因唯一约束冲突（IntegrityError）失败，则再次执行 UPDATE
         :param table: 表
         :param data: 数据
         :param condition: 条件
         """
-        # 判断是否存在符合条件的数据
-        exist = (await self.execute(select(table).where(*condition))).all()
-        if exist:
-            # 如果存在，则执行更新操作
-            stmt = update(table).where(*condition).values(**data)
-            stmt = stmt.execution_options(synchronize_session="fetch")
-            await self.execute(stmt)
-        else:
-            # 否则执行插入操作
-            await self.execute(insert(table).values(**data))
+        # 优先 UPDATE，避免大多数情况下的唯一约束冲突
+        update_stmt = self._build_update_stmt(table, data, condition)
+        result = await self.execute(update_stmt)
+
+        # 如果 UPDATE 影响了行，则成功返回
+        if getattr(result, "rowcount", 0) > 0:
+            return result
+
+        # 受影响行数为 0，说明不存在，尝试 INSERT
+        try:
+            return await self.execute(insert(table).values(**data))
+        except IntegrityError:
+            # 竞争条件：在我们尝试 INSERT 前，其他协程/进程已插入相同唯一键
+            # 回退到 UPDATE，确保数据按预期更新
+            retry_update_stmt = self._build_update_stmt(table, data, condition)
+            return await self.execute(retry_update_stmt)
 
     async def insert_or_update_batch(self, table, data_list, conditions_list):
         """
@@ -230,28 +251,35 @@ async def insert_or_update_batch(self, table, data_list, conditions_list):
         :param data_list: 数据列表，每个元素是一个dict，表示一条记录的数据
         :param conditions_list: 条件列表，每个元素是一个tuple或list，表示该记录的条件，与data_list中的元素一一对应
         """
-        stmts = []
+        # 为了避免批量操作中的约束冲突，我们逐个处理每条记录
         for data, condition in zip(data_list, conditions_list):
-            exist = (await self.execute(select(table).where(*condition))).all()
-            if exist:
-                # 如果存在符合条件的数据，则更新
-                stmts.append(update(table).where(*condition).values(**data))
-            else:
-                # 否则插入
-                stmts.append(insert(table).values(**data))
-        # 执行批量插入或更新操作
-        await self.execute_all(stmts)
+            try:
+                await self.insert_or_update(table, data, condition)
+            except Exception as e:
+                logger.error(f"Failed to insert_or_update in batch operation: {e}")
+                # 继续处理其他记录，不中断整个批量操作
+                continue
 
     async def insert_or_ignore(self, table, data, condition):
         """
-        不满足条件则插入,否则跳过
+        不满足条件则插入,否则跳过（并发安全）
+        直接尝试插入，若 UNIQUE 约束冲突则忽略
         :param table: 表
         :param data: 数据
-        :param condition: 条件
+        :param condition: 条件（用于日志记录，实际插入依赖唯一约束）
         """
-        if not (await self.execute(select(table).where(*condition))).all():
+        try:
+            # 直接尝试插入，让数据库的唯一约束来保证不重复
             return await self.execute(insert(table).values(**data))
-        return None
+        except IntegrityError as e:
+            # 如果是 UNIQUE 约束冲突，说明记录已存在，直接忽略
+            if "UNIQUE constraint failed" in str(e):
+                logger.debug(f"Record already exists, ignoring insert: {e}")
+                return None
+            else:
+                # 其他完整性错误直接抛出
+                logger.error(f"IntegrityError in insert_or_ignore: {e}")
+                raise
 
     async def select(self, el, condition=None):
         """