"""SQL queries."""

import concurrent.futures
import itertools
from contextlib import nullcontext
from typing import Any, cast

from flask import g
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from sqlalchemy.sql import functions

from video import config
from video.connectors import mysql
from video.constants import job_statuses
from video.models.sql.classes.context import Context
from video.models.sql.classes.context_field import ContextField
from video.models.sql.classes.job import Job
from video.models.sql.classes.job_input import JobInput
from video.models.sql.classes.job_output import JobOutput
from video.models.sql.classes.job_status import JobStatus
from video.models.sql.classes.release import Release
from video.models.sql.classes.track import Track
from video.models.sql.classes.track_video import TrackVideo
from video.models.sql.classes.video_asset import VideoAsset
from video.models.sql.classes.video_dashboard_item import VideoDashboardItem
from video.utils import typer


def get_model_instance(model_class: type[Any], values: dict[str, Any]) -> Any:
    """
    Get model instance.

    :param model_class:
    :param values:
    :return: model instance
    """
    g_context_id = g.get("context", {}).get("id")
    if "context_id" not in values.keys() and g_context_id:
        values["context_id"] = g_context_id
    model = model_class(**values)
    return model


def create(
    model_class: type[Any],
    values: dict[str, Any] | None = None,
    *,
    session: Session,
) -> Any:
    """Create a record using given model class.

    Args:
        model_class (sqlalchemy.ext.declarative.declarative_base):
            A sqlalchemy model class.
        values (dict): Values to write to model.
        session (Session): db session.

    Returns:
        Any: The created model instance.
    """
    if values is None:
        values = {}
    model = get_model_instance(model_class, values)
    session.add(model)
    session.flush()

    return model


def _get_jobs(
    job_id: int | None = None,
    parent_id: int | None = None,
    *,
    session: Session,
) -> list[dict[str, Any]]:
    stmt = select(Job)

    if job_id:
        stmt = stmt.where(Job.primary_key == job_id)
    if parent_id:
        stmt = stmt.where(Job.parent_id == parent_id)

    stmt = stmt.order_by(Job.primary_key).limit(config.QUERY_RECORD_LIMIT)

    return [job_row.to_dict() for job_row in session.execute(stmt).scalars().all()]


def _get_contexts(
    context_ids: list[int] | None = None,
) -> list[dict[str, Any]]:
    with mysql.db_session() as session:
        stmt = select(Context)

        if context_ids:
            stmt = stmt.where(Context.primary_key.in_(context_ids))

        stmt = stmt.order_by(Context.primary_key)

        return [
            contexts_row.to_dict()
            for contexts_row in session.execute(stmt).scalars().all()
        ]


def _get_context_fields(
    context_ids: list[int] | None = None,
) -> list[dict[str, Any]]:
    with mysql.db_session() as session:
        stmt = select(ContextField)

        if context_ids:
            stmt = stmt.where(ContextField.context_id.in_(context_ids))

        stmt = stmt.order_by(ContextField.primary_key)

        return [
            context_fields_row.to_dict()
            for context_fields_row in session.execute(stmt).scalars().all()
        ]


def _get_statuses_by_job_ids(
    job_ids: list[int] | None = None,
) -> list[dict[str, Any]]:
    with mysql.db_session() as session:
        stmt = select(JobStatus)

        if job_ids:
            stmt = stmt.where(JobStatus.job_id.in_(job_ids))

        stmt = stmt.order_by(JobStatus.primary_key)

        return [
            job_status_row.to_dict()
            for job_status_row in session.execute(stmt).scalars().all()
        ]


def _get_inputs_by_job_ids(
    job_ids: list[int] | None = None,
) -> list[dict[str, Any]]:
    with mysql.db_session() as session:
        stmt = select(JobInput)

        if job_ids:
            stmt = stmt.where(JobInput.job_id.in_(job_ids))

        stmt = stmt.order_by(JobInput.primary_key)

        return [
            job_input_row.to_dict()
            for job_input_row in session.execute(stmt).scalars().all()
        ]


def _get_outputs_by_job_ids(
    job_ids: list[int] | None = None,
) -> list[dict[str, Any]]:
    with mysql.db_session() as session:
        stmt = select(JobOutput)

        if job_ids:
            stmt = stmt.where(JobOutput.job_id.in_(job_ids))

        stmt = stmt.order_by(JobOutput.primary_key)

        return [
            job_output_row.to_dict()
            for job_output_row in session.execute(stmt).scalars().all()
        ]


def get_jobs(
    job_data_filters: dict[str, Any],
    session: Session | None = None,
) -> list[dict[str, Any]]:
    """Get jobs.

    Args:
        job_data_filters (dict): Filters for jobs.
        session: Optional SQLAlchemy session for the initial _get_jobs fetch.
            The parallel helpers (inputs/outputs/statuses) always open their
            own sessions because they run in separate threads.
    """
    job_id = job_data_filters.get("id")
    parent_id = job_data_filters.get("parent_id")

    with nullcontext(session) if session is not None else mysql.db_session() as s:
        jobs = _get_jobs(job_id=job_id, parent_id=parent_id, session=s)
    if not jobs:
        return []

    context_ids = [job["context_id"] for job in jobs]
    job_id_to_job_map: dict[int, dict[str, Any]] = {
        job["id"]: {
            "id": job["id"],
            "type": job["type"],
            "parent_id": job["parent_id"],
            "status": None,
            "inputs": {},
            "outputs": {},
            "context": {
                "id": job["context_id"],
            },
        }
        for job in jobs
    }
    job_ids = list(job_id_to_job_map.keys())

    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        contexts_future = executor.submit(_get_contexts, context_ids=context_ids)
        context_fields_future = executor.submit(
            _get_context_fields, context_ids=context_ids
        )
        job_statuses_future = executor.submit(_get_statuses_by_job_ids, job_ids=job_ids)
        job_inputs_future = executor.submit(_get_inputs_by_job_ids, job_ids=job_ids)
        job_outputs_future = executor.submit(_get_outputs_by_job_ids, job_ids=job_ids)

    context_id_to_context_map = {
        context["id"]: context for context in contexts_future.result()
    }
    context_id_to_context_fields_map: dict[int, dict[str, Any]] = {}
    context_fields = context_fields_future.result()
    for context_field in context_fields:
        context_id = context_field["context_id"]

        if not context_id_to_context_fields_map.get(context_id):
            context_id_to_context_fields_map[context_id] = {
                "datetime": context_id_to_context_map[context_id]["datetime"]
            }

        context_id_to_context_fields_map[context_id][context_field["name"]] = (
            typer.unstringify(context_field["value"])
        )

    job_statuses = job_statuses_future.result()
    job_inputs = job_inputs_future.result()
    job_outputs = job_outputs_future.result()

    for job_id, job in job_id_to_job_map.items():
        context_id = job["context"]["id"]
        job["context"].update(context_id_to_context_fields_map[context_id])

    for js, ji, jo in itertools.zip_longest(job_statuses, job_inputs, job_outputs):
        if js:
            job_id_to_job_map[js["job_id"]]["status"] = js["status"]
        if ji:
            job_id_to_job_map[ji["job_id"]]["inputs"][ji["name"]] = typer.unstringify(
                ji["value"]
            )
        if jo:
            job_id_to_job_map[jo["job_id"]]["outputs"][jo["name"]] = typer.unstringify(
                jo["value"]
            )

    jobs_response = list(job_id_to_job_map.values())

    for job in jobs_response:
        if not job["inputs"]:
            job["inputs"] = None
        if not job["outputs"]:
            job["outputs"] = None

    return jobs_response


def persist_job_data(
    jobs: dict[str, Any] | list[dict[str, Any]],
    session: Session | None = None,
) -> list[int]:
    """Persist job data.

    Args:
        jobs (dict | list[dict]): A job description or list of job descriptions.
        session (Session): db session.

    Returns:
        list[int]: list of job_ids
    """
    if not isinstance(jobs, list):
        jobs = [jobs]

    if not jobs:
        raise ValueError("persist_job_data requires at least one job")

    job_models: list[Any] = []
    job_status_models: list[Any] = []
    job_input_models: list[Any] = []
    job_output_models: list[Any] = []

    jobs_to_create = [job for job in jobs if not job.get("id")]
    job_data_to_add = [job for job in jobs if job.get("id")]

    with nullcontext(session) if session is not None else mysql.db_session() as s:
        for job in jobs_to_create:
            job_models.append(
                get_model_instance(
                    Job,
                    {
                        "parent_id": job.get("parent_id"),
                        "job_type": job["type"],
                    },
                )
            )
        s.add_all(job_models)
        s.flush()

        for job_model, job in zip(job_models, jobs_to_create, strict=True):
            job_status = job.get("status") or job_statuses.SUBMITTED
            job_data_to_add.append(
                {
                    **job,
                    "id": job_model.primary_key,
                    "status": job_status,
                }
            )

        for job in job_data_to_add:
            job_id = job.get("id")
            job_status = job.get("status")

            if job_status:
                job_status_models.append(
                    get_model_instance(
                        JobStatus,
                        {
                            "job_id": job_id,
                            "status": job_status,
                        },
                    )
                )

            for name, value in job.get("inputs", {}).items():
                job_input_models.append(
                    get_model_instance(
                        JobInput,
                        {
                            "job_id": job_id,
                            "name": name,
                            "value": typer.stringify(value),
                            "field_type": typer.typer(value),
                        },
                    )
                )

            for name, value in job.get("outputs", {}).items():
                job_output_models.append(
                    get_model_instance(
                        JobOutput,
                        {
                            "job_id": job_id,
                            "name": name,
                            "value": typer.stringify(value),
                            "field_type": typer.typer(value),
                        },
                    )
                )

        s.add_all([*job_status_models, *job_input_models, *job_output_models])

    return [job_data["id"] for job_data in job_data_to_add]


def persist_context(
    context: dict[str, Any] | None = None,
    *,
    session: Session | None = None,
) -> int:
    """Persist context.

    Args:
        context (dict): Context to be persisted.
        session (Session): db session.

    Returns:
        int: context_id
    """
    if context is None:
        context = {}
    context_id = context.pop("id", None)
    with nullcontext(session) if session is not None else mysql.db_session() as s:
        if not context_id:
            context_model = create(
                model_class=Context,
                session=s,
            )
            context_id = context_model.primary_key

        for name, value in context.items():
            create(
                ContextField,
                {
                    "context_id": context_id,
                    "name": name,
                    "value": typer.stringify(value),
                    "field_type": typer.typer(value),
                },
                session=s,
            )

    return cast(int, context_id)


def get_context(job_id: int) -> dict[str, Any]:
    """Get context by job_id.

    Args:
        job_id (int): Job id.

    Returns:
        dict: Context.
    """
    with mysql.db_session() as session:
        jobs = _get_jobs(job_id=job_id, session=session)
    if not jobs:
        return {}
    context_id = jobs[0]["context_id"]
    context: dict[str, Any] = {
        "id": context_id,
    }

    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        contexts_future = executor.submit(_get_contexts, context_ids=[context_id])
        context_fields_future = executor.submit(
            _get_context_fields, context_ids=[context_id]
        )

    context["datetime"] = contexts_future.result()[0]["datetime"]
    context_fields = context_fields_future.result()
    for context_field in context_fields:
        context[context_field["name"]] = typer.unstringify(context_field["value"])

    return context


def get_new_assets(dashboard_item_id: int, offset: int) -> list[dict[str, Any]]:
    """Get new assets from video dashboard item table.

    Args:
        dashboard_item_id (int): ID of last video dashboard item encoded.
        offset (int): Pagination offset.
    """
    with mysql.ar_db_session() as session:
        stmt = (
            select(VideoDashboardItem, Release)
            .join(Release, VideoDashboardItem.upc == Release.upc)
            .where(VideoDashboardItem.dashboard_item_id > dashboard_item_id)
            .order_by(VideoDashboardItem.dashboard_item_id)
            .offset(offset)
            .limit(config.QUERY_RECORD_LIMIT)
        )
        rows = session.execute(stmt).all()
        video_dashboard_items: list[dict[str, Any]] = []
        for row in rows:
            video_dashboard_item = {
                "dashboard_item_id": row.VideoDashboardItem.dashboard_item_id,
                "asset_id": row.VideoDashboardItem.asset_id,
                "product_id": row.Release.release_id,
            }
            video_dashboard_items.append(video_dashboard_item)
    return video_dashboard_items


def get_new_assets_count(dashboard_item_id: int) -> int | None:
    """Get number of new assets from video dashboard item table.

    Args:
        dashboard_item_id (int): ID of last video dashboard item encoded.
    """
    with mysql.ar_db_session() as session:
        return session.execute(
            select(functions.count(VideoDashboardItem.asset_id))
            .join(Release, VideoDashboardItem.upc == Release.upc)
            .where(VideoDashboardItem.dashboard_item_id > dashboard_item_id)
        ).scalar()


def get_max_dashboard_item_id() -> int | None:
    """Get most recent id from video dashboard item table."""
    with mysql.ar_db_session() as session:
        return session.execute(
            select(functions.max(VideoDashboardItem.dashboard_item_id))
        ).scalar()


def get_product_assets(
    product_ids: list[int],
    asset_types: list[str] | None = None,
) -> list[dict[str, Any]] | None:
    """Get the S3 keys for the assets associated with a product.

    Args:
        product_ids (list): Product ID's.
        asset_types (list): Asset types (VIDEO_MASTER, H264_HD, etc.).

    Returns:
        list: Query result set.
    """
    if asset_types is None:
        asset_types = []
    with mysql.db_session() as session:
        stmt = select(
            VideoAsset.product_id,
            VideoAsset.asset_type,
            VideoAsset.asset_path,
            VideoAsset.tuid,
            VideoAsset.last_updated,
            func.convert_tz(VideoAsset.last_updated, "UTC", "US/Eastern").label(
                "last_updated_us_east"
            ),
            VideoAsset.duration,
        ).where(VideoAsset.product_id.in_(product_ids))

        if asset_types:
            stmt = stmt.where(VideoAsset.asset_type.in_(asset_types))

        rows = session.execute(stmt).all()

    if not rows:
        return None

    return [
        {
            "product_id": row.product_id,
            "asset_type": row.asset_type,
            "s3_bucket": config.VIDEO_BUCKET_NAME,
            "s3_key": row.asset_path,
            "tuid": row.tuid,
            "last_updated": row.last_updated.strftime("%Y-%m-%d %H:%M:%S"),
            "last_updated_us_east": row.last_updated_us_east.strftime(
                "%Y-%m-%d %H:%M:%S"
            ),
            "duration": row.duration,
        }
        for row in rows
    ]


def get_video_tracks_metadata(track_ids: list[int]) -> list[dict[str, Any]] | None:
    """Get the video tracks metadata associated with a video/bundle product.

    Args:
        track_ids (list): Track ID's.

    Returns:
        list: Query result set.
    """
    with mysql.ar_db_session() as session:
        stmt = (
            select(
                Track.duration_minutes,
                Track.duration_seconds,
                Track.closed_caption_exists,
                Track.closed_caption_reason_id,
                TrackVideo.aspect_ratio,
                TrackVideo.channel,
                TrackVideo.color,
                TrackVideo.resolution,
                TrackVideo.fps,
                TrackVideo.track_id,
            )
            .join(TrackVideo, TrackVideo.track_id == Track.track_id)
            .where(TrackVideo.track_id.in_(track_ids))
        )

        rows = session.execute(stmt).all()

    if not rows:
        return None

    return [
        {
            "track_id": row.track_id,
            "duration_minutes": row.duration_minutes,
            "duration_seconds": row.duration_seconds,
            "closed_caption_exists": row.closed_caption_exists,
            "closed_caption_reason_id": row.closed_caption_reason_id,
            "color": row.color,
            "channel": row.channel,
            "fps": row.fps,
            "resolution": row.resolution,
            "aspect_ratio": row.aspect_ratio,
        }
        for row in rows
    ]