#   Copyright 2021 Modelyst LLC
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.

"""Objects related to the running of Models and Generators."""
from bdb import BdbQuit
from datetime import datetime, timedelta
from time import time
from typing import Any, Dict, List, Mapping, Optional, Set, Tuple
from uuid import UUID

from psycopg import connect as pg3_connect
from pydantic.fields import Field, PrivateAttr
from pydasher import hasher
from sqlalchemy.future import Engine
from sqlmodel import Session, select
from tqdm import tqdm

import dbgen.exceptions as exceptions
from dbgen.core.base import Base, encoders
from dbgen.core.generator import Generator
from dbgen.core.metadata import (
    GeneratorEntity,
    GeneratorRunEntity,
    GensToRun,
    ModelEntity,
    Repeats,
    RunEntity,
    Status,
)
from dbgen.core.model import Model
from dbgen.core.node.extract import Extract
from dbgen.core.node.query import BaseQuery
from dbgen.exceptions import DBgenExternalError, DBgenSkipException, RepeatException, SerializationError
from dbgen.utils.log import LogLevel


class RunConfig(Base):
    """Configuration for the running of a Generator and Model"""

    retry: bool = False
    include: Set[str] = Field(default_factory=set)
    exclude: Set[str] = Field(default_factory=set)
    start: Optional[str]
    until: Optional[str]
    batch_size: Optional[int]
    progress_bar: bool = True
    log_level: LogLevel = LogLevel.INFO

    def should_gen_run(self, generator: Generator) -> bool:
        """Check a generator against include/exclude to see if it should run."""
        markers = [generator.name, *generator.tags]
        should_run = any(
            map(
                lambda x: (not self.include or x in self.include) and x not in self.exclude,
                markers,
            )
        )
        return should_run

    def get_invalid_markers(self, model: Model) -> Dict[str, List[str]]:
        """Check that all inputs to RunConfig are meaningful for the model."""
        invalid_marks = {}
        gen_names = model.gens().keys()
        # Validate start and until
        for attr in ("start", "until"):
            val: str = getattr(self, attr)
            if val is not None and val not in gen_names:
                invalid_marks[attr] = [val]
        # Validate include and exclude as sets
        for attr in ("include", "exclude"):
            set_val: Set[str] = getattr(self, attr)
            invalid_vals = [x for x in set_val if not model.validate_marker(x)]
            if invalid_vals:
                invalid_marks[attr] = invalid_vals

        return invalid_marks


def update_run_by_id(run_id, status: Status, session: Session):
    run = session.get(RunEntity, run_id)
    assert run, f"No run found with id {run_id}"
    run.status = status
    session.commit()


class RunInitializer(Base):
    """Intializes a run by syncing the database and getting the run_id."""

    def execute(self, engine: Engine, run_config: RunConfig) -> int:
        # Use some metadatabase connection to initialize initialize the run
        # Store the details of the run on the metadatabase so downstream GeneratorRuns can pick them up
        # Sync the database with the registries
        with Session(engine) as session:
            run = RunEntity(status=Status.initialized)
            session.add(run)
            session.commit()
            session.refresh(run)
            assert isinstance(run.id, int)
            run.status = Status.running
            session.commit()
            run_id = run.id

        return run_id


class BaseGeneratorRun(Base):
    """A lightwieght wrapper for the Generator that grabs a specific Generator from metadatabase and runs it."""

    _old_repeats: Set[UUID] = PrivateAttr(default_factory=set)
    _new_repeats: Set[UUID] = PrivateAttr(default_factory=set)

    def get_gen(self, meta_engine: Engine, *args, **kwargs) -> Generator:
        raise NotImplementedError

    def execute(
        self,
        main_engine: Engine,
        meta_engine: Engine,
        run_id: Optional[int],
        run_config: Optional[RunConfig],
        ordering: Optional[int],
    ):

        # Set default values for run_config if none provided
        if run_config is None:
            run_config = RunConfig()

        generator = self.get_gen(meta_engine=meta_engine)
        # Initialize the generator_row in the meta database
        meta_session = Session(meta_engine)

        gen_run = self._initialize_gen_run(
            generator=generator, session=meta_session, run_id=run_id, ordering=ordering
        )
        # Check if our run config excludes our generator
        if not run_config.should_gen_run(generator):
            self._logger.info(f'Excluding generator {generator.name!r}')
            gen_run.status = Status.excluded
            meta_session.commit()
            return
        # Start the Generator
        self._logger.info(f'Running generator {generator.name!r}...')
        gen_run.status = Status.running
        meta_session.commit()
        start = time()
        # Set the extractor
        self._logger.debug('Initializing extractor')
        extractor_connection = main_engine.connect()
        extract = generator.extract
        if isinstance(extract, BaseQuery):
            extract.set_extractor(connection=extractor_connection)
        else:
            extract.set_extractor()
        self._logger.debug('Fetching extractor length')
        row_count = extract.length(connection=extractor_connection)
        gen_run.inputs_extracted = row_count
        meta_session.commit()

        self._logger.debug('Fetching repeats')
        # Query the repeats table for input_hashes that match this generator's hash
        self._old_repeats = set(
            meta_session.exec(select(Repeats.input_hash).where(Repeats.generator_id == generator.uuid)).all()
        )
        # The batch_size is set either on the run_config or the generator
        batch_size = run_config.batch_size or generator.batch_size
        assert batch_size is None or batch_size > 0, f"Invalid batch size batch_size must be >0: {batch_size}"
        # Open raw connections for fast loading
        main_raw_connection = pg3_connect(str(main_engine.url))
        meta_raw_connection = meta_engine.raw_connection()
        batch_done = lambda x: x % batch_size == 0 if batch_size is not None else False
        # Start while loop to iterate through the nodes
        self._logger.info('Looping through extracted rows...')
        progress_bar = tqdm(
            total=row_count,
            position=1,
            leave=False,
            desc="Transforming...",
            disable=not run_config.progress_bar,
        )
        try:
            while True:
                gen_run.inputs_processed += 1
                row: Dict[str, Mapping[str, Any]] = {}
                try:
                    for node in generator._sort_graph():
                        output = node.run(row)
                        # Extract outputs need to be fed to our repeat checker and need to be checked for stop iterations
                        if isinstance(node, Extract):
                            if output is None or batch_done(gen_run.inputs_processed):
                                self._logger.debug('loading batch...')
                                self._load_repeats(meta_raw_connection, generator)
                                rows_inserted, rows_updated = self._load(main_raw_connection, generator)
                                gen_run.rows_inserted += rows_inserted
                                gen_run.rows_updated += rows_updated
                                meta_session.commit()
                                self._logger.debug('done loading batch.')
                                self._logger.debug(f'inserted {rows_inserted} rows.')
                                self._logger.debug(f'updated {rows_updated} rows.')
                            # if we are out of rows break out of while loop
                            if output is None:
                                raise StopIteration
                            is_repeat, input_hash = self._check_repeat(output, generator.uuid)
                            if not run_config.retry and is_repeat:
                                raise RepeatException()
                        row[node.hash] = output  # type: ignore
                    if not is_repeat:
                        self._new_repeats.add(input_hash)
                        gen_run.unique_inputs += 1
                    progress_bar.update()
                # Stop iteration is used to catch the empty extractor
                except StopIteration:
                    break
                # A repeated input from the extract will also cause a row to be skipped
                except RepeatException:
                    continue
                # Any node can raise a skip exception to skip the input before loading
                except DBgenSkipException as exc:
                    self._logger.debug(f"Skipped Row: {exc.msg}")
                    gen_run.inputs_skipped += 1
                # External errors are raised whenever a node fails due to internal logic
                except DBgenExternalError as e:
                    msg = f"\n\nError when running generator {generator.name}\n"
                    self._logger.error(msg)
                    self._logger.error(f"\n{e}")
                    gen_run.status = Status.failed
                    gen_run.error = str(e)
                    run = meta_session.get(RunEntity, run_id)
                    assert run
                    run.errors = run.errors + 1 if run.errors else 1
                    meta_session.commit()
                    meta_session.close()
                    return 2
                except (
                    Exception,
                    KeyboardInterrupt,
                    SystemExit,
                    BdbQuit,
                ) as e:
                    gen_run.status = Status.failed
                    gen_run.error = (
                        f"Uncaught Error encountered during running of generator {generator.name}: {e!r}"
                    )
                    update_run_by_id(run_id, Status.failed, meta_session)
                    raise
        # Close all connections
        finally:
            gen_run.runtime = round(time() - start, 3)
            meta_session.commit()
            main_raw_connection.close()
            meta_raw_connection.close()
            extractor_connection.close()

        gen_run.status = Status.completed
        gen_run.runtime = round(time() - start, 3)
        self._logger.info(
            f"Finished running generator {generator.name}({generator.uuid}) in {gen_run.runtime}(s)."
        )
        self._logger.info(f"Loaded approximately {gen_run.rows_inserted} rows")
        meta_session.commit()
        meta_session.close()
        return 0

    def _initialize_gen_run(
        self,
        session: Session,
        generator: Generator,
        run_id: Optional[int],
        ordering: Optional[int],
    ) -> GeneratorRunEntity:
        # if no run_id is provided create one and mark it as a testing run
        if run_id is None:
            run = RunEntity(status='testing')
            session.add(run)
            session.commit()
            session.refresh(run)
            ordering = 0
            run_id = run.id
        gen_row = generator._get_gen_row()
        session.merge(gen_row)
        session.commit()
        query = generator.extract.query if isinstance(generator.extract, BaseQuery) else ''
        gen_run = GeneratorRunEntity(
            run_id=run_id,
            generator_id=gen_row.id,
            status=Status.initialized,
            ordering=ordering,
            query=query,
        )
        session.add(gen_run)
        session.commit()
        session.refresh(gen_run)
        return gen_run

    def _load(self, connection, generator: Generator) -> Tuple[int, int]:
        rows_inserted = 0
        rows_updated = 0
        for load in generator._sorted_loads():
            if load.insert:
                rows_inserted += len(load._output)
            else:
                rows_updated += len(load._output)
            load.load(connection, gen_id=self.uuid)
        return (rows_inserted, rows_updated)

    def _load_repeats(self, connection, generator: Generator) -> None:
        rows = ((generator.uuid, input_hash) for input_hash in self._new_repeats)
        Repeats._quick_load(connection, rows, column_names=["generator_id", "input_hash"])
        self._old_repeats = self._old_repeats.union(self._new_repeats)
        self._new_repeats = set()

    def _check_repeat(self, extracted_dict: Dict[str, Any], generator_uuid: UUID) -> Tuple[bool, UUID]:
        # Convert Row to a dictionary so we can hash it for repeat-checking
        input_hash = UUID(hasher((generator_uuid, extracted_dict), encoders=encoders))
        # If the input_hash has been seen and we don't have retry=True skip row
        is_repeat = input_hash in self._old_repeats or input_hash in self._new_repeats
        return (is_repeat, input_hash)


class GeneratorRun(BaseGeneratorRun):
    generator: Generator

    def get_gen(self, meta_engine: Engine, *args, **kwargs):
        return self.generator


class RemoteGeneratorRun(BaseGeneratorRun):
    generator_id: UUID

    def get_gen(self, meta_engine, *args, **kwargs):
        with Session(meta_engine) as sess:
            gen_json = sess.exec(
                select(GeneratorEntity.gen_json).where(GeneratorEntity.id == self.generator_id)
            ).one()
            try:
                generator = Generator.deserialize(gen_json)
            except ModuleNotFoundError as exc:
                import os

                raise SerializationError(
                    f"While deserializing generator id {self.generator_id} an unknown module was encountered. Are you using custom dbgen objects reachable by your python environment? Make sure any custom extractors or code can be found in your PYTHONPATH environment variable\nError: {exc}\nPYTHONPATH={os.environ.get('PYTHONPATH')}"
                ) from exc
            if generator.uuid != self.generator_id:
                error = f"Deserialization Failed the generator hash has changed for generator named {generator.name}!\n{generator}\n{self.generator_id}"
                raise exceptions.SerializationError(error)
        return generator


class ModelRun(Base):
    model: Model

    def get_gen_run(self, generator: Generator) -> BaseGeneratorRun:
        return GeneratorRun(generator=generator)

    def execute(
        self,
        main_engine: Engine,
        meta_engine: Engine,
        run_config: RunConfig = None,
        nuke: bool = False,
        rerun_failed: bool = False,
    ) -> RunEntity:
        start = time()
        if run_config is None:
            run_config = RunConfig()
        # Sync the Database statew with the model state
        self.model.sync(main_engine, meta_engine, nuke=nuke)

        # If doing last failed run query for gens to run and add to include
        if rerun_failed:
            with meta_engine.connect() as conn:
                result = conn.execute(select(GensToRun.__table__.c.name))
                for (gen_name,) in result:
                    run_config.include.add(gen_name)

        # Initialize the run
        run_init = RunInitializer()
        run_id = run_init.execute(meta_engine, run_config)
        sorted_generators = self.model._sort_graph()
        # Add generators to metadb
        with Session(meta_engine) as meta_session:
            model_row = self.model._get_model_row()
            model_row.last_run = datetime.now()
            existing_model = meta_session.get(ModelEntity, model_row.id)
            if not existing_model:
                meta_session.merge(model_row)
            else:
                existing_model.last_run = datetime.now()
            meta_session.commit()

        # Apply start and until to exclude generators not between start_idx and until_idx
        if run_config.start or run_config.until:
            gen_names = [gen.name for gen in sorted_generators]
            start_idx = gen_names.index(run_config.start) if run_config.start else 0
            until_idx = gen_names.index(run_config.until) + 1 if run_config.until else len(gen_names)
            # Modify include to only include the gen_names that pass the test
            run_config.include = run_config.include.union(gen_names[start_idx:until_idx])
            print(f"Only running generators: {gen_names[start_idx:until_idx]} due to start/until")
            self._logger.debug(
                f"Only running generators: {gen_names[start_idx:until_idx]} due to start/until"
            )
        with tqdm(total=len(sorted_generators), position=0, disable=not run_config.progress_bar) as tq:
            for i, generator in enumerate(sorted_generators):
                tq.set_description(generator.name)
                gen_run = self.get_gen_run(generator)
                gen_run.execute(main_engine, meta_engine, run_id, run_config, ordering=i)
                tq.update()

        # Complete run
        with Session(meta_engine) as session:
            update_run_by_id(run_id, Status.completed, session)
            run = session.get(RunEntity, run_id)
            assert run
            run.runtime = timedelta(seconds=time() - start)
            session.commit()
            session.refresh(run)
        return run


class RemoteModelRun(ModelRun):
    def get_gen_run(self, generator):
        return RemoteGeneratorRun(generator_id=generator.uuid)

pydantic.fields.PrivateAttr tqdm.tqdm dbgen.exceptions.RepeatException sqlmodel.Session pydasher.hasher sqlmodel.select dbgen.core.metadata.GeneratorRunEntity dbgen.core.metadata.Repeats._quick_load time.time dbgen.core.metadata.RunEntity os.environ.get pydantic.fields.Field dbgen.core.generator.Generator.deserialize dbgen.exceptions.SerializationError datetime.datetime.now
[((1732, 1758), 'pydantic.fields.Field', 'Field', ([], {'default_factory': 'set'}), '(default_factory=set)\n', (1737, 1758), False, 'from pydantic.fields import Field, PrivateAttr\n'), ((1783, 1809), 'pydantic.fields.Field', 'Field', ([], {'default_factory': 'set'}), '(default_factory=set)\n', (1788, 1809), False, 'from pydantic.fields import Field, PrivateAttr\n'), ((4267, 4299), 'pydantic.fields.PrivateAttr', 'PrivateAttr', ([], {'default_factory': 'set'}), '(default_factory=set)\n', (4278, 4299), False, 'from pydantic.fields import Field, PrivateAttr\n'), ((4330, 4362), 'pydantic.fields.PrivateAttr', 'PrivateAttr', ([], {'default_factory': 'set'}), '(default_factory=set)\n', (4341, 4362), False, 'from pydantic.fields import Field, PrivateAttr\n'), ((4946, 4966), 'sqlmodel.Session', 'Session', (['meta_engine'], {}), '(meta_engine)\n', (4953, 4966), False, 'from sqlmodel import Session, select\n'), ((5577, 5583), 'time.time', 'time', ([], {}), '()\n', (5581, 5583), False, 'from time import time\n'), ((7051, 7162), 'tqdm.tqdm', 'tqdm', ([], {'total': 'row_count', 'position': '(1)', 'leave': '(False)', 'desc': '"""Transforming..."""', 'disable': '(not run_config.progress_bar)'}), "(total=row_count, position=1, leave=False, desc='Transforming...',\n    disable=not run_config.progress_bar)\n", (7055, 7162), False, 'from tqdm import tqdm\n'), ((12159, 12281), 'dbgen.core.metadata.GeneratorRunEntity', 'GeneratorRunEntity', ([], {'run_id': 'run_id', 'generator_id': 'gen_row.id', 'status': 'Status.initialized', 'ordering': 'ordering', 'query': 'query'}), '(run_id=run_id, generator_id=gen_row.id, status=Status.\n    initialized, ordering=ordering, query=query)\n', (12177, 12281), False, 'from dbgen.core.metadata import GeneratorEntity, GeneratorRunEntity, GensToRun, ModelEntity, Repeats, RunEntity, Status\n'), ((13037, 13123), 'dbgen.core.metadata.Repeats._quick_load', 'Repeats._quick_load', (['connection', 'rows'], {'column_names': "['generator_id', 'input_hash']"}), "(connection, rows, column_names=['generator_id',\n    'input_hash'])\n", (13056, 13123), False, 'from dbgen.core.metadata import GeneratorEntity, GeneratorRunEntity, GensToRun, ModelEntity, Repeats, RunEntity, Status\n'), ((15403, 15409), 'time.time', 'time', ([], {}), '()\n', (15407, 15409), False, 'from time import time\n'), ((3754, 3769), 'sqlmodel.Session', 'Session', (['engine'], {}), '(engine)\n', (3761, 3769), False, 'from sqlmodel import Session, select\n'), ((3800, 3836), 'dbgen.core.metadata.RunEntity', 'RunEntity', ([], {'status': 'Status.initialized'}), '(status=Status.initialized)\n', (3809, 3836), False, 'from dbgen.core.metadata import GeneratorEntity, GeneratorRunEntity, GensToRun, ModelEntity, Repeats, RunEntity, Status\n'), ((11778, 11805), 'dbgen.core.metadata.RunEntity', 'RunEntity', ([], {'status': '"""testing"""'}), "(status='testing')\n", (11787, 11805), False, 'from dbgen.core.metadata import GeneratorEntity, GeneratorRunEntity, GensToRun, ModelEntity, Repeats, RunEntity, Status\n'), ((13432, 13491), 'pydasher.hasher', 'hasher', (['(generator_uuid, extracted_dict)'], {'encoders': 'encoders'}), '((generator_uuid, extracted_dict), encoders=encoders)\n', (13438, 13491), False, 'from pydasher import hasher\n'), ((13992, 14012), 'sqlmodel.Session', 'Session', (['meta_engine'], {}), '(meta_engine)\n', (13999, 14012), False, 'from sqlmodel import Session, select\n'), ((16141, 16161), 'sqlmodel.Session', 'Session', (['meta_engine'], {}), '(meta_engine)\n', (16148, 16161), False, 'from sqlmodel import Session, select\n'), ((16264, 16278), 'datetime.datetime.now', 'datetime.now', ([], {}), '()\n', (16276, 16278), False, 'from datetime import datetime, timedelta\n'), ((17768, 17788), 'sqlmodel.Session', 'Session', (['meta_engine'], {}), '(meta_engine)\n', (17775, 17788), False, 'from sqlmodel import Session, select\n'), ((11151, 11157), 'time.time', 'time', ([], {}), '()\n', (11155, 11157), False, 'from time import time\n'), ((14217, 14248), 'dbgen.core.generator.Generator.deserialize', 'Generator.deserialize', (['gen_json'], {}), '(gen_json)\n', (14238, 14248), False, 'from dbgen.core.generator import Generator\n'), ((14953, 14989), 'dbgen.exceptions.SerializationError', 'exceptions.SerializationError', (['error'], {}), '(error)\n', (14982, 14989), True, 'import dbgen.exceptions as exceptions\n'), ((16493, 16507), 'datetime.datetime.now', 'datetime.now', ([], {}), '()\n', (16505, 16507), False, 'from datetime import datetime, timedelta\n'), ((10902, 10908), 'time.time', 'time', ([], {}), '()\n', (10906, 10908), False, 'from time import time\n'), ((15783, 15817), 'sqlmodel.select', 'select', (['GensToRun.__table__.c.name'], {}), '(GensToRun.__table__.c.name)\n', (15789, 15817), False, 'from sqlmodel import Session, select\n'), ((17981, 17987), 'time.time', 'time', ([], {}), '()\n', (17985, 17987), False, 'from time import time\n'), ((6308, 6334), 'sqlmodel.select', 'select', (['Repeats.input_hash'], {}), '(Repeats.input_hash)\n', (6314, 6334), False, 'from sqlmodel import Session, select\n'), ((8828, 8845), 'dbgen.exceptions.RepeatException', 'RepeatException', ([], {}), '()\n', (8843, 8845), False, 'from dbgen.exceptions import DBgenExternalError, DBgenSkipException, RepeatException, SerializationError\n'), ((14072, 14104), 'sqlmodel.select', 'select', (['GeneratorEntity.gen_json'], {}), '(GeneratorEntity.gen_json)\n', (14078, 14104), False, 'from sqlmodel import Session, select\n'), ((14668, 14696), 'os.environ.get', 'os.environ.get', (['"""PYTHONPATH"""'], {}), "('PYTHONPATH')\n", (14682, 14696), False, 'import os\n')]