每天40分玩转Django：Django实战 - 在线打印服务系统

Django实战 - 在线打印服务系统

一、系统功能概览表

模块	主要功能	技术要点
文件上传	PDF/Word文件上传、文件验证	文件处理、MIME类型验证
异步处理	文件转换、打印队列	Celery、Redis
通知邮件	打印状态通知、订单确认	SMTP、邮件模板

二、系统架构设计

2.1 模型设计

# models.py
from django.db import models
from django.contrib.auth.models import User
from django.core.validators import FileExtensionValidatorclass PrintJob(models.Model):STATUS_CHOICES = (('pending', '等待处理'),('processing', '处理中'),('completed', '已完成'),('failed', '失败'),)user = models.ForeignKey(User, on_delete=models.CASCADE)document = models.FileField(upload_to='documents/%Y/%m/%d/',validators=[FileExtensionValidator(allowed_extensions=['pdf', 'doc', 'docx'])])copies = models.IntegerField(default=1)double_sided = models.BooleanField(default=True)color = models.BooleanField(default=False)status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending')created_at = models.DateTimeField(auto_now_add=True)updated_at = models.DateTimeField(auto_now=True)notes = models.TextField(blank=True)page_count = models.IntegerField(null=True, blank=True)estimated_cost = models.DecimalField(max_digits=10, decimal_places=2, null=True)def __str__(self):return f"Print Job #{self.id} - {self.user.username}"class PrintNotification(models.Model):print_job = models.ForeignKey(PrintJob, on_delete=models.CASCADE)sent_at = models.DateTimeField(auto_now_add=True)message = models.TextField()is_read = models.BooleanField(default=False)

2.2 文件上传处理

# forms.py
from django import forms
from .models import PrintJobclass PrintJobForm(forms.ModelForm):class Meta:model = PrintJobfields = ['document', 'copies', 'double_sided', 'color', 'notes']def clean_document(self):document = self.cleaned_data.get('document')if document:if document.size > 20 * 1024 * 1024:  # 20MB限制raise forms.ValidationError('文件大小不能超过20MB')return document# views.py
from django.shortcuts import render, redirect
from django.contrib.auth.decorators import login_required
from django.contrib import messages
from .forms import PrintJobForm
from .tasks import process_print_job@login_required
def submit_print_job(request):if request.method == 'POST':form = PrintJobForm(request.POST, request.FILES)if form.is_valid():print_job = form.save(commit=False)print_job.user = request.userprint_job.save()# 启动异步任务process_print_job.delay(print_job.id)messages.success(request, '打印任务已提交，请等待处理')return redirect('print_job_status', job_id=print_job.id)else:form = PrintJobForm()return render(request, 'print/submit.html', {'form': form})

2.3 Celery异步任务配置

# celery.py
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from django.conf import settingsos.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')app = Celery('print_service')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)# tasks.py
from celery import shared_task
from django.core.mail import send_mail
from django.template.loader import render_to_string
from .models import PrintJob@shared_task
def process_print_job(job_id):try:print_job = PrintJob.objects.get(id=job_id)print_job.status = 'processing'print_job.save()# 处理文件并计算页数page_count = calculate_page_count(print_job.document)print_job.page_count = page_count# 计算估计成本cost = calculate_cost(page_count, print_job.color, print_job.double_sided)print_job.estimated_cost = cost# 模拟打印过程import timetime.sleep(5)  # 模拟打印时间print_job.status = 'completed'print_job.save()# 发送完成通知send_completion_email.delay(job_id)except Exception as e:print_job.status = 'failed'print_job.notes = str(e)print_job.save()@shared_task
def send_completion_email(job_id):print_job = PrintJob.objects.get(id=job_id)context = {'job': print_job,'user': print_job.user,}html_message = render_to_string('print/email/completion.html', context)send_mail(subject='您的打印任务已完成',message=f'打印任务 #{print_job.id} 已完成，请及时取件。',html_message=html_message,from_email='printservice@example.com',recipient_list=[print_job.user.email],fail_silently=False,)

2.4 邮件模板设计

<!-- templates/print/email/completion.html -->
<!DOCTYPE html>
<html>
<head><meta charset="utf-8"><style>.container { padding: 20px; }.header { color: #333; }.details { margin: 20px 0; }.footer { color: #666; font-size: 12px; }</style>
</head>
<body><div class="container"><h2 class="header">打印任务完成通知</h2><div class="details"><p>尊敬的 {{ user.username }}：</p><p>您的打印任务已经完成，详情如下：</p><ul><li>任务编号：#{{ job.id }}</li><li>页数：{{ job.page_count }}</li><li>费用：¥{{ job.estimated_cost }}</li><li>完成时间：{{ job.updated_at|date:"Y-m-d H:i" }}</li></ul><p>请尽快到打印点取件，谢谢！</p></div><div class="footer"><p>此邮件为系统自动发送，请勿回复。</p></div></div>
</body>
</html>

三、流程图设计

在这里插入图片描述

四、实用工具函数

# utils.py
import os
import PyPDF2
from docx import Documentdef calculate_page_count(document):"""计算文档页数"""file_extension = os.path.splitext(document.name)[1].lower()if file_extension == '.pdf':with document.open('rb') as pdf_file:pdf_reader = PyPDF2.PdfReader(pdf_file)return len(pdf_reader.pages)elif file_extension in ['.doc', '.docx']:doc = Document(document)return len(doc.paragraphs) // 40  # 估算页数return 0def calculate_cost(page_count, is_color, is_double_sided):"""计算打印费用"""base_price = 0.5  # 单面黑白价格if is_color:base_price *= 3  # 彩印价格是黑白的3倍if is_double_sided:# 双面打印优惠10%total_pages = (page_count + 1) // 2return round(total_pages * base_price * 0.9, 2)return round(page_count * base_price, 2)

五、单元测试

# tests.py
from django.test import TestCase, Client
from django.contrib.auth.models import User
from django.core import mail
from django.core.files.uploadedfile import SimpleUploadedFile
from .models import PrintJob
from .tasks import process_print_jobclass PrintServiceTests(TestCase):def setUp(self):self.client = Client()self.user = User.objects.create_user(username='testuser',email='test@example.com',password='testpass123')# 创建测试PDF文件self.test_file = SimpleUploadedFile("test.pdf",b"file_content",content_type="application/pdf")def test_print_job_creation(self):"""测试打印任务创建"""self.client.login(username='testuser', password='testpass123')response = self.client.post('/print/submit/', {'document': self.test_file,'copies': 1,'double_sided': True,'color': False,})self.assertEqual(response.status_code, 302)self.assertEqual(PrintJob.objects.count(), 1)def test_email_notification(self):"""测试邮件通知"""print_job = PrintJob.objects.create(user=self.user,document=self.test_file,status='processing')process_print_job(print_job.id)self.assertEqual(len(mail.outbox), 1)self.assertEqual(mail.outbox[0].to[0], self.user.email)

六、部署配置

# settings.py
# Celery配置
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'# 邮件配置
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
EMAIL_HOST = 'smtp.example.com'
EMAIL_PORT = 587
EMAIL_USE_TLS = True
EMAIL_HOST_USER = 'your-email@example.com'
EMAIL_HOST_PASSWORD = 'your-email-password'# 文件上传配置
MEDIA_URL = '/media/'
MEDIA_ROOT = os.path.join(BASE_DIR, 'media')
FILE_UPLOAD_MAX_MEMORY_SIZE = 20 * 1024 * 1024  # 20MB
FILE_UPLOAD_PERMISSIONS = 0o644