每天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 FileExtensionValidator

class 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 PrintJob

class PrintJobForm(forms.ModelForm):
    class Meta:
        model = PrintJob
        fields = ['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.user
            print_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 settings

os.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 time
        time.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 Document

def 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 0

def 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) // 2
        return 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_job

class 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