企业官网建设流程全解析

简介：这是一份基于Django开发的运维管理系统平台源代码，面向运维工程师和Django学习者，用来解决操作日志集中收集、设备远程管理与端口批量探测等日常运维需求。整个资源包共389个文件，以Python源码、JavaScript/CSS静态资源及HTML模板为主，同时包含SQLite数据库文件、配置文件和第三方库依赖，压缩包仅13.89MB，结构清晰便于快速部署学习。系统实现了操作日志的简要与详细展示、设备的增删改查、基于传统方式的设备登录配置与维护，并利用socket和asyncio实现异步端口扫描；前端基于Bootstrap与Django-SimpleUI搭建，后端集成paramiko、telnetlib等库，技术栈清晰完整。通过学习这份代码，可以掌握Django项目结构、异步编程、设备交互及后台管理界面的定制方法，适合作为运维自动化开发的实战参考。目前已有518人学习下载。

1. 基于Django的运维管理系统源代码包，拿到手先看这四个模块

一个「基于Django的运维管理系统」源代码包解压后，通常是完整的 Django 工程：manage.py、requirements.txt、几个 app，外加一个体积不小的 db.sqlite3。这类项目的核心价值不在代码行数，而在四个模块的配合——设备台账的增删改查、操作日志的自动收集、端口扫描任务的异步化，以及初始化好的数据库文件。直接 runserver 能看到页面，但那只是把别人做好的轮子转起来；真正有价值的是拆开设备模型看字段设计、顺着中间件和信号捋清日志链路、把 Celery 任务和 Redis 队列接起来之后的取舍。下面按读代码和二次开发的顺序，把这四块的实现思路、可复现命令和踩坑点写清楚，适合想改成公司内部资产平台的工程师，也适合面试前快速过一遍典型的 Django 运维系统设计。

2. 设备增删改查怎么落地：Device 模型、通用视图与操作日志串起来的完整链路

2.1 拆设备模型：字段类型、choices 和索引决定后续功能怎么挂

读这类源码的第一步是打开device/models.py，把设备表字段逐个过一遍。常见做法是建一张Device表，字段覆盖名称、管理 IP、设备类型、状态、机房位置、备注和两个时间戳，下面这份是典型写法：

# device/models.py from django.db import models class Device(models.Model): DEVICE_TYPE = ( ('switch', '交换机'), ('router', '路由器'), ('server', '服务器'), ('firewall', '防火墙'), ) STATUS = ( ('online', '在线'), ('offline', '离线'), ('maintain', '维护中'), ) name = models.CharField(max_length=64, verbose_name='设备名称') ip_address = models.GenericIPAddressField(unique=True, verbose_name='管理IP') device_type = models.CharField(max_length=16, choices=DEVICE_TYPE, default='server') status = models.CharField(max_length=16, choices=STATUS, default='offline') location = models.CharField(max_length=128, blank=True, verbose_name='机房位置') remark = models.TextField(blank=True, verbose_name='备注') created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: db_table = 'ops_device' ordering = ['-created_at'] indexes = [models.Index(fields=['ip_address'])] def __str__(self): return f'{self.name} ({self.ip_address})'

ip_address用GenericIPAddressField而不是CharField，Django 会在表单校验和 ORM 反序列化两层做 IPv4/IPv6 合法性验证，省掉手写正则。unique=True防止同一台设备被录进去两次，这是资产类系统最容易被忽略的约束。device_type和status用choices而不是外键，因为枚举项变化频率低，建外键只会平白多出两张表和关联查询。created_at用auto_now_add只在插入时写入，updated_at用auto_now每次save()都会刷新，但要注意queryset.update()批量更新不会触发auto_now，审计依赖时间戳时要留意这个差异。

大部分此类源码还会把 admin 暴露出来做快速演示。admin.py里如果只写了admin.site.register(Device)，可以补成带list_display和list_filter的配置，后台界面会好维护得多：

# device/admin.py from django.contrib import admin from .models import Device @admin.register(Device) class DeviceAdmin(admin.ModelAdmin): list_display = ('name', 'ip_address', 'device_type', 'status', 'location') list_filter = ('device_type', 'status') search_fields = ('name', 'ip_address') list_editable = ('status',) list_per_page = 20

search_fields会生成基于icontains的搜索条件，list_editable允许在列表页直接改状态，这两个配置在有人手动录入几十台设备时能省大量点击。设备模型定下来之后，增删改查的视图层才有东西可操作。

2.2 用通用视图搭 CRUD：函数视图和类视图怎么选

读源码时会看到两种风格：老一点的项目用def device_add(request)手写函数视图，请求方法判断、表单实例化、重定向全挤在一个函数里；新一点的项目用CreateView、UpdateView这类通用视图。函数视图可读性好、适合改内部逻辑，但增删改查这种高度模板化的接口，类视图维护成本更低。如果拿到的是「基于Django的运维管理系统」函数视图版本，二次开发时建议逐步迁到通用视图，模板目录不用动，只换视图层的写法：

# device/views.py from django.contrib.auth.mixins import LoginRequiredMixin from django.urls import reverse_lazy from django.views.generic import ListView, CreateView, UpdateView, DeleteView from .models import Device from .forms import DeviceForm class DeviceList(LoginRequiredMixin, ListView): model = Device template_name = 'device/list.html' context_object_name = 'devices' paginate_by = 20 class DeviceCreate(LoginRequiredMixin, CreateView): model = Device form_class = DeviceForm template_name = 'device/form.html' success_url = reverse_lazy('device:list') class DeviceUpdate(LoginRequiredMixin, UpdateView): model = Device form_class = DeviceForm template_name = 'device/form.html' success_url = reverse_lazy('device:list') class DeviceDelete(LoginRequiredMixin, DeleteView): model = Device template_name = 'device/confirm_delete.html' success_url = reverse_lazy('device:list')

# device/urls.py from django.urls import path from . import views app_name = 'device' urlpatterns = [ path('', views.DeviceList.as_view(), name='list'), path('add/', views.DeviceCreate.as_view(), name='add'), path('<int:pk>/edit/', views.DeviceUpdate.as_view(), name='edit'), path('<int:pk>/delete/', views.DeviceDelete.as_view(), name='delete'), ]

这里把LoginRequiredMixin放在最左边，Django MRO 会保证未登录请求先被拦下来再进视图逻辑，比在函数里手写if not request.user.is_authenticated干净。success_url用reverse_lazy('device:list')而不是直接写死/device/，这样后来改路由前缀时不用回改视图。如果目标是前后端分离的项目形态，DeviceList和DeviceCreate这套思路可以直接换成 DRF 的ModelViewSet，模型和表单校验逻辑不需要推翻。

2.3 操作日志收集：中间件、Signal 和装饰器三条路对比

运维管理系统的操作日志收集一般要回答三个问题：谁在什么时间改了什么、改之前的值是什么、请求来自哪个 IP。读源码时最常见的实现有三种，落地时按记录粒度和侵入程度选。

post_saveSignal 方案记录的是模型层变更，不依赖 HTTP 视图，任何地方调用Device.objects.create()都会被记录：

# device/signals.py import json from django.db.models.signals import post_save from django.dispatch import receiver from .models import Device, OperationLog @receiver(post_save, sender=Device) def log_device_save(sender, instance, created, **kwargs): OperationLog.objects.create( model_name='Device', object_id=instance.pk, action='create' if created else 'update', detail=json.dumps({ 'name': instance.name, 'ip_address': instance.ip_address, 'status': instance.status, }, ensure_ascii=False)[:500], )

中间件方案记录的是每次 HTTP 请求，能拿到登录用户、请求路径、方法和状态码：

# opsmanage/middleware.py import json from django.utils.deprecation import MiddlewareMixin from device.models import OperationLog class AuditLogMiddleware(MiddlewareMixin): SAFE_METHODS = ('GET', 'HEAD', 'OPTIONS', 'TRACE') def process_response(self, request, response): if request.method in self.SAFE_METHODS: return response if not (request.user and request.user.is_authenticated): return response OperationLog.objects.create( user=request.user, method=request.method, path=request.path[:200], status_code=response.status_code, ip_address=self._get_client_ip(request), body=json.dumps(getattr(request, 'data', {}), ensure_ascii=False)[:1000], ) return response def _get_client_ip(self, request): xff = request.META.get('HTTP_X_FORWARDED_FOR') return xff.split(',')[0].strip() if xff else request.META.get('REMOTE_ADDR')

两种方案的区别可以量化对比：

Signal 方案的短板是拿不到当前登录用户，因为模型层不知道 HTTP 请求上下文。常见做法是用threading.local()在中间件里把当前用户塞进去，Signal 里再取出来，代价是要额外维护一个线程局部变量。中小规模内网系统我一般选中间件方案，把request.path和白名单路径比对，静态资源、查询接口直接放行，写库动作集中在一个位置，排查时只用盯一类代码。

2.4 日志批量落库：避免每次请求阻塞写库

中间件方案在process_response里同步建一条OperationLog，高并发下等于每次写操作都多一次 insert。设备管理页面人不多还好，一旦接了定时任务做批量巡检，日志表会迅速膨胀。可以把写库改成批量落库：请求期间把日志对象攒在threading.local()的列表里，响应结束时用bulk_create一次写一批。

# opsmanage/log_buffer.py import threading _local = threading.local() def get_logs(): if not hasattr(_local, 'logs'): _local.logs = [] return _local.logs def add_log(log): get_logs().append(log) def flush_logs(): from device.models import OperationLog logs = get_logs() if logs: OperationLog.objects.bulk_create(logs) logs.clear()

bulk_create单条 insert 改成一个多值 INSERT，SQL 次数从 N 次降到 1 次，代价是拿不到auto_now_add之外的时间戳回填和主键，日志模型里不要依赖实例主键做后续关联。批量落库的另一个好处是日志失败不影响主流程：flush_logs放在process_response最后，即使写库异常也可以吞掉并打到logging，业务请求的响应码不会因为日志失败变成 500。

3. 异步端口扫描：Celery 任务队列加并发探测的落地方案

3.1 端口扫描为什么必须挪出请求线程

端口扫描本质上是对目标 IP 的一组 TCP connect 探测。用socket.connect_ex((ip, port))判断端口是否开放，每个端口都要等一个超时时间：探测 10 个端口、超时设 1 秒，最坏情况串行要 10 秒；探测 100 个端口就是 100 秒。这段耗时如果放在 Django 视图里同步执行，前端请求卡死、worker 线程被占满，运维系统自己的管理页面也跟着不可用。所以这类源码里只要写了「异步端口扫描」，实现路径基本都是任务队列加并发 socket 探测，而不是在视图里开线程。

另一个常被忽略的点是网络状况不可控。跨机房探测时丢包会造成 socket 一直等到超时，即使每个端口只等 0.5 秒，几十个端口累计的等待也是秒级。把超时、并发数、端口列表做成任务参数，比写死在视图里更符合运维场景——有人想扫 22 到 10000 的端口段，有人只想确认 22、80、443 三个常用端口，参数不同，任务执行时间差一个数量级。

3.2 最小 Celery + Redis 配置：任务从下发到执行的标准姿势

异步任务选 Celery 加 Redis 是这个场景的默认组合。Celery 负责任务分发、重试和结果管理，Redis 同时承担 broker 和 result backend。安装依赖：

pip install celery redis

工程里的opsmanage/celery.py负责创建 Celery 实例，settings.py里的命名空间配置让 Celery 直接读 Django 配置：

# opsmanage/celery.py import os from celery import Celery os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'opsmanage.settings') app = Celery('opsmanage') app.config_from_object('django.conf:settings', namespace='CELERY') app.autodiscover_tasks()

# opsmanage/__init__.py from .celery import app as celery_app __all__ = ('celery_app',)

# settings.py 中新增 CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0' CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1' CELERY_TASK_TRACK_STARTED = True CELERY_TASK_TIME_LIMIT = 600

namespace='CELERY'表示直接读 settings 里以CELERY_开头的配置，不用再单独建 celeryconfig 文件。CELERY_TASK_TRACK_STARTED打开后任务会多出STARTED状态，方便排查任务是否真的在跑而不是卡在队列里。CELERY_TASK_TIME_LIMIT是硬性超时，扫描任务容易因为网络问题挂住，这个必须设。启动 worker：

celery -A opsmanage worker -l info

Windows 上开发调试需要加-P solo，否则 prefork 进程池在 Windows 上可能起不来。任务定义放在各 app 下的tasks.py，autodiscover_tasks()会自动注册。

3.3 扫描任务里的并发控制与超时参数：asyncio 加信号量

核心任务写法如下，用asyncio.Semaphore控制并发连接数，把阻塞的socket调用丢到线程池执行：

# device/tasks.py import asyncio import logging import socket from celery import shared_task from django.utils import timezone from .models import Device, ScanResult logger = logging.getLogger(__name__) def tcp_check(ip, port, timeout): try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.settimeout(timeout) return sock.connect_ex((ip, port)) == 0 except OSError: return False async def scan_many(ip, ports, timeout, concurrent): sem = asyncio.Semaphore(concurrent) loop = asyncio.get_running_loop() async def one(port): async with sem: ok = await loop.run_in_executor(None, tcp_check, ip, port, timeout) return port, ok results = await asyncio.gather(*(one(p) for p in ports)) return {port: ok for port, ok in results if ok} @shared_task(bind=True) def scan_device_ports(self, device_id, ports, timeout=1.0, concurrent=50): device = Device.objects.get(pk=device_id) open_ports = asyncio.run(scan_many(device.ip_address, ports, timeout, concurrent)) result, _ = ScanResult.objects.update_or_create( device=device, defaults={'open_ports': open_ports, 'scanned_at': timezone.now()} ) return result.open_ports

Semaphore保证同一时刻最多只有 50 个 socket 在等回复，避免几百个端口同时把文件描述符打满。run_in_executor把阻塞的connect_ex挪到线程池，事件循环不被卡死。返回值只保留开放端口，未开放的端口在字典里被过滤掉，减少落库体积。任务失败时bind=True让self可用，可以在self.retry(countdown=30)处做重试。

参数对扫描结果的影响可以参考这张表：

需要注意asyncio.run()在 Celery 里使用时，worker 的 pool 不能是eventlet或gevent，这两个协程池和 asyncio 的事件循环会互相干扰；prefork 和 solo pool 都安全。

3.4 扫描结果回写模型，页面上怎么查最近一次结果

扫描结果要有地方存。ScanResult用一个 JSONField 存开放端口列表，外键关联设备，每次扫描用update_or_create覆盖，保证一台设备只保留最新结果的历史入口：

# device/models.py class ScanResult(models.Model): device = models.ForeignKey(Device, on_delete=models.CASCADE, related_name='scan_results') open_ports = models.JSONField(default=dict, verbose_name='开放端口') scanned_at = models.DateTimeField(auto_now_add=True, verbose_name='扫描时间') class Meta: ordering = ['-scanned_at']

视图里下发任务时先返回 task_id，前端轮询任务状态，这是任务型操作的标准交互：

# device/views.py from django.http import JsonResponse from .tasks import scan_device_ports def trigger_scan(request, pk): ports = [int(p) for p in request.POST.getlist('ports')] or [22, 80, 443] task = scan_device_ports.delay(pk, ports) return JsonResponse({'task_id': task.id, 'status': 'queued'})

request.POST.getlist('ports')拿到的是表单里的多选值，没有传就用默认端口。task.id返回给前端后，用一个带setInterval的接口查AsyncResult.state，状态从PENDING变成SUCCESS再刷新结果列表。这套交互在运维管理系统的端口扫描页面里基本是固定打法。

4. 内含数据库文件怎么处理：SQLite 迁移、初始数据与生产切换

4.1 db.sqlite3 的正确打开方式：先迁移再核对，别直接改

标题里「内含数据库文件」指的一般是工程根目录的db.sqlite3。拿到手第一件事不是打开看数据，而是先确认数据库和当前 Django 版本的迁移状态是否一致：

python manage.py migrate --plan

migrate --plan只输出将要执行的迁移和依赖顺序，不真正执行，适合先过一眼有没有未应用的历史迁移。如果项目源码是别人打包的，很可能db.sqlite3里的django_migrations表记录和代码里的 migrations 文件对不上，常见的报错是Table 'ops_device' already exists，这时用migrate --fake-initial跳过已存在的建表语句。确认迁移状态后，再用 SQLite 自带工具核对数据：

sqlite3 db.sqlite3 ".tables" sqlite3 db.sqlite3 "select name, sql from sqlite_master where type='table' limit 10;" sqlite3 db.sqlite3 "select count(*) from ops_device;"

.tables列出所有表名，select count(*)确认设备表里有没有预置数据。如果db.sqlite3因为 Django 版本升级打不开或缺字段，可以用inspectdb把现有表反向生成模型参考，但那个输出只能当草稿，字段类型命名规则和手写模型差距很大，不建议直接复制进 models.py。

4.2 dumpdata 导出、loaddata 导入：初始数据的标准搬运法

「内含数据库文件」的价值在于预置数据，迁移到新环境时不要整包拷贝体积几十 MB 的db.sqlite3，用 Django fixture 搬运更干净：

python manage.py dumpdata device --indent 2 > device_fixture.json python manage.py loaddata device_fixture.json

dumpdata device导出deviceapp 下所有模型的数据，--indent 2让 JSON 具有可读性，方便在版本控制里 diff。只导出部分记录用--pks指定主键：

python manage.py dumpdata device.Device --pks 1,2,3 --indent 2 > core_devices.json

导入时注意两点：loaddata默认不做幂等，重复执行会插入重复数据，所以脚本里要先清空目标表；另一个是--ignorenonexistent参数，数据库里有 fixture 里不存在的字段时不会报错，只跳过。刷数据时的标准顺序是python manage.py flush清空后再loaddata，flush会重置所有自增主键和django_content_type表，比手工 delete 干净。

4.3 从 SQLite 切到 MySQL 的五个注意点

内网部署时大多数运维管理系统最终要换到 MySQL 或 PostgreSQL。切库的代码改动集中在 settings：

# settings.py DATABASES = { 'default': { 'ENGINE': 'django.db.backends.mysql', 'NAME': 'opsmanage', 'USER': 'ops', 'PASSWORD': 'your_password', 'HOST': '127.0.0.1', 'PORT': '3306', 'OPTIONS': {'charset': 'utf8mb4'}, } }

切库时最容易踩的五个坑，按顺序列出：

1. mysqlclient在 Linux 上需要系统里装libmysqlclient-dev，装不上可以用pip install pymysql，然后在__init__.py里import pymysql; pymysql.install_as_MySQLdb()作为兼容替代。
2. 先python manage.py makemigrations再migrate，把字段映射差异显式化，SQLite 的AutoField到 MySQL 会变成自增主键，这一步可以自动处理。
3. 切库前的旧数据用上一节的dumpdata，切完在 MySQL 上migrate之后再loaddata，不要在 SQLite 和 MySQL 之间做库级迁移工具转换，不用引入第三方同步工具。
4. SQLite 的布尔字段在库内是0/1，导入 MySQL 后filter(status=True)的语义不变，但JSONField在 MySQL 5.7 以下版本不支持，需要确认 MySQL 版本高于 5.7.8。
5. 数据库配置里有密码，提交代码前用环境变量或local_settings.py覆盖，不要把内网数据库口令写进仓库，这是运维管理系统代码包最常见的安全问题。

4.4 数据库文件报错或损坏时的恢复路径

本地开发环境里db.sqlite3报database disk image is malformed时，先检查物理损坏：

sqlite3 db.sqlite3 "PRAGMA integrity_check;" sqlite3 db.sqlite3 ".backup backup.sqlite3"

integrity_check返回ok表示结构完好，查询报错通常是索引或 wal 文件问题，.backup到新文件后再拿新文件跑迁移。如果项目代码和数据库文件是不同人、不同 Django 版本制作的，打开时还会遇到OperationalError: no such column: ...这类字段缺失错误，原因是db.sqlite3用的是旧版 schema 而代码里的 ORM 查询引用了新字段。这时候优先看django_migrations表和实际表结构的差异，比强行改表结构快得多；实在对不上，直接用上面 dumpdata 导出有数据的表，然后删掉库文件重新 migrate 再 loaddata，比手工补列省时间。

5. 从源码到跑通：启动顺序与任务排错的验证技巧

5.1 三分钟启动顺序清单

拿到源码先按固定顺序跑，任何一步报错都能定位到是环境问题还是代码问题：

python -m venv venv && source venv/bin/activate pip install -r requirements.txt redis-server --daemonize yes python manage.py migrate python manage.py createsuperuser python manage.py runserver 0.0.0.0:8000 celery -A opsmanage worker -l info -P solo

redis-server --daemonize yes让 Redis 后台运行，Windows 上换成redis-server --service-start。worker 启动后日志出现ready (celery@主机名)才算注册成功。

5.2 端口扫描任务不执行的排查顺序

任务delay()了但端口一直不出来，按这个顺序查：

celery -A opsmanage inspect registered celery -A opsmanage inspect active redis-cli -n 0 LLEN celery

inspect registered确认任务类有没有被autodiscover_tasks注册进去，没有的话检查tasks.py是否在 app 目录下、__init__.py是否导入了celery_app。LLEN celery看默认队列里积压了多少条消息，数字一直涨说明 worker 消费速度跟不上或 worker 根本没连上同一个 Redis 库。还有一个很容易踩的配置：settings 里如果把CELERY_TASK_ALWAYS_EAGER设成True，任务会在本地同步执行，生产环境忘了注释会掩盖异步排队的问题。

5.3 用 Django shell 手动发任务验证整条链路

不用点页面，直接在 shell 里验证任务、Broker、Worker、数据库回写这四段链路：

python manage.py shell

from device.tasks import scan_device_ports from device.models import Device d = Device.objects.first() task = scan_device_ports.delay(d.pk, [22, 80, 443, 3306]) print(task.id) from celery.result import AsyncResult r = AsyncResult(task.id) print(r.ready(), r.state, r.result)

r.ready()为False说明还在队列或执行中，r.state显示STARTED表示任务已经进入执行函数。如果.delay()后 state 一直是PENDING，问题出在 worker 没起来或 broker 连不上。调试任务内部报错时，换.apply()同步执行，异常会直接抛到 shell 里：

scan_device_ports.apply(args=[d.pk, [22, 80, 443]])

.apply()不走队列，在调用进程里直接跑任务，Device.DoesNotExist、socket 拼写错误这类问题会立刻暴露调用栈。用这个方式验证完任务本体，再上.delay()测队列链路，一套组合拳下来基本能定位 90% 的异步任务故障。

本文还有配套的精品资源，点击获取