简介这是一份面向计算机专业本科生的毕业设计实战项目聚焦Web开发与自动化运维融合实践适用于Python全栈初学者及DevOps入门者。项目基于Django构建可视化管理后台集成Ansible实现远程服务器批量部署、配置管理与任务调度切实解决中小团队IT资源手工运维效率低、易出错的痛点。压缩包含2000个文件主体为518个Python脚本含Django视图、模型、Ansible调用逻辑、1321个JS/CSS前端资源支撑数据表格、响应式界面与交互控制及68个HTML模板辅以MD文档说明与JSON配置整体30.09MB结构完整、模块清晰开箱即见典型WebAnsible工程目录。已有341人学习下载读者可直接复用其用户权限体系、Ansible Playbook触发机制、前后端联调方案及安全加固实践快速掌握从Web界面到底层运维的端到端开发链路。1. 这不是又一个 Django CRUD 页面——它用 Python 把 Ansible 的 YAML 剧本变成可点击的 Web 按钮你见过能直接在浏览器里点“一键部署 Redis 集群”、点完立刻弹出执行日志、失败时高亮报错行、成功后自动刷新服务器状态列表的毕业设计吗这不是 Demo而是这个基于 Python Django Ansible 的自动化管理系统的真实能力。它跳出了传统毕设常见的“用户增删改查简单表单提交”套路把 Ansible 的playbook.yml文件从运维工程师的终端命令变成了前端按钮背后的可审计、可授权、可回滚的 Web 操作流。系统不替代 Ansible而是封装 AnsibleDjango 负责身份校验、任务排队、日志归档和界面渲染Ansible 负责 SSH 连接、幂等执行、模块调用和结果解析。适合两类人一是正在写毕设、需要体现“工程闭环能力”的本科生二是刚转岗 DevOps、想快速理解 Web 化运维平台底层逻辑的初级工程师。它不追求炫酷 UI但所有交互都对应真实运维场景——比如“执行前校验目标主机连通性”“限制普通用户仅能触发预审批剧本”“Playbook 执行超时自动 kill 并释放锁”。2. Django 后端如何安全调用 Ansible进程隔离、权限控制与结果捕获2.1 为什么不能直接os.system(ansible-playbook ...)初学者常犯的错误是在 Django 视图里用os.system()或subprocess.run()直接调用 Ansible 命令。这看似简单实则埋下三颗雷权限失控Web 进程如gunicorn通常以低权限用户运行但 Ansible 需要读取inventory、roles和私钥文件硬编码路径易导致 Permission Denied进程阻塞Ansible 执行可能耗时数分钟同步调用会卡死整个 Django 请求线程导致后续请求排队甚至超时结果丢失subprocess返回的是原始 stdout/stderr 字符串无法结构化提取 task 名称、changed/count、failed 状态更难做前端实时日志流。提示Django 官方明确建议耗时操作必须异步化。Ansible 本身无内置 Web API因此需构建一层轻量胶水层。2.2 推荐方案Celery ansible-runner 封装调用链本项目采用ansible-runnerRed Hat 官方维护的 Python 封装库替代裸subprocess配合 Celery 实现异步解耦。ansible-runner的核心价值在于它将 Ansible 执行抽象为Runner对象自动处理环境隔离、stdout/stderr 解析、事件回调和结果序列化且原生支持json输出格式。安装依赖要求 Python 3.8# 创建虚拟环境并激活 python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 安装核心组件 pip install django4.2.13 celery5.3.6 ansible-runner2.3.3 redis4.6.0配置 Celerymyproject/celery.pyimport os from celery import Celery os.environ.setdefault(DJANGO_SETTINGS_MODULE, myproject.settings) app Celery(myproject) app.config_from_object(django.conf:settings, namespaceCELERY) app.autodiscover_tasks()编写异步任务tasks.pyfrom celery import shared_task import ansible_runner import json from django.conf import settings shared_task(bindTrue, max_retries3) def run_ansible_playbook(self, playbook_path, inventory_path, extra_varsNone): 异步执行 Ansible Playbook :param playbook_path: 相对路径如 playbooks/deploy_redis.yml :param inventory_path: 如 inventories/production :param extra_vars: dict动态传入变量如 {redis_version: 7.2} try: # ansible-runner 自动处理工作目录、环境变量、SSH 密钥路径 r ansible_runner.run( playbookplaybook_path, inventoryinventory_path, extravarsextra_vars or {}, verbosity2, # 控制日志详细程度 json_modeTrue, # 强制输出 JSON 格式便于解析 limitNone, # 可传入 host pattern 限制目标 event_handlerlambda event: self.update_state( statePROGRESS, meta{event: event[event], stdout: event.get(stdout, )} ) ) # 结构化结果status, rc, stats, stdout, stderr result { status: r.status, rc: r.rc, stats: r.stats, stdout: r.stdout.read(), stderr: r.stderr.read(), events: list(r.events) # 所有事件流含 task start/end } return result except Exception as exc: # 重试机制网络波动或临时文件锁 raise self.retry(excexc, countdown60 * (2 ** self.request.retries))在视图中触发任务views.pyfrom django.http import JsonResponse from django.views.decorators.csrf import csrf_exempt from .tasks import run_ansible_playbook csrf_exempt def trigger_playbook(request): if request.method POST: data json.loads(request.body) playbook data.get(playbook) inventory data.get(inventory, inventories/localhost) extra_vars data.get(extra_vars, {}) # Celery 异步提交立即返回 task_id task run_ansible_playbook.delay(playbook, inventory, extra_vars) return JsonResponse({ task_id: task.id, status: submitted, message: fPlaybook {playbook} queued for execution }) return JsonResponse({error: Only POST allowed}, status405)关键参数说明参数作用实际值示例注意事项playbookPlaybook 文件路径相对于 runner 工作目录playbooks/nginx_setup.yml必须存在路径需在 Django 项目内可访问inventoryInventory 文件或目录路径inventories/staging支持 ini/yaml 格式推荐用目录结构管理多环境extravars动态注入变量覆盖 playbook 中vars{domain: test.example.com}避免硬编码敏感信息应通过 Django 配置或数据库读取verbosity日志级别0~4影响 stdout 详细度2生产环境建议设为 1避免日志爆炸json_modeTrue强制 ansible-runner 输出结构化 JSONTrue是解析r.stats和r.events的前提注意ansible-runner默认使用当前用户环境执行因此需确保venv中已安装ansiblepip install ansible且inventory目录下有合法的 SSH 密钥或密码配置。若用密码登录需在inventory中启用ansible_ssh_pass并设置--ask-pass不推荐应优先用密钥。2.3 权限控制谁能在 Web 上触发哪个 Playbook单纯异步还不够——必须限制用户操作范围。Django 的Group和Permission机制可直接映射到 Playbook 级别定义自定义权限models.pyfrom django.db import models class Playbook(models.Model): name models.CharField(max_length100, uniqueTrue) # 如 deploy_redis path models.CharField(max_length255) # 相对路径如 playbooks/deploy_redis.yml description models.TextField() is_active models.BooleanField(defaultTrue) class Meta: permissions [ (can_run_deploy_redis, Can run deploy_redis playbook), (can_run_backup_mysql, Can run backup_mysql playbook), ]在视图中校验权限views.pyfrom django.contrib.auth.decorators import login_required from django.core.exceptions import PermissionDenied login_required def safe_trigger_playbook(request): if request.method POST: data json.loads(request.body) playbook_name data.get(playbook_name) # 如 deploy_redis # 查询 Playbook 对象并检查用户权限 try: pb Playbook.objects.get(nameplaybook_name) perm_codename fcan_run_{playbook_name} if not request.user.has_perm(fmyapp.{perm_codename}): raise PermissionDenied(You dont have permission to run this playbook) except Playbook.DoesNotExist: return JsonResponse({error: Playbook not found}, status404) # 权限通过提交任务 task run_ansible_playbook.delay(pb.path, inventories/production) return JsonResponse({task_id: task.id})后台分配权限Django Admin在/admin/auth/group/中创建组如devops_team为其勾选Can run deploy_redis权限再将用户加入该组。无需修改代码即可动态调整权限策略。3. 前端如何实时渲染 Ansible 执行日志WebSocket EventSource 双方案对比3.1 为什么轮询Polling不适合 Ansible 日志流Ansible 执行时间跨度大秒级到分钟级且日志是逐行输出的。若用 AJAX 每 2 秒轮询一次/api/task-status/id/会产生大量无效请求且无法保证日志顺序——因为stdout是缓冲区可能整块返回而非逐行。真正的解决方案是服务端推送。3.2 方案一Django Channels WebSocket推荐用于生产Channels是 Django 官方支持的异步通信扩展可将 Celery 任务事件实时推送到前端。安装与配置pip install channels redis在settings.py中添加INSTALLED_APPS [channels] ASGI_APPLICATION myproject.asgi.application CHANNEL_LAYERS { default: { BACKEND: channels_redis.core.RedisChannelLayer, CONFIG: { hosts: [(127.0.0.1, 6379)], }, }, }创建消费者consumers.pyimport json from channels.generic.websocket import AsyncWebsocketConsumer from asgiref.sync import async_to_sync from celery.result import AsyncResult class AnsibleLogConsumer(AsyncWebsocketConsumer): async def connect(self): self.task_id self.scope[url_route][kwargs][task_id] await self.channel_layer.group_add(ftask_{self.task_id}, self.channel_name) await self.accept() async def disconnect(self, close_code): await self.channel_layer.group_discard(ftask_{self.task_id}, self.channel_name) async def receive(self, text_data): pass # 本例只接收不发送 # Celery 任务通过 channel_layer 发送事件 async def task_log(self, event): await self.send(text_datajson.dumps({ type: log, data: event[log_line] }))Celery 任务中推送日志tasks.py更新from channels.layers import get_channel_layer from asgiref.sync import async_to_sync shared_task(bindTrue) def run_ansible_playbook(self, playbook_path, inventory_path, extra_varsNone): # ... 执行 ansible-runner ... # 每捕获一个 event推送到 WebSocket 组 channel_layer get_channel_layer() for event in r.events: if stdout in event and event[stdout].strip(): async_to_sync(channel_layer.group_send)( ftask_{self.request.id}, { type: task_log, log_line: event[stdout].strip() } )前端连接JavaScriptconst taskId abc123; const ws new WebSocket(ws://localhost:8000/ws/logs/${taskId}/); ws.onmessage function(e) { const data JSON.parse(e.data); if (data.type log) { document.getElementById(log-output).innerHTML div classlog-line${data.data}/div; } };3.3 方案二Server-Sent EventsSSE——轻量级备选若暂不引入 Channels可用 Django 的 StreamingHttpResponse 实现 SSE视图views.pyfrom django.http import StreamingHttpResponse import json from celery.result import AsyncResult def stream_task_logs(request, task_id): def event_stream(): task AsyncResult(task_id) while not task.ready(): if task.info and hasattr(task.info, get): log_lines task.info.get(log_lines, []) for line in log_lines: yield fdata: {json.dumps(line)}\n\n time.sleep(1) # 最终结果 yield fdata: {json.dumps({status: task.status, result: task.result})}\n\n return StreamingHttpResponse( event_stream(), content_typetext/event-stream )前端监听JavaScriptconst eventSource new EventSource(/api/logs/${taskId}/); eventSource.onmessage function(event) { const log JSON.parse(event.data); document.getElementById(log-output).innerHTML div classlog-line${log.stdout || log.message}/div; };方案对比表维度WebSocketChannelsServer-Sent EventsSSE双向通信✅ 支持客户端发指令如取消任务❌ 仅服务端→客户端单向连接开销较低长连接复用中等HTTP/1.1 持久连接浏览器兼容性IE10现代浏览器全支持IE 不支持需 polyfillDjango 集成复杂度需配置 ASGI、Redis、Channels仅需视图函数零额外依赖适用场景生产环境需交互控制快速验证、教学演示、低并发提示本毕设项目默认采用 SSE 方案因其无需额外中间件StreamingHttpResponse在 Django 4.x 中稳定可靠且足够覆盖“查看执行日志”这一核心需求。若后续需支持“终止任务”“重试失败步骤”再平滑升级至 WebSocket。4. Playbook 设计规范与 Django 模板集成让 YAML 文件可被 Web 动态渲染4.1 Playbook 不是黑盒——它必须可读、可配、可审计很多毕设把playbook.yml当作静态文件扔进项目导致用户无法在 Web 界面看到剧本干了什么只有“执行中”无法动态修改变量如目标端口、版本号审计时无法追溯某次执行对应的 exact YAML 内容。正确做法将 Playbook 拆解为元数据描述 Jinja2 模板 变量注入三层结构。示例playbooks/deploy_nginx.yml.j2Jinja2 模板--- - name: Deploy Nginx on {{ target_os }} hosts: {{ target_hosts }} become: true vars: nginx_port: {{ nginx_port | default(80) }} nginx_version: {{ nginx_version | default(1.24) }} tasks: - name: Install nginx package ansible.builtin.apt: name: nginx{{ nginx_version }} state: present when: target_os ubuntu - name: Configure nginx port ansible.builtin.lineinfile: path: /etc/nginx/sites-available/default regexp: ^listen.*; line: listen {{ nginx_port }} default_server; notify: restart nginx handlers: - name: restart nginx ansible.builtin.service: name: nginx state: restarted元数据文件playbooks/deploy_nginx.yml.meta{ name: deploy_nginx, title: 部署 Nginx 服务, description: 在 Ubuntu/CentOS 主机上安装指定版本 Nginx 并配置监听端口, required_vars: [ {name: target_hosts, label: 目标主机, type: text, default: webservers}, {name: nginx_port, label: 监听端口, type: number, default: 80}, {name: nginx_version, label: Nginx 版本, type: text, default: 1.24} ], allowed_inventories: [staging, production] }4.2 Django 视图动态渲染 Playbook 表单加载元数据并生成表单views.pyimport json import os from django.shortcuts import render from django.conf import settings def playbook_form(request, playbook_name): meta_path os.path.join(settings.BASE_DIR, playbooks, f{playbook_name}.yml.meta) with open(meta_path, r) as f: meta json.load(f) # 构建初始表单数据 initial_data {} for var in meta[required_vars]: initial_data[var[name]] var.get(default, ) return render(request, playbook_form.html, { meta: meta, initial_data: initial_data, playbook_name: playbook_name })模板playbook_form.htmlh2{{ meta.title }}/h2 p{{ meta.description }}/p form idplaybook-form {% for var in meta.required_vars %} div classform-group label{{ var.label }}/label input type{{ var.type }} name{{ var.name }} value{{ initial_data|default_if_none:|default:var.default }} required /div {% endfor %} button typesubmit执行/button /form script document.getElementById(playbook-form).onsubmit async function(e) { e.preventDefault(); const formData new FormData(this); const data Object.fromEntries(formData); const res await fetch(/api/trigger/, { method: POST, headers: {Content-Type: application/json}, body: JSON.stringify({ playbook_name: {{ playbook_name }}, extra_vars: data }) }); const result await res.json(); window.location.href /task/status/{{ playbook_name }}?task_id${result.task_id}; }; /script4.3 执行时生成可审计的 YAML 快照为满足“每次执行都有据可查”在任务启动前用 Jinja2 渲染出本次执行的完整 YAML并存入数据库from jinja2 import Environment, FileSystemLoader from django.conf import settings def render_playbook_template(playbook_name, extra_vars): env Environment(loaderFileSystemLoader(os.path.join(settings.BASE_DIR, playbooks))) template env.get_template(f{playbook_name}.yml.j2) rendered template.render(**extra_vars) # 保存快照到 media 目录供审计下载 snapshot_path os.path.join(settings.MEDIA_ROOT, snapshots, f{playbook_name}_{int(time.time())}.yml) os.makedirs(os.path.dirname(snapshot_path), exist_okTrue) with open(snapshot_path, w) as f: f.write(rendered) return snapshot_path调用位置在run_ansible_playbook任务开头将snapshot_path作为任务元数据存入AsyncResult前端可通过/api/snapshot/task_id/下载本次执行的精确 YAML。5. 毕设答辩高频问题与防御性代码实践从“能跑”到“能讲清楚”5.1 答辩必问为什么选 Ansible 而不是 SaltStack / Puppet这不是技术选型题而是考察你是否理解工具定位。标准回答应包含三层学习成本Ansible 无 agent纯 SSHPython 编写与本项目 Django 技术栈天然契合SaltStack 需部署 minionPuppet 用 Ruby DSL增加额外学习负担生态适配ansible-runner提供成熟 Python API而 Salt 的salt-runPython 封装文档稀疏Puppet 的 REST API 需额外启服务毕设边界Ansible Playbook 本质是 YAML Jinja2易于在 Django 模板中解析和渲染符合“Web 化运维”的核心命题Salt/Puppet 更侧重大规模集群超出本科毕设复杂度。提示若被追问“Ansible 性能差怎么办”可答“本系统定位是中小规模 IT 环境50 台服务器Ansible 的串行执行完全满足若需并行可在 Playbook 中用strategy: free或拆分host_pattern非架构瓶颈。”5.2 安全红线如何防止用户通过 Web 注入恶意 Ansible 变量extra_vars是最大攻击面。例如若用户传入{host: localhost; rm -rf /}虽 Ansible 不会执行 shell 命令但若你在 Playbook 中用shell模块拼接变量就构成命令注入。防御三原则白名单变量名在*.yml.meta中严格定义required_vars后端只接受列表中声明的 key类型强校验对number类型字段在 Django 视图中用int()强转并捕获ValueError禁用危险模块在ansible.cfg中设置[defaults] allow_world_readable_tmpfiles False remote_tmp ~/.ansible/tmp local_tmp /tmp/.ansible/tmp示例校验代码views.pydef validate_extra_vars(playbook_name, user_vars): meta_path os.path.join(settings.BASE_DIR, playbooks, f{playbook_name}.yml.meta) with open(meta_path) as f: meta json.load(f) allowed_keys {var[name] for var in meta[required_vars]} for key in user_vars.keys(): if key not in allowed_keys: raise ValueError(fInvalid variable: {key}) # 类型校验 for var in meta[required_vars]: val user_vars.get(var[name]) if var[type] number: try: int(val) except (TypeError, ValueError): raise ValueError(f{var[label]} must be a number) return user_vars5.3 一个能立刻加分的细节用 Django Admin 管理 Inventory不要手动编辑inventories/production文件。在admin.py中注册Inventory模型from django.contrib import admin from .models import Inventory, Host admin.register(Inventory) class InventoryAdmin(admin.ModelAdmin): list_display (name, description, is_active) prepopulated_fields {slug: (name,)} admin.register(Host) class HostAdmin(admin.ModelAdmin): list_display (hostname, ip_address, inventory, is_active) list_filter (inventory, is_active) search_fields (hostname, ip_address)这样答辩时可现场演示在后台新增一台测试服务器勾选“加入 production Inventory”刷新 Web 页面该主机即出现在 Playbook 执行目标列表中——直观体现“配置即代码”的闭环。最后检查你的requirements.txt是否包含精确版本Django4.2.13 ansible8.7.0 ansible-runner2.3.3 celery5.3.6 redis4.6.0版本锁定是工程化基本素养也是答辩老师一眼能看出你是否真部署过的关键证据。本文还有配套的精品资源点击获取