Python 物品库存管理系统 (PyCharm项目/Django框架/HTML页面/MySQL数据库)

import datetime import json from django.db import connection from django.http import JsonResponse from django.shortcuts import render, redirect from django.views.decorators.csrf import csrf_exempt from user_management.connect_db import database_initialization, insert_data, delete_data, update_data, query_data,query_default config = { 'host': 'localhost', 'user': 'root', 'password': '123456', # 数据库密码 'database': 'inventory' } # 初始化函数 def initialization(request): database_initialization(config) return redirect('/LoginPage/') # 登录页面 def login_view(request): return render(request, 'LoginPage.html') # 用户页面 def user_view(request): return render(request, 'UserPage.html') # 管理员页面 def admin_view(request): return render(request, 'AdminPage.html') # 用户登录 def user_login(request): if request.method == 'POST': # 获取请求中的数据 username = request.POST.get('username') password = request.POST.get('password') # 检查输入是否为空 if not username or not password: return JsonResponse({"result": "false", "message": "账号和密码不能为空"}) # 查询数据库检查用户名和密码 result = query_data(config, 'users', '*', f"username = '{username}'") if result: user_info = result["message"][0] # 提取存储的密码、real_name 和 position # 检查密码是否匹配（假设密码是经过哈希存储的，下面为直接对比明文密码） if password == user_info["password"]: # 密码匹配，返回成功，并带上用户信息 return JsonResponse({ "result": "true", "message": "登录成功", "id": user_info["id"], "name": user_info["name"], "position": user_info["position"] }) else: # 密码错误 return JsonResponse({"result": "false", "message": "密码错误"}) else: # 用户名不存在 return JsonResponse({"result": "false", "message": "用户名不存在"}) else: return JsonResponse({"result": "false", "message": "请求方法错误"}) # 添加用户 def add_user(request): try: data = json.loads(request.body.decode('utf-8')) # 提取数据字段 user_data = { "name": data.get("name"), "username": data.get("username"), "password": data.get("password"), "position": data.get("position") } # 插入数据库 result = insert_data(config, "users", user_data) if result["result"]: return JsonResponse({"result": True, "message": f"用户添加成功"}) else: return JsonResponse({"result": False, "message": f"添加失败: {result['message']}"}) except Exception as e: return JsonResponse({"result": "error", "message": f"请求处理失败: {str(e)}"}) # 获取用户数据 def get_user(request): try: result = query_data(config, 'users') return JsonResponse(result) except Exception as e: return JsonResponse({ 'result': False, 'message': str(e) }) # 删除用户 def del_user(request): try: data = json.loads(request.body.decode('utf-8')) user_id = data.get('id') if not user_id: return JsonResponse({'result': False, 'message': '缺少用户ID'}) conditions = {'id': user_id} # 传字典 res = delete_data(config, "users", conditions) return JsonResponse(res) except Exception as e: return JsonResponse({'result': False, 'message': str(e)}) # 添加库存 def add_resource(request): try: data = json.loads(request.body.decode('utf-8')) resource_name = data.get('resource_name') quantity = data.get('quantity') if not resource_name or quantity is None: return JsonResponse({'result': False, 'message': '资源名称和数量不能为空'}) quantity = int(quantity) query_result = query_data(config, 'resource', columns='*', conditions=f"resource_name = '{resource_name}'") if not query_result['result']: # 查询失败，直接返回失败信息 return JsonResponse({'result': False, 'message': query_result['message']}) # 查询成功，判断是否有数据 if query_result['message'] and len(query_result['message']) > 0: # 找到了对应资源，进行更新库存 current_stock = query_result['message'][0]['stock_quantity'] new_stock = current_stock + quantity # 用字典形式传入更新条件和新值 update_result = update_data( config, 'resource', conditions={'resource_name': resource_name}, new_values={'stock_quantity': new_stock} ) if update_result['result']: return JsonResponse({'result': True, 'message': '库存更新成功'}) else: return JsonResponse({'result': False, 'message': update_result['message']}) else: # 未找到资源，插入新记录 insert_result = insert_data(config, 'resource', data={'resource_name': resource_name, 'stock_quantity': quantity}) if insert_result['result']: return JsonResponse({'result': True, 'message': '库存添加成功'}) else: return JsonResponse({'result': False, 'message': insert_result['message']}) except Exception as e: return JsonResponse({'result': False, 'message': f'服务器错误: {str(e)}'}) # 显示所有库存 def show_all_resource(request): try: result = query_data(config, 'resource') return JsonResponse(result) except Exception as e: return JsonResponse({ 'result': False, 'message': str(e) }) # 获取所有申请记录 def get_all_resource(request): try: result = query_data(config, 'resource_usage_log') return JsonResponse(result) except Exception as e: return JsonResponse({ 'result': False, 'message': str(e) }) # 查询申请记录 def search_resource(request): data = json.loads(request.body.decode('utf-8')) resource_name = data.get('resource_name') try: result = query_data(config, 'resource_usage_log', "*", f"resource_name like '%{resource_name}%'") return JsonResponse(result) except Exception as e: return JsonResponse({ 'result': False, 'message': str(e) }) # 删除库存 def del_resource(request): try: data = json.loads(request.body.decode('utf-8')) resource_id = data.get('id') if not resource_id: return JsonResponse({'result': False, 'message': '缺少用户ID'}) conditions = {'id': resource_id} # 传字典 res = delete_data(config, "resource", conditions) return JsonResponse(res) except Exception as e: return JsonResponse({'result': False, 'message': str(e)}) # 工具函数：将字典条件转换成 SQL 条件字符串 def dict_to_conditions_str(conditions: dict) -> str: if not conditions: return '' return " AND ".join([f"{k} = '{v}'" if isinstance(v, str) else f"{k} = {v}" for k, v in conditions.items()]) # 添加用户权限 def add_permission(request):