简介这是一个基于Django构建的运维管理系统平台完整源代码面向运维工程师、Python后端开发者及Django学习者用于快速搭建具备操作日志收集、设备增删改查、传统方式设备登录配置管理以及基于socket的asyncio异步端口扫描等核心功能的运维后台。前端采用HTML/CSS/Bootstrap与少量JavaScript并集成Django-SimpleUI作为Admin后台后端使用Django内置paramiko、telnetlib、asyncio、socket等第三方模块代码结构清晰方便二次开发与学习。资源包共包含389个文件以JavaScript、CSS、HTML等前端静态文件为主同时有20个Python源码文件、13个HTML模板、17个编译后的pyc文件以及1个SQLite数据库文件含初始数据压缩包整体大小为13.89MB目录组织规范便于按模块检索。目前已有518人学习下载。使用者可直接导入SQLite数据库快速启动项目也可参照源码学习异步端口扫描、操作日志采集等实战技巧适合需要参考完整Django项目实践、了解运维自动化实现思路的开发者。1. Django运维管理系统模块边界与异步任务的选型起点运维系统最尴尬的追问往往不是“功能做没做”而是“页面卡住的时候后台到底在干什么”。不少基于Django搭建的运维平台把设备台账、操作日志、端口扫描全揉进同步请求里扫几百个端口要等几分钟前端超时用户只能刷新重试最后谁也不知道任务是成功了还是卡死了。实际上Django做运维系统难点不在增删改查本身而在三个交叉问题操作日志要记到“人、路径、状态码”这一级需要中间件和模型信号配合设备管理必须把“谁建的、能不能删”落进ORM和权限体系端口扫描一旦异步化还得有一套稳定的任务队列和并发探测方案。这篇文章只讲这三条主链路外加“数据库文件怎么从开发快照变成可复现的初始化数据”适合刚接手Django运维项目的开发者也适合准备把分散脚本重构成平台需要一个可落地架构的读者。文中命令基于Django 3.2、Celery 5.x、Python 3.9以上环境代码可以直接抄进项目改着跑。2. Django操作日志收集中间件拦截与模型信号双通道实现2.1 先按日志类型建表别把访问日志和数据变更混成一坨运维日志最怕的是“全”而不“分”。全量日志进了同一张表按用户查得到但按“谁删了设备”“哪次操作把状态改成了离线”这种语义去查时SQL越写越长。建议先做类型拆分统一存一张表用log_type字段区分请求访问和数据变更这样既能看一次完整操作链路又能单独过滤。# ops_audit/models.py import uuid from django.db import models class OperationLog(models.Model): LOG_TYPES ( (req, 请求访问), (model, 数据变更), (scan, 端口扫描), ) log_type models.CharField(日志类型, max_length16, choicesLOG_TYPES, defaultreq) user models.CharField(操作用户, max_length64, defaultanonymous) method models.CharField(HTTP方法, max_length8, blankTrue) path models.CharField(请求路径, max_length500, blankTrue) status_code models.PositiveSmallIntegerField(状态码, nullTrue, blankTrue) request_body models.JSONField(请求体摘要, nullTrue, blankTrue) request_id models.UUIDField(关联ID, defaultuuid.uuid4, db_indexTrue) created_at models.DateTimeField(发生时间, auto_now_addTrue) class Meta: db_table ops_operation_log ordering [-created_at]字段里值得注意的参数是request_id。它是UUIDField且加了db_indexTrue用来把一次请求产生的访问日志、后续的模型变更日志、端口扫描日志串成同一条链路。运维排查时拿着前端跑接口返回的request_id就能把这条请求在所有环节留下的痕迹全部捞出来。request_body用JSONField而不是在代码里拼字符串是为了保留结构化字段后面做“敏感信息打码”时逐 key 处理更方便。2.2 用中间件记录访问日志再把当前用户塞进线程变量中间件拦截请求很简单但还有一个更实际的问题模型信号里拿不到“当前登录用户”。常规做法是在中间件里把request对象存到线程局部变量信号处理函数再从线程局部变量里取用户。注意必须在process_response的finally里清空引用否则线程复用时会记错操作人这个坑排查起来很隐蔽。# ops_audit/middleware.py import json import threading from django.utils.deprecation import MiddlewareMixin _thread_locals threading.local() def get_current_request(): return getattr(_thread_locals, request, None) def get_current_user(): request get_current_request() if request is None: return None user request.user if user and user.is_authenticated: return user return None class OperationLogMiddleware(MiddlewareMixin): SENSITIVE_FIELDS {password, token, secret, authorization, private_key} def process_request(self, request): # 请求进入时保存上下文后续 model signal 中也能取到用户 _thread_locals.request request def process_response(self, request, response): try: self._record(request, response) finally: # 必须清空线程池复用时才能避免用户串号 _thread_locals.request None return response def _safe_body(self, body): if not isinstance(body, dict): return {_raw_body_len: len(body)} safe {} for key, value in body.items(): if key.lower() in self.SENSITIVE_FIELDS: safe[key] *** elif isinstance(value, (dict, list)): safe[key] self._safe_body(value) else: safe[key] value return safe def _record(self, request, response): from .models import OperationLog body {} if request.method not in (GET, HEAD) and hasattr(request, body): try: if request.content_type application/json: body json.loads(request.body[:65536].decode(utf-8, errorsignore)) else: body request.POST.dict() except Exception: body {} user get_current_user() OperationLog.objects.create( log_typereq, useruser.username if user else anonymous, methodrequest.method, pathrequest.get_full_path()[:500], status_codegetattr(response, status_code, 500), request_bodyself._safe_body(body), )这段代码的关键在_safe_body。很多生产事故源于日志系统把密码、token 明文落库等数据库被拖走时才发现敏感信息全在里面。这里约定了password、token、secret、authorization、private_key五个敏感 key遇到嵌套 dict 和 list 会递归打码。process_response中finally的清理动作是线程安全的兜底Django 开发服务器和 gunicorn/uvicorn 都是多线程处理请求线程池复用时如果不清空下一个请求信号里读到的request很可能是上一个请求的残留对象。写完模型和中间件后记得在settings.py的MIDDLEWARE列表最后加上ops_audit.middleware.OperationLogMiddleware并执行python manage.py makemigrations ops_audit python manage.py migrate。2.3 模型信号记录数据变更必须跳过 loaddata 产生的无主操作中间件只能覆盖“经过 Django 视图的请求”但系统内部定时任务、Celery worker、Admin 里直接调用 ORM 的批量操作不会经过视图。异常捕获代码适合放主流程太破碎。常规做法是用post_save和post_delete信号把设备类核心模型的每一次 create、update、delete 都转为log_typemodel的日志。# ops_audit/signal_handlers.py from django.db.models.signals import post_save, post_delete from .middleware import get_current_user, get_current_request from .models import OperationLog def _snapshot_instance(instance): return { model: instance._meta.label, pk: instance.pk, } def record_change(instance, createdFalse, rawFalse, **kwargs): # rawTrue 表示 loaddata 等场景直接写入不记录操作日志 if raw: return user get_current_user() request get_current_request() payload _snapshot_instance(instance) payload[action] create if created else update OperationLog.objects.create( log_typemodel, useruser.username if user else anonymous, methodrequest.method if request else signal, pathrequest.path if request else , status_code201 if created else 200, request_bodypayload, ) def record_delete(instance, **kwargs): user get_current_user() request get_current_request() payload _snapshot_instance(instance) payload[action] delete OperationLog.objects.create( log_typemodel, useruser.username if user else anonymous, methodrequest.method if request else signal, pathrequest.path if request else , status_code204, request_bodypayload, )干货在raw参数上。loaddata导入初始化数据时每条记录都会触发post_save但此时没有用户上下文也没有请求上下文强行写日志只会让日志表里充满anonymous噪音。Django 传入了rawTrue来标记这种批量导入信号里直接 return 即可。除了这个点我还会在apps.py的ready()方法里显式绑定到Device模型而不是听网上“对全模型全局注册信号”全局注册会让每张表的每次变更都触发日志写入字典、Session 表的读写会瞬间把日志表撑爆而且对 DBA 定位问题没有任何帮助。# ops_audit/apps.py from django.apps import AppConfig class OpsAuditConfig(AppConfig): default_auto_field django.db.models.BigAutoField name ops_audit def ready(self): from django.db.models.signals import post_save, post_delete from device.models import Device from .signal_handlers import record_change, record_delete post_save.connect(record_change, senderDevice) post_delete.connect(record_delete, senderDevice)2.4 读旧值写新值用 pre_save 记录字段变更差异post_save记录的是操作结果是一个“变更后”的瞬间快照但这不够——设备改 IP、改负责人时“从哪个值改成哪个值”是审计关键。更细的做法是再挂一个pre_save信号在保存前查出旧记录并比较字段差异暂存在实例属性上等post_save时把它合进request_body。# ops_audit/diff_mixin.py 建议挂到 Device 信号链中 from django.db.models.signals import pre_save def diff_on_save(sender, instance, **kwargs): if not instance.pk: return try: old sender.objects.get(pkinstance.pk) except sender.DoesNotExist: return changed {} for field in instance._meta.fields: name field.name if name in (updated_at,): continue old_value getattr(old, name) new_value getattr(instance, name) if old_value ! new_value: # 只记录真正变化的核心业务字段 changed[name] {old: old_value, new: new_value} if changed: setattr(instance, _field_diff, changed)这一段不是必须但做了之后运维审计会轻松很多。注意这里跳过了updated_at因为自动更新时间每次都会变记录下来全是噪音更通用的做法是维护一个白名单字段列表只有ip_address、port、status、owner等核心字段进入比较范围。如果项目里用django-simple-history可以做整表全量历史但引入成本和存储开销都偏高小平台先做字段级 diff性价比最高。3. Django设备增删改查从ORM建模到DRF与Admin权限控制3.1 设备表建模独立 IP 字段别把“IP:PORT”塞进一个字符串设备表是整个运维系统的主数据来源。建模时最容易埋的雷是把 IP 和端口拼成一个192.168.1.10:22字符串存进 CharField等要做按 IP 段过滤、按端口分组统计时只能写正则表达式索引也永远建不上。正确做法是把ip_address拆成GenericIPAddressField端口单独放port字段并用db_indexTrue支撑按状态查询。# device/models.py from django.db import models from django.contrib.auth.models import User class Device(models.Model): STATUS_CHOICES ( (up, 在线), (down, 离线), (unknown, 未知), ) hostname models.CharField(主机名, max_length128) ip_address models.GenericIPAddressField(管理IP, uniqueTrue) port models.PositiveIntegerField(管理端口, default22, help_textSSH/带外管理端口) os_type models.CharField(操作系统, max_length32, defaultlinux, help_textlinux/windows/network 等尽量用枚举值 ) status models.CharField(资产状态, max_length16, choicesSTATUS_CHOICES, defaultunknown, db_indexTrue) owner models.CharField(负责人, max_length64, blankTrue) create_by models.ForeignKey(User, on_deletemodels.SET_NULL, nullTrue, editableFalse, related_namecreated_devices) last_scan_at models.DateTimeField(最近扫描时间, nullTrue, blankTrue, db_indexTrue) scan_report models.JSONField(扫描结果摘要, nullTrue, blankTrue) remark models.TextField(备注, blankTrue) created_at models.DateTimeField(auto_now_addTrue) updated_at models.DateTimeField(auto_nowTrue) class Meta: db_table ops_device ordering [-updated_at] indexes [ models.Index(fields[status, last_scan_at]), ] def __str__(self): return f{self.hostname}({self.ip_address})这里create_by是不可编辑字段由视图自动写入普通用户即便拿到序列化接口也改不了“创建人”。os_type建议统一小写枚举别在项目里一会儿写Linux一会儿写linux否则按系统类型做筛选时会漏数据。设备状态status加db_index是因为列表页最常见的过滤就是按在线/离线筛后续扫描任务也依赖它定位待扫描设备。3.2 用 DRF ViewSet 快速搭出设备增删改查接口设备管理接口用viewsets.ModelViewSet最省事一个类就把 GET 列表、GET 详情、POST 新增、PUT/PATCH 修改、DELETE 删除全包了。为了让列表支持按 IP 模糊过滤和状态精确过滤重写get_queryset。# device/api.py from rest_framework import viewsets, permissions from .models import Device from .serializers import DeviceSerializer class DeviceViewSet(viewsets.ModelViewSet): serializer_class DeviceSerializer permission_classes [permissions.IsAuthenticated] def get_queryset(self): qs Device.objects.select_related(create_by).all() status self.request.query_params.get(status) ip self.request.query_params.get(ip) hostname self.request.query_params.get(hostname) if status: qs qs.filter(statusstatus) if ip: qs qs.filter(ip_address__icontainsip) if hostname: qs qs.filter(hostname__icontainshostname) return qs# device/serializers.py from rest_framework import serializers from .models import Device class DeviceSerializer(serializers.ModelSerializer): create_by_name serializers.CharField(sourcecreate_by.username, read_onlyTrue) class Meta: model Device fields [ id, hostname, ip_address, port, os_type, status, owner, create_by, create_by_name, last_scan_at, scan_report, remark, created_at, updated_at, ] read_only_fields [create_by, last_scan_at, scan_report, created_at, updated_at]read_only_fields里的last_scan_at和scan_report只能由后台扫描任务写入普通用户改设备资料时不可能顺手把扫描结果也改了避免脏数据。select_related(create_by)是为了列表接口少发一条 SQL方便一次查出创建人用户名。注册路由用DefaultRouterrouter.register(devices, DeviceViewSet)后DRF 自动给出/api/devices/、/api/devices/{id}/两套资源地址和对应的 OPTIONS 描述前端可以直接对接。增删改查里还有一个运维平台特有的决策删除。建议默认不物理删除设备而是增加is_active或archived字段把删除变成停用。资产台账通常有保留价值物理删掉后很难追溯“这台设备曾经存在过这一事实”。真要走 ModelViewSet 自带的 DELETE至少要重写perform_destroy把status为up的设备拦截下来。3.3 Django Admin后台的权限裁剪在线设备禁删、按负责人过滤不是每个运维同事都会直接调 APIAdmin 后台依然是高频入口。Django Admin 的权限控制比 DRF 更细可以通过get_queryset做行级过滤也能通过has_delete_permission做对象级删除控制。# device/admin.py from django.contrib import admin from .models import Device admin.register(Device) class DeviceAdmin(admin.ModelAdmin): list_display [hostname, ip_address, port, status, owner, last_scan_at] list_filter [status, os_type] search_fields [hostname, ip_address] actions [mark_down_selected] def get_queryset(self, request): qs super().get_queryset(request) if request.user.is_superuser: return qs # 非超级管理员只能看到自己负责的设备 return qs.filter(ownerrequest.user.username) def has_delete_permission(self, request, objNone): # 在线设备不允许直接从 Admin 后台删除 if obj is not None and obj.status up: return False return super().has_delete_permission(request, obj) admin.action(description标记为离线谨慎) def mark_down_selected(self, request, queryset): if not request.user.is_superuser: self.message_user(request, 只有超级管理员可以批量标记离线, levelwarning) return queryset.update(statusdown)get_queryset里只用了ownerrequest.user.username这一行就把普通管理员的数据范围锁在自己的设备上。has_delete_permission之所以按对象判断是因为 Admin 的批量删除会遍历get_deleted_objects逐个对象调用这个权限在线设备返回False用户只会看到“没有删除权限”而不会出现误删生产设备的问题。批量 action 则演示了“操作前二次确认”和“按角色限制”的写法。3.4 设备操作日志串通删除、修改、新增都走信号不用在视图里手写搭建完这部分后设备增删改查与前一章的操作日志自动打通post_save和post_delete信号会捕获设备表每个操作日志里带上了用户、路径、状态码。这样在设备视图里不需要再写一行OperationLog.objects.create。如果需要排查“谁在哪个时间改了设备 IP”直接查操作日志表即可无需前端配合。日志的request_id对不上时多半是中间件没在MIDDLEWARE里注册回settings.py检查顺序即可。4. Django异步端口扫描Celery任务编排与asyncio并发探测实现4.1 “异步”要先拆成两件事请求不阻塞探测更高效端口扫描的“异步”在运维平台里经常被误会成“Django 里的 async 语法”。实际上要拆成两层第一层是任务队列异步把扫描任务丢给 Celery workerHTTP 请求立即返回前端不用傻等第二层才是探测本身的并发扫描目标端口之间没有依赖可以用 asyncio 协程并发连接也可以用 nmap 的多线程端口扫描。两者不是替代关系是编排关系。表格对比一下常见可落地方案方案请求侧体验扫描效率部署依赖适合场景视图内直接跑 socket 扫描阻塞等待受 GIL 影响效率低无扫 110 个端口调试用Celery worker 调 nmap 命令行立即返回 task_id高nmap 内部多线程宿主机安装 nmap大批量、需要 OS 指纹识别Celery worker 内用 asyncio 协程探测立即返回 task_id中高可精确控制并发数无额外依赖大多数内网设备存活和开放端口检测一般默认选第三种因为把扫描结果只定位到“端口是否开放”这个层面时纯 Python 就能完成只有需要服务版本识别、操作系统指纹时才返回去接 nmap。这样项目在离线环境、无 root 权限的 worker 节点上也能跑。4.2 Celery任务队列与Django配置的要点Celery 5.x 的配置走namespaceCELERY所有配置项都放到 settings.py 里的CELERY_*前缀下。Redis 作为 broker 和 backend一套服务两用。# ops_platform/celery.py import os from celery import Celery os.environ.setdefault(DJANGO_SETTINGS_MODULE, ops_platform.settings) app Celery(ops_platform) app.config_from_object(django.conf:settings, namespaceCELERY) app.autodiscover_tasks()# settings.py 片段 import os CELERY_BROKER_URL os.getenv(CELERY_BROKER_URL, redis://127.0.0.1:6379/1) CELERY_RESULT_BACKEND os.getenv(CELERY_RESULT_BACKEND, redis://127.0.0.1:6379/2) CELERY_TASK_TIME_LIMIT 1200 CELERY_TASK_ACKS_LATE False CELERY_WORKER_PREFETCH_MULTIPLIER 1time_limit1200是硬超时防止某个目标主机不可达时任务挂 20 分钟不退prefetch_multiplier1让每个 worker 少量预取任务任务间可更均衡对短扫描任务更公平。在__init__.py里记得加入from .celery import app as celery_app否则celery -A ops_platform worker可能找不到入口。worker 启动命令python manage.py runserver 0.0.0.0:8000 celery -A ops_platform worker -l info -c 4-c 4控制并发 worker 数本地调试时可以降低到2避免每个任务里的 asyncio 同时拉起大量连接。4.3 端口探测协程实现与并发参数控制4.3.1 解析“22,80,8000-8010”这样的端口表达式场景里最常见的是前端传一个“22,80,443,8000-8010”这样的表达式需要先解析成端口列表。这个小函数值得抽出来单独处理避免在任务主流程里写正则堆垃圾代码。# device/port_parser.py def parse_ports(port_expr: str) - list[int]: 解析端口表达式支持逗号、中划线范围例如 22,80,8000-8010 ports set() for item in str(port_expr).split(,): item item.strip() if not item: continue if - in item: start, end item.split(-, 1) for port in range(int(start), int(end) 1): if 1 port 65535: ports.add(port) else: port int(item) if 1 port 65535: ports.add(port) return sorted(ports)parse_ports用set去重防止用户传80,80或8000-8010,8005导致同一端口被探测两次。排序后返回便于后续日志展示和结果对比。端口范围越界直接过滤避免生成 0 或 65536 这种非法值。4.3.2 asyncio 并发探测主逻辑传统同步 socket 循环会一个个端口等超时扫 11024 个内网端口通常要几十秒。改用 asyncio 后在协程里并发发起 TCP 连接通过Semaphore控制并发数。# device/scanner.py import asyncio from .port_parser import parse_ports async def _probe(host, port, timeout): try: reader, writer await asyncio.wait_for( asyncio.open_connection(host, port), timeouttimeout ) writer.close() await writer.wait_closed() return port, True except Exception: return port, False async def async_tcp_scan(host, port_expr, timeout1.0, max_concurrency200): ports parse_ports(port_expr) semaphore asyncio.Semaphore(max_concurrency) async def guarded(port): async with semaphore: return await _probe(host, port, timeout) tasks [asyncio.create_task(guarded(port)) for port in ports] results await asyncio.gather(*tasks) open_ports [port for port, ok in results if ok] return { host: host, open_ports: open_ports, open_count: len(open_ports), scanned_count: len(ports), }timeout1.0是单次 TCP 连接的等待时间内网建议 0.51 秒就够公网扫描要调到 23 秒否则丢端口严重。max_concurrency200表示同一时刻最多 200 个连接在飞如果目标设备是防火墙或老交换机并发太高会被设备直接丢包保守可以降到 50。需要注意这个方案只判断“TCP 端口能否建立连接”无法识别“端口被防火墙 drop 时算 closed 还是 filtered”对绝大多数设备巡检来说已经够用。4.4 扫描任务落地与结果回写Celery 任务里扫描结束后要把状态和结果写回Device让列表页和详情页能直接展示。任务直接接收device_idworker 里去查库。# device/tasks.py import asyncio import logging from celery import shared_task from django.utils import timezone from .models import Device from .scanner import async_tcp_scan logger logging.getLogger(__name__) shared_task(ignore_resultFalse) def async_port_scan(device_id, port_expr1-1024, timeout1.0, max_concurrency200): try: device Device.objects.get(pkdevice_id) except Device.DoesNotExist: return {ok: False, msg: device not found} result asyncio.run(async_tcp_scan(device.ip_address, port_expr, timeouttimeout, max_concurrencymax_concurrency)) device.status up if result[open_count] 0 else down device.last_scan_at timezone.now() device.scan_report result device.save(update_fields[status, last_scan_at, scan_report]) logger.info( device%s host%s open_ports%s, device.id, device.ip_address, result[open_ports][:20] ) return {ok: True, device_id: device.id, result: result}save(update_fields...)是性能控制点只回写变化字段避免把整个Device重新存一遍触发冗余更新。logger.info只打前 20 个端口避免大量端口时日志行过长。asyncio.run()在任务函数内部调用确保每个 worker 进程里协程事件循环是被当前线程单独管理不受 Celery worker 自身事件循环干扰。任务入队入口直接在 API 视图中实现。在DeviceViewSet加一个自定义scanaction# device/api.py 追加 from rest_framework.decorators import action from rest_framework.response import Response from .tasks import async_port_scan class DeviceViewSet(viewsets.ModelViewSet): # ... 前面已有代码 action(detailTrue, methods[post]) def scan(self, request, pkNone): device self.get_object() port_expr request.data.get(ports, 22,80,443,8080) task async_port_scan.delay( device.id, port_exprport_expr, timeoutfloat(request.data.get(timeout, 1.0)), max_concurrencyint(request.data.get(max_concurrency, 200)), ) return Response({ task_id: task.id, device_id: device.id, message: 扫描任务已提交, }, status202)action(detailTrue, methods[post])让 DRF 生成POST /api/devices/{id}/scan/接口。task.id返回给前端前端轮询“扫描任务状态”或直接刷新设备详情就能看到结果。接口必须先返回 202表示请求已接受异步执行不占用 HTTP worker。5. 把数据库文件转换成生产可复用初始数据5.1 项目里自带的 SQLite 文件是开发快照不是运维资产“内含数据库文件”通常指源码包里放了一个db.sqlite3启动项目就能看到设备表和日志表里有数据。作为从零开始的学习项目这种交付方式很方便但把它直接搬到生产环境有隐患SQLite 的并发写锁在多人同时操作用户、日志表时会报database is locked备份策略也不透明。正确姿势是把这份 SQLite 当作“初始数据的逻辑来源”导出成 JSON fixture再让生产环境自己用 migrate 建表、loaddata 导入。不用在 Git 仓库里长期维护一个二进制数据库文件。导出初始数据的常用命令# 在项目开发环境执行 python manage.py dumpdata \ --excludecontenttypes \ --excludeauth.permission \ --excludesessions \ --excludeadmin.logentry \ -o fixtures/init_data.json这里--exclude掉了几类不需要跨环境复制的数据contenttypes和auth.permission会随 migrate 自动重建复制过去容易造成权限主键冲突sessions是运行时数据没意义admin.logentry是 Admin 操作记录不属于业务初始化范围。夹具文件应包含的典型内容是auth.User和ops_device.Device以及字典表。如果已经有操作日志是否需要导出看实际需求日志仓库建议留空从上线日重新积累。5.2 从 SQLite 迁到 MySQLdumpdata loaddata 的完整迁移路径生产环境建议 MySQL/PostgreSQL。安装驱动时最常见的坑是pip install mysqlclient报编译错误Debian/Ubuntu 下需要先装系统库sudo apt-get update sudo apt-get install -y default-libmysqlclient-dev build-essential pkg-config pip install mysqlclient如果不想折腾编译可以使用pymysql作为 MySQL 后端在 Django 的数据库配置里用django.db.backends.mysql并在__init__.py里执行pymysql.install_as_MySQLdb()。不过长期运行建议用官方数据库驱动性能和异常信息更完整。settings 里把数据库配置抽成环境变量import os DATABASES { default: { ENGINE: os.getenv(DB_ENGINE, django.db.backends.mysql), NAME: os.getenv(DB_NAME, ops_platform), USER: os.getenv(DB_USER, ops), PASSWORD: os.getenv(DB_PASSWORD, ), HOST: os.getenv(DB_HOST, 127.0.0.1), PORT: os.getenv(DB_PORT, 3306), OPTIONS: {charset: utf8mb4}, } }这样换环境时改环境变量即可不需要改代码。先用python manage.py migrate在 MySQL 里建全部表结构再执行python manage.py loaddata fixtures/init_data.json导入初始数据。loaddata 导入的数据会触发pre_save和post_save这就是前一章为什么要对rawTrue做判断否则日志表会被 fixture 数据刷一遍垃圾日志。迁移完成后验证 Django 是否真正连到 MySQL用一条查询确认python manage.py shell -c from django.db import connection; cursorconnection.cursor(); cursor.execute(select count(*) from ops_device); print(cursor.fetchone())列表里能输出设备总数说明表结构、连接、驱动三件事都正常。5.3 上线前集成验证一次访问、一次修改、一次扫描最后把整个链路跑一遍验证“操作日志收集、设备增删改查、异步端口扫描”三块功能在同一个环境里配合正常。第一步验证访问日志落库。登录 Django Admin打开设备列表页再刷新一次数据库中的ops_operation_log表执行以下 SQLSELECT log_type, user, path, status_code, created_at FROM ops_operation_log ORDER BY id DESC LIMIT 10;如果中间件没注册这里查出来会空库如果user字段都是anonymous说明中间件里取request.user的逻辑有问题优先检查中间件在MIDDLEWARE里的位置确保它在django.contrib.auth.middleware.AuthenticationMiddleware之后。第二步验证数据变更日志。通过 Admin 修改一台设备的负责人再查询日志表应该看到log_typemodelrequest_body里包含action:update和主键信息。第三步验证异步端口扫描。先确认 Redis 可用worker 已启动celery -A ops_platform worker -l info -c 2后端调用前也可以直接用 Django shell 入队一个测试任务避免前端接口网络干扰python manage.py shell -c from device.tasks import async_port_scan; rasync_port_scan.delay(1, 80,443, timeout1.0, max_concurrency50); print(r.id)任务执行后回到数据库查询ops_device表SELECT hostname, ip_address, status, last_scan_at, scan_report FROM ops_device ORDER BY last_scan_at DESC LIMIT 5;last_scan_at有值、scan_report里能看到open_ports列表说明 Celery worker 扫描完成后成功把结果写回设备记录。全链路到这一步就是闭环数据库迁移干净、日志两条通道可用、扫描不阻塞前端请求这套基于 Django 的运维管理系统就可以放心交给使用方了。本文还有配套的精品资源点击获取
Django运维平台实战：日志审计、设备管理与异步端口扫描
简介这是一个基于Django构建的运维管理系统平台完整源代码面向运维工程师、Python后端开发者及Django学习者用于快速搭建具备操作日志收集、设备增删改查、传统方式设备登录配置管理以及基于socket的asyncio异步端口扫描等核心功能的运维后台。前端采用HTML/CSS/Bootstrap与少量JavaScript并集成Django-SimpleUI作为Admin后台后端使用Django内置paramiko、telnetlib、asyncio、socket等第三方模块代码结构清晰方便二次开发与学习。资源包共包含389个文件以JavaScript、CSS、HTML等前端静态文件为主同时有20个Python源码文件、13个HTML模板、17个编译后的pyc文件以及1个SQLite数据库文件含初始数据压缩包整体大小为13.89MB目录组织规范便于按模块检索。目前已有518人学习下载。使用者可直接导入SQLite数据库快速启动项目也可参照源码学习异步端口扫描、操作日志采集等实战技巧适合需要参考完整Django项目实践、了解运维自动化实现思路的开发者。1. Django运维管理系统模块边界与异步任务的选型起点运维系统最尴尬的追问往往不是“功能做没做”而是“页面卡住的时候后台到底在干什么”。不少基于Django搭建的运维平台把设备台账、操作日志、端口扫描全揉进同步请求里扫几百个端口要等几分钟前端超时用户只能刷新重试最后谁也不知道任务是成功了还是卡死了。实际上Django做运维系统难点不在增删改查本身而在三个交叉问题操作日志要记到“人、路径、状态码”这一级需要中间件和模型信号配合设备管理必须把“谁建的、能不能删”落进ORM和权限体系端口扫描一旦异步化还得有一套稳定的任务队列和并发探测方案。这篇文章只讲这三条主链路外加“数据库文件怎么从开发快照变成可复现的初始化数据”适合刚接手Django运维项目的开发者也适合准备把分散脚本重构成平台需要一个可落地架构的读者。文中命令基于Django 3.2、Celery 5.x、Python 3.9以上环境代码可以直接抄进项目改着跑。2. Django操作日志收集中间件拦截与模型信号双通道实现2.1 先按日志类型建表别把访问日志和数据变更混成一坨运维日志最怕的是“全”而不“分”。全量日志进了同一张表按用户查得到但按“谁删了设备”“哪次操作把状态改成了离线”这种语义去查时SQL越写越长。建议先做类型拆分统一存一张表用log_type字段区分请求访问和数据变更这样既能看一次完整操作链路又能单独过滤。# ops_audit/models.py import uuid from django.db import models class OperationLog(models.Model): LOG_TYPES ( (req, 请求访问), (model, 数据变更), (scan, 端口扫描), ) log_type models.CharField(日志类型, max_length16, choicesLOG_TYPES, defaultreq) user models.CharField(操作用户, max_length64, defaultanonymous) method models.CharField(HTTP方法, max_length8, blankTrue) path models.CharField(请求路径, max_length500, blankTrue) status_code models.PositiveSmallIntegerField(状态码, nullTrue, blankTrue) request_body models.JSONField(请求体摘要, nullTrue, blankTrue) request_id models.UUIDField(关联ID, defaultuuid.uuid4, db_indexTrue) created_at models.DateTimeField(发生时间, auto_now_addTrue) class Meta: db_table ops_operation_log ordering [-created_at]字段里值得注意的参数是request_id。它是UUIDField且加了db_indexTrue用来把一次请求产生的访问日志、后续的模型变更日志、端口扫描日志串成同一条链路。运维排查时拿着前端跑接口返回的request_id就能把这条请求在所有环节留下的痕迹全部捞出来。request_body用JSONField而不是在代码里拼字符串是为了保留结构化字段后面做“敏感信息打码”时逐 key 处理更方便。2.2 用中间件记录访问日志再把当前用户塞进线程变量中间件拦截请求很简单但还有一个更实际的问题模型信号里拿不到“当前登录用户”。常规做法是在中间件里把request对象存到线程局部变量信号处理函数再从线程局部变量里取用户。注意必须在process_response的finally里清空引用否则线程复用时会记错操作人这个坑排查起来很隐蔽。# ops_audit/middleware.py import json import threading from django.utils.deprecation import MiddlewareMixin _thread_locals threading.local() def get_current_request(): return getattr(_thread_locals, request, None) def get_current_user(): request get_current_request() if request is None: return None user request.user if user and user.is_authenticated: return user return None class OperationLogMiddleware(MiddlewareMixin): SENSITIVE_FIELDS {password, token, secret, authorization, private_key} def process_request(self, request): # 请求进入时保存上下文后续 model signal 中也能取到用户 _thread_locals.request request def process_response(self, request, response): try: self._record(request, response) finally: # 必须清空线程池复用时才能避免用户串号 _thread_locals.request None return response def _safe_body(self, body): if not isinstance(body, dict): return {_raw_body_len: len(body)} safe {} for key, value in body.items(): if key.lower() in self.SENSITIVE_FIELDS: safe[key] *** elif isinstance(value, (dict, list)): safe[key] self._safe_body(value) else: safe[key] value return safe def _record(self, request, response): from .models import OperationLog body {} if request.method not in (GET, HEAD) and hasattr(request, body): try: if request.content_type application/json: body json.loads(request.body[:65536].decode(utf-8, errorsignore)) else: body request.POST.dict() except Exception: body {} user get_current_user() OperationLog.objects.create( log_typereq, useruser.username if user else anonymous, methodrequest.method, pathrequest.get_full_path()[:500], status_codegetattr(response, status_code, 500), request_bodyself._safe_body(body), )这段代码的关键在_safe_body。很多生产事故源于日志系统把密码、token 明文落库等数据库被拖走时才发现敏感信息全在里面。这里约定了password、token、secret、authorization、private_key五个敏感 key遇到嵌套 dict 和 list 会递归打码。process_response中finally的清理动作是线程安全的兜底Django 开发服务器和 gunicorn/uvicorn 都是多线程处理请求线程池复用时如果不清空下一个请求信号里读到的request很可能是上一个请求的残留对象。写完模型和中间件后记得在settings.py的MIDDLEWARE列表最后加上ops_audit.middleware.OperationLogMiddleware并执行python manage.py makemigrations ops_audit python manage.py migrate。2.3 模型信号记录数据变更必须跳过 loaddata 产生的无主操作中间件只能覆盖“经过 Django 视图的请求”但系统内部定时任务、Celery worker、Admin 里直接调用 ORM 的批量操作不会经过视图。异常捕获代码适合放主流程太破碎。常规做法是用post_save和post_delete信号把设备类核心模型的每一次 create、update、delete 都转为log_typemodel的日志。# ops_audit/signal_handlers.py from django.db.models.signals import post_save, post_delete from .middleware import get_current_user, get_current_request from .models import OperationLog def _snapshot_instance(instance): return { model: instance._meta.label, pk: instance.pk, } def record_change(instance, createdFalse, rawFalse, **kwargs): # rawTrue 表示 loaddata 等场景直接写入不记录操作日志 if raw: return user get_current_user() request get_current_request() payload _snapshot_instance(instance) payload[action] create if created else update OperationLog.objects.create( log_typemodel, useruser.username if user else anonymous, methodrequest.method if request else signal, pathrequest.path if request else , status_code201 if created else 200, request_bodypayload, ) def record_delete(instance, **kwargs): user get_current_user() request get_current_request() payload _snapshot_instance(instance) payload[action] delete OperationLog.objects.create( log_typemodel, useruser.username if user else anonymous, methodrequest.method if request else signal, pathrequest.path if request else , status_code204, request_bodypayload, )干货在raw参数上。loaddata导入初始化数据时每条记录都会触发post_save但此时没有用户上下文也没有请求上下文强行写日志只会让日志表里充满anonymous噪音。Django 传入了rawTrue来标记这种批量导入信号里直接 return 即可。除了这个点我还会在apps.py的ready()方法里显式绑定到Device模型而不是听网上“对全模型全局注册信号”全局注册会让每张表的每次变更都触发日志写入字典、Session 表的读写会瞬间把日志表撑爆而且对 DBA 定位问题没有任何帮助。# ops_audit/apps.py from django.apps import AppConfig class OpsAuditConfig(AppConfig): default_auto_field django.db.models.BigAutoField name ops_audit def ready(self): from django.db.models.signals import post_save, post_delete from device.models import Device from .signal_handlers import record_change, record_delete post_save.connect(record_change, senderDevice) post_delete.connect(record_delete, senderDevice)2.4 读旧值写新值用 pre_save 记录字段变更差异post_save记录的是操作结果是一个“变更后”的瞬间快照但这不够——设备改 IP、改负责人时“从哪个值改成哪个值”是审计关键。更细的做法是再挂一个pre_save信号在保存前查出旧记录并比较字段差异暂存在实例属性上等post_save时把它合进request_body。# ops_audit/diff_mixin.py 建议挂到 Device 信号链中 from django.db.models.signals import pre_save def diff_on_save(sender, instance, **kwargs): if not instance.pk: return try: old sender.objects.get(pkinstance.pk) except sender.DoesNotExist: return changed {} for field in instance._meta.fields: name field.name if name in (updated_at,): continue old_value getattr(old, name) new_value getattr(instance, name) if old_value ! new_value: # 只记录真正变化的核心业务字段 changed[name] {old: old_value, new: new_value} if changed: setattr(instance, _field_diff, changed)这一段不是必须但做了之后运维审计会轻松很多。注意这里跳过了updated_at因为自动更新时间每次都会变记录下来全是噪音更通用的做法是维护一个白名单字段列表只有ip_address、port、status、owner等核心字段进入比较范围。如果项目里用django-simple-history可以做整表全量历史但引入成本和存储开销都偏高小平台先做字段级 diff性价比最高。3. Django设备增删改查从ORM建模到DRF与Admin权限控制3.1 设备表建模独立 IP 字段别把“IP:PORT”塞进一个字符串设备表是整个运维系统的主数据来源。建模时最容易埋的雷是把 IP 和端口拼成一个192.168.1.10:22字符串存进 CharField等要做按 IP 段过滤、按端口分组统计时只能写正则表达式索引也永远建不上。正确做法是把ip_address拆成GenericIPAddressField端口单独放port字段并用db_indexTrue支撑按状态查询。# device/models.py from django.db import models from django.contrib.auth.models import User class Device(models.Model): STATUS_CHOICES ( (up, 在线), (down, 离线), (unknown, 未知), ) hostname models.CharField(主机名, max_length128) ip_address models.GenericIPAddressField(管理IP, uniqueTrue) port models.PositiveIntegerField(管理端口, default22, help_textSSH/带外管理端口) os_type models.CharField(操作系统, max_length32, defaultlinux, help_textlinux/windows/network 等尽量用枚举值 ) status models.CharField(资产状态, max_length16, choicesSTATUS_CHOICES, defaultunknown, db_indexTrue) owner models.CharField(负责人, max_length64, blankTrue) create_by models.ForeignKey(User, on_deletemodels.SET_NULL, nullTrue, editableFalse, related_namecreated_devices) last_scan_at models.DateTimeField(最近扫描时间, nullTrue, blankTrue, db_indexTrue) scan_report models.JSONField(扫描结果摘要, nullTrue, blankTrue) remark models.TextField(备注, blankTrue) created_at models.DateTimeField(auto_now_addTrue) updated_at models.DateTimeField(auto_nowTrue) class Meta: db_table ops_device ordering [-updated_at] indexes [ models.Index(fields[status, last_scan_at]), ] def __str__(self): return f{self.hostname}({self.ip_address})这里create_by是不可编辑字段由视图自动写入普通用户即便拿到序列化接口也改不了“创建人”。os_type建议统一小写枚举别在项目里一会儿写Linux一会儿写linux否则按系统类型做筛选时会漏数据。设备状态status加db_index是因为列表页最常见的过滤就是按在线/离线筛后续扫描任务也依赖它定位待扫描设备。3.2 用 DRF ViewSet 快速搭出设备增删改查接口设备管理接口用viewsets.ModelViewSet最省事一个类就把 GET 列表、GET 详情、POST 新增、PUT/PATCH 修改、DELETE 删除全包了。为了让列表支持按 IP 模糊过滤和状态精确过滤重写get_queryset。# device/api.py from rest_framework import viewsets, permissions from .models import Device from .serializers import DeviceSerializer class DeviceViewSet(viewsets.ModelViewSet): serializer_class DeviceSerializer permission_classes [permissions.IsAuthenticated] def get_queryset(self): qs Device.objects.select_related(create_by).all() status self.request.query_params.get(status) ip self.request.query_params.get(ip) hostname self.request.query_params.get(hostname) if status: qs qs.filter(statusstatus) if ip: qs qs.filter(ip_address__icontainsip) if hostname: qs qs.filter(hostname__icontainshostname) return qs# device/serializers.py from rest_framework import serializers from .models import Device class DeviceSerializer(serializers.ModelSerializer): create_by_name serializers.CharField(sourcecreate_by.username, read_onlyTrue) class Meta: model Device fields [ id, hostname, ip_address, port, os_type, status, owner, create_by, create_by_name, last_scan_at, scan_report, remark, created_at, updated_at, ] read_only_fields [create_by, last_scan_at, scan_report, created_at, updated_at]read_only_fields里的last_scan_at和scan_report只能由后台扫描任务写入普通用户改设备资料时不可能顺手把扫描结果也改了避免脏数据。select_related(create_by)是为了列表接口少发一条 SQL方便一次查出创建人用户名。注册路由用DefaultRouterrouter.register(devices, DeviceViewSet)后DRF 自动给出/api/devices/、/api/devices/{id}/两套资源地址和对应的 OPTIONS 描述前端可以直接对接。增删改查里还有一个运维平台特有的决策删除。建议默认不物理删除设备而是增加is_active或archived字段把删除变成停用。资产台账通常有保留价值物理删掉后很难追溯“这台设备曾经存在过这一事实”。真要走 ModelViewSet 自带的 DELETE至少要重写perform_destroy把status为up的设备拦截下来。3.3 Django Admin后台的权限裁剪在线设备禁删、按负责人过滤不是每个运维同事都会直接调 APIAdmin 后台依然是高频入口。Django Admin 的权限控制比 DRF 更细可以通过get_queryset做行级过滤也能通过has_delete_permission做对象级删除控制。# device/admin.py from django.contrib import admin from .models import Device admin.register(Device) class DeviceAdmin(admin.ModelAdmin): list_display [hostname, ip_address, port, status, owner, last_scan_at] list_filter [status, os_type] search_fields [hostname, ip_address] actions [mark_down_selected] def get_queryset(self, request): qs super().get_queryset(request) if request.user.is_superuser: return qs # 非超级管理员只能看到自己负责的设备 return qs.filter(ownerrequest.user.username) def has_delete_permission(self, request, objNone): # 在线设备不允许直接从 Admin 后台删除 if obj is not None and obj.status up: return False return super().has_delete_permission(request, obj) admin.action(description标记为离线谨慎) def mark_down_selected(self, request, queryset): if not request.user.is_superuser: self.message_user(request, 只有超级管理员可以批量标记离线, levelwarning) return queryset.update(statusdown)get_queryset里只用了ownerrequest.user.username这一行就把普通管理员的数据范围锁在自己的设备上。has_delete_permission之所以按对象判断是因为 Admin 的批量删除会遍历get_deleted_objects逐个对象调用这个权限在线设备返回False用户只会看到“没有删除权限”而不会出现误删生产设备的问题。批量 action 则演示了“操作前二次确认”和“按角色限制”的写法。3.4 设备操作日志串通删除、修改、新增都走信号不用在视图里手写搭建完这部分后设备增删改查与前一章的操作日志自动打通post_save和post_delete信号会捕获设备表每个操作日志里带上了用户、路径、状态码。这样在设备视图里不需要再写一行OperationLog.objects.create。如果需要排查“谁在哪个时间改了设备 IP”直接查操作日志表即可无需前端配合。日志的request_id对不上时多半是中间件没在MIDDLEWARE里注册回settings.py检查顺序即可。4. Django异步端口扫描Celery任务编排与asyncio并发探测实现4.1 “异步”要先拆成两件事请求不阻塞探测更高效端口扫描的“异步”在运维平台里经常被误会成“Django 里的 async 语法”。实际上要拆成两层第一层是任务队列异步把扫描任务丢给 Celery workerHTTP 请求立即返回前端不用傻等第二层才是探测本身的并发扫描目标端口之间没有依赖可以用 asyncio 协程并发连接也可以用 nmap 的多线程端口扫描。两者不是替代关系是编排关系。表格对比一下常见可落地方案方案请求侧体验扫描效率部署依赖适合场景视图内直接跑 socket 扫描阻塞等待受 GIL 影响效率低无扫 110 个端口调试用Celery worker 调 nmap 命令行立即返回 task_id高nmap 内部多线程宿主机安装 nmap大批量、需要 OS 指纹识别Celery worker 内用 asyncio 协程探测立即返回 task_id中高可精确控制并发数无额外依赖大多数内网设备存活和开放端口检测一般默认选第三种因为把扫描结果只定位到“端口是否开放”这个层面时纯 Python 就能完成只有需要服务版本识别、操作系统指纹时才返回去接 nmap。这样项目在离线环境、无 root 权限的 worker 节点上也能跑。4.2 Celery任务队列与Django配置的要点Celery 5.x 的配置走namespaceCELERY所有配置项都放到 settings.py 里的CELERY_*前缀下。Redis 作为 broker 和 backend一套服务两用。# ops_platform/celery.py import os from celery import Celery os.environ.setdefault(DJANGO_SETTINGS_MODULE, ops_platform.settings) app Celery(ops_platform) app.config_from_object(django.conf:settings, namespaceCELERY) app.autodiscover_tasks()# settings.py 片段 import os CELERY_BROKER_URL os.getenv(CELERY_BROKER_URL, redis://127.0.0.1:6379/1) CELERY_RESULT_BACKEND os.getenv(CELERY_RESULT_BACKEND, redis://127.0.0.1:6379/2) CELERY_TASK_TIME_LIMIT 1200 CELERY_TASK_ACKS_LATE False CELERY_WORKER_PREFETCH_MULTIPLIER 1time_limit1200是硬超时防止某个目标主机不可达时任务挂 20 分钟不退prefetch_multiplier1让每个 worker 少量预取任务任务间可更均衡对短扫描任务更公平。在__init__.py里记得加入from .celery import app as celery_app否则celery -A ops_platform worker可能找不到入口。worker 启动命令python manage.py runserver 0.0.0.0:8000 celery -A ops_platform worker -l info -c 4-c 4控制并发 worker 数本地调试时可以降低到2避免每个任务里的 asyncio 同时拉起大量连接。4.3 端口探测协程实现与并发参数控制4.3.1 解析“22,80,8000-8010”这样的端口表达式场景里最常见的是前端传一个“22,80,443,8000-8010”这样的表达式需要先解析成端口列表。这个小函数值得抽出来单独处理避免在任务主流程里写正则堆垃圾代码。# device/port_parser.py def parse_ports(port_expr: str) - list[int]: 解析端口表达式支持逗号、中划线范围例如 22,80,8000-8010 ports set() for item in str(port_expr).split(,): item item.strip() if not item: continue if - in item: start, end item.split(-, 1) for port in range(int(start), int(end) 1): if 1 port 65535: ports.add(port) else: port int(item) if 1 port 65535: ports.add(port) return sorted(ports)parse_ports用set去重防止用户传80,80或8000-8010,8005导致同一端口被探测两次。排序后返回便于后续日志展示和结果对比。端口范围越界直接过滤避免生成 0 或 65536 这种非法值。4.3.2 asyncio 并发探测主逻辑传统同步 socket 循环会一个个端口等超时扫 11024 个内网端口通常要几十秒。改用 asyncio 后在协程里并发发起 TCP 连接通过Semaphore控制并发数。# device/scanner.py import asyncio from .port_parser import parse_ports async def _probe(host, port, timeout): try: reader, writer await asyncio.wait_for( asyncio.open_connection(host, port), timeouttimeout ) writer.close() await writer.wait_closed() return port, True except Exception: return port, False async def async_tcp_scan(host, port_expr, timeout1.0, max_concurrency200): ports parse_ports(port_expr) semaphore asyncio.Semaphore(max_concurrency) async def guarded(port): async with semaphore: return await _probe(host, port, timeout) tasks [asyncio.create_task(guarded(port)) for port in ports] results await asyncio.gather(*tasks) open_ports [port for port, ok in results if ok] return { host: host, open_ports: open_ports, open_count: len(open_ports), scanned_count: len(ports), }timeout1.0是单次 TCP 连接的等待时间内网建议 0.51 秒就够公网扫描要调到 23 秒否则丢端口严重。max_concurrency200表示同一时刻最多 200 个连接在飞如果目标设备是防火墙或老交换机并发太高会被设备直接丢包保守可以降到 50。需要注意这个方案只判断“TCP 端口能否建立连接”无法识别“端口被防火墙 drop 时算 closed 还是 filtered”对绝大多数设备巡检来说已经够用。4.4 扫描任务落地与结果回写Celery 任务里扫描结束后要把状态和结果写回Device让列表页和详情页能直接展示。任务直接接收device_idworker 里去查库。# device/tasks.py import asyncio import logging from celery import shared_task from django.utils import timezone from .models import Device from .scanner import async_tcp_scan logger logging.getLogger(__name__) shared_task(ignore_resultFalse) def async_port_scan(device_id, port_expr1-1024, timeout1.0, max_concurrency200): try: device Device.objects.get(pkdevice_id) except Device.DoesNotExist: return {ok: False, msg: device not found} result asyncio.run(async_tcp_scan(device.ip_address, port_expr, timeouttimeout, max_concurrencymax_concurrency)) device.status up if result[open_count] 0 else down device.last_scan_at timezone.now() device.scan_report result device.save(update_fields[status, last_scan_at, scan_report]) logger.info( device%s host%s open_ports%s, device.id, device.ip_address, result[open_ports][:20] ) return {ok: True, device_id: device.id, result: result}save(update_fields...)是性能控制点只回写变化字段避免把整个Device重新存一遍触发冗余更新。logger.info只打前 20 个端口避免大量端口时日志行过长。asyncio.run()在任务函数内部调用确保每个 worker 进程里协程事件循环是被当前线程单独管理不受 Celery worker 自身事件循环干扰。任务入队入口直接在 API 视图中实现。在DeviceViewSet加一个自定义scanaction# device/api.py 追加 from rest_framework.decorators import action from rest_framework.response import Response from .tasks import async_port_scan class DeviceViewSet(viewsets.ModelViewSet): # ... 前面已有代码 action(detailTrue, methods[post]) def scan(self, request, pkNone): device self.get_object() port_expr request.data.get(ports, 22,80,443,8080) task async_port_scan.delay( device.id, port_exprport_expr, timeoutfloat(request.data.get(timeout, 1.0)), max_concurrencyint(request.data.get(max_concurrency, 200)), ) return Response({ task_id: task.id, device_id: device.id, message: 扫描任务已提交, }, status202)action(detailTrue, methods[post])让 DRF 生成POST /api/devices/{id}/scan/接口。task.id返回给前端前端轮询“扫描任务状态”或直接刷新设备详情就能看到结果。接口必须先返回 202表示请求已接受异步执行不占用 HTTP worker。5. 把数据库文件转换成生产可复用初始数据5.1 项目里自带的 SQLite 文件是开发快照不是运维资产“内含数据库文件”通常指源码包里放了一个db.sqlite3启动项目就能看到设备表和日志表里有数据。作为从零开始的学习项目这种交付方式很方便但把它直接搬到生产环境有隐患SQLite 的并发写锁在多人同时操作用户、日志表时会报database is locked备份策略也不透明。正确姿势是把这份 SQLite 当作“初始数据的逻辑来源”导出成 JSON fixture再让生产环境自己用 migrate 建表、loaddata 导入。不用在 Git 仓库里长期维护一个二进制数据库文件。导出初始数据的常用命令# 在项目开发环境执行 python manage.py dumpdata \ --excludecontenttypes \ --excludeauth.permission \ --excludesessions \ --excludeadmin.logentry \ -o fixtures/init_data.json这里--exclude掉了几类不需要跨环境复制的数据contenttypes和auth.permission会随 migrate 自动重建复制过去容易造成权限主键冲突sessions是运行时数据没意义admin.logentry是 Admin 操作记录不属于业务初始化范围。夹具文件应包含的典型内容是auth.User和ops_device.Device以及字典表。如果已经有操作日志是否需要导出看实际需求日志仓库建议留空从上线日重新积累。5.2 从 SQLite 迁到 MySQLdumpdata loaddata 的完整迁移路径生产环境建议 MySQL/PostgreSQL。安装驱动时最常见的坑是pip install mysqlclient报编译错误Debian/Ubuntu 下需要先装系统库sudo apt-get update sudo apt-get install -y default-libmysqlclient-dev build-essential pkg-config pip install mysqlclient如果不想折腾编译可以使用pymysql作为 MySQL 后端在 Django 的数据库配置里用django.db.backends.mysql并在__init__.py里执行pymysql.install_as_MySQLdb()。不过长期运行建议用官方数据库驱动性能和异常信息更完整。settings 里把数据库配置抽成环境变量import os DATABASES { default: { ENGINE: os.getenv(DB_ENGINE, django.db.backends.mysql), NAME: os.getenv(DB_NAME, ops_platform), USER: os.getenv(DB_USER, ops), PASSWORD: os.getenv(DB_PASSWORD, ), HOST: os.getenv(DB_HOST, 127.0.0.1), PORT: os.getenv(DB_PORT, 3306), OPTIONS: {charset: utf8mb4}, } }这样换环境时改环境变量即可不需要改代码。先用python manage.py migrate在 MySQL 里建全部表结构再执行python manage.py loaddata fixtures/init_data.json导入初始数据。loaddata 导入的数据会触发pre_save和post_save这就是前一章为什么要对rawTrue做判断否则日志表会被 fixture 数据刷一遍垃圾日志。迁移完成后验证 Django 是否真正连到 MySQL用一条查询确认python manage.py shell -c from django.db import connection; cursorconnection.cursor(); cursor.execute(select count(*) from ops_device); print(cursor.fetchone())列表里能输出设备总数说明表结构、连接、驱动三件事都正常。5.3 上线前集成验证一次访问、一次修改、一次扫描最后把整个链路跑一遍验证“操作日志收集、设备增删改查、异步端口扫描”三块功能在同一个环境里配合正常。第一步验证访问日志落库。登录 Django Admin打开设备列表页再刷新一次数据库中的ops_operation_log表执行以下 SQLSELECT log_type, user, path, status_code, created_at FROM ops_operation_log ORDER BY id DESC LIMIT 10;如果中间件没注册这里查出来会空库如果user字段都是anonymous说明中间件里取request.user的逻辑有问题优先检查中间件在MIDDLEWARE里的位置确保它在django.contrib.auth.middleware.AuthenticationMiddleware之后。第二步验证数据变更日志。通过 Admin 修改一台设备的负责人再查询日志表应该看到log_typemodelrequest_body里包含action:update和主键信息。第三步验证异步端口扫描。先确认 Redis 可用worker 已启动celery -A ops_platform worker -l info -c 2后端调用前也可以直接用 Django shell 入队一个测试任务避免前端接口网络干扰python manage.py shell -c from device.tasks import async_port_scan; rasync_port_scan.delay(1, 80,443, timeout1.0, max_concurrency50); print(r.id)任务执行后回到数据库查询ops_device表SELECT hostname, ip_address, status, last_scan_at, scan_report FROM ops_device ORDER BY last_scan_at DESC LIMIT 5;last_scan_at有值、scan_report里能看到open_ports列表说明 Celery worker 扫描完成后成功把结果写回设备记录。全链路到这一步就是闭环数据库迁移干净、日志两条通道可用、扫描不阻塞前端请求这套基于 Django 的运维管理系统就可以放心交给使用方了。本文还有配套的精品资源点击获取
