引言
企业资源管理系统(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 models
class 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 messages
def 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_required
def permission_required(permission):
def decorator(view_func):
@wraps(view_func)
@login_required
def _wrapped_view(request, *args, **kwargs):
if request.user.has_perm(permission):
return view_func(request, *args, **kwargs)
return HttpResponseForbidden('权限不足')
return _wrapped_view
return 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 - expense
context = {
'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 cache
def 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-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["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
- redis
environment:
- DEBUG=False
- DATABASE_URL=postgresql://erp_user:password@db:5432/erp_db
- REDIS_URL=redis://redis:6379/0
db:
image: postgres:13
environment:
- POSTGRES_DB=erp_db
- POSTGRES_USER=erp_user
- POSTGRES_PASSWORD=password
volumes:
- postgres_data:/var/lib/postgresql/data
redis:
image: redis:6-alpine
nginx:
image: nginx:alpine
ports:
- "80:80"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
depends_on:
- web
volumes:
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 re
def 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('手机号码格式不正确')
总结
本文详细介绍了如何使用Django框架构建一个功能完整的企业资源管理系统。从项目架构设计到具体功能实现,从前端界面到后端优化,我们涵盖了ERP系统开发的各个方面。
项目亮点
- 模块化设计:采用Django的应用模式,实现了良好的代码组织和模块解耦
- 权限管理:实现了细粒度的权限控制,确保数据安全
- 性能优化:通过缓存、数据库优化等手段提升系统性能
- 响应式设计:使用Bootstrap实现了适配多设备的用户界面
- 容器化部署:使用Docker实现了便捷的部署和扩展
扩展建议
- API接口:使用Django REST Framework提供RESTful API
- 微服务架构:将不同模块拆分为独立的微服务
- 消息队列:使用Celery处理异步任务和定时任务
- 监控系统:集成Prometheus和Grafana进行系统监控
- 移动端支持:开发移动端应用或PWA
通过本项目的实践,开发者可以掌握Django框架的高级特性,学习企业级应用的开发模式,为后续的大型项目开发奠定坚实基础。