Python物流運輸管理系統源代碼，基於Django實現，實現了運單錄入、發車出庫、到貨簽收、客戶簽收等基本功能，含測試賬號。擁有較為完善的報表功能和財務管理功能。可以通過後台界面對各個用戶進行權限管理。
程序運行截圖
核心代碼
""" Django settings for PPWuLiu project. Generated by 'django-admin startproject' using Django 3.1.2. For more information on this file, see https://docs.djangoproject.com/en/3.1/topics/settings/ For the full list of settings and their values, see https://docs.djangoproject.com/en/3.1/ref/settings/ """
import os
from pathlib import Path
from django.contrib.messages import constants
# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve().parent.parent
# 創建日志的路徑
LOG_PATH = os.path.join(BASE_DIR, "logs")
if not os.path.isdir(LOG_PATH):
if os.path.isfile(LOG_PATH):
os.remove(LOG_PATH)
os.makedirs(LOG_PATH)
# 啟用Django Debug Toolbar
ENABLE_DJANGO_TOOLBAR = False
# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/3.1/howto/deployment/checklist/
# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = os.getenv("DJANGO_TMS_SECRET_KEY")
#SECRET_KEY = os.environ["SECRET_KEY"]
PBKDF2_ITERATIONS = 3
# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = True
ALLOWED_HOSTS = []
# Application definition
INSTALLED_APPS = [
# 'channels',
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
# 'rest_framework',
# 'django_filters',
'wuliu.apps.WuliuConfig',
'utils',
]
if DEBUG:
try:
import django_extensions
INSTALLED_APPS.append('django_extensions')
# print SQL queries in shell_plus
SHELL_PLUS_PRINT_SQL = True
except ImportError:
pass
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
# Pyinstrument, 暫時不用, 因為Django Debug Toolbar更好用更強大
# if DEBUG:
# try:
# import pyinstrument
# MIDDLEWARE.append('pyinstrument.middleware.ProfilerMiddleware')
# except ImportError:
# pass
# Django Debug Toolbar
if ENABLE_DJANGO_TOOLBAR:
# debug_toolbar依賴於django.contrib.staticfiles
if DEBUG and 'django.contrib.staticfiles' in INSTALLED_APPS:
try:
import debug_toolbar
INSTALLED_APPS.append('debug_toolbar')
MIDDLEWARE.append('debug_toolbar.middleware.DebugToolbarMiddleware')
# Internal IPs
INTERNAL_IPS = [
'127.0.0.1',
]
except ImportError:
pass
ROOT_URLCONF = 'PPWuLiu.urls'
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.template.context_processors.debug',
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
'wuliu.processor.pros'
],
},
},
]
WSGI_APPLICATION = 'PPWuLiu.wsgi.application'
# Database
# https://docs.djangoproject.com/en/3.1/ref/settings/#databases
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.mysql',
'NAME': os.getenv("DJANGO_TMS_MYSQL_DATABASE_NAME"),
'USER': os.getenv("DJANGO_TMS_MYSQL_USER"),
'PASSWORD': os.getenv("DJANGO_TMS_MYSQL_PASSWORD"),
'HOST': os.getenv("DJANGO_TMS_MYSQL_HOST"),
'PORT': os.getenv("DJANGO_TMS_MYSQL_PORT"),
}
}
# Password validation
# https://docs.djangoproject.com/en/3.1/ref/settings/#auth-password-validators
AUTH_PASSWORD_VALIDATORS = [
{
'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator',
},
{
'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator',
},
]
# Internationalization
# https://docs.djangoproject.com/en/3.1/topics/i18n/
LANGUAGE_CODE = 'zh-hans'
TIME_ZONE = 'Asia/Shanghai'
USE_I18N = True
USE_L10N = True
USE_TZ = True
# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/3.1/howto/static-files/
STATIC_URL = '/static/'
STATICFILES_DIRS = [
os.path.join(BASE_DIR, "static")
]
# Custom
CACHES = {
'default': {
# 基於本地內存的緩存
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
}
}
# 關閉浏覽器使會話立即過期
SESSION_EXPIRE_AT_BROWSER_CLOSE = True
MESSAGE_TAGS = {
constants.ERROR: "danger",
}
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'filters': {
'require_debug_false': {
'()': 'django.utils.log.RequireDebugFalse',
},
'require_debug_true': {
'()': 'django.utils.log.RequireDebugTrue',
},
},
'handlers': {
'console': {
'level': 'INFO',
'filters': ['require_debug_true'],
'class': 'logging.StreamHandler',
},
# Unused
'file': {
'level': 'INFO',
'class': 'logging.FileHandler',
'filename': '%s/log.txt' % LOG_PATH,
'encoding': 'utf-8',
},
'request_exceptions': {
'level': 'INFO',
'class': 'logging.FileHandler',
'filename': '%s/request_exceptions.txt' % LOG_PATH,
'encoding': 'utf-8',
},
'mail_admins': {
'level': 'ERROR',
'filters': ['require_debug_false'],
'class': 'django.utils.log.AdminEmailHandler'
}
},
'loggers': {
'django': {
'handlers': ['console', 'mail_admins'],
'level': 'INFO',
},
'wuliu.apps': {
'handlers': ['request_exceptions'],
'level': 'INFO',
'propagate': True,
},
'utils.common.expire_lru_cache': {
'handlers': ['console'],
'level': 'INFO',
'propagate': True,
},
},
}
''' # channels相關配置, 暫時不用 ASGI_APPLICATION = "PPWuLiu.asgi.application" CHANNEL_LAYERS = { 'default': { 'BACKEND': 'channels_redis.core.RedisChannelLayer', 'CONFIG': { 'hosts': [('127.0.0.1', 6379)], }, }, } '''
commn.py
from functools import wraps
from django.shortcuts import redirect
from django.http import Http404, HttpResponseForbidden
from django.utils import timezone
from .models import (
User, Waybill, TransportOut, DepartmentPayment, CargoPricePayment, Permission, PermissionGroup,
_get_global_settings,
)
from utils.common import ExpireLruCache, model_to_dict_
expire_lru_cache_three_hours = ExpireLruCache(expire_time=timezone.timedelta(hours=3))
expire_lru_cache_one_minute = ExpireLruCache(expire_time=timezone.timedelta(minutes=1))
get_global_settings = expire_lru_cache_three_hours(_get_global_settings)
@expire_lru_cache_one_minute
def _get_logged_user_by_id(user_id: int) -> User:
""" 根據用戶名返回用戶模型對象 """
return User.objects.get(id=user_id)
def get_logged_user(request) -> User:
""" 獲取已登錄的用戶對象 """
return _get_logged_user_by_id(request.session["user"]["id"])
def get_logged_user_type(request) -> User.Types:
""" 獲取已登錄的用戶的用戶類型 """
return get_logged_user(request).get_type
@expire_lru_cache_one_minute
def _get_user_permissions(user: User) -> set:
""" 獲取用戶擁有的權限, 注意該方法返回的是集合而不是QuerySet """
return set(user.permission.all().values_list("name", flat=True))
def is_logged_user_has_perm(request, perm_name: str) -> bool:
""" 檢查已登錄用戶是否具有perm_name權限 :return: True或False """
if not perm_name:
return True
return perm_name in _get_user_permissions(get_logged_user(request))
def is_logged_user_is_goods_yard(request) -> bool:
""" 判斷已登錄的用戶是否屬於貨場 """
return get_logged_user_type(request) == User.Types.GoodsYard
def _gen_permission_tree_list(root_pg_=PermissionGroup.objects.get(father__isnull=True)) -> list:
""" 根據所有的權限組和權限的層級結構生成列表, 用於前端渲染 """
tree_list = []
for pg in PermissionGroup.objects.filter(father=root_pg_):
tree_list.append({
"id": pg.id, "name": pg.name, "print_name": pg.print_name, "children": _gen_permission_tree_list(pg)
})
for p in Permission.objects.filter(father=root_pg_):
tree_list.append({
"id": p.id, "name": p.name, "print_name": p.print_name,
})
return tree_list
PERMISSION_TREE_LIST = _gen_permission_tree_list()
def login_required(raise_404=False):
""" 自定義裝飾器, 用於裝飾路由方法 若用戶未登錄, 則跳轉到登錄頁面 raise_404為True時, 則跳轉到404頁面 """
def _login_required(func):
@wraps(func)
def login_check(request, *args, **kwargs):
if not request.session.get("user"):
if raise_404:
raise Http404
return redirect("wuliu:login")
return func(request, *args, **kwargs)
return login_check
return _login_required
def check_permission(perm_name: str):
""" 自定義裝飾器, 用於在請求前檢查用戶是否具有perm_name權限 若無perm_name權限則跳轉至403頁面 """
def _check_permission(func):
@wraps(func)
def perm_check(request, *args, **kwargs):
if perm_name and not is_logged_user_has_perm(request, perm_name):
return HttpResponseForbidden()
return func(request, *args, **kwargs)
return perm_check
return _check_permission
def check_administrator(func):
""" 自定義裝飾器, 用於在請求前檢查用戶是否為管理員 若不是管理員則跳轉至403頁面 """
@wraps(func)
def admin_check(request, *args, **kwargs):
if not get_logged_user(request).administrator:
return HttpResponseForbidden()
return func(request, *args, **kwargs)
return admin_check
def waybill_to_dict(waybill_obj: Waybill) -> dict:
""" 將Waybill對象轉為字典 """
waybill_dic = model_to_dict_(waybill_obj)
waybill_dic["id_"] = waybill_obj.get_full_id
waybill_dic["fee_type_id"] = waybill_dic["fee_type"]
waybill_dic["fee_type"] = waybill_obj.get_fee_type_display()
waybill_dic["status_id"] = waybill_dic["status"]
waybill_dic["status"] = waybill_obj.get_status_display()
if waybill_obj.return_waybill is not None:
waybill_dic["return_waybill"] = waybill_to_dict(waybill_obj.return_waybill)
else:
waybill_dic["return_waybill"] = None
return waybill_dic
def transport_out_to_dict(transport_out_obj: TransportOut) -> dict:
""" 將TransportOut對象轉為字典 """
to_dic = model_to_dict_(transport_out_obj)
to_dic["id_"] = transport_out_obj.get_full_id
to_dic["status_id"] = to_dic["status"]
to_dic["status"] = transport_out_obj.get_status_display()
to_dic.update(transport_out_obj.gen_waybills_info())
return to_dic
def department_payment_to_dict(department_payment_obj: DepartmentPayment) -> dict:
""" 將DepartmentPayment對象轉為字典 """
dp_dic = model_to_dict_(department_payment_obj)
dp_dic["id_"] = department_payment_obj.get_full_id
dp_dic["status_id"] = dp_dic["status"]
dp_dic["status"] = department_payment_obj.get_status_display()
total_fee_dic = department_payment_obj.gen_total_fee()
dp_dic["total_fee_now"] = total_fee_dic["fee_now"]
dp_dic["total_fee_sign_for"] = total_fee_dic["fee_sign_for"]
dp_dic["total_cargo_price"] = total_fee_dic["cargo_price"]
dp_dic["final_total_fee"] = sum(total_fee_dic.values())
return dp_dic
def cargo_price_payment_to_dict(cargo_price_payment_obj: CargoPricePayment) -> dict:
""" 將CargoPricePayment對象轉為字典 """
cpp_dic = model_to_dict_(cargo_price_payment_obj)
cpp_dic["id_"] = cargo_price_payment_obj.get_full_id
cpp_dic["status_id"] = cpp_dic["status"]
cpp_dic["status"] = cargo_price_payment_obj.get_status_display()
total_fee_dic = cargo_price_payment_obj.gen_total_fee()
cpp_dic["total_cargo_price"] = total_fee_dic["cargo_price"]
cpp_dic["total_deduction_fee"] = total_fee_dic["deduction_fee"]
cpp_dic["total_cargo_handling_fee"] = total_fee_dic["cargo_handling_fee"]
cpp_dic["final_fee"] = (
total_fee_dic["cargo_price"] - total_fee_dic["deduction_fee"] - total_fee_dic["cargo_handling_fee"]
)
return cpp_dic
測試數據賬號密碼如下:
#用戶名 密碼 用戶類型
#csshd 666666 收貨點
#csfgs 666666 分公司（到貨點）
#kj_1 666666 會計（財務部）
#zxg_1 666666 貨物裝卸工1（貨台）
#zxg_2 666666 貨物裝卸工2（貨台）
#pzqqt 88888888 管理員
完整Python物流運輸管理系統源代碼下載地址：物流運輸管理系統
