一、Django 中使用 django-apscheduler 实现定时任务
可以方便地管理周期性任务（如每天清理缓存、定时发送邮件等）
1. 安装
pip install django-apscheduler -i https://pypi.tuna.tsinghua.edu.cn/simple #0.7.0
2.添加到应用，python manage.py migrate
# settings.py
INSTALLED_APPS = [# ...'django_apscheduler',
]
python manage.py migrate
执行后，mysql中，多了两个django_apscheduler开头的表。
3. 一个django应用中创建定时任务
可以是一个utils包下，创建一个叫task.py的，我是直接在相关应用下建了一个task.py
也有人也在view.py中， 还有人习惯专门改个app来实现定时任务。
# tasks.py
from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore
import uuid
# 初始化调度器
scheduler = BackgroundScheduler()
scheduler.add_jobstore(DjangoJobStore(), "default")# --- 定义定时任务函数 ---
def cleanup_temp_data():print("清理临时数据...")def send_daily_report():print("发送日报...")def generate_job_id( jobname):return f"{jobname}_{uuid.uuid4().hex[:6]}"# --- 将任务添加到调度器 ---
scheduler.add_job(cleanup_temp_data,"interval", # 间隔性任务seconds=10, # 每隔 60 秒执行一次# days=1, # 每天执行一次id=generate_job_id("cleanup_job"),# id="cleanup_job",# replace_existing=True, # 允许覆盖同名任务 (即数据库已有id为 cleanup_job，不添加这个就会报错)
)scheduler.add_job(send_daily_report,"cron", # 定时任务（类似 crontab）hour=9, # 每天 9 点执行minute=10, # 加了这行就表示，每天9点 10 分钟id="report_job",replace_existing=True, # 允许覆盖同名任务
)# 注意：这里暂时不启动调度器！在 apps.py 中启动。
# 启动调度器 (这里也启动不了呢)
# scheduler.start()
方式了，在方法上写装饰器@register_job(scheduler,‘interval’,seconds=20,id=‘clenauo_job2’)，就不用下方的 scheduler.add_job（）部分了
启动项目时的定时任务
4.配置启动
一般定义任务启动是配置在应用中app.py中的，表示启动项目时，来启动定义是任务。
from django.apps import AppConfig
import filelockclass HrunnerConfig(AppConfig):default_auto_field = "django.db.models.BigAutoField"name = "hrunner"# 在 Django 开发环境中，AppConfig.ready() 方法被执行两次是.正常现象，但会导致定时任务重复注册# 生产环境下是没问题的。def ready(self):# 仅在应用加载完成后初始化调度器if not hasattr(self, 'scheduler_started'):from .task import scheduler # 导入当前应用的调度器# 启动调度器scheduler.start()self.scheduler_started = True
5.启动观察第三步代码
python managy runserver 因为开发环境，会导致执行了两次ready。有点点问题。
开发环境下，解决办法。
先清空两个表，然后这里就这样写，但是启动时会报。（原因就是ready执行了两次，第二次报错，故意让第二次没有加成功。。就只有一个任务）
生成环境下。还没尝试
好像，必须要 replace_existing,不然测试环境时，项目启动不起来。
定时任务方式，有两种
后续定时任务
接口添加任务。
定时任务应用（没行）
感觉就把定时任务写死，启动时执行就好
django+ celery异步任务配置
1. 安装celery 和redis
pip install celery redis -i https://pypi.tuna.tsinghua.edu.cn/simple
2. 需要有一个redis服务
3. django项目结构
假设项目名为 myproject，应用名为 myapp：
myproject/
├── myproject/
│ ├── init.py
│ ├── celery.py # 新增 Celery 配置文件
│ ├── settings.py
│ └── urls.py
└── myapp/
├── tasks.py # 存放异步任务
└── …
4. 配置celery
创建 myproject/celery.py
# myproject/celery.py
import os
from celery import Celery# 设置 Django 环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')# 创建 Celery 实例
app = Celery('myproject')# 从 Django 配置中读取 Celery 设置（以 CELERY_ 开头的配置）
app.config_from_object('django.conf:settings', namespace='CELERY')# 自动发现所有 Django 应用中的 tasks.py 文件
app.autodiscover_tasks()
修改 myproject/init.py
# myproject/__init__.py
from .celery import app as celery_app__all__ = ['celery_app'] # 确保 Celery 应用被正确加载
在settings.py中添加配置
# myproject/settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0' # 消息代理用 Redis
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' # 任务结果存储
CELERY_TIMEZONE = 'Asia/Shanghai' # 时区
# 注： TIME_ZONE = 'Asia/Shanghai' # 要和这个时区一致
# 启动celery worker
# celery -A myproject worker --loglevel=info# windows电脑还要添加这个，不然会有点问题
CELERY_WORKER_POOL = 'solo'
# 启动celery worker
# celery -A your_project worker --loglevel=info -P solo
5. 编写异步任务
在应用 myapp 中创建 tasks.py
# myapp/tasks.py
from celery import shared_task@shared_task
def send_welcome_email(email):print(f"模拟发送邮件到 {email}...")# 这里可以写实际发邮件的代码（如调用 Django 的 send_mail）return f"邮件已发送至 {email}"
6. 配置异步任务调用
在视图或任何地方调用
# myapp/views.py
from django.http import JsonResponse
from .tasks import send_welcome_emaildef register(request):email = "user@example.com"# 异步调用任务（.delay() 是快捷方法）send_welcome_email.delay(email)return JsonResponse({"status": "邮件发送中..."})
7.启动服务
a.启动celery worker （注意换成你的项目名）
celery -A myproject worker --loglevel=info
# windows电脑用下面的，并且settings中还要加那个solo
celery -A your_project worker --loglevel=info -P solo
b.启动django服务器
python manage.py runserver
8. 测试
访问注册接口后，观察 Celery Worker 的日志输出：
9. 原理理解
异步调用playwright操控浏览器。
任务方法
from celery import shared_task
from playwright.sync_api import sync_playwright
from django.core.cache import cache@shared_task
def execute_browser_operation( debugger_url, actions):""":param debugger_url: 远程浏览器调试地址（如 http://172.16.1.4:9222）:param actions: 操作步骤列表（JSON序列化）"""try:with sync_playwright() as p:# 连接远程浏览器browser = p.chromium.connect_over_cdp(debugger_url)page = browser.contexts[0].pages[0]# 操作浏览器（示例：打开百度）# page.goto("https://www.baidu.com")# page.fill("#kw", "自动化测试")"""action 的结构。[{"type": "navigate", "url": "https://www.baidu.com"},{"type": "fill", "selector": "#kw", "text": "刘亦菲"},{"type": "click", "selector": "#su"},{"type": "select_dropdown", "selector": "#country-select", "value": "china"}]"""# 执行操作步骤result = {'steps': []}for action in actions:if action['type'] == 'navigate':page.goto(action['url'])result['steps'].append(f"导航至 {action['url']}")elif action['type'] == 'fill':page.fill(action['selector'], action['text'])result['steps'].append(f"填写 {action['selector']}")elif action['type'] == 'click':page.click(action['selector'])elif action['type'] == 'select_dropdown':page.select_option(action['selector'], action['value'])# 添加更多操作类型...# 保存截图到媒体目录# screenshot_path = f"media/screenshots/{self.request.id}.png"# page.screenshot(path=screenshot_path)return {'status': 'success',# 'screenshot': screenshot_path,'details': result}except Exception as e:return {'status': 'error', 'message': str(e)}finally:browser.close()views
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import permissions
from .celery_tasks import execute_browser_operationclass BrowserControlView(APIView):def post(self, request):# 验证参数required_fields = ['debugger_url', 'actions']if not all(field in request.data for field in required_fields):return Response({'error': '缺少必要参数'}, status=400)# 提交异步任务task = execute_browser_operation.delay(debugger_url=request.data['debugger_url'],actions=request.data['actions'])return Response({'task_id': task.id,'status_endpoint': f'/api/task-status/{task.id}/'})# 任务状态查询接口
# myapp/views.py
from celery.result import AsyncResultclass TaskStatusView(APIView):def get(self, request, task_id):task = AsyncResult(task_id)return Response({'task_id': task_id,'status': task.status,'result': task.result if task.ready() else None})url.py
path('control/', views.BrowserControlView.as_view(), name='browser-control'),path('task-status/<str:task_id>/', views.TaskStatusView.as_view(), name='task-status'),效果
调接口，可以执行本地浏览器
执行端开放端口方式：
开放成功后，在打开的浏览器访问
Chrome：
访问 http://0.0.0.0:9222/json/version，若返回浏览器信息则成功
http://127.0.0.1:9222/json/version 这样是不对的 （0.0.0.0）才能是ip+端口访问
可能需要
# 将本地9222端口暴露到所有网络接口（需管理员权限）
netsh interface portproxy add v4tov4 listenaddress=0.0.0.0 listenport=9222 connectaddress=127.0.0.1 connectport=9222
#验证
netsh interface portproxy show all
#删除
netsh interface portproxy delete v4tov4 listenaddress=0.0.0.0 listenport=9222# 启动浏览器
cd "C:\Program Files\Google\Chrome\Application"
chrome.exe --remote-debugging-port=9222 --remote-debugging-address=0.0.0.0 --user-data-dir="C:\playwright_debug"
# 验证 需要看到 （tcp 0.0.0.0:9222别人才能通过ip访问到你）
netstat -ano | findstr :9222开端口
# 查看有没有
netsh advfirewall firewall show rule name=all | findstr "9222"
# 给一个
netsh advfirewall firewall add rule name="Playwright_9222" dir=in action=allow protocol=TCP localport=9222
# 删除
netsh advfirewall firewall delete rule name="Playwright_9222"
bat文件（参考-未成熟1）
@echo off
setlocal enabledelayedexpansion:: 检查管理员权限
net session >nul 2>&1
if %errorlevel% neq 0 (echo 请右键点击此脚本，选择"以管理员身份运行"pauseexit /b 1
):: 1. 设置防火墙规则
echo 正在配置防火墙规则...
netsh advfirewall firewall show rule name="Playwright_9222" >nul 2>&1
if !errorlevel! equ 0 (echo [跳过] 防火墙规则 Playwright_9222 已存在
) else (netsh advfirewall firewall add rule name="Playwright_9222" dir=in action=allow protocol=TCP localport=9222if !errorlevel! neq 0 (echo [错误] 无法创建防火墙规则exit /b 1)echo [成功] 防火墙规则已添加
):: 2. 配置端口转发
echo 正在设置端口代理...
netsh interface portproxy show v4tov4 | findstr "0.0.0.0:9222" >nul
if !errorlevel! equ 0 (echo [跳过] 端口代理规则已存在
) else (netsh interface portproxy add v4tov4 listenaddress=0.0.0.0 listenport=9222 connectaddress=127.0.0.1 connectport=9222if !errorlevel! neq 0 (echo [错误] 无法创建端口代理规则exit /b 1)echo [成功] 端口代理已配置
):: 3. 启动 Chrome
echo 正在启动 Chrome 浏览器...
cd /d "C:\Program Files\Google\Chrome\Application"
if not exist "chrome.exe" (echo [错误] 未找到 Chrome 安装路径echo 请检查路径: "C:\Program Files\Google\Chrome\Application"exit /b 1
)if not exist "C:\playwright_debug" (mkdir "C:\playwright_debug"
)
start "" chrome.exe --remote-debugging-port=9222 --remote-debugging-address=0.0.0.0 --user-data-dir="C:\playwright_debug"echo 所有操作已完成
pause
bat2,供参考
@echo off
setlocal enabledelayedexpansion:: 强制管理员权限
if not "%1"=="admin" (powershell start -verb runas '%0' admin & exit):: 清理旧进程和规则
taskkill /im chrome.exe /f >nul 2>&1
netsh interface portproxy delete v4tov4 listenport=9222 >nul 2>&1:: 启动 Chrome（核心修改）
echo 启动 Chrome 调试实例...
cd /d "C:\Program Files\Google\Chrome\Application"
start "" chrome.exe ^
--remote-debugging-port=9222 ^
--remote-debugging-address=0.0.0.0 ^
--user-data-dir="C:\playwright_debug" ^
--no-first-run --disable-extensions:: 延迟等待服务启动
timeout /t 5 /nobreak >nul:: 验证监听状态
echo 当前端口监听状态：
netstat -ano | findstr :9222pause
palywright操作别人电脑调试
from playwright.sync_api import sync_playwrightwith sync_playwright() as p:# 替换为同事电脑的IP地址（如192.168.1.100）# browser = p.chromium.connect_over_cdp("http://127.0.0.1:9222")browser = p.chromium.connect_over_cdp("http://172.16.1.4:9222")default_context = browser.contexts[0] # 获取默认上下文page = default_context.pages[0] # 获取已打开的页面# 操作浏览器（示例：打开百度）page.goto("https://www.baidu.com")page.fill("#kw", "自动化测试")# from playwright.sync_api import sync_playwright
#
# with sync_playwright() as p:
# # 替换为同事电脑的实际IP（如192.168.1.100）
# browser = p.chromium.connect_over_cdp("http://192.168.1.100:9222")
# default_context = browser.contexts[0] # 获取默认上下文
# page = default_context.pages[0] # 获取已打开的页面
#
# # 操作浏览器示例：访问百度并搜索
# page.goto("https://www.baidu.com")
# page.fill("#kw", "自动化测试")
# page.click("#su")
# page.wait_for_timeout(3000) # 等待3秒观察结果
前端端vue代码和效果
点击执行，会操作本地电脑浏览器
uicasedemo.vue
<template><div class="container"><h1 class="title">UI自动化Demo</h1><!-- 操作按钮区域 --><div class="action-bar"><!-- 新增执行端地址输入框 --><el-inputv-model="form.debugger_url"placeholder="请输入执行端IP端口"style="width: 300px; margin-right: 20px"/><el-button type="primary" @click="dialogVisible = true">新增步骤</el-button><el-button type="success" @click="handleExecute">执行测试</el-button></div><!-- 步骤列表 --><div class="step-list"><div v-for="(step, index) in steps" :key="index" class="step-item"><span class="step-index">步骤 {{ index + 1 }}</span><div class="step-content"><div class="step-name">{{ step.action }}操作</div><div class="step-params" v-if="step.params">{{ step.params }}</div></div><div class="step-actions"><el-button size="small" @click="handleEditStep(index)">编辑</el-button><el-button size="small" type="danger" @click="handleDeleteStep(index)">删除</el-button></div></div></div><!-- 新增步骤弹窗 --><el-dialog v-model="dialogVisible" title="新增操作步骤" width="30%"><el-form :model="formData"><el-form-item label="操作类型"><el-selectv-model="formData.action"placeholder="请选择操作"@change="handleActionChange"><el-option label="点击" value="click" /><el-option label="输入" value="fill" /><el-option label="访问地址" value="goto" /><el-option label="等待" value="wait" /></el-select></el-form-item><!-- 动态参数区域 --><template v-if="formData.action === 'goto'"><el-form-item label="访问地址"><el-input v-model="formData.params.url" placeholder="请输入URL" /></el-form-item></template><template v-if="formData.action === 'click'"><el-form-item label="元素定位"><el-input v-model="formData.params.selector" placeholder="请输入元素选择器" /></el-form-item></template><template v-if="formData.action === 'fill'"><el-form-item label="元素定位"><el-input v-model="formData.params.selector" placeholder="请输入元素选择器" /></el-form-item><el-form-item label="输入内容"><el-input v-model="formData.params.text" placeholder="请输入要输入的内容" /></el-form-item></template><!-- 新增等待参数区域 --><template v-if="formData.action === 'wait'"><el-form-item label="等待时间（毫秒）"><el-inputv-model="formData.params.time"type="number"placeholder="请输入等待时间"min="1"/></el-form-item></template></el-form><template #footer><el-button @click="dialogVisible = false">取消</el-button><el-button type="primary" @click="handleAddStep">确认</el-button></template></el-dialog></div>
</template><script setup>
import { ref, reactive } from 'vue'
import { uidemosend } from '@/utils/api.js'// 步骤数据
const steps = ref([{action: 'goto',params: {type: 'goto',url: 'https://www.baidu.com',selector: '',text: ''}},{action: 'fill',params: {type: 'fill',url: '',selector: '#kw',text: '刘亦菲'}},{action: 'click',params: {type: 'click',url: '',selector: '#su',text: ''}}
])const dialogVisible = ref(false)
const formData = ref({editingIndex: -1,action: '',params: {type: '',url: '',selector: '',text: '',time: 0}
})// 切换操作类型时清空参数
const handleActionChange = (value) => {formData.value.params = {type: value,url: '',selector: '',text: '',time: value === 'wait' ? 2000 : 0}
}// 处理编辑步骤
const handleEditStep = (index) => {const step = steps.value[index]formData.value = {editingIndex: index,action: step.action,params: {type: step.params.type,url: step.params.url || '',selector: step.params.selector || '',text: step.params.text || '',time: Number(step.params.time) || 0}}dialogVisible.value = true
}// 处理删除步骤
const handleDeleteStep = (index) => {steps.value.splice(index, 1)
}// 修改后的添加步骤方法
const handleAddStep = () => {if (formData.value.action) {const newStep = {action: formData.value.action,params: {type: formData.value.params.type,url: formData.value.params.url || '',selector: formData.value.params.selector || '',text: formData.value.params.text || '',time: Number(formData.value.params.time) || 0}}if (formData.value.editingIndex >= 0) {// 更新现有步骤steps.value.splice(formData.value.editingIndex, 1, newStep)} else {// 添加新步骤steps.value.push(newStep)}// 重置表单formData.value = {editingIndex: -1,action: '',params: {type: '',url: '',selector: '',text: '',time: 0}}dialogVisible.value = false}
}const form = reactive({debugger_url: 'http://127.0.0.1:9222',actions: []
})const handleExecute = () => {console.log('执行')// 将steps中的params提取到form.actionsform.actions = steps.value.map((step) => ({ ...step.params }))console.log(form)uidemosend(form)
}
</script><style lang="scss" scoped>
.container {padding: 20px;.title {text-align: center;margin-bottom: 30px;}.action-bar {margin-bottom: 20px;display: flex;justify-content: flex-end;}.step-list {border: 1px solid #ebeef5;border-radius: 4px;.step-item {padding: 12px 20px;border-bottom: 1px solid #ebeef5;display: flex;align-items: center;&:last-child {border-bottom: none;}.step-index {width: 80px;color: #909399;}.step-content {flex: 1;.step-name {font-weight: 500;margin-bottom: 4px;}.step-params {color: #666;font-size: 0.9em;}}.step-actions {margin-left: auto;display: flex;align-items: center;.el-button {margin-left: 10px;}}}}
}
</style>
)
 设置触发器)
