feat：部分处理notify，自动同步数据库null约束变更

notify存储至message库

scripts/import_openie.py

@@ -15,7 +15,6 @@ from src.chat.knowledge.kg_manager import KGManager
 from src.common.logger import get_logger
 from src.chat.knowledge.utils.hash import get_sha256
 from src.manager.local_store_manager import local_storage
-from dotenv import load_dotenv
 
 
 # 添加项目根目录到 sys.path

src/chat/message_receive/bot.py

@@ -146,7 +146,10 @@ class ChatBot:
 
     async def hanle_notice_message(self, message: MessageRecv):
         if message.message_info.message_id == "notice":
-            logger.info("收到notice消息，暂时不支持处理")
+            message.is_notify = True
+            logger.info("notice消息")
+            print(message)
+
             return True
 
     async def do_s4u(self, message_data: Dict[str, Any]):
@@ -207,7 +210,8 @@ class ChatBot:
             message = MessageRecv(message_data)
 
             if await self.hanle_notice_message(message):
-                return
+                # return
+                pass
 
             group_info = message.message_info.group_info
             user_info = message.message_info.user_info

src/chat/message_receive/chat_stream.py

@@ -217,7 +217,8 @@ class ChatManager:
                 # 更新用户信息和群组信息
                 stream.update_active_time()
                 stream = copy.deepcopy(stream)  # 返回副本以避免外部修改影响缓存
-                stream.user_info = user_info
+                if user_info.platform and user_info.user_id:
+                    stream.user_info = user_info
                 if group_info:
                     stream.group_info = group_info
                 from .message import MessageRecv  # 延迟导入，避免循环引用

src/chat/message_receive/message.py

@@ -109,6 +109,7 @@ class MessageRecv(Message):
         self.has_picid = False
         self.is_voice = False
         self.is_mentioned = None
+        self.is_notify = False
 
         self.is_command = False
 

src/chat/message_receive/storage.py

@@ -43,6 +43,7 @@ class MessageStorage:
                 priority_info = {}
                 is_emoji = False
                 is_picid = False
+                is_notify = False
                 is_command = False
             else:
                 filtered_display_message = ""
@@ -53,6 +54,7 @@ class MessageStorage:
                 priority_info = message.priority_info
                 is_emoji = message.is_emoji
                 is_picid = message.is_picid
+                is_notify = message.is_notify
                 is_command = message.is_command
 
             chat_info_dict = chat_stream.to_dict()
@@ -98,6 +100,7 @@ class MessageStorage:
                 priority_info=priority_info,
                 is_emoji=is_emoji,
                 is_picid=is_picid,
+                is_notify=is_notify,
                 is_command=is_command,
             )
         except Exception:

src/common/database/database_model.py

@@ -146,9 +146,9 @@ class Messages(BaseModel):
     chat_info_last_active_time = DoubleField()
 
     # 从顶层 user_info 扁平化而来的字段 (消息发送者信息)
-    user_platform = TextField()
-    user_id = TextField()
-    user_nickname = TextField()
+    user_platform = TextField(null=True)
+    user_id = TextField(null=True)
+    user_nickname = TextField(null=True)
     user_cardname = TextField(null=True)
 
     processed_plain_text = TextField(null=True)  # 处理后的纯文本消息
@@ -162,6 +162,7 @@ class Messages(BaseModel):
     is_emoji = BooleanField(default=False)
     is_picid = BooleanField(default=False)
     is_command = BooleanField(default=False)
+    is_notify = BooleanField(default=False)
 
     class Meta:
         # database = db # 继承自 BaseModel
@@ -252,7 +253,7 @@ class PersonInfo(BaseModel):
     name_reason = TextField(null=True)  # 名称设定的原因
     platform = TextField()  # 平台
     user_id = TextField(index=True)  # 用户ID
-    nickname = TextField()  # 用户昵称
+    nickname = TextField(null=True)  # 用户昵称
     impression = TextField(null=True)  # 个人印象
     short_impression = TextField(null=True)  # 个人印象的简短描述
     points = TextField(null=True)  # 个人印象的点
@@ -378,10 +379,14 @@ def create_tables():
         )
 
 
-def initialize_database():
+def initialize_database(sync_constraints=False):
     """
     检查所有定义的表是否存在，如果不存在则创建它们。
     检查所有表的所有字段是否存在，如果缺失则自动添加。
+    
+    Args:
+        sync_constraints (bool): 是否同步字段约束。默认为 False。
+                               如果为 True，会检查并修复字段的 NULL 约束不一致问题。
     """
 
     models = [
@@ -462,6 +467,13 @@ def initialize_database():
                         logger.info(f"字段 '{field_name}' 删除成功")
                     except Exception as e:
                         logger.error(f"删除字段 '{field_name}' 失败: {e}")
+                        
+        # 如果启用了约束同步，执行约束检查和修复
+        if sync_constraints:
+            logger.debug("开始同步数据库字段约束...")
+            sync_field_constraints()
+            logger.debug("数据库字段约束同步完成")
+            
     except Exception as e:
         logger.exception(f"检查表或字段是否存在时出错: {e}")
         # 如果检查失败（例如数据库不可用），则退出
@@ -470,5 +482,259 @@ def initialize_database():
     logger.info("数据库初始化完成")
 
 
+def sync_field_constraints():
+    """
+    同步数据库字段约束，确保现有数据库字段的 NULL 约束与模型定义一致。
+    如果发现不一致，会自动修复字段约束。
+    """
+    
+    models = [
+        ChatStreams,
+        LLMUsage,
+        Emoji,
+        Messages,
+        Images,
+        ImageDescriptions,
+        OnlineTime,
+        PersonInfo,
+        Expression,
+        Memory,
+        ThinkingLog,
+        GraphNodes,
+        GraphEdges,
+        ActionRecords,
+    ]
+
+    try:
+        with db:
+            for model in models:
+                table_name = model._meta.table_name
+                if not db.table_exists(model):
+                    logger.warning(f"表 '{table_name}' 不存在，跳过约束检查")
+                    continue
+
+                logger.debug(f"检查表 '{table_name}' 的字段约束...")
+                
+                # 获取当前表结构信息
+                cursor = db.execute_sql(f"PRAGMA table_info('{table_name}')")
+                current_schema = {row[1]: {'type': row[2], 'notnull': bool(row[3]), 'default': row[4]} 
+                                for row in cursor.fetchall()}
+                
+                # 检查每个模型字段的约束
+                constraints_to_fix = []
+                for field_name, field_obj in model._meta.fields.items():
+                    if field_name not in current_schema:
+                        continue  # 字段不存在，跳过
+                    
+                    current_notnull = current_schema[field_name]['notnull']
+                    model_allows_null = field_obj.null
+                    
+                    # 如果模型允许 null 但数据库字段不允许 null，需要修复
+                    if model_allows_null and current_notnull:
+                        constraints_to_fix.append({
+                            'field_name': field_name,
+                            'field_obj': field_obj,
+                            'action': 'allow_null',
+                            'current_constraint': 'NOT NULL',
+                            'target_constraint': 'NULL'
+                        })
+                        logger.warning(f"字段 '{field_name}' 约束不一致: 模型允许NULL，但数据库为NOT NULL")
+                    
+                    # 如果模型不允许 null 但数据库字段允许 null，也需要修复（但要小心）
+                    elif not model_allows_null and not current_notnull:
+                        constraints_to_fix.append({
+                            'field_name': field_name,
+                            'field_obj': field_obj,
+                            'action': 'disallow_null',
+                            'current_constraint': 'NULL',
+                            'target_constraint': 'NOT NULL'
+                        })
+                        logger.warning(f"字段 '{field_name}' 约束不一致: 模型不允许NULL，但数据库允许NULL")
+                
+                # 修复约束不一致的字段
+                if constraints_to_fix:
+                    logger.info(f"表 '{table_name}' 需要修复 {len(constraints_to_fix)} 个字段约束")
+                    _fix_table_constraints(table_name, model, constraints_to_fix)
+                else:
+                    logger.debug(f"表 '{table_name}' 的字段约束已同步")
+                    
+    except Exception as e:
+        logger.exception(f"同步字段约束时出错: {e}")
+
+
+def _fix_table_constraints(table_name, model, constraints_to_fix):
+    """
+    修复表的字段约束。
+    对于 SQLite，由于不支持直接修改列约束，需要重建表。
+    """
+    try:
+        # 备份表名
+        backup_table = f"{table_name}_backup_{int(datetime.datetime.now().timestamp())}"
+        
+        logger.info(f"开始修复表 '{table_name}' 的字段约束...")
+        
+        # 1. 创建备份表
+        db.execute_sql(f"CREATE TABLE {backup_table} AS SELECT * FROM {table_name}")
+        logger.info(f"已创建备份表 '{backup_table}'")
+        
+        # 2. 删除原表
+        db.execute_sql(f"DROP TABLE {table_name}")
+        logger.info(f"已删除原表 '{table_name}'")
+        
+        # 3. 重新创建表（使用当前模型定义）
+        db.create_tables([model])
+        logger.info(f"已重新创建表 '{table_name}' 使用新的约束")
+        
+        # 4. 从备份表恢复数据
+        # 获取字段列表
+        fields = list(model._meta.fields.keys())
+        fields_str = ', '.join(fields)
+        
+        # 对于需要从 NOT NULL 改为 NULL 的字段，直接复制数据
+        # 对于需要从 NULL 改为 NOT NULL 的字段，需要处理 NULL 值
+        insert_sql = f"INSERT INTO {table_name} ({fields_str}) SELECT {fields_str} FROM {backup_table}"
+        
+        # 检查是否有字段需要从 NULL 改为 NOT NULL
+        null_to_notnull_fields = [
+            constraint['field_name'] for constraint in constraints_to_fix 
+            if constraint['action'] == 'disallow_null'
+        ]
+        
+        if null_to_notnull_fields:
+            # 需要处理 NULL 值，为这些字段设置默认值
+            logger.warning(f"字段 {null_to_notnull_fields} 将从允许NULL改为不允许NULL，需要处理现有的NULL值")
+            
+            # 构建更复杂的 SELECT 语句来处理 NULL 值
+            select_fields = []
+            for field_name in fields:
+                if field_name in null_to_notnull_fields:
+                    field_obj = model._meta.fields[field_name]
+                    # 根据字段类型设置默认值
+                    if isinstance(field_obj, (TextField,)):
+                        default_value = "''"
+                    elif isinstance(field_obj, (IntegerField, FloatField, DoubleField)):
+                        default_value = "0"
+                    elif isinstance(field_obj, BooleanField):
+                        default_value = "0"
+                    elif isinstance(field_obj, DateTimeField):
+                        default_value = f"'{datetime.datetime.now()}'"
+                    else:
+                        default_value = "''"
+                    
+                    select_fields.append(f"COALESCE({field_name}, {default_value}) as {field_name}")
+                else:
+                    select_fields.append(field_name)
+            
+            select_str = ', '.join(select_fields)
+            insert_sql = f"INSERT INTO {table_name} ({fields_str}) SELECT {select_str} FROM {backup_table}"
+        
+        db.execute_sql(insert_sql)
+        logger.info(f"已从备份表恢复数据到 '{table_name}'")
+        
+        # 5. 验证数据完整性
+        original_count = db.execute_sql(f"SELECT COUNT(*) FROM {backup_table}").fetchone()[0]
+        new_count = db.execute_sql(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]
+        
+        if original_count == new_count:
+            logger.info(f"数据完整性验证通过: {original_count} 行数据")
+            # 删除备份表
+            db.execute_sql(f"DROP TABLE {backup_table}")
+            logger.info(f"已删除备份表 '{backup_table}'")
+        else:
+            logger.error(f"数据完整性验证失败: 原始 {original_count} 行，新表 {new_count} 行")
+            logger.error(f"备份表 '{backup_table}' 已保留，请手动检查")
+        
+        # 记录修复的约束
+        for constraint in constraints_to_fix:
+            logger.info(f"已修复字段 '{constraint['field_name']}': "
+                       f"{constraint['current_constraint']} -> {constraint['target_constraint']}")
+                       
+    except Exception as e:
+        logger.exception(f"修复表 '{table_name}' 约束时出错: {e}")
+        # 尝试恢复
+        try:
+            if db.table_exists(backup_table):
+                logger.info(f"尝试从备份表 '{backup_table}' 恢复...")
+                db.execute_sql(f"DROP TABLE IF EXISTS {table_name}")
+                db.execute_sql(f"ALTER TABLE {backup_table} RENAME TO {table_name}")
+                logger.info(f"已从备份恢复表 '{table_name}'")
+        except Exception as restore_error:
+            logger.exception(f"恢复表失败: {restore_error}")
+
+
+def check_field_constraints():
+    """
+    检查但不修复字段约束，返回不一致的字段信息。
+    用于在修复前预览需要修复的内容。
+    """
+    
+    models = [
+        ChatStreams,
+        LLMUsage,
+        Emoji,
+        Messages,
+        Images,
+        ImageDescriptions,
+        OnlineTime,
+        PersonInfo,
+        Expression,
+        Memory,
+        ThinkingLog,
+        GraphNodes,
+        GraphEdges,
+        ActionRecords,
+    ]
+    
+    inconsistencies = {}
+    
+    try:
+        with db:
+            for model in models:
+                table_name = model._meta.table_name
+                if not db.table_exists(model):
+                    continue
+
+                # 获取当前表结构信息
+                cursor = db.execute_sql(f"PRAGMA table_info('{table_name}')")
+                current_schema = {row[1]: {'type': row[2], 'notnull': bool(row[3]), 'default': row[4]} 
+                                for row in cursor.fetchall()}
+                
+                table_inconsistencies = []
+                
+                # 检查每个模型字段的约束
+                for field_name, field_obj in model._meta.fields.items():
+                    if field_name not in current_schema:
+                        continue
+                    
+                    current_notnull = current_schema[field_name]['notnull']
+                    model_allows_null = field_obj.null
+                    
+                    if model_allows_null and current_notnull:
+                        table_inconsistencies.append({
+                            'field_name': field_name,
+                            'issue': 'model_allows_null_but_db_not_null',
+                            'model_constraint': 'NULL',
+                            'db_constraint': 'NOT NULL',
+                            'recommended_action': 'allow_null'
+                        })
+                    elif not model_allows_null and not current_notnull:
+                        table_inconsistencies.append({
+                            'field_name': field_name,
+                            'issue': 'model_not_null_but_db_allows_null',
+                            'model_constraint': 'NOT NULL',
+                            'db_constraint': 'NULL',
+                            'recommended_action': 'disallow_null'
+                        })
+                
+                if table_inconsistencies:
+                    inconsistencies[table_name] = table_inconsistencies
+                    
+    except Exception as e:
+        logger.exception(f"检查字段约束时出错: {e}")
+    
+    return inconsistencies
+
+
+
 # 模块加载时调用初始化函数
-initialize_database()
+initialize_database(sync_constraints=True)

src/mais4u/mais4u_chat/s4u_stream_generator.py

@@ -1,7 +1,6 @@
-import os
 from typing import AsyncGenerator
 from src.mais4u.openai_client import AsyncOpenAIClient
-from src.config.config import global_config, model_config
+from src.config.config import model_config
 from src.chat.message_receive.message import MessageRecvS4U
 from src.mais4u.mais4u_chat.s4u_prompt import prompt_builder
 from src.common.logger import get_logger