在Django或Sanic中集成Magika进行文件类型检测

Magika是Google开发的一款高效文件类型识别工具，能够准确识别各种文件格式。本文将介绍如何在Python Web框架（如Django或Sanic）中集成Magika来实现文件上传时的类型检测功能。

问题背景

开发者在尝试将Magika集成到Django或Sanic API端点时遇到了文件路径相关的问题。虽然命令行版本工作正常，但在Web框架中直接调用时会出现"文件或目录未找到"的错误。

解决方案

通过分析发现，Magika在处理大文件时可能会遇到内存问题。有效的解决方案是采用分块读取文件的方式，逐块传递给Magika进行检测。这种方法不仅解决了内存问题，还能保持较高的检测准确性。

实现细节

以下是基于Django Ninja框架的实现示例：

import os
from magika import Magika
from ninja import File, NinjaAPI
from ninja.files import UploadedFile
from typing_extensions import Any

# 初始化Magika实例
m = Magika()

# 创建API实例
api = NinjaAPI()

# 设置分块大小，默认为100字节
chunk_size = int(os.getenv("CHUNK_SIZE", 100))

def check_file_type_magika(chunked_file: bytes) -> str:
    """使用Magika检测文件类型"""
    check_file_type = m.identify_bytes(content=chunked_file)
    return f"文件类型: {check_file_type.output.ct_label}, MIME类型: {check_file_type.output.mime_type}"

@api.post("/upload")
def upload(request, file: UploadedFile = File(...)) -> dict[str, Any]:
    """文件上传端点"""
    data = file
    for chunk in data.chunks(chunk_size):
        detect_filetype = check_file_type_magika(chunk)
    
    if detect_filetype is not None:
        return {"检测结果": str(detect_filetype), "文件大小": len(data)}
    return {"检测结果": "未知类型", "文件大小": 0}

技术要点

1. 分块处理：通过设置合理的chunk_size参数，可以平衡内存使用和检测效率。对于大文件，建议适当增大分块大小。

2. 内存优化：使用生成器方式逐块读取文件，避免一次性加载大文件到内存中。

3. 类型检测：Magika的identify_bytes方法可以直接处理字节数据，非常适合Web应用中的文件上传场景。

4. 环境配置：通过环境变量CHUNK_SIZE可以灵活调整分块大小，适应不同部署环境的需求。

性能考虑

• 对于小文件（<1MB），可以直接读取整个文件进行检测
• 中等文件（1MB-100MB）建议使用100-500字节的分块
• 大文件（>100MB）可以适当增大分块到1KB左右

扩展应用

此方案不仅适用于Django Ninja，也可以轻松适配到其他Python Web框架如Flask、FastAPI或Sanic。核心思想是将上传的文件流分块处理，然后利用Magika的字节检测功能。

通过这种集成方式，开发者可以在Web应用中实现高效、准确的文件类型检测功能，为后续的文件处理流程提供可靠的类型信息。