简介这是一套面向计算机专业毕业设计场景的自动化运维平台源码基于Python与Django构建适合正在准备毕设、需要完整Web项目参考的学生以及希望入门运维自动化的开发者。项目围绕服务器监控、日志分析、任务调度、配置管理、权限控制与RESTful API等模块展开将Python的paramiko、psutil等库与Django的MVT架构结合覆盖从界面交互到后端数据存储的完整链路。压缩包共1132个文件约13.78MB其中191个py文件承载核心业务逻辑223个html与144个css、184个js构成前端页面另含png、gif等静态资源及txt、md说明文档目录结构清晰便于按模块拆解学习。已有312人浏览学习。对于毕设选题、答辩演示或二次开发而言这份源码可作为可运行的参考方案帮助理解运维平台的功能划分与实现思路也能为Python Web开发实践提供较完整的案例支撑。1. 从一份 Django 运维平台源码说起它到底能替你省下哪几小时凌晨两点被告警叫醒登机器、看日志、重启服务、截图发群这套动作重复到第三遍时人就会开始想能不能把这些操作收进一个网页里点一下就跑完。基于 Python Django 的自动化运维平台解决的正是这件事——把散落在各台机器上的脚本、巡检项、发布流程收拢成一个有权限、有记录、有界面的 Web 系统。它适合已经会写点 Python、但每次操作还靠手工 SSH 的运维和开发也适合想拿一个真实项目练 Django 的后端新手。核心不是 Django 本身多强而是它把「任务下发、结果回传、执行留痕」这条链路串起来了。下面按我实际搭过的一套思路从环境到任务执行再到踩坑一步步拆开讲。2. 环境与项目骨架把 Django 运维平台在本地跑起来2.1 为什么选 Django 而不是 Flask 或 FastAPI做运维平台绕不开三件事用户和权限、后台管理、ORM 操作数据库。Django 自带 admin、认证系统和成熟的 ORM这三样开箱即用省掉大量胶水代码。Flask 更轻但权限和后台要自己拼FastAPI 异步性能好适合做纯 API 网关可一旦要带页面和运营后台还是得补一堆东西。运维平台的使用者通常就几十人并发压力集中在任务执行而非页面渲染Django 的同步模型完全够用。常见做法是把耗时任务丢给 Celery 或后台线程Web 层只负责接收请求和展示结果这样 Django 的短板就被绕开了。选型确定后目录结构要提前想清楚。我一般按「配置、应用、任务、工具」四块分config放 settings 和 urlsapps下按功能拆hosts、tasks、accountsutils放 SSH 封装和通用函数。这样后期加模块不会全堆在一个 app 里。2.2 本地跑通的最小命令序列先确认 Python 版本。Django 4.x 要求 Python 3.8 以上我习惯用 3.10 或 3.11兼容性和库支持都稳。Windows 和 Linux 安装 Python 的步骤不同Windows 记得勾选「Add Python to PATH」Linux 一般自带或走包管理器。装完验证python --version pip --version接着建虚拟环境并装依赖。虚拟环境是后悔药装崩了直接删掉重建不污染系统环境python -m venv venv # Windows venv\Scripts\activate # Linux / macOS source venv/bin/activate pip install django paramiko celery redisdjango是框架本体paramiko负责 SSH 连接远程主机celery和redis用于异步任务和结果存储。如果只是先跑通页面celery和redis可以后装但任务执行模块迟早要用。创建项目和应用django-admin startproject config . python manage.py startapp hosts python manage.py startapp tasks注意startproject后面那个点它让项目文件直接生成在当前目录避免多套一层文件夹。hosts管主机资产tasks管任务模板和执行记录这是运维平台最核心的两个 app。2.3 settings 里必须改的几处配置打开config/settings.py以下几项不改后面必翻车# 注册应用 INSTALLED_APPS [ django.contrib.admin, django.contrib.auth, django.contrib.contenttypes, django.contrib.sessions, django.contrib.messages, django.contrib.staticfiles, hosts, tasks, ] # 数据库开发阶段用 sqlite上线换 mysql 或 postgresql DATABASES { default: { ENGINE: django.db.backends.sqlite3, NAME: BASE_DIR / db.sqlite3, } } # 中文和时区 LANGUAGE_CODE zh-hans TIME_ZONE Asia/Shanghai USE_TZ False # 静态文件 STATIC_URL /static/ STATIC_ROOT BASE_DIR / staticfilesUSE_TZ False是为了让任务执行时间直接按本地时间存省得每次展示还要转换。STATIC_ROOT是collectstatic的收集目录上线时 Nginx 指向它。数据库开发阶段用 sqlite 足够数据量上来或要多人协作再换 MySQL改ENGINE和连接参数即可ORM 层代码不用动。配置完执行迁移生成 Django 自带表python manage.py migrate python manage.py createsuperuser python manage.py runserver浏览器打开127.0.0.1:8000/admin能登录就说明骨架通了。这一步看着简单但环境变量、Python 路径、端口占用经常在这里出问题后面避坑章节会细说。3. 主机资产与 SSH 执行运维平台的手和眼3.1 主机模型怎么设计才够用主机表不是越全越好字段多了录入成本高少了又不够用。我一般保留这几列别名、IP、SSH 端口、认证方式、用户名、密钥或密码、分组、备注。认证方式分密码和密钥两种密钥更安全但分发麻烦密码方便但要在数据库里加密存。Django 里可以用django-fernet-fields或自己写个加解密工具别明文存密码。# hosts/models.py from django.db import models class Host(models.Model): AUTH_CHOICES ( (password, 密码), (key, 密钥), ) alias models.CharField(别名, max_length64, uniqueTrue) ip models.GenericIPAddressField(IP地址) port models.IntegerField(SSH端口, default22) username models.CharField(用户名, max_length64, defaultroot) auth_type models.CharField(认证方式, max_length16, choicesAUTH_CHOICES, defaultpassword) password models.CharField(密码, max_length256, blankTrue) private_key models.TextField(私钥, blankTrue) group models.CharField(分组, max_length64, blankTrue) remark models.CharField(备注, max_length256, blankTrue) class Meta: verbose_name 主机 verbose_name_plural verbose_name def __str__(self): return f{self.alias}({self.ip})alias加uniqueTrue防止重名GenericIPAddressField自带格式校验。password和private_key都留空允许因为一台主机只用一种认证方式。group用普通字符字段而不是外键是为了让分组可以随手填不用先去建分组表后期要规范化再迁。写完模型执行python manage.py makemigrations hosts python manage.py migrate然后在hosts/admin.py注册就能在 admin 里录主机了from django.contrib import admin from .models import Host admin.register(Host) class HostAdmin(admin.ModelAdmin): list_display (alias, ip, port, username, auth_type, group) search_fields (alias, ip, group) list_filter (auth_type, group)list_display决定列表页显示哪些列search_fields支持按别名和 IP 搜索list_filter右侧出筛选器。主机一多没有搜索和筛选根本没法用。3.2 用 paramiko 封装一个能复用的执行函数SSH 执行是平台的核心动作封装得好后面任务模块直接调封装得差每加一个功能就复制一遍连接代码。我一般写一个utils/ssh.py# utils/ssh.py import paramiko def run_command(host, command, timeout30): 在远程主机执行命令返回 (是否成功, 输出内容) host: Host 模型实例 command: 要执行的 shell 命令 timeout: 命令超时秒数 client paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: if host.auth_type key: key paramiko.RSAKey.from_private_key_file(host.private_key) client.connect(host.ip, porthost.port, usernamehost.username, pkeykey, timeout10) else: client.connect(host.ip, porthost.port, usernamehost.username, passwordhost.password, timeout10) stdin, stdout, stderr client.exec_command(command, timeouttimeout) out stdout.read().decode(utf-8, errorsignore) err stderr.read().decode(utf-8, errorsignore) exit_code stdout.channel.recv_exit_status() return exit_code 0, out err except Exception as e: return False, str(e) finally: client.close()AutoAddPolicy自动接受未知主机指纹方便但降低了安全性内网环境可接受公网机器建议换成手动确认或预置 known_hosts。timeout分两层connect的 10 秒是建连超时exec_command的 timeout 是命令执行超时两个都要设否则卡死会拖垮整个请求。recv_exit_status拿退出码判断命令是否真的成功只看输出有没有内容是不够的。errorsignore防止远程输出里有非 UTF-8 字符导致解码报错。这个函数返回元组而不是抛异常是为了让调用方自己决定怎么处理失败任务记录里能存下错误信息。3.3 把执行结果落库任务记录表设计光执行不记录出了问题就是黑匣子。任务记录表至少要有关联主机、执行的命令、开始时间、结束时间、是否成功、输出内容、触发人。# tasks/models.py from django.db import models from django.contrib.auth.models import User from hosts.models import Host class TaskRecord(models.Model): host models.ForeignKey(Host, on_deletemodels.CASCADE, verbose_name主机) command models.TextField(命令) success models.BooleanField(是否成功, defaultFalse) output models.TextField(输出, blankTrue) created_by models.ForeignKey(User, on_deletemodels.SET_NULL, nullTrue, verbose_name触发人) created_at models.DateTimeField(开始时间, auto_now_addTrue) finished_at models.DateTimeField(结束时间, nullTrue, blankTrue) class Meta: verbose_name 任务记录 verbose_name_plural verbose_name ordering [-created_at] def __str__(self): return f{self.host.alias} - {self.command[:20]}on_deletemodels.CASCADE表示主机删了记录也删如果想让记录保留改成SET_NULL并把host设为可空。created_by用SET_NULL用户注销后记录还在只是触发人变空。ordering [-created_at]让列表默认按时间倒序最新的在最上面符合运维查看习惯。视图里串起来# tasks/views.py from django.shortcuts import render, redirect from django.contrib.auth.decorators import login_required from django.utils import timezone from .models import TaskRecord from hosts.models import Host from utils.ssh import run_command login_required def execute(request): if request.method POST: host_id request.POST.get(host_id) command request.POST.get(command, ).strip() if not host_id or not command: return render(request, tasks/execute.html, {hosts: Host.objects.all(), error: 主机和命令不能为空}) host Host.objects.get(idhost_id) record TaskRecord.objects.create(hosthost, commandcommand, created_byrequest.user) success, output run_command(host, command) record.success success record.output output record.finished_at timezone.now() record.save() return redirect(task_list) return render(request, tasks/execute.html, {hosts: Host.objects.all()})login_required保证只有登录用户能执行这是最低限度的权限控制。先建记录再执行是为了即使执行过程中进程崩了也能看到有一条未完成记录而不是什么都没留下。执行完更新成功状态、输出和结束时间。真实场景里这段应该丢给 Celery 异步跑否则命令一慢页面就转圈用户体验很差。4. 任务编排与异步执行别让页面卡在一条命令上4.1 为什么同步执行迟早要换掉上面那个视图命令跑 30 秒浏览器就等 30 秒Nginx 或 Gunicorn 的超时一过直接 502。运维命令里yum update、tar打包、日志清理随便一个都超过这个时间。所以任务执行必须异步化Web 层只负责创建任务、返回任务 ID后台 worker 去跑前端轮询或走 WebSocket 拿结果。Celery Redis 是最常见的组合。Redis 既做 broker 又做 result backend一套顶两套。安装后建config/celery.py# config/celery.py import os from celery import Celery os.environ.setdefault(DJANGO_SETTINGS_MODULE, config.settings) app Celery(ops) app.config_from_object(django.conf:settings, namespaceCELERY) app.autodiscover_tasks()namespaceCELERY表示 settings 里所有CELERY_开头的配置都归 Celery 管。autodiscover_tasks()自动去每个 app 下找tasks.py不用手动注册。settings 里加CELERY_BROKER_URL redis://127.0.0.1:6379/0 CELERY_RESULT_BACKEND redis://127.0.0.1:6379/1 CELERY_TASK_SERIALIZER json CELERY_RESULT_SERIALIZER json CELERY_TIMEZONE Asia/Shanghaibroker 用 0 号库result 用 1 号库分开避免键冲突。序列化统一用 json比 pickle 安全跨语言也方便。4.2 把执行逻辑搬进 Celery 任务在tasks/tasks.py里定义异步任务# tasks/tasks.py from celery import shared_task from django.utils import timezone from .models import TaskRecord from utils.ssh import run_command shared_task def async_run_command(record_id): record TaskRecord.objects.get(idrecord_id) success, output run_command(record.host, record.command) record.success success record.output output record.finished_at timezone.now() record.save() return {success: success, output: output[:500]}shared_task让任务不依赖具体 app 实例方便复用。传record_id而不是整个对象是因为 Celery 序列化模型实例容易出问题传主键最稳。返回输出截断到 500 字符避免结果后端存太大。视图改成login_required def execute(request): if request.method POST: host_id request.POST.get(host_id) command request.POST.get(command, ).strip() if not host_id or not command: return render(request, tasks/execute.html, {hosts: Host.objects.all(), error: 主机和命令不能为空}) host Host.objects.get(idhost_id) record TaskRecord.objects.create(hosthost, commandcommand, created_byrequest.user) async_run_command.delay(record.id) return redirect(task_list) return render(request, tasks/execute.html, {hosts: Host.objects.all()}).delay()把任务丢进队列立刻返回页面不再等待。启动 workercelery -A config worker -l infoWindows 上 Celery 4 之后对多进程支持不好要加-P solo参数celery -A config worker -l info -P solo这是血泪经验不加会报PermissionError或直接卡住不执行。4.3 批量任务与并发控制单机执行跑通后下一步是批量。比如给一个分组下所有主机统一执行一条命令。做法是循环创建记录再逐个丢进 Celeryshared_task def batch_run(host_ids, command, user_id): from hosts.models import Host from django.contrib.auth.models import User user User.objects.get(iduser_id) for hid in host_ids: host Host.objects.get(idhid) record TaskRecord.objects.create(hosthost, commandcommand, created_byuser) async_run_command.delay(record.id)批量任务要注意并发上限。几十台机器同时 SSH本机文件描述符和网络连接会吃紧。Celery 可以用--concurrency限制 worker 并发数也可以在任务里加信号量。我一般把并发控制在 10 到 20 之间具体看机器配置和目标主机承受能力。目标主机如果是老旧服务器SSH 并发太高会被 sshd 拒绝表现为Connection reset by peer这时候降并发比调超时有用。5. 避坑与排查这套平台最容易翻车的五个地方5.1 现象页面报 DisallowedHostadmin 打不开原因ALLOWED_HOSTS没配。Django 默认只允许localhost和127.0.0.1用服务器 IP 或域名访问就被拦。解决settings 里加ALLOWED_HOSTS [*]仅限开发上线必须写具体域名或 IP比如[ops.example.com, 10.0.0.5]。用*上线等于把 Host 头校验关了有安全风险。5.2 现象paramiko 连接超时但手动 SSH 能连上原因多半是目标主机端口不是 22或者防火墙只放行了特定来源。还有一种情况是密钥格式不对paramiko 只认 RSA/DSA/ECDSA 等特定格式OpenSSH 新格式的私钥可能读不了。解决先确认端口再在平台所在机器上用ssh -i key userip -p port手动验证。密钥问题用ssh-keygen -p -m PEM -f your_key转成 PEM 格式paramiko 就能读了。连接超时设 10 秒别设太长否则批量任务里一台卡住拖慢整批。5.3 现象Celery 任务一直 pendingworker 没反应原因broker 连不上或者 worker 没启动或者任务没被 autodiscover 到。Windows 上还多一个多进程问题。解决先redis-cli ping确认 Redis 活着再确认 worker 启动命令带没带-P solo。任务没被发现检查tasks.py是否在已注册的 app 目录下函数是否带shared_task。用celery -A config inspect registered能看到当前 worker 注册了哪些任务。5.4 现象命令执行成功但输出为空原因有些命令把内容输出到 stderr或者命令本身是交互式的exec_command拿不到。还有编码问题远程输出 GBK 编码用 UTF-8 解码就乱码或空。解决输出拼接out err已经覆盖 stderr。交互式命令要加参数让它非交互比如yum -y、apt-get -y。编码问题在decode时指定errorsignore或尝试gbk解码。判断成功与否看退出码不看输出长度。5.5 现象数据库密码明文存被审计挑出来原因图省事直接CharField存了明文密码。解决用对称加密存读取时解密。简单做法是引入cryptography库写个encrypt/decrypt工具密钥放环境变量不写进代码。更规范的是接 Vault 之类的密钥管理但小团队用对称加密加环境变量已经够。admin 里展示密码字段时用readonly_fields别让它在列表页直接显示。6. 进阶把任务模板和定时巡检做成可复用的资产平台跑通单次执行后真正省时间的是把常用操作沉淀成模板。我一般建一张TaskTemplate表存模板名、命令内容、适用分组、参数占位符。执行时选模板、填参数、选主机一键下发。占位符用{param}形式执行前用str.format替换比让用户手敲命令靠谱得多。class TaskTemplate(models.Model): name models.CharField(模板名, max_length128, uniqueTrue) command models.TextField(命令模板) group models.CharField(适用分组, max_length64, blankTrue) description models.CharField(说明, max_length256, blankTrue) def render(self, **kwargs): return self.command.format(**kwargs)定时巡检用 Celery Beat。settings 里配CELERY_BEAT_SCHEDULE或者用 django-celery-beat 在 admin 里动态加周期任务。巡检结果建议单独建表按天分区或定期清理不然记录表几个月就膨胀到几百万行查询变慢。清理用一条TaskRecord.objects.filter(created_at__ltcutoff).delete()挂到定时任务里比手动删省心。验证平台是否可靠我的习惯是拿一台测试机故意制造失败场景关掉 SSH、写错命令、断网看记录表里是否如实记下失败和错误信息。能正确记录失败的平台才敢用在生产上。这套东西我从最早的手工脚本一路改到现在的 Django 版本最大的教训是别一上来就追求功能全先把「执行—记录—查看」这条最小闭环跑稳再往上加模板、加定时、加权限。骨架不稳功能越多越难维护。希望帮到你。本文还有配套的精品资源点击获取