Django¶

# settings.py
STATIC_ROOT = os.path.join(BASE_DIR, "static")

STATIC_URL = '/static/'
STATICFILES_DIRS = [
    BASE_DIR / 'static',
]

# 模板文件存放在 BASE_DIR/templates
TEMPLATES = [
    {
        ...
        'DIRS': [BASE_DIR / 'templates'],
        ...
    },
]

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.sqlite3',
        'NAME': BASE_DIR / 'db.sqlite3',
    }
}

LANGUAGE_CODE = 'zh-hans'
TIME_ZONE = 'Asia/Shanghai'
USE_I18N = True
USE_L10N = True
USE_TZ = False

from django.template import Library
import datetime

register = Library()

def show_time(v):
    # 首页返回底部时间的过滤器
    return datetime.datetime.now().date()

register.filter('show_time', show_time)

from django.utils.deprecation import MiddlewareMixin


class letdo(MiddlewareMixin):
    # 它应该返回 None 或 HttpResponse 对象。如果返回 None，Django 将继续处理这个请求
    ''' 顺序
    调用 self.process_request(request)（如果被定义过）。
    调用 self.get_response(request) 来从后续的中间件和视图得到响应。
    调用 self.process_response(request, response)（如果被定义过）。
    返回响应。
    '''
    def __call__(self, request):
        '''
        :param request: 是中间件的主要入口点。请求到达时 Django 调用中间件的 __call__ 方法，
            传递 request 对象作为参数。中间件可在此执行任何预处理，例如记录请求信息、
            检查用户身份、设置请求标志等。
        :return:
        '''
        ...

    def process_view(self, request, view_func, view_args, view_kwargs):
        ''' 只在 Django 调用视图前被调用；请求被路由到视图函数之前调用
        request 是一个 HttpRequest 对象。
        view_func 是一个 Django 将要使用的 Python 函数（真实的函数对象，不是函数名）；
        view_args 是传递给视图的位置参数列表；
        view_kwargs 是传递给视图的关键字参数字典。
        view_args 和 view_kwargs 都不包含第一个视图参数 (request)。
        '''
        ...

    def process_exception(self, request, exception):
        '''
        当视图引发异常时，Django 会调用
        :param request:
        :param exception:
        :return:
        '''
        ...

    def process_template_response(self, request, response):
        '''
        在视图被完全执行后调用；它必须返回一个实现了 render 方法的响应对象。
        当视图函数返回一个模板响应时，这个方法会被调用。中间件可在这里处理模板响应，
        例如添加额外的上下文变量或修改响应内容。
        :param request:
        :param response:
        :return:
        '''
        ...

    def process_request(self, request):
        '''
        这个方法在 __call__ 方法之前调用，允许中间件在请求处理之前进行一些操作
        '''
        ...

    def process_response(self, request, response):
        # 这个方法在视图函数处理请求并生成响应后调用
        ...

AUTHENTICATION_BACKENDS = [
    # 'django_auth_ldap.backend.LDAPBackend',
    'hoaofun.backend.SettingsBackend',
    'django.contrib.auth.backends.ModelBackend',
]
LDAP_CONFIG = {"HOST": "ldap://192.168.1.1:389"}

CACHES = {
    "default": {
        "BACKEND": "django_redis.cache.RedisCache",
        "LOCATION": "redis://127.0.0.1:6379/1",
        "OPTIONS": {
            "CLIENT_CLASS": "django_redis.client.DefaultClient",
        },
    }
}

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = "Asia/Shanghai"
CELERY_MAX_TASKS_PER_CHILD = 100
CELERY_LOG_FILE = os.path.join(BASE_DIR, 'celery.log')
CELERYBEAT_LOG_FILE = os.path.join(BASE_DIR, 'celery.log.beta')

import os
from celery import Celery

# 为 'celery' 程序设置默认的 Django settings 模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'Assetinventory.settings')  # 注意自己的 setting

app = Celery('tech')  # 注意替换 app

# 用字符串配置，worker 无需把配置对象序列化给子进程
# namespace='CELERY' 表示所有 celery 相关配置项都以 CELERY_ 开头
app.config_from_object('django.conf:settings', namespace='CELERY')

# 加载所有已注册 Django app 的任务模块
app.autodiscover_tasks()  # 自动遍历目录下的 tasks.py

@app.task(bind=True, ignore_result=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

from celery import shared_task
from .about_qiwei import wxsendmsg

@shared_task
def shared_task_demo(username, txt):
    print('shared_task_demo')
    wxsendmsg(username, txt)

shared_task_demo.delay(c.userinfo_obj.username,
                       txt=f"{request.user.first_name}申请使用你在用的设备")

@shared_task
def check_our_dev():
    check_devalive()

# settings.py
CELERY_BEAT_SCHEDULE = {
    'run-every-minute': {
        'task': 'tech.tasks.check_our_dev',
        'schedule': 60.0,  # 每分钟执行一次
    },
}

# 生产环境不要用关系型数据库做 Broker；用 RabbitMQ 或 Redis
worker_prefetch_multiplier = 0    # 禁用预取，避免分配不均导致重复执行
worker_max_tasks_per_child = 50   # 每个 worker 执行 50 次即销毁，防卡死

# 不要用复杂对象做任务参数（会序列化失败/查询不到）
# Good
@app.task
def my_task(user_id):
    user = User.objects.get(id=user_id)

# Bad
@app.task
def my_task(user):
    print(user.name)

# 异步视图
async def index(request):
    return HttpResponse("Hello, Django!")

async def async_view(request):
    loop = asyncio.get_event_loop()
    loop.create_task(async_task())
    return HttpResponse("Hello, async Django!")

# asgi.py
import os
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application
from django.urls import re_path
from app import consumers

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'Assetinventory.settings')
application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter([re_path("msg/chat/", consumers.MyConsumer.as_asgi())])
    ),
})

# consumers.py
from channels.generic.websocket import AsyncWebsocketConsumer

class MyConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_group_name = "meet"
        await self.channel_layer.group_add(self.room_group_name, self.channel_name)
        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(self.room_group_name, self.channel_name)

    async def chat_message(self, event):
        await self.send(text_data=json.dumps({'message': event['message']}))

from channels.layers import get_channel_layer

async def send_message_to_group(room_group_name, message):
    channel_layer = get_channel_layer()
    await channel_layer.group_send(
        room_group_name,
        {'type': 'chat_message', 'message': message}
    )

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {"hosts": [("127.0.0.1", 6379)]},
    },
}

import json
from django.http import JsonResponse

def saveaiconf(request):
    info = json.loads(request.body)
    remark = info.get("remark")
    return JsonResponse({"success": '更新成功'})

import os
import django

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'tech.settings')
django.setup()