django扩展练习记录

2025-05-20 13:00:44 0 阅读

一、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 filelock


class 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_email

def 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_operation


class 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 AsyncResult

class 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//', 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 FilesGoogleChromeApplication"
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 请右键点击此脚本，选择"以管理员身份运行"
    pause
    exit /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=9222
    if !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=9222
    if !errorlevel! neq 0 (
        echo [错误] 无法创建端口代理规则
        exit /b 1
    )
    echo [成功] 端口代理已配置
)

:: 3. 启动 Chrome
echo 正在启动 Chrome 浏览器...
cd /d "C:Program FilesGoogleChromeApplication"
if not exist "chrome.exe" (
    echo [错误] 未找到 Chrome 安装路径
    echo 请检查路径: "C:Program FilesGoogleChromeApplication"
    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 FilesGoogleChromeApplication"
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 :9222

pause

palywright操作别人电脑调试

from playwright.sync_api import sync_playwright

with 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秒观察结果