import logging
import os
import tempfile
import threading
from collections import defaultdict
from http import HTTPStatus
from multiprocessing import shared_memory
from pathlib import Path
from typing import Any, List, Optional

import joblib
import numpy as np
from diskcache import Cache
from filelock import FileLock, Timeout
from flask import Flask, jsonify, request
from scipy.stats import rankdata
from sentence_transformers import SentenceTransformer
from sklearn.gaussian_process import GaussianProcessRegressor
from sklearn.gaussian_process.kernels import DotProduct, WhiteKernel
from sklearn.linear_model import Ridge

logging.config.fileConfig("logging.conf")

logger = logging.getLogger(__name__)

PATH_CACHE = Path(__file__).parent.absolute() / "app_data" / "sbert_cache"
PATH_MODELS = Path(__file__).parent.absolute() / "app_data" / "models"


class Server:
    def __init__(self):
        self._app = Flask(__name__)
        self._lock_directory: Path = Path(tempfile.gettempdir()) / ".locks"

        self._app.add_url_rule("/rerank/<user_id>", "sort", self._rerank, methods=["POST"])
        self._app.add_url_rule("/train/<user_id>", "train", self._train, methods=["POST"])

        self._featurizer = CachedSentenceTransformer("paraphrase-distilroberta-base-v1")

        self._generations = Cache(PATH_CACHE / "generations.db")
        self._generations.clear()

    def start(self, debug: bool = False, host: str = "0.0.0.0", port: int = 5000):
        self._app.run(debug=debug, host=host, port=port)

    def _rerank(self, user_id: str):
        logger.info("Got ranking request for [%s]", user_id)

        texts = request.get_json()

        model = self._load_model(user_id)

        if model is None:
            logger.info("No model for user [%s] yet", user_id)
            ranks = np.arange(len(texts))
            rng = np.random.default_rng()
            rng.shuffle(ranks)
        else:
            logger.info("Predicting for user [%s] with generation [%s]", user_id, self._generations.get(user_id, 0))
            Xf = self._featurizer.featurize(texts)
            y = model.predict(Xf)
            ranks = np.argsort(y)

        return jsonify([int(r) for r in ranks])

    def _train(self, user_id: str):
        logger.info("Got training request for [%s]", user_id)
        json_data = request.get_json()
        texts = json_data["texts"]
        times = json_data["times"]

        try:
            # The lock needs to be acquired out here, not in the fn scope, else it would
            # just throw the Timeout inside fn.
            lock = self._get_lock(user_id)
            lock.acquire()

            def _fn():
                try:
                    Xf = self._featurizer.featurize(texts)
                    model = GaussianProcessRegressor(kernel=(DotProduct() + WhiteKernel())).fit(Xf, times)
                    self._save_model(user_id, model)
                finally:
                    lock.release()

            # We spawn a thread and run the training in there so that this HTTP request can return directly
            threading.Thread(target=_fn, daemon=True).start()
            return HTTPStatus.NO_CONTENT.description, HTTPStatus.NO_CONTENT.value
        except Timeout:
            logger.info("Already training for user [%s], skipping!", user_id)
            return HTTPStatus.TOO_MANY_REQUESTS.description, HTTPStatus.TOO_MANY_REQUESTS.value

    def _get_lock(self, user_id: str) -> FileLock:
        self._lock_directory.mkdir(parents=True, exist_ok=True)
        lock_path = self._lock_directory / f"{user_id}.lock"
        return FileLock(lock_path, timeout=1)

    def _load_model(self, user_id: str) -> Optional[Any]:
        model_path = self._get_model_path(user_id)
        logger.info("Reading model from [%s]", model_path)
        if model_path.is_file():
            logger.debug("Model found for [%s]", model_path)
            return joblib.load(model_path)
        else:
            logger.debug("No model found for [%s]", model_path)
            return None

    def _save_model(self, user_id: str, model: Any):
        model_path = self._get_model_path(user_id)
        logger.info("Writing model to [%s]", model_path)
        model_path.parent.mkdir(parents=True, exist_ok=True)
        tmp_model_path = model_path.with_suffix(".joblib.tmp")
        joblib.dump(model, tmp_model_path)
        os.replace(tmp_model_path, model_path)

        self._generations.incr(user_id)

    def _get_model_path(self, user_id: str) -> Path:
        return PATH_MODELS / f"model_{user_id}.joblib"


class CachedSentenceTransformer:
    def __init__(self, model_name: str):
        super().__init__()
        self._model = SentenceTransformer(model_name)
        self._cache = Cache(PATH_CACHE / model_name)

    def featurize(self, sentences: List[str]) -> np.ndarray:
        not_in_cache = [s for s in sentences if s not in self._cache]
        vecs_not_in_cache = list(self._model.encode(not_in_cache))

        assert len(not_in_cache) == len(vecs_not_in_cache)

        for s, vec in zip(not_in_cache, vecs_not_in_cache):
            self._cache[s] = vec

        return np.array([self._cache[s].squeeze() for s in sentences])


server = Server()
app = server._app

if __name__ == "__main__":
    server.start(debug=True, port=5000)

import logging
import os
import tempfile
import threading
from collections import defaultdict
from http import HTTPStatus
from multiprocessing import shared_memory
from pathlib import Path
from typing import Any, List, Optional

import joblib
import numpy as np
from diskcache import Cache
from filelock import FileLock, Timeout
from flask import Flask, jsonify, request
from scipy.stats import rankdata
from sentence_transformers import SentenceTransformer
from sklearn.gaussian_process import GaussianProcessRegressor
from sklearn.gaussian_process.kernels import DotProduct, WhiteKernel
from sklearn.linear_model import Ridge

logging.config.fileConfig("logging.conf")

logger = logging.getLogger(__name__)

PATH_CACHE = Path(__file__).parent.absolute() / "app_data" / "sbert_cache"
PATH_MODELS = Path(__file__).parent.absolute() / "app_data" / "models"


class Server:
    def __init__(self):
        self._app = Flask(__name__)
        self._lock_directory: Path = Path(tempfile.gettempdir()) / ".locks"

        self._app.add_url_rule("/rerank/<user_id>", "sort", self._rerank, methods=["POST"])
        self._app.add_url_rule("/train/<user_id>", "train", self._train, methods=["POST"])

        self._featurizer = CachedSentenceTransformer("paraphrase-distilroberta-base-v1")

        self._generations = Cache(PATH_CACHE / "generations.db")
        self._generations.clear()

    def start(self, debug: bool = False, host: str = "0.0.0.0", port: int = 5000):
        self._app.run(debug=debug, host=host, port=port)

    def _rerank(self, user_id: str):
        logger.info("Got ranking request for [%s]", user_id)

        texts = request.get_json()

        model = self._load_model(user_id)

        if model is None:
            logger.info("No model for user [%s] yet", user_id)
            ranks = np.arange(len(texts))
            rng = np.random.default_rng()
            rng.shuffle(ranks)
        else:
            logger.info("Predicting for user [%s] with generation [%s]", user_id, self._generations.get(user_id, 0))
            Xf = self._featurizer.featurize(texts)
            y = model.predict(Xf)
            ranks = np.argsort(y)

        return jsonify([int(r) for r in ranks])

    def _train(self, user_id: str):
        logger.info("Got training request for [%s]", user_id)
        json_data = request.get_json()
        texts = json_data["texts"]
        times = json_data["times"]

        try:
            # The lock needs to be acquired out here, not in the fn scope, else it would
            # just throw the Timeout inside fn.
            lock = self._get_lock(user_id)
            lock.acquire()

            def _fn():
                try:
                    Xf = self._featurizer.featurize(texts)
                    model = GaussianProcessRegressor(kernel=(DotProduct() + WhiteKernel())).fit(Xf, times)
                    self._save_model(user_id, model)
                finally:
                    lock.release()

            # We spawn a thread and run the training in there so that this HTTP request can return directly
            threading.Thread(target=_fn, daemon=True).start()
            return HTTPStatus.NO_CONTENT.description, HTTPStatus.NO_CONTENT.value
        except Timeout:
            logger.info("Already training for user [%s], skipping!", user_id)
            return HTTPStatus.TOO_MANY_REQUESTS.description, HTTPStatus.TOO_MANY_REQUESTS.value

    def _get_lock(self, user_id: str) -> FileLock:
        self._lock_directory.mkdir(parents=True, exist_ok=True)
        lock_path = self._lock_directory / f"{user_id}.lock"
        return FileLock(lock_path, timeout=1)

    def _load_model(self, user_id: str) -> Optional[Any]:
        model_path = self._get_model_path(user_id)
        logger.info("Reading model from [%s]", model_path)
        if model_path.is_file():
            logger.debug("Model found for [%s]", model_path)
            return joblib.load(model_path)
        else:
            logger.debug("No model found for [%s]", model_path)
            return None

    def _save_model(self, user_id: str, model: Any):
        model_path = self._get_model_path(user_id)
        logger.info("Writing model to [%s]", model_path)
        model_path.parent.mkdir(parents=True, exist_ok=True)
        tmp_model_path = model_path.with_suffix(".joblib.tmp")
        joblib.dump(model, tmp_model_path)
        os.replace(tmp_model_path, model_path)

        self._generations.incr(user_id)

    def _get_model_path(self, user_id: str) -> Path:
        return PATH_MODELS / f"model_{user_id}.joblib"


class CachedSentenceTransformer:
    def __init__(self, model_name: str):
        super().__init__()
        self._model = SentenceTransformer(model_name)
        self._cache = Cache(PATH_CACHE / model_name)

    def featurize(self, sentences: List[str]) -> np.ndarray:
        not_in_cache = [s for s in sentences if s not in self._cache]
        vecs_not_in_cache = list(self._model.encode(not_in_cache))

        assert len(not_in_cache) == len(vecs_not_in_cache)

        for s, vec in zip(not_in_cache, vecs_not_in_cache):
            self._cache[s] = vec

        return np.array([self._cache[s].squeeze() for s in sentences])


server = Server()
app = server._app

if __name__ == "__main__":
    server.start(debug=True, port=5000)

# The lock needs to be acquired out here, not in the fn scope, else it would # just throw the Timeout inside fn. # We spawn a thread and run the training in there so that this HTTP request can return directly