бэта сайта, убрал мусор, чуть-чуть допилил код
This commit is contained in:
@@ -1,198 +1,71 @@
|
||||
from flask import Flask, request, jsonify
|
||||
import logging
|
||||
import random
|
||||
from typing import Dict, Any, List, Tuple, Optional
|
||||
import os
|
||||
from fastapi import FastAPI, Request, Form, Depends, Cookie, responses, templating, HTTPException
|
||||
from starlette.staticfiles import StaticFiles
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
|
||||
cities = {
|
||||
'москва': [['1540737/daa6e420d33102bf6947', '213044/7df73ae4cc715175059e'], "Россия"],
|
||||
'нью-йорк': [['1652229/728d5c86707054d4745f', '1030494/aca7ed7acefde2606bdc']],
|
||||
'париж': [["1652229/f77136c2364eb90a3ea8", '123494/aca7ed7acefd12e606bdc']]
|
||||
}
|
||||
|
||||
sessionStorage = {}
|
||||
app = FastAPI()
|
||||
templates = templating.Jinja2Templates(directory="static/templates")
|
||||
app.mount("/static", StaticFiles(directory="static"), name="static")
|
||||
|
||||
|
||||
@app.route('/post', methods=['POST'])
|
||||
def main():
|
||||
logging.info('Request: %r', request.json)
|
||||
response = {
|
||||
'session': request.json['session'],
|
||||
'version': request.json['version'],
|
||||
'response': {
|
||||
'end_session': False
|
||||
}
|
||||
}
|
||||
handle_dialog(response, request.json)
|
||||
logging.info('Response: %r', response)
|
||||
return jsonify(response)
|
||||
def is_logged_in(logged_in: Optional[str] = Cookie(None)):
|
||||
return bool(logged_in)
|
||||
|
||||
|
||||
def handle_dialog(res, req):
|
||||
user_id = req['session']['user_id']
|
||||
if req['session']['new']:
|
||||
res['response']['text'] = 'Привет! Назови своё имя!'
|
||||
sessionStorage[user_id] = {
|
||||
'first_name': None,
|
||||
'game_started': False
|
||||
}
|
||||
return
|
||||
@app.get("/", response_class=responses.HTMLResponse)
|
||||
def index(request: Request):
|
||||
return templates.TemplateResponse("index.html", {"request": request})
|
||||
|
||||
if sessionStorage[user_id]['first_name'] is None:
|
||||
first_name = get_first_name(req)
|
||||
if first_name is None:
|
||||
res['response']['text'] = 'Не расслышала имя. Повтори, пожалуйста!'
|
||||
else:
|
||||
sessionStorage[user_id]['first_name'] = first_name
|
||||
sessionStorage[user_id]['guessed_cities'] = []
|
||||
res['response']['text'] = f'Приятно познакомиться, {first_name.title()}. Я Алиса. Отгадаешь город по фото?'
|
||||
res['response']['buttons'] = [
|
||||
{
|
||||
'title': 'Да',
|
||||
'hide': True
|
||||
},
|
||||
{
|
||||
'title': 'Нет',
|
||||
'hide': True
|
||||
}
|
||||
]
|
||||
|
||||
@app.post("/login")
|
||||
async def login(password: str = Form(...)):
|
||||
if password == "password":
|
||||
response = responses.RedirectResponse(url="/logs", status_code=303)
|
||||
response.set_cookie(key="logged_in", value="true")
|
||||
return response
|
||||
return {"message": "Invalid password"}
|
||||
|
||||
|
||||
@app.get("/logs", response_class=responses.HTMLResponse)
|
||||
async def logs(request: Request, logged_in: bool = Depends(is_logged_in)):
|
||||
if logged_in:
|
||||
return templates.TemplateResponse("logs.html", {"request": request, "logs": get_logs()})
|
||||
else:
|
||||
if not sessionStorage[user_id]['game_started']:
|
||||
if 'да' in req['request']['nlu']['tokens']:
|
||||
if len(sessionStorage[user_id]['guessed_cities']) == 3:
|
||||
res['response']['text'] = 'Ты отгадал все города!'
|
||||
res['end_session'] = True
|
||||
else:
|
||||
sessionStorage[user_id]['game_started'] = True
|
||||
sessionStorage[user_id]['attempt'] = 1
|
||||
play_game(res, req)
|
||||
elif 'нет' in req['request']['nlu']['tokens']:
|
||||
res['response']['text'] = 'Ну и ладно!'
|
||||
res['end_session'] = True
|
||||
else:
|
||||
res['response']['text'] = 'Не поняла ответа! Так да или нет?'
|
||||
res['response']['buttons'] = [
|
||||
{
|
||||
'title': 'Да',
|
||||
'hide': True
|
||||
},
|
||||
{
|
||||
'title': 'Нет',
|
||||
'hide': True
|
||||
}
|
||||
]
|
||||
else:
|
||||
play_game(res, req)
|
||||
responses.RedirectResponse(url="/login", status_code=303)
|
||||
|
||||
|
||||
def play_game(res, req):
|
||||
user_id = req['session']['user_id']
|
||||
city = random.choice(list(cities.keys()))
|
||||
while city in sessionStorage[user_id]['guessed_cities']:
|
||||
city = random.choice(list(cities.keys()))
|
||||
sessionStorage[user_id]['city'] = city
|
||||
sessionStorage[user_id]['country'] = cities[city][1]
|
||||
sessionStorage[user_id]['guessed_cities'].append(city)
|
||||
images = cities[city][0]
|
||||
res['response']['card'] = {}
|
||||
res['response']['card']['type'] = 'BigImage'
|
||||
res['response']['card']['title'] = 'Что это за город?'
|
||||
res['response']['card']['image_id'] = random.choice(images)
|
||||
res['response']['text'] = 'Тогда сыграем! Как называется этот город?'
|
||||
|
||||
|
||||
def check_city(res, req):
|
||||
user_id = req['session']['user_id']
|
||||
if req['request']['original_utterance'].lower() == sessionStorage[user_id]['city']:
|
||||
if len(sessionStorage[user_id]['guessed_cities']) == 3:
|
||||
res['response']['text'] = f'Правильно! Ты отгадал все города!'
|
||||
res['end_session'] = True
|
||||
else:
|
||||
res['response']['text'] = f'Правильно! Следующий город - {play_game(res, req)}'
|
||||
sessionStorage[user_id]['game_started'] = False
|
||||
del sessionStorage[user_id]['attempt']
|
||||
|
||||
# Добавляем вопрос о стране
|
||||
res['response']['buttons'] = [
|
||||
{
|
||||
'title': 'Да',
|
||||
'hide': True,
|
||||
},
|
||||
{
|
||||
'title': 'Нет',
|
||||
'hide': True,
|
||||
}
|
||||
]
|
||||
@app.get("/logs/{log_id}", response_class=responses.HTMLResponse)
|
||||
async def log(request: Request, log_id: int, logged_in: bool = Depends(is_logged_in)):
|
||||
if logged_in:
|
||||
if not get_log(log_id):
|
||||
raise HTTPException(status_code=400)
|
||||
return templates.TemplateResponse("log.html", {"request": request,
|
||||
"log": get_log(log_id)})
|
||||
else:
|
||||
if sessionStorage[user_id]['attempt'] == 3:
|
||||
res['response'][
|
||||
'text'] = f'Вы проиграли! Правильный ответ - {sessionStorage[user_id]["city"]}. Сыграем еще?'
|
||||
del sessionStorage[user_id]['attempt']
|
||||
sessionStorage[user_id]['game_started'] = False
|
||||
res['response']['buttons'] = [
|
||||
{
|
||||
'title': 'Да',
|
||||
'hide': True,
|
||||
},
|
||||
{
|
||||
'title': 'Нет',
|
||||
'hide': True,
|
||||
}
|
||||
]
|
||||
else:
|
||||
sessionStorage[user_id]['attempt'] += 1
|
||||
res['response']['text'] = f'Неправильно. Попробуй еще раз!'
|
||||
responses.RedirectResponse(url="/login", status_code=303)
|
||||
|
||||
|
||||
def check_country(res, req):
|
||||
user_id = req['session']['user_id']
|
||||
if req['request']['original_utterance'].lower() == sessionStorage[user_id]['country']:
|
||||
res['response']['text'] = f'Правильно! Следующий город - {play_game(res, req)}'
|
||||
del sessionStorage[user_id]['attempt']
|
||||
res['response']['buttons'] = [
|
||||
{
|
||||
'title': 'Да',
|
||||
'hide': True,
|
||||
},
|
||||
{
|
||||
'title': 'Нет',
|
||||
'hide': True,
|
||||
}
|
||||
]
|
||||
else:
|
||||
if sessionStorage[user_id]['attempt'] == 3:
|
||||
res['response'][
|
||||
'text'] = f'Вы проиграли! Правильный ответ - {sessionStorage[user_id]["country"]}. Сыграем еще?'
|
||||
del sessionStorage[user_id]['attempt']
|
||||
sessionStorage[user_id]['game_started'] = False
|
||||
res['response']['buttons'] = [
|
||||
{
|
||||
'title': 'Да',
|
||||
'hide': True,
|
||||
},
|
||||
{
|
||||
'title': 'Нет',
|
||||
'hide': True,
|
||||
}
|
||||
]
|
||||
else:
|
||||
sessionStorage[user_id]['attempt'] += 1
|
||||
res['response']['text'] = f'Неправильно. Попробуй еще раз!'
|
||||
|
||||
def get_logs() -> List[Tuple[int, str]]:
|
||||
return [(int(os.path.basename(dir_path)), dir_path.split("/")[2].strip())
|
||||
for dir_path, _, filenames in os.walk("static/logs")
|
||||
if dir_path != "static/logs" and len(dir_path.split("/")) == 3]
|
||||
|
||||
|
||||
def get_city(req):
|
||||
for entity in req['request']['nlu']['entities']:
|
||||
if entity['type'] == 'YANDEX.GEO':
|
||||
return entity['value'].get('city', None)
|
||||
|
||||
|
||||
def get_first_name(req):
|
||||
for entity in req['request']['nlu']['entities']:
|
||||
if entity['type'] == 'YANDEX.FIO':
|
||||
return entity['value'].get('first_name', None)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
app.run(port=52520)
|
||||
def get_log(log_id: int) -> List[Dict[str, Any]]:
|
||||
log_dir = os.path.join("static/logs", str(log_id))
|
||||
return_dir = []
|
||||
for dir_path, _, filenames in os.walk(log_dir):
|
||||
if dir_path != log_dir:
|
||||
audio_file = os.path.join(dir_path, "audio.ogg")
|
||||
if not os.path.exists(audio_file):
|
||||
audio_file = os.path.join(dir_path, "audio.wav")
|
||||
text_file = os.path.join(dir_path, "yandex-text.txt")
|
||||
if not os.path.exists(text_file):
|
||||
text_file = os.path.join(dir_path, "google-text.txt")
|
||||
try:
|
||||
return_dir.append({"id": log_id, "audio_file": f"/{audio_file}",
|
||||
"text": open(text_file).read().split("\n")})
|
||||
except UnicodeDecodeError:
|
||||
pass
|
||||
return return_dir
|
||||
|
||||
Reference in New Issue
Block a user