Class File

apps/knowledge/models/knowledge.py:365–470

class File(AppModelMixin):
    id = models.UUIDField(primary_key=True, max_length=128, default=uuid.uuid7, editable=False, verbose_name="主键id")
    file_name = models.CharField(max_length=256, verbose_name="文件名称", default="")
    file_size = models.IntegerField(verbose_name="文件大小", default=0)
    sha256_hash = models.CharField(verbose_name="文件sha256_hash标识", default="")
    source_type = models.CharField(
        verbose_name="资源类型",
        choices=FileSourceType,
        default=FileSourceType.TEMPORARY_120_MINUTE.value,
        db_index=True,
    )
    source_id = models.CharField(
        verbose_name="资源id", default=FileSourceType.TEMPORARY_120_MINUTE.value, db_index=True
    )
    loid = models.IntegerField(verbose_name="loid")
    meta = models.JSONField(verbose_name="文件关联数据", default=dict)

    class Meta:
        db_table = "file"

    def save(self, bytea=None, force_insert=False, force_update=False, using=None, update_fields=None):
        if bytea is None:
            raise ValueError("bytea参数不能为空")

        sha256_hash = get_sha256_hash(bytea)
        self.sha256_hash = sha256_hash
        existing_file = QuerySet(File).filter(sha256_hash=sha256_hash).first()
        if existing_file:
            self.loid = existing_file.loid
            self.file_size = existing_file.file_size
            return super().save()

        compressed_data = self._compress_data(bytea)
        self.file_size = len(compressed_data)

        self.loid = self._create_large_object()

        self._write_compressed_data(compressed_data)
        # 调用父类保存
        return super().save()

    def _compress_data(self, data, compression_level=9):
        """压缩数据到内存"""
        buffer = io.BytesIO()
        with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as zip_file:
            zipinfo = zipfile.ZipInfo(self.file_name)
            zipinfo.compress_type = zipfile.ZIP_DEFLATED
            zip_file.writestr(zipinfo, data, compresslevel=compression_level)

        return buffer.getvalue()

    def _create_large_object(self):
        result = select_one("SELECT lo_creat(-1)::int8 as lo_id;", [])
        return result["lo_id"]

    def _write_compressed_data(self, data, block_size=64 * 1024):
        buffer = io.BytesIO(data)
        offset = 0