准备工作
需2类工具,均为开源免费,直接执行命令安装:
- Python3.10+:执行命令
sudo apt install python3 python3-pip -y(Linux)或从官网下载安装包(Windows/Mac)
- MinIO对象存储服务:执行命令
curl https://dl.min.io/server/minio/release/linux-amd64/minio -o /usr/local/bin/minio && chmod +x /usr/local/bin/minio(Linux;Mac替换为darwin-amd64,Windows替换为win-amd64.exe)
环境部署
1. MinIO存储服务启动
执行以下命令创建数据目录并启动服务:
```
mkdir -p ~/minio-data
minio server ~/minio-data --console-address ":9001" --address ":9000"
```
登录控制台:打开浏览器输入 http://localhost:9001,默认账号密码为 minioadmin/minioadmin。点击「Create Bucket」,输入Bucket名称 food-archive,勾选「Private」,确认创建完成。
2. Python项目环境搭建
创建项目目录并初始化虚拟环境:
```
mkdir food-archive-system && cd food-archive-system
python3 -m venv venv
source venv/bin/activate Windows替换为 venv\Scripts\activate
```
创建依赖文件 requirements.txt,内容如下:
```
fastapi==0.104.1
uvicorn==0.24.0.post1
minio==7.2.5
python-multipart==0.0.6
python-magic==0.4.27
pydantic==2.5.2
```
安装依赖:执行 pip install -r requirements.txt。注意:Linux下需额外执行 sudo apt install libmagic1 -y,否则文件类型检测会失败。
系统核心功能实现
1. 配置文件编写

创建 config.py,内容为固定配置,无需修改(本地测试用):
```
config.py
MINIO_ENDPOINT = "localhost:9000"
MINIO_ACCESS_KEY = "minioadmin"
MINIO_SECRET_KEY = "minioadmin"
MINIO_SECURE = False 生产环境需改为True
BUCKET_NAME = "food-archive"
```
2. 档案上传接口
创建 main.py,实现带批次、产品类型标识的档案上传,限制仅支持食品相关文件:
```
main.py
from fastapi import FastAPI, File, UploadFile, HTTPException
from minio import Minio
from minio.error import S3Error
import magic
import uvicorn
from config import MINIO_ENDPOINT, MINIO_ACCESS_KEY, MINIO_SECRET_KEY, MINIO_SECURE, BUCKET_NAME
app = FastAPI(title="食品工程数字档案馆系统")
初始化MinIO客户端
minio_client = Minio(
MINIO_ENDPOINT, access_key=MINIO_ACCESS_KEY,
secret_key=MINIO_SECRET_KEY, secure=MINIO_SECURE
)
允许的食品档案文件类型
ALLOWED_MIMETYPES = {
"application/pdf", "image/jpeg", "image/png",
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
"application/vnd.openxmlformats-officedocument.wordprocessingml.document"
}
@app.post("/upload-archive/")
async def upload_archive(batch_number: str, product_type: str, file: UploadFile = File(...)):
验证文件类型
file_content = await file.read()
mime_type = magic.from_buffer(file_content, mime=True)
if mime_type not in ALLOWED_MIMETYPES:
raise HTTPException(status_code=400, detail="仅支持PDF、JPG、PNG、Excel、Word格式")
生成唯一档案名:批次号_产品类型_原文件名
filename = f"{batch_number}_{product_type}_{file.filename}"
上传到MinIO
try:
minio_client.put_object(
bucket_name=BUCKET_NAME, object_name=filename,
data=file_content, length=len(file_content), content_type=mime_type
)
except S3Error as e:
raise HTTPException(status_code=500, detail=f"上传失败:{str(e)}")
return {"code":200, "message":"上传成功", "archive_path":filename, "batch":batch_number}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
```
3. 档案查询接口
在 main.py 中追加查询逻辑,按批次号获取档案带有效期链接:
```
@app.get("/query-archives/{batch_number}")
async def query_archives(batch_number: str):
try:
objects = minio_client.list_objects(BUCKET_NAME, recursive=True)
result = []
for obj in objects:
if obj.object_name.startswith(f"{batch_number}_"):
生成1小时有效期的下载链接
presigned_url = minio_client.presigned_get_object(
BUCKET_NAME, obj.object_name, expires=3600
)
result.append({
"name": obj.object_name, "size": obj.size,
"last_modified": obj.last_modified.strftime("%Y-%m-%d %H:%M:%S"),
"download_url": presigned_url
})
if not result:
raise HTTPException(status_code=404, detail="未找到对应批次档案")
return {"code":200, "data":result}
except S3Error as e:
raise HTTPException(status_code=500, detail=f"查询失败:{str(e)}")
```
系统启动与测试
启动Python服务:执行 python main.py,服务默认运行在 http://localhost:8000。
调试接口:直接访问 http://localhost:8000/docs,所有接口可在线调试,无需前端。
上传测试步骤
- 在/docs页面找到POST /upload-archive/,点击「Try it out」
- 填写batch_number(如SP20240501)、product_type(如乳制品),选择要上传的食品档案文件
- 点击「Execute」,返回code:200则上传成功
查询测试步骤
- 在/docs页面找到GET /query-archives/{batch_number},点击「Try it out」
- 输入刚才的批次号SP20240501,点击「Execute」
- 返回结果包含该批次所有档案的名称、大小、时间及1小时有效期下载链接
关键注意事项
1. 本地测试时,MinIO和Python服务需同时运行,关闭任意一个则功能失效;2. 生产环境需将MINIO_SECURE设为True,修改MinIO账号密码为强密码,禁止使用默认凭证;3. 文件类型限制可根据需求在ALLOWED_MIMETYPES中增减格式;4. 若遇MinIO连接失败,检查端口9000/9001是否被占用,或MinIO服务是否启动。