一、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>
)
 設置觸發器)
