如何在 Django 中生成 Excel 文件并上传至 FastDFS

时间：2024-11-09 07:21:30

文章目录

  • 如何在 Django 中生成 Excel 文件并上传至 FastDFS
    • 需求背景
      • 主要任务
    • 实现步骤
  • 创建 Excel 文件
  • 上传 Excel 文件到 FastDFS
    • client.conf
  • 保存文件 URL 到数据库
  • 组合完整的流程
  • 总结

如何在 Django 中生成 Excel 文件并上传至 FastDFS

在很多实际应用场景中，我们需要批量处理数据并生成 Excel 文件，然后将这些文件上传到远程存储系统，比如 FastDFS。本文将详细介绍如何在 Django 中实现这一过程，尤其是如何生成多个工作表，每个工作表存放一个提示词的内容，并将生成的文件上传至 FastDFS，最后将文件 URL 存储到数据库中。

需求背景

假设我们有一个任务管理系统，每个任务都有多个提示词。每个提示词需要在一个新的工作表中生成，最终将这些 Excel 文件上传到 FastDFS，然后将文件的 URL 保存到数据库中，以便后续使用。

主要任务

  1. 从数据库中获取任务和提示词数据。
  2. 生成包含多个工作表的 Excel 文件，每个工作表包含一个提示词的名称和内容。
  3. 上传生成的 Excel 文件到 FastDFS。
  4. 将返回的文件 URL 保存到数据库中的相应字段。

实现步骤

创建 Excel 文件

首先，我们需要根据提示词的数量生成一个包含多个工作表的 Excel 文件。每个工作表的名称为 Sheet1, Sheet2, 等等，内容为提示词的名称和详情。

import openpyxl
from django.utils import timezone
import os

def generate_excel_and_upload(cue_data, job_name):
    """
    生成 Excel 文件并上传到 FastDFS
    每个提示词在一个新的工作表中
    """
    # 创建一个新的 Excel 文件
    wb = openpyxl.Workbook()
    
    # 使用默认的命名方式，例如 Sheet1, Sheet2, ...
    for index, cue in enumerate(cue_data, 1):
        sheet_name = f"Sheet{index}"  # 默认命名为 Sheet1, Sheet2, ...
        sheet = wb.create_sheet(title=sheet_name)  # 创建新的工作表
        
        # 在工作表中写入提示词详情
        sheet['A1'] = "提示词名称"
        sheet['B1'] = cue['name']
        sheet['A2'] = "提示词详情"
        sheet['B2'] = cue['prompt_details']
        
        # 你可以根据需要继续添加更多的行或列
        
    # 删除默认创建的工作表
    del wb['Sheet']
    
    # 生成临时文件名
    file_path = f'temp_{job_name}_{timezone.now().strftime("%Y%m%d%H%M%S")}.xlsx'
    
    # 保存为本地 Excel 文件
    wb.save(file_path)
    
    # 返回文件路径
    return file_path

在这段代码中：

  • openpyxl.Workbook() 用于创建新的 Excel 工作簿。
  • 我们使用 enumerate 来遍历提示词数据，并为每个提示词创建一个新的工作表，命名为 Sheet1, Sheet2，依此类推。
  • sheet['A1'] 和 sheet['B1'] 用于写入提示词的名称和详情。

上传 Excel 文件到 FastDFS

FastDFS 是一个高性能的分布式文件系统，我们可以将生成的 Excel 文件上传到 FastDFS，并返回一个文件 URL。为了上传文件，我们需要使用 FastDFS 客户端。

from fdfs_client.client import Fdfs_client

def uploadFile(file_path):
    """
    上传文件到 FastDFS 并返回文件 URL
    """
    client = Fdfs_client('client.conf')  # 配置文件路径
    ret = client.upload(filepath=file_path)
    if ret:
        return ret['Remote file_id']  # 返回文件 ID 或 URL
    else:
        raise Exception("文件上传失败")

在这段代码中：

  • Fdfs_client('client.conf') 用于初始化 FastDFS 客户端，client.conf 是 FastDFS 客户端的配置文件。
  • client.upload(filepath=file_path) 用于上传文件并返回上传结果。

client.conf

tracker_server=192.168.1.1:22122
http.tracker_server.port=8888
connect_timeout=30
network_timeout=60

保存文件 URL 到数据库

上传文件后，我们会得到一个文件 URL，需要将该 URL 保存到数据库中的相应字段。在本例中，保存文件 URL 的字段是 cue_txttitle。

from Workers.models import JobId
from django.utils import timezone

def save_job_to_db(job_name, file_url):
    """
    保存任务信息到数据库，包括文件 URL
    """
    job_instance = JobId(
        job_name=job_name,
        status=JobId.StatusChoices.RUNNING,
        create_time=timezone.now(),
        update_time=timezone.now(),
        cue_txttitle=file_url,  # 将文件 URL 存储到 cue_txttitle 字段
        jobtotal=0  # 初始总数为 0
    )
    job_instance.save()  # 保存任务记录
    
    return job_instance.id  # 返回任务的 ID

在这段代码中：

  • JobId 是任务记录的模型，我们将文件 URL 存储到 cue_txttitle 字段。
  • timezone.now() 获取当前时间，用于记录任务的创建和更新时间。

组合完整的流程

最后，我们将所有步骤组合成一个完整的流程来处理请求。假设用户发送一个包含任务信息的请求，我们将从数据库中获取提示词信息，生成 Excel 文件，上传文件到 FastDFS，并将文件 URL 保存到数据库中。

import json
from django.http import JsonResponse
from django.views import View
from utils.InitMq import InitMq
from Workers.models import jobmanage
from promptword.models import PromptData

class Execute(View):
    def post(self, request):
        try:
            data = json.loads(request.body)
            job_name = data.get('name')
            job_id = data.get('jobid')
            job_type = int(data.get('taskType'))
            
            # 参数验证
            if not (job_name, job_id, job_type):
                return JsonResponse({'code': 400, 'message': "参数不全！"})
            
            # 获取任务数据
            db_alias = 'prod' if job_type == 0 else 'default'
            job_data = jobmanage.objects.using(db_alias).filter(id=job_id).first()
            
            if not job_data:
                return JsonResponse({'code': 400, 'message': '未找到相应的任务数据'})
            
            # 获取提示词数据
            cue_ids = eval(job_data.cue_id)
            cue_data = []
            if cue_ids:
                names = PromptData.objects.filter(id__in=cue_ids).values_list('id', 'name', 'prompt_details')
                for cue_id, name, prompt_details in names:
                    cue_data.append({'id': cue_id, 'name': name, 'prompt_details': prompt_details})
            
            # 生成 Excel 文件并上传
            file_path = generate_excel_and_upload(cue_data, job_name)
            file_url = uploadFile(file_path)
            
            # 保存 URL 到数据库
            job_id = save_job_to_db(job_name, file_url)
            
            return JsonResponse({
                'code': 200,
                'message': '成功生成并上传 Excel 文件',
                'data': file_url
            })
        
        except Exception as e:
            return JsonResponse({'code': 500, 'message': str(e)})

总结

通过这篇文章，我们展示了如何在 Django 中实现以下功能：

  1. 从数据库中获取提示词数据。
  2. 生成包含多个工作表的 Excel 文件，每个工作表保存一个提示词的名称和详情。
  3. 上传生成的 Excel 文件到 FastDFS 并获取文件 URL。
  4. 将文件 URL 保存到数据库中的任务记录中。

这种方法不仅能处理批量数据，还能灵活地将文件上传到分布式文件系统，实现高效的文件管理。
在这里插入图片描述

相关文章