Django 채 팅 로봇 실현
실현 원리
사용자 가 채 팅 인터페이스 에서 Celery 비동기 작업 을 호출 하고 Celery 비동기 작업 이 실 행 된 후에 channels 에 결 과 를 보 낸 다음 에 channels 는 websocket 을 통 해 결 과 를 실시 간 으로 사용자 에 게 전달 합 니 다.간단 한 산술 연산 에 대해 서 는 셀 러 리 가 일반적으로 스스로 계산 하면 된다.인터넷 에서 시인 의 프로필 을 찾 는 임무 에 대해 서 는 Celery 가 Python 파충류(requests+parsel)를 이용 해 고시 문 사이트 의 시인 프로필 을 올 라 가 결 과 를 실시 간 으로 사용자 에 게 되 돌려 준다.
다음은 구체 적 인 코드 실현 을 살 펴 보 자.
첫 번 째 설치 환경 의존
우선 가상 환경 에 django 와 아래 의 주요 항목 의존 을 설치 합 니 다.이 항목 은 3.X 버 전 으로 최신 버 전 을 사 용 했 습 니 다.
#
pip install django
pip install channels
pip install channels_redis
pip install celery
pip install redis
pip install eventlet # windows only
#
pip install requests
pip install parsel
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'channels', # channels
'bots', # bots
]
# ASGI
ASGI_APPLICATION = 'myproject.asgi.application'
# redis ， channels_redis
import os
CHANNEL_LAYERS = {
"default": {
"BACKEND": "channels_redis.core.RedisChannelLayer",
"CONFIG": {
"hosts": [os.environ.get('REDIS_URL', 'redis://127.0.0.1:6379/2')],
},
},
}
# myproject/urls.py
from django.conf.urls import include
from django.urls import path
from django.contrib import admin
urlpatterns = [
path('bots/', include('bots.urls')),
path('admin/', admin.site.urls),
]pip 가 Celery 와 redis 를 설치 한 후에 우 리 는 그것 을 설정 해 야 합 니 다.my procject 디 렉 터 리 의 을 각각 수정 합 니 다.init__.py 와 celery.py(새로 만 들 기)는 다음 코드 를 추가 합 니 다.
# __init__.py
from .celery import app as celery_app
__all__ = ('celery_app',)
# celery.py
import os
from celery import Celery
#
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
#
app = Celery('myproject')
# namespace='CELERY' Django Celery
# Celery CELERY ，
app.config_from_object('django.conf:settings', namespace='CELERY')
# Django app
app.autodiscover_tasks()
#
@app.task(bind=True)
def debug_task(self):
print(f'Request: {self.request!r}')
# Celery
CELERY_BROKER_URL = "redis://127.0.0.1:6379/0"
CELERY_TIMEZONE = TIME_ZONE
# celery ， json
CELERY_ACCEPT_CONTENT = ['application/json', ]
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'세 번 째 단 계 는 로봇 채 팅 홈 페이지 를 작성 합 니 다.
이 예 는 django 일반 보기 함 수 를 이용 하여 첫 페이지(index)와 사용자 가 대화 하 는 채 팅 페이지 를 보 여 주 는 데 만 사용 할 수 있 습 니 다.이 페이지 에 대응 하 는 경로 와 보기 함 수 는 다음 과 같 습 니 다.
# bots/urls.py
from django.urls import path
from . import views
urlpatterns = [
path('', views.index, name='index'),
]
# bots/views.py
from django.shortcuts import render
def index(request):
return render(request, 'bots/index.html', {})
bots/
__init__.py
templates/
bots/
index.html
urls.py
views.py
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8"/>
<title>Django+Channels+Celery </title>
</head>
<body>
<textarea id="chat-log" cols="100" rows="20" readonly></textarea>
<br/>
<input id="chat-message-input" type="text" size="100"
placeholder=" `help` ."/><br/><input id="chat-message-submit" type="button" value="Send"/>
<script>
var wss_protocol = (window.location.protocol == 'https:') ? 'wss://': 'ws://';
var chatSocket = new WebSocket(
wss_protocol + window.location.host + '/ws/bots/'
);
chatSocket.onopen = function(e) {
document.querySelector('#chat-log').value +=
(' Django . `help` .
')}
chatSocket.onmessage = function(e) {
var data = JSON.parse(e.data);
var message = data['message'];
document.querySelector('#chat-log').value += (message + '
');
};
chatSocket.onclose = function(e) {
document.querySelector('#chat-log').value +=
('Socket closed unexpectedly, please reload the page.
')};
document.querySelector('#chat-message-input').focus();
document.querySelector('#chat-message-input').onkeyup = function(e) {
if (e.keyCode === 13) { // enter, return
document.querySelector('#chat-message-submit').click();
}
};
document.querySelector('#chat-message-submit').onclick = function(e) {
var messageInputDom = document.querySelector('#chat-message-input');
var message = messageInputDom.value;
chatSocket.send(JSON.stringify({
'message': message
}));
messageInputDom.value = '';
};
</script>
</body>
</html>channels 가 웹 소켓 연결 을 받 아들 일 때 루트 설정 에 따라 해당 하 는 처리 방법 을 찾 습 니 다.다만 channels 의 websocket 길 은 urls.py 에 설정 되 어 있 지 않 고 처리 함수 도 views.py 에 쓰 여 있 지 않 습 니 다.채널 에서 이 두 파일 은 각각 routing.py 와 consumers.py 로 바 뀌 었 다.
bots 에 새 routing.py 를 적용 하여 다음 코드 를 추가 합 니 다.뉴스/bots/에 보 내 는 웹 소켓 요청 을 BotConsumer 로 처리 하 는 역할 을 합 니 다.
from django.urls import re_path
from . import consumers
websocket_urlpatterns = [
re_path(r'ws/bots/$', consumers.BotConsumer.as_asgi()),
]Django 와 유사 합 니 다.이 app 의 websocket 경 로 를 프로젝트 의 루트 에 추가 해 야 합 니 다.my procject/asgi.py 를 편집 하고 다음 코드 를 추가 합 니 다.
# myproject/asgi.py
import os
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application
import chat.routing
import bots.routing
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")
application = ProtocolTypeRouter({
"http": get_asgi_application(),
# websocket
"websocket": AuthMiddlewareStack(
URLRouter(
bots.routing.websocket_urlpatterns
)
)
})
import json
from asgiref.sync import async_to_sync
from channels.generic.websocket import WebsocketConsumer
from . import tasks
COMMANDS = {
'help': {
'help': ' .',
},
'add': {
'args': 2,
'help': ' , : `add 12 32`.',
'task': 'add'
},
'search': {
'args': 1,
'help': ' ， : `search `.',
'task': 'search'
},
}
class BotConsumer(WebsocketConsumer):
def receive(self, text_data):
text_data_json = json.loads(text_data)
message = text_data_json['message']
response_message = ' `help` 。'
message_parts = message.split()
if message_parts:
command = message_parts[0].lower()
if command == 'help':
response_message = ' :
' + '
'.join(
[f'{command} - {params["help"]} ' for command, params in COMMANDS.items()])
elif command in COMMANDS:
if len(message_parts[1:]) != COMMANDS[command]['args']:
response_message = f' `{command}` ， .'
else:
getattr(tasks, COMMANDS[command]['task']).delay(self.channel_name, *message_parts[1:])
response_message = f' `{message}` .'
async_to_sync(self.channel_layer.send)(
self.channel_name,
{
'type': 'chat.message',
'message': response_message
}
)
def chat_message(self, event):
message = event['message']
# Send message to WebSocket
self.send(text_data=json.dumps({
'message': f'[ ]: {message}'
}))
getattr(tasks, COMMANDS[command]['task']).delay(self.channel_name, *message_parts[1:])bots 디 렉 터 리 에'tasks.py'를 새로 만 들 고 다음 코드 를 추가 합 니 다.
from asgiref.sync import async_to_sync
from celery import shared_task
from channels.layers import get_channel_layer
from parsel import Selector
import requests
channel_layer = get_channel_layer()
@shared_task
def add(channel_name, x, y):
message = '{}+{}={}'.format(x, y, int(x) + int(y))
async_to_sync(channel_layer.send)(channel_name, {"type": "chat.message", "message": message})
print(message)
@shared_task
def search(channel_name, name):
spider = PoemSpider(name)
result = spider.parse_page()
async_to_sync(channel_layer.send)(channel_name, {"type": "chat.message", "message": str(result)})
print(result)
class PoemSpider(object):
def __init__(self, keyword):
self.keyword = keyword
self.url = "https://so.gushiwen.cn/search.aspx"
def parse_page(self):
params = {'value': self.keyword}
response = requests.get(self.url, params=params)
if response.status_code == 200:
# Selector
selector = Selector(response.text)
# xpath
intro = selector.xpath('//textarea[starts-with(@id,"txtareAuthor")]/text()').get()
print("{} :{}".format(self.keyword, intro))
if intro:
return intro
print(" status:{}".format(response.status_code))
return " 。"주의:
-기본 획득 채널layer 방식 은 인터페이스 호출:channels.layers.getchannel_layer()。consumer 에서 인 터 페 이 스 를 호출 하 는 경우 self.channel 을 직접 사용 할 수 있 습 니 다.layer。
-channel layer 에 대한 방법(send(),group 포함send()，group_add()등 은 모두 비동기 적 인 방법 에 속 합 니 다.이 는 호출 할 때 await 를 사용 해 야 한 다 는 뜻 입 니 다.동기 코드 에서 사용 하려 면 장식 기 asgiref.sync.async 를 사용 해 야 합 니 다.to_sync
6 단계 운행 시 효과 보기
만약 의외 의 사고 가 발생 하지 않 는 다 면,당신 의 현재 프로젝트 구 조 는 아래 와 같 아야 합 니 다.솔직히 전체 프로젝트 에 파일 이 별로 없어 서 Python 의 간결 함 과 효율 성 이 정말 좋 기로 유명 합 니 다.
다음 명령 을 계속 실행 하면 우리 문장의 첫 효 과 를 볼 수 있 습 니 다.
# django
python manage.py makemigrations
python manage.py migrate
python manage.py runserver
# windows Celery eventlet
# Celery redis
Celery -A myproject worker -l info -P eventlet본 고 는 Django+Channel s+Celery+Redis 를 이용 하여 채 팅 로봇 을 만 들 었 습 니 다.산술 도 할 줄 알 고 고대 시문 도 찾 을 수 있 습 니 다.이 실현 원 리 를 빌려 아주 재 미 있 는 실시 간 채 팅 애플 리 케 이 션 을 만 들 수 있 습 니 다.예 를 들 어 온라인 실시 간 문답,온라인 고객 센터,실시 간 으로 주문 서 를 조회 하고 Django 버 전의 siri 미녀 등 입 니 다.
Django 채널+Websocket+Celery 채 팅 로봇 프로젝트 소스 주소:https://github.com/shiyunbo/django-channels-chatbot
이상 은 Django 가 채 팅 로봇 을 실현 하 는 상세 한 내용 입 니 다.더 많은 Django 채 팅 로봇 에 관 한 자 료 는 우리 의 다른 관련 글 을 주목 하 세 요!
이 내용에 흥미가 있습니까?
현재 기사가 여러분의 문제를 해결하지 못하는 경우 AI 엔진은 머신러닝 분석(스마트 모델이 방금 만들어져 부정확한 경우가 있을 수 있음)을 통해 가장 유사한 기사를 추천합니다:
Django의 질문 및 답변 웹사이트환영 친구, 이것은 우리의 새로운 블로그입니다. 이 블로그에서는 , 과 같은 Question-n-Answer 웹사이트를 만들고 있습니다. 이 웹사이트는 회원가입 및 로그인이 가능합니다. 로그인 후 사용자는 사용자의 ...
텍스트를 자유롭게 공유하거나 복사할 수 있습니다.하지만 이 문서의 URL은 참조 URL로 남겨 두십시오.
CC BY-SA 2.5, CC BY-SA 3.0 및 CC BY-SA 4.0에 따라 라이센스가 부여됩니다.
좋은 웹페이지 즐겨찾기
