dream
97 строк · 2.6 Кб
1#!/usr/bin/env python
2
3import logging4import time5import os6import random7
8import sentry_sdk9from flask import Flask, request, jsonify10from sentry_sdk.integrations.logging import ignore_logger11
12from common.dff.integration.actor import load_ctxs, get_response13from scenario.main import actor14
15import test_server16
17
18ignore_logger("root")19
20sentry_sdk.init(os.getenv("SENTRY_DSN"))21SERVICE_NAME = os.getenv("SERVICE_NAME")22SERVICE_PORT = int(os.getenv("SERVICE_PORT"))23RANDOM_SEED = int(os.getenv("RANDOM_SEED", 2718))24
25logging.basicConfig(26format="%(asctime)s - %(pathname)s - %(lineno)d - %(levelname)s - %(message)s",27level=logging.DEBUG,28)
29logger = logging.getLogger(__name__)30
31
32app = Flask(__name__)33logging.getLogger("werkzeug").setLevel("WARNING")34
35
36def handler(requested_data, random_seed=None):37st_time = time.time()38ctxs = load_ctxs(requested_data)39random_seed = requested_data.get("random_seed", random_seed) # for tests40
41responses = []42for ctx in ctxs:43try:44# for tests45if random_seed:46random.seed(int(random_seed))47ctx = actor(ctx)48responses.append(get_response(ctx, actor))49except Exception as exc:50sentry_sdk.capture_exception(exc)51logger.exception(exc)52responses.append(("", 0.0, {}, {}, {}))53
54total_time = time.time() - st_time55logger.info(f"{SERVICE_NAME} exec time = {total_time:.3f}s")56return responses57
58
59try:60test_server.run_test(handler)61logger.info("test query processed")62except Exception as exc:63sentry_sdk.capture_exception(exc)64logger.exception(exc)65raise exc66
67
68logger.info(f"{SERVICE_NAME} is loaded and ready")69
70# import pathlib
71# import json
72
73# for in_file in pathlib.Path("tests").glob("./*_in.json"):
74# logger.error(in_file)
75# test_in = json.load(in_file.open())
76# responses = handler(test_in, RANDOM_SEED)
77# out_file = str(in_file).replace("in.json", "out.json")
78# import common.test_utils as t_utils
79
80# t_utils.save_to_test(responses, out_file, indent=4) # TEST
81
82
83@app.route("/respond", methods=["POST"])84def respond():85# import common.test_utils as t_utils86
87# t_utils.save_to_test(request.json, "tests/lets_talk_in.json", indent=4) # TEST88# responses = handler(request.json, RANDOM_SEED) # TEST89# import common.test_utils as t_utils90
91# t_utils.save_to_test(responses, "tests/lets_talk_out.json", indent=4) # TEST92responses = handler(request.json)93return jsonify(responses)94
95
96if __name__ == "__main__":97app.run(debug=False, host="0.0.0.0", port=SERVICE_PORT)98