企业官网建设流程全解析

简介：这是一个基于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_length=16, choices=LOG_TYPES, default='req') user = models.CharField("操作用户", max_length=64, default='anonymous') method = models.CharField("HTTP方法", max_length=8, blank=True) path = models.CharField("请求路径", max_length=500, blank=True) status_code = models.PositiveSmallIntegerField("状态码", null=True, blank=True) request_body = models.JSONField("请求体摘要", null=True, blank=True) request_id = models.UUIDField("关联ID", default=uuid.uuid4, db_index=True) created_at = models.DateTimeField("发生时间", auto_now_add=True) class Meta: db_table = 'ops_operation_log' ordering = ['-created_at']

字段里值得注意的参数是request_id。它是UUIDField且加了db_index=True，用来把一次请求产生的访问日志、后续的模型变更日志、端口扫描日志串成同一条链路。运维排查时，拿着前端跑接口返回的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', errors='ignore')) else: body = request.POST.dict() except Exception: body = {} user = get_current_user() OperationLog.objects.create( log_type='req', user=user.username if user else 'anonymous', method=request.method, path=request.get_full_path()[:500], status_code=getattr(response, 'status_code', 500), request_body=self._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_type='model'的日志。

# 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, created=False, raw=False, **kwargs): # raw=True 表示 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_type='model', user=user.username if user else 'anonymous', method=request.method if request else 'signal', path=request.path if request else '', status_code=201 if created else 200, request_body=payload, ) def record_delete(instance, **kwargs): user = get_current_user() request = get_current_request() payload = _snapshot_instance(instance) payload['action'] = 'delete' OperationLog.objects.create( log_type='model', user=user.username if user else 'anonymous', method=request.method if request else 'signal', path=request.path if request else '', status_code=204, request_body=payload, )

干货在raw参数上。loaddata导入初始化数据时，每条记录都会触发post_save，但此时没有用户上下文，也没有请求上下文，强行写日志只会让日志表里充满anonymous噪音。Django 传入了raw=True来标记这种批量导入，信号里直接 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, sender=Device) post_delete.connect(record_delete, sender=Device)

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(pk=instance.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_index=True支撑按状态查询。

# 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_length=128) ip_address = models.GenericIPAddressField("管理IP", unique=True) port = models.PositiveIntegerField("管理端口", default=22, help_text="SSH/带外管理端口") os_type = models.CharField("操作系统", max_length=32, default='linux', help_text="linux/windows/network 等，尽量用枚举值） ) status = models.CharField("资产状态", max_length=16, choices=STATUS_CHOICES, default='unknown', db_index=True) owner = models.CharField("负责人", max_length=64, blank=True) create_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, editable=False, related_name="created_devices") last_scan_at = models.DateTimeField("最近扫描时间", null=True, blank=True, db_index=True) scan_report = models.JSONField("扫描结果摘要", null=True, blank=True) remark = models.TextField("备注", blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) 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(status=status) if ip: qs = qs.filter(ip_address__icontains=ip) if hostname: qs = qs.filter(hostname__icontains=hostname) return qs

# device/serializers.py from rest_framework import serializers from .models import Device class DeviceSerializer(serializers.ModelSerializer): create_by_name = serializers.CharField(source='create_by.username', read_only=True) 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，方便一次查出创建人用户名。注册路由用DefaultRouter，router.register('devices', DeviceViewSet)后，DRF 自动给出/api/devices/、/api/devices/{id}/两套资源地址和对应的 OPTIONS 描述，前端可以直接对接。

增删改查里还有一个运维平台特有的决策：删除。建议默认不物理删除设备，而是增加is_active或archived字段，把删除变成停用。资产台账通常有保留价值，物理删掉后很难追溯“这台设备曾经存在过这一事实”。真要走 ModelViewSet 自带的 DELETE，至少要重写perform_destroy，把status为up的设备拦截下来。

3.3 Django Admin后台的权限裁剪：在线设备禁删、按负责人过滤

不是每个运维同事都会直接调 API，Admin 后台依然是高频入口。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(owner=request.user.username) def has_delete_permission(self, request, obj=None): # 在线设备不允许直接从 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, "只有超级管理员可以批量标记离线", level='warning') return queryset.update(status='down')

get_queryset里只用了owner=request.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 worker，HTTP 请求立即返回，前端不用傻等；第二层才是探测本身的并发扫描，目标端口之间没有依赖，可以用 asyncio 协程并发连接，也可以用 nmap 的多线程端口扫描。两者不是替代关系，是编排关系。

表格对比一下常见可落地方案：

一般默认选第三种，因为把扫描结果只定位到“端口是否开放”这个层面时，纯 Python 就能完成；只有需要服务版本识别、操作系统指纹时，才返回去接 nmap。这样项目在离线环境、无 root 权限的 worker 节点上也能跑。

4.2 Celery任务队列与Django配置的要点

Celery 5.x 的配置走namespace='CELERY'，所有配置项都放到 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', namespace='CELERY') 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 = 1

time_limit=1200是硬超时，防止某个目标主机不可达时任务挂 20 分钟不退；prefetch_multiplier=1让每个 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 循环会一个个端口等超时，扫 1～1024 个内网端口通常要几十秒。改用 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), timeout=timeout ) writer.close() await writer.wait_closed() return port, True except Exception: return port, False async def async_tcp_scan(host, port_expr, timeout=1.0, max_concurrency=200): 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), }

timeout=1.0是单次 TCP 连接的等待时间，内网建议 0.5～1 秒就够，公网扫描要调到 2～3 秒，否则丢端口严重。max_concurrency=200表示同一时刻最多 200 个连接在飞；如果目标设备是防火墙或老交换机，并发太高会被设备直接丢包，保守可以降到 50。需要注意这个方案只判断“TCP 端口能否建立连接”，无法识别“端口被防火墙 drop 时算 closed 还是 filtered”，对绝大多数设备巡检来说已经够用。

4.4 扫描任务落地与结果回写

Celery 任务里扫描结束后要把状态和结果写回Device，让列表页和详情页能直接展示。任务直接接收device_id，worker 里去查库。

# 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_result=False) def async_port_scan(device_id, port_expr='1-1024', timeout=1.0, max_concurrency=200): try: device = Device.objects.get(pk=device_id) except Device.DoesNotExist: return {'ok': False, 'msg': 'device not found'} result = asyncio.run(async_tcp_scan(device.ip_address, port_expr, timeout=timeout, max_concurrency=max_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(detail=True, methods=['post']) def scan(self, request, pk=None): device = self.get_object() port_expr = request.data.get('ports', '22,80,443,8080') task = async_port_scan.delay( device.id, port_expr=port_expr, timeout=float(request.data.get('timeout', 1.0)), max_concurrency=int(request.data.get('max_concurrency', 200)), ) return Response({ 'task_id': task.id, 'device_id': device.id, 'message': '扫描任务已提交', }, status=202)

@action(detail=True, 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 \ --exclude=contenttypes \ --exclude=auth.permission \ --exclude=sessions \ --exclude=admin.logentry \ -o fixtures/init_data.json

这里--exclude掉了几类不需要跨环境复制的数据：contenttypes和auth.permission会随 migrate 自动重建，复制过去容易造成权限主键冲突；sessions是运行时数据，没意义；admin.logentry是 Admin 操作记录，不属于业务初始化范围。夹具文件应包含的典型内容是auth.User和ops_device.Device，以及字典表。如果已经有操作日志，是否需要导出看实际需求，日志仓库建议留空从上线日重新积累。

5.2 从 SQLite 迁到 MySQL：dumpdata + 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，这就是前一章为什么要对raw=True做判断，否则日志表会被 fixture 数据刷一遍垃圾日志。

迁移完成后，验证 Django 是否真正连到 MySQL，用一条查询确认：

python manage.py shell -c "from django.db import connection; cursor=connection.cursor(); cursor.execute('select count(*) from ops_device'); print(cursor.fetchone())"

列表里能输出设备总数，说明表结构、连接、驱动三件事都正常。

5.3 上线前集成验证：一次访问、一次修改、一次扫描

最后把整个链路跑一遍，验证“操作日志收集、设备增删改查、异步端口扫描”三块功能在同一个环境里配合正常。

第一步验证访问日志落库。登录 Django Admin，打开设备列表页，再刷新一次数据库中的ops_operation_log表，执行以下 SQL：

SELECT 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_type='model'，request_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; r=async_port_scan.delay(1, '80,443', timeout=1.0, max_concurrency=50); 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 的运维管理系统就可以放心交给使用方了。

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