openclaw-memory/backend/memory_app/models.py

from django.db import models
from django.core.validators import FileExtensionValidator
import hashlib


class LobsterMemory(models.Model):
    """龙虾记忆文件模型"""

    STATUS_CHOICES = [
        ('consistent', '一致'),
        ('local_newer', '本地更新'),
        ('db_newer', '数据库更新'),
        ('conflict', '冲突'),
        ('hard_conflict', '严重冲突'),  # 新增：严重冲突状态
    ]

    lobster_id = models.CharField(max_length=50, db_index=True, help_text='龙虾ID')

    file_path = models.CharField(max_length=500, help_text='文件相对路径')

    content = models.TextField(help_text='文件内容')

    hash = models.CharField(max_length=64, help_text='SHA256哈希')

    status = models.CharField(
        max_length=20,
        choices=STATUS_CHOICES,
        default='consistent',
        db_index=True,
        help_text='同步状态'
    )

    version = models.IntegerField(default=1, help_text='版本号')

    size = models.IntegerField(default=0, help_text='文件大小（字节）')

    summary = models.TextField(null=True, blank=True, max_length=1000, help_text='语义摘要')

    has_attributes = models.BooleanField(default=False, help_text='是否有属性')

    created_at = models.DateTimeField(auto_now_add=True, db_index=True, help_text='创建时间')

    updated_at = models.DateTimeField(auto_now=True, db_index=True, help_text='更新时间')

    class Meta:
        db_table = 'lobster_memory'
        unique_together = ('lobster_id', 'file_path', 'version')
        ordering = ['-updated_at']
        indexes = [
            models.Index(fields=['lobster_id', 'file_path']),
            models.Index(fields=['status']),
            models.Index(fields=['updated_at']),
            models.Index(fields=['lobster_id', 'updated_at']),
        ]

    def __str__(self):
        return f"{self.lobster_id}/{self.file_path} (v{self.version})"

    def compute_hash(self, content: str) -> str:
        """
        计算 SHA256 哈希

        Args:
            content: 文件内容

        Returns:
            哈希值
        """
        return hashlib.sha256(content.encode('utf-8')).hexdigest()

    def save(self, *args, **kwargs):
        """保存时自动计算哈希和大小"""
        if self.content:
            self.hash = self.compute_hash(self.content)
            self.size = len(self.content.encode('utf-8'))
        super().save(*args, **kwargs)


class FileAttribute(models.Model):
    """文件属性模型（支持属性目录结构）"""

    VALUE_TYPE_CHOICES = [
        ('string', '字符串'),
        ('integer', '整数'),
        ('float', '浮点数'),
        ('boolean', '布尔值'),
        ('json', 'JSON'),
    ]

    lobster_id = models.CharField(max_length=50, db_index=True, help_text='龙虾ID')

    file_path = models.CharField(max_length=500, db_index=True, help_text='文件相对路径')

    key = models.CharField(max_length=200, db_index=True, help_text='属性键（支持点号分隔的嵌套路径）')

    value = models.TextField(null=True, blank=True, help_text='属性值')

    value_type = models.CharField(
        max_length=20,
        choices=VALUE_TYPE_CHOICES,
        default='string',
        help_text='值类型'
    )

    category = models.CharField(max_length=100, db_index=True, null=True, blank=True, help_text='属性分类')

    metadata = models.JSONField(default=dict, help_text='元数据')

    created_at = models.DateTimeField(auto_now_add=True, db_index=True, help_text='创建时间')

    updated_at = models.DateTimeField(auto_now=True, help_text='更新时间')

    class Meta:
        db_table = 'file_attribute'
        unique_together = ('lobster_id', 'file_path', 'key')
        ordering = ['lobster_id', 'file_path', 'key']
        indexes = [
            models.Index(fields=['lobster_id', 'file_path']),
            models.Index(fields=['lobster_id', 'category']),
            models.Index(fields=['lobster_id', 'updated_at']),
        ]

    def __str__(self):
        return f"{self.lobster_id}/{self.file_path}.{self.key} = {self.value}"

    def get_parsed_value(self):
        """根据类型解析值"""
        if self.value_type == 'string':
            return self.value
        elif self.value_type == 'integer':
            return int(self.value) if self.value else None
        elif self.value_type == 'float':
            return float(self.value) if self.value else None
        elif self.value_type == 'boolean':
            return self.value.lower() in ('true', '1', 'yes') if self.value else False
        elif self.value_type == 'json':
            import json
            return json.loads(self.value) if self.value else None
        return self.value


class SyncHistory(models.Model):
    """同步操作历史记录"""

    ACTION_CHOICES = [
        ('sync_to_db', '同步到数据库'),
        ('sync_to_local', '同步到本地'),
        ('auto_sync', '自动同步'),
        ('manual_merge', '手动合并'),
        ('conflict_resolved', '冲突解决'),
        ('attribute_sync', '属性同步'),
    ]

    STATUS_CHOICES = [
        ('success', '成功'),
        ('failed', '失败'),
        ('partial', '部分成功'),
    ]

    SOURCE_CHOICES = [
        ('local', '本地文件'),
        ('database', '数据库'),
        ('manual', '手动操作'),
    ]

    lobster_id = models.CharField(max_length=50, db_index=True, help_text='龙虾ID')

    file_path = models.CharField(max_length=500, db_index=True, help_text='文件相对路径')

    action = models.CharField(
        max_length=20,
        choices=ACTION_CHOICES,
        help_text='操作类型'
    )

    status = models.CharField(
        max_length=20,
        choices=STATUS_CHOICES,
        help_text='操作状态'
    )

    source = models.CharField(
        max_length=20,
        choices=SOURCE_CHOICES,
        default='local',
        help_text='数据源'
    )

    old_version = models.IntegerField(null=True, blank=True, help_text='操作前版本')

    new_version = models.IntegerField(null=True, blank=True, help_text='操作后版本')

    old_hash = models.CharField(max_length=64, null=True, blank=True, help_text='操作前哈希')

    new_hash = models.CharField(max_length=64, null=True, blank=True, help_text='操作后哈希')

    file_size = models.IntegerField(default=0, help_text='文件大小（字节）')

    lines_changed = models.IntegerField(default=0, help_text='变动行数（+新增/-删除）')

    operator = models.CharField(max_length=50, default='system', help_text='操作者')

    error_message = models.TextField(null=True, blank=True, help_text='错误信息')

    execution_time = models.FloatField(default=0, help_text='执行时间（秒）')

    attributes_changed = models.JSONField(default=dict, help_text='属性变更记录')

    is_attribute_sync = models.BooleanField(default=False, help_text='是否为属性同步')

    created_at = models.DateTimeField(auto_now_add=True, db_index=True, help_text='操作时间')

    class Meta:
        db_table = 'sync_history'
        ordering = ['-created_at']
        indexes = [
            models.Index(fields=['lobster_id', 'file_path']),
            models.Index(fields=['action']),
            models.Index(fields=['status']),
            models.Index(fields=['created_at']),
            models.Index(fields=['lobster_id', 'created_at']),
        ]

    def __str__(self):
        return f"{self.action} - {self.lobster_id}/{self.file_path} ({self.status})"