当前位置：首页 > news >正文

Python实战--基于Django的企业资源管理系统

news 2025/8/19 11:32:32

引言

企业资源管理系统（ERP）是现代企业运营的核心工具，它整合了企业的各个业务流程，包括人力资源、财务管理、库存管理、客户关系管理等。本文将详细介绍如何使用Python的Django框架构建一个功能完整的企业资源管理系统，从项目架构设计到具体功能实现，为开发者提供一个完整的实战指南。

项目概述

系统功能模块

我们的ERP系统将包含以下核心模块：

用户管理模块：用户注册、登录、权限管理
人力资源模块：员工信息管理、考勤管理、薪资管理
财务管理模块：收支记录、财务报表、预算管理
库存管理模块：商品管理、库存监控、采购管理
客户关系模块：客户信息、销售记录、客户服务
报表分析模块：数据可视化、业务分析

技术栈选择

后端框架：Django 4.2+
数据库：PostgreSQL
前端技术：Bootstrap 5 + jQuery
缓存系统：Redis
任务队列：Celery
部署方案：Docker + Nginx

项目架构设计

目录结构

erp_system/
├── manage.py
├── requirements.txt
├── docker-compose.yml
├── erp_system/
│   ├── __init__.py
│   ├── settings/
│   │   ├── __init__.py
│   │   ├── base.py
│   │   ├── development.py
│   │   └── production.py
│   ├── urls.py
│   └── wsgi.py
├── apps/
│   ├── __init__.py
│   ├── users/
│   ├── hr/
│   ├── finance/
│   ├── inventory/
│   ├── crm/
│   └── reports/
├── static/
├── media/
├── templates/
└── utils/

数据库设计

用户模型

from django.contrib.auth.models import AbstractUser
from django.db import modelsclass User(AbstractUser):employee_id = models.CharField(max_length=20, unique=True)department = models.CharField(max_length=100)position = models.CharField(max_length=100)phone = models.CharField(max_length=20)hire_date = models.DateField()is_active = models.BooleanField(default=True)class Meta:db_table = 'users'

员工信息模型

class Employee(models.Model):user = models.OneToOneField(User, on_delete=models.CASCADE)salary = models.DecimalField(max_digits=10, decimal_places=2)bank_account = models.CharField(max_length=50)emergency_contact = models.CharField(max_length=100)address = models.TextField()created_at = models.DateTimeField(auto_now_add=True)updated_at = models.DateTimeField(auto_now=True)class Meta:db_table = 'employees'

核心功能实现

1. 用户认证与权限管理

自定义用户认证

# apps/users/views.py
from django.contrib.auth import authenticate, login
from django.contrib.auth.decorators import login_required
from django.shortcuts import render, redirect
from django.contrib import messagesdef user_login(request):if request.method == 'POST':username = request.POST['username']password = request.POST['password']user = authenticate(request, username=username, password=password)if user is not None:login(request, user)return redirect('dashboard')else:messages.error(request, '用户名或密码错误')return render(request, 'users/login.html')@login_required
def dashboard(request):context = {'user': request.user,'recent_activities': get_recent_activities(request.user)}return render(request, 'dashboard.html', context)

权限装饰器

# utils/decorators.py
from functools import wraps
from django.http import HttpResponseForbidden
from django.contrib.auth.decorators import login_requireddef permission_required(permission):def decorator(view_func):@wraps(view_func)@login_requireddef _wrapped_view(request, *args, **kwargs):if request.user.has_perm(permission):return view_func(request, *args, **kwargs)return HttpResponseForbidden('权限不足')return _wrapped_viewreturn decorator

2. 人力资源管理

员工信息管理

# apps/hr/views.py
from django.shortcuts import render, get_object_or_404
from django.contrib.auth.decorators import login_required
from .models import Employee, Attendance
from .forms import EmployeeForm@login_required
@permission_required('hr.view_employee')
def employee_list(request):employees = Employee.objects.select_related('user').all()return render(request, 'hr/employee_list.html', {'employees': employees})@login_required
@permission_required('hr.add_employee')
def employee_create(request):if request.method == 'POST':form = EmployeeForm(request.POST)if form.is_valid():form.save()messages.success(request, '员工信息添加成功')return redirect('employee_list')else:form = EmployeeForm()return render(request, 'hr/employee_form.html', {'form': form})

考勤管理

# apps/hr/models.py
class Attendance(models.Model):employee = models.ForeignKey(Employee, on_delete=models.CASCADE)date = models.DateField()check_in = models.TimeField()check_out = models.TimeField(null=True, blank=True)work_hours = models.DecimalField(max_digits=4, decimal_places=2, default=0)status = models.CharField(max_length=20, choices=[('present', '出勤'),('absent', '缺勤'),('late', '迟到'),('leave', '请假')])class Meta:db_table = 'attendance'unique_together = ['employee', 'date']

3. 财务管理模块

财务记录模型

# apps/finance/models.py
class FinanceRecord(models.Model):TRANSACTION_TYPES = [('income', '收入'),('expense', '支出'),]transaction_type = models.CharField(max_length=10, choices=TRANSACTION_TYPES)amount = models.DecimalField(max_digits=12, decimal_places=2)description = models.TextField()category = models.CharField(max_length=100)date = models.DateField()created_by = models.ForeignKey(User, on_delete=models.CASCADE)created_at = models.DateTimeField(auto_now_add=True)class Meta:db_table = 'finance_records'

财务报表生成

# apps/finance/views.py
from django.db.models import Sum, Q
from django.utils import timezone
from datetime import datetime, timedelta@login_required
@permission_required('finance.view_report')
def financial_report(request):end_date = timezone.now().date()start_date = end_date - timedelta(days=30)# 收入统计income = FinanceRecord.objects.filter(transaction_type='income',date__range=[start_date, end_date]).aggregate(total=Sum('amount'))['total'] or 0# 支出统计expense = FinanceRecord.objects.filter(transaction_type='expense',date__range=[start_date, end_date]).aggregate(total=Sum('amount'))['total'] or 0# 净利润profit = income - expensecontext = {'income': income,'expense': expense,'profit': profit,'start_date': start_date,'end_date': end_date}return render(request, 'finance/report.html', context)

4. 库存管理系统

商品模型

# apps/inventory/models.py
class Product(models.Model):name = models.CharField(max_length=200)sku = models.CharField(max_length=50, unique=True)description = models.TextField()unit_price = models.DecimalField(max_digits=10, decimal_places=2)stock_quantity = models.IntegerField(default=0)min_stock_level = models.IntegerField(default=10)category = models.CharField(max_length=100)supplier = models.CharField(max_length=200)created_at = models.DateTimeField(auto_now_add=True)class Meta:db_table = 'products'

库存预警系统

# apps/inventory/tasks.py
from celery import shared_task
from django.core.mail import send_mail
from .models import Product@shared_task
def check_low_stock():low_stock_products = Product.objects.filter(stock_quantity__lte=models.F('min_stock_level'))if low_stock_products.exists():message = "以下商品库存不足：\n"for product in low_stock_products:message += f"- {product.name}: 当前库存 {product.stock_quantity}\n"send_mail('库存预警通知',message,'system@company.com',['manager@company.com'],fail_silently=False,)

前端界面设计

响应式布局

<!-- templates/base.html -->
<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>企业资源管理系统</title><link href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css" rel="stylesheet"><link href="{% static 'css/custom.css' %}" rel="stylesheet">
</head>
<body><nav class="navbar navbar-expand-lg navbar-dark bg-primary"><div class="container"><a class="navbar-brand" href="{% url 'dashboard' %}">ERP系统</a><div class="navbar-nav ms-auto"><a class="nav-link" href="{% url 'logout' %}">退出</a></div></div></nav><div class="container-fluid"><div class="row"><nav class="col-md-2 d-md-block bg-light sidebar">{% include 'partials/sidebar.html' %}</nav><main class="col-md-10 ms-sm-auto px-md-4">{% block content %}{% endblock %}</main></div></div><script src="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/js/bootstrap.bundle.min.js"></script><script src="{% static 'js/app.js' %}"></script>
</body>
</html>

数据可视化

// static/js/charts.js
function renderFinancialChart(data) {const ctx = document.getElementById('financialChart').getContext('2d');new Chart(ctx, {type: 'line',data: {labels: data.labels,datasets: [{label: '收入',data: data.income,borderColor: 'rgb(75, 192, 192)',tension: 0.1}, {label: '支出',data: data.expense,borderColor: 'rgb(255, 99, 132)',tension: 0.1}]},options: {responsive: true,plugins: {title: {display: true,text: '财务趋势图'}}}});
}

系统优化与部署

性能优化

数据库优化

# settings/base.py
DATABASES = {'default': {'ENGINE': 'django.db.backends.postgresql','NAME': 'erp_db','USER': 'erp_user','PASSWORD': 'password','HOST': 'localhost','PORT': '5432','OPTIONS': {'MAX_CONNS': 20,'CONN_MAX_AGE': 600,}}
}# 缓存配置
CACHES = {'default': {'BACKEND': 'django_redis.cache.RedisCache','LOCATION': 'redis://127.0.0.1:6379/1','OPTIONS': {'CLIENT_CLASS': 'django_redis.client.DefaultClient',}}
}

查询优化

# 使用select_related和prefetch_related优化查询
def get_employees_with_attendance():return Employee.objects.select_related('user').prefetch_related('attendance_set').all()# 使用缓存
from django.core.cache import cachedef get_dashboard_data(user):cache_key = f'dashboard_data_{user.id}'data = cache.get(cache_key)if data is None:data = {'total_employees': Employee.objects.count(),'total_products': Product.objects.count(),'monthly_revenue': calculate_monthly_revenue(),}cache.set(cache_key, data, 300)  # 缓存5分钟return data

Docker部署

# Dockerfile
FROM python:3.11-slimWORKDIR /appCOPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txtCOPY . .EXPOSE 8000CMD ["gunicorn", "--bind", "0.0.0.0:8000", "erp_system.wsgi:application"]

# docker-compose.yml
version: '3.8'services:web:build: .ports:- "8000:8000"depends_on:- db- redisenvironment:- DEBUG=False- DATABASE_URL=postgresql://erp_user:password@db:5432/erp_db- REDIS_URL=redis://redis:6379/0db:image: postgres:13environment:- POSTGRES_DB=erp_db- POSTGRES_USER=erp_user- POSTGRES_PASSWORD=passwordvolumes:- postgres_data:/var/lib/postgresql/dataredis:image: redis:6-alpinenginx:image: nginx:alpineports:- "80:80"volumes:- ./nginx.conf:/etc/nginx/nginx.confdepends_on:- webvolumes:postgres_data:

安全性考虑

数据安全

# settings/production.py
import os# 安全设置
SECRET_KEY = os.environ.get('SECRET_KEY')
DEBUG = False
ALLOWED_HOSTS = ['your-domain.com']# HTTPS设置
SECURE_SSL_REDIRECT = True
SECURE_HSTS_SECONDS = 31536000
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_HSTS_PRELOAD = True# 会话安全
SESSION_COOKIE_SECURE = True
CSRF_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True# 数据库连接加密
DATABASES['default']['OPTIONS']['sslmode'] = 'require'

输入验证

# apps/common/validators.py
from django.core.exceptions import ValidationError
import redef validate_employee_id(value):if not re.match(r'^EMP\d{6}$', value):raise ValidationError('员工ID格式不正确，应为EMP+6位数字')def validate_phone_number(value):if not re.match(r'^1[3-9]\d{9}$', value):raise ValidationError('手机号码格式不正确')