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

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

一、体系功能概览表

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

2.1 模型设计

1. # models.py
2. from django.db import models
3. from django.contrib.auth.models import User
4. from django.core.validators import FileExtensionValidator
6. class PrintJob(models.Model):
7.     STATUS_CHOICES = (
8.         ('pending', '等待处理'),
9.         ('processing', '处理中'),
10.         ('completed', '已完成'),
11.         ('failed', '失败'),
12.     )
13.    
14.     user = models.ForeignKey(User, on_delete=models.CASCADE)
15.     document = models.FileField(
16.         upload_to='documents/%Y/%m/%d/',
17.         validators=[FileExtensionValidator(allowed_extensions=['pdf', 'doc', 'docx'])]
18.     )
19.     copies = models.IntegerField(default=1)
20.     double_sided = models.BooleanField(default=True)
21.     color = models.BooleanField(default=False)
22.     status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending')
23.     created_at = models.DateTimeField(auto_now_add=True)
24.     updated_at = models.DateTimeField(auto_now=True)
25.     notes = models.TextField(blank=True)
26.     page_count = models.IntegerField(null=True, blank=True)
27.     estimated_cost = models.DecimalField(max_digits=10, decimal_places=2, null=True)
28.    
29.     def __str__(self):
30.         return f"Print Job #{self.id} - {self.user.username}"
32. class PrintNotification(models.Model):
33.     print_job = models.ForeignKey(PrintJob, on_delete=models.CASCADE)
34.     sent_at = models.DateTimeField(auto_now_add=True)
35.     message = models.TextField()
36.     is_read = models.BooleanField(default=False)

复制代码

2.2 文件上传处理惩罚

1. # forms.py
2. from django import forms
3. from .models import PrintJob
5. class PrintJobForm(forms.ModelForm):
6.     class Meta:
7.         model = PrintJob
8.         fields = ['document', 'copies', 'double_sided', 'color', 'notes']
9.         
10.     def clean_document(self):
11.         document = self.cleaned_data.get('document')
12.         if document:
13.             if document.size > 20 * 1024 * 1024:  # 20MB限制
14.                 raise forms.ValidationError('文件大小不能超过20MB')
15.         return document
17. # views.py
18. from django.shortcuts import render, redirect
19. from django.contrib.auth.decorators import login_required
20. from django.contrib import messages
21. from .forms import PrintJobForm
22. from .tasks import process_print_job
24. @login_required
25. def submit_print_job(request):
26.     if request.method == 'POST':
27.         form = PrintJobForm(request.POST, request.FILES)
28.         if form.is_valid():
29.             print_job = form.save(commit=False)
30.             print_job.user = request.user
31.             print_job.save()
32.             
33.             # 启动异步任务
34.             process_print_job.delay(print_job.id)
35.             
36.             messages.success(request, '打印任务已提交，请等待处理')
37.             return redirect('print_job_status', job_id=print_job.id)
38.     else:
39.         form = PrintJobForm()
40.    
41.     return render(request, 'print/submit.html', {'form': form})

复制代码

2.3 Celery异步使命设置

1. # celery.py
2. from __future__ import absolute_import, unicode_literals
3. import os
4. from celery import Celery
5. from django.conf import settings
7. os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
9. app = Celery('print_service')
10. app.config_from_object('django.conf:settings', namespace='CELERY')
11. app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
13. # tasks.py
14. from celery import shared_task
15. from django.core.mail import send_mail
16. from django.template.loader import render_to_string
17. from .models import PrintJob
19. @shared_task
20. def process_print_job(job_id):
21.     try:
22.         print_job = PrintJob.objects.get(id=job_id)
23.         print_job.status = 'processing'
24.         print_job.save()
25.         
26.         # 处理文件并计算页数
27.         page_count = calculate_page_count(print_job.document)
28.         print_job.page_count = page_count
29.         
30.         # 计算估计成本
31.         cost = calculate_cost(page_count, print_job.color, print_job.double_sided)
32.         print_job.estimated_cost = cost
33.         
34.         # 模拟打印过程
35.         import time
36.         time.sleep(5)  # 模拟打印时间
37.         
38.         print_job.status = 'completed'
39.         print_job.save()
40.         
41.         # 发送完成通知
42.         send_completion_email.delay(job_id)
43.         
44.     except Exception as e:
45.         print_job.status = 'failed'
46.         print_job.notes = str(e)
47.         print_job.save()
48.         
49. @shared_task
50. def send_completion_email(job_id):
51.     print_job = PrintJob.objects.get(id=job_id)
52.     context = {
53.         'job': print_job,
54.         'user': print_job.user,
55.     }
56.    
57.     html_message = render_to_string('print/email/completion.html', context)
58.    
59.     send_mail(
60.         subject='您的打印任务已完成',
61.         message=f'打印任务 #{print_job.id} 已完成，请及时取件。',
62.         html_message=html_message,
63.         from_email='printservice@example.com',
64.         recipient_list=[print_job.user.email],
65.         fail_silently=False,
66.     )

复制代码

2.4 邮件模板设计

1. <!-- templates/print/email/completion.html -->
2. <!DOCTYPE html>
3. <html>
4. <head>
5.     <meta charset="utf-8">
6.     <style>
7.         .container { padding: 20px; }
8.         .header { color: #333; }
9.         .details { margin: 20px 0; }
10.         .footer { color: #666; font-size: 12px; }
11.     </style>
12. </head>
13. <body>
14.     <div class="container">
15.         <h2 class="header">打印任务完成通知</h2>
16.         <div class="details">
17.             <p>尊敬的 {{ user.username }}：</p>
18.             <p>您的打印任务已经完成，详情如下：</p>
19.             <ul>
20.                 <li>任务编号：#{{ job.id }}</li>
21.                 <li>页数：{{ job.page_count }}</li>
22.                 <li>费用：¥{{ job.estimated_cost }}</li>
23.                 <li>完成时间：{{ job.updated_at|date:"Y-m-d H:i" }}</li>
24.             </ul>
25.             <p>请尽快到打印点取件，谢谢！</p>
26.         </div>
27.         <div class="footer">
28.             <p>此邮件为系统自动发送，请勿回复。</p>
29.         </div>
30.     </div>
31. </body>
32. </html>

复制代码

三、流程图设计

四、实用工具函数

1. # utils.py
2. import os
3. import PyPDF2
4. from docx import Document
6. def calculate_page_count(document):
7.     """计算文档页数"""
8.     file_extension = os.path.splitext(document.name)[1].lower()
9.    
10.     if file_extension == '.pdf':
11.         with document.open('rb') as pdf_file:
12.             pdf_reader = PyPDF2.PdfReader(pdf_file)
13.             return len(pdf_reader.pages)
14.             
15.     elif file_extension in ['.doc', '.docx']:
16.         doc = Document(document)
17.         return len(doc.paragraphs) // 40  # 估算页数
18.    
19.     return 0
21. def calculate_cost(page_count, is_color, is_double_sided):
22.     """计算打印费用"""
23.     base_price = 0.5  # 单面黑白价格
24.    
25.     if is_color:
26.         base_price *= 3  # 彩印价格是黑白的3倍
27.    
28.     if is_double_sided:
29.         # 双面打印优惠10%
30.         total_pages = (page_count + 1) // 2
31.         return round(total_pages * base_price * 0.9, 2)
32.    
33.     return round(page_count * base_price, 2)

复制代码

五、单元测试

1. # tests.py
2. from django.test import TestCase, Client
3. from django.contrib.auth.models import User
4. from django.core import mail
5. from django.core.files.uploadedfile import SimpleUploadedFile
6. from .models import PrintJob
7. from .tasks import process_print_job
9. class PrintServiceTests(TestCase):
10.     def setUp(self):
11.         self.client = Client()
12.         self.user = User.objects.create_user(
13.             username='testuser',
14.             email='test@example.com',
15.             password='testpass123'
16.         )
17.         
18.         # 创建测试PDF文件
19.         self.test_file = SimpleUploadedFile(
20.             "test.pdf",
21.             b"file_content",
22.             content_type="application/pdf"
23.         )
24.         
25.     def test_print_job_creation(self):
26.         """测试打印任务创建"""
27.         self.client.login(username='testuser', password='testpass123')
28.         
29.         response = self.client.post('/print/submit/', {
30.             'document': self.test_file,
31.             'copies': 1,
32.             'double_sided': True,
33.             'color': False,
34.         })
35.         
36.         self.assertEqual(response.status_code, 302)
37.         self.assertEqual(PrintJob.objects.count(), 1)
38.         
39.     def test_email_notification(self):
40.         """测试邮件通知"""
41.         print_job = PrintJob.objects.create(
42.             user=self.user,
43.             document=self.test_file,
44.             status='processing'
45.         )
46.         
47.         process_print_job(print_job.id)
48.         
49.         self.assertEqual(len(mail.outbox), 1)
50.         self.assertEqual(mail.outbox[0].to[0], self.user.email)

复制代码

六、部署设置

1. # settings.py
2. # Celery配置
3. CELERY_BROKER_URL = 'redis://localhost:6379/0'
4. CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
5. CELERY_ACCEPT_CONTENT = ['json']
6. CELERY_TASK_SERIALIZER = 'json'
7. CELERY_RESULT_SERIALIZER = 'json'
8. CELERY_TIMEZONE = 'Asia/Shanghai'
10. # 邮件配置
11. EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend'
12. EMAIL_HOST = 'smtp.example.com'
13. EMAIL_PORT = 587
14. EMAIL_USE_TLS = True
15. EMAIL_HOST_USER = 'your-email@example.com'
16. EMAIL_HOST_PASSWORD = 'your-email-password'
18. # 文件上传配置
19. MEDIA_URL = '/media/'
20. MEDIA_ROOT = os.path.join(BASE_DIR, 'media')
21. FILE_UPLOAD_MAX_MEMORY_SIZE = 20 * 1024 * 1024  # 20MB
22. FILE_UPLOAD_PERMISSIONS = 0o644

复制代码

七、安全考虑

• 文件上传安全：
  
  
  
  • 限制文件大小和类型
    
  • 使用安全的文件存储位置
    
  • 实行病毒扫描
    
  
  
• 用户认证：
  
  
  
  • 要求用户登录
    
  • 实行访问控制
    
  • 防止未授权访问
    
  
  
• 异步使命安全：
  
  
  
  • 使命队列加密
    
  • 错误处理惩罚和重试机制
    
  • 监控和日志记录
    
  
  

八、性能优化发起

• 文件处理惩罚优化：
  
  
  
  • 使用分块上传
    
  • 实现断点续传
    
  • 文件压缩处理惩罚
    
  
  
• 使命队列优化：
  
  
  
  • 公道设置并发数
    
  • 实现使命优先级
    
  • 添加使命超时机制
    
  
  
• 邮件发送优化：
  
  
  
  • 使用邮件队列
    
  • 批量处理惩罚通知
    
  • 模板预编译
    
  
  

九、后续扩展发起

• 功能扩展：
  
  
  
  • 添加打印预览
    
  • 支持更多文件格式
    
  • 实现打印参数设置
    
  
  
• 用户体验提升：
  
  
  
  • 添加进度条显示
    
  • 实时状态更新
    
  • 打印汗青记录
    
  
  
• 管理功能：
  
  
  
  • 打印机管理
    
  • 耗材管理
    
  • 统计报表
    
  
  

本节课程介绍了如何使用Django框架构建一个在线打印服务体系，包括文件上传、异步处理惩罚和邮件通知等核心功能。发起学习者在理解根本代码的同时，注意体系的安全性和性能优化，并根据实际需求举行功能扩展。

---

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。