"""
AssetUpload Model.

This AssetUpload model uses sqlalchemy. It's used to store about newly
uploaded assets.
"""

from datetime import datetime, timezone
from typing import Any

import sentry_sdk
from sqlalchemy import (
    TIMESTAMP,
    Boolean,
    Column,
    ForeignKey,
    Index,
    Integer,
    LargeBinary,
    String,
    false,
    func,
    select,
    text,
    true,
    tuple_,
    update,
)
from sqlalchemy.dialects.mysql import INTEGER, TINYINT
from sqlalchemy.orm.session import Session

from assets import config
from assets.connectors import mysql
from assets.constants import (
    api,
    asset_status as asset_status_constants,
    error as error_constants,
)
from assets.exceptions import (
    AssetUploadNotFound,
    DuplicateAssetUploadFilename,
    MissingContext,
    NoAssetsFound,
)
from assets.models.asset_status import AssetStatus
from assets.models.asset_upload_type import resolve_asset_upload_type_id
from assets.utils import raw_queries

ORDER_ASC = "ASC"
ORDER_DESC = "DESC"


class AssetUpload(mysql.AuModel):
    """Table definition for asset_upload table."""

    __tablename__ = "asset_upload"

    asset_upload_id: int = Column("id", Integer, primary_key=True, autoincrement=True)
    user_id = Column(String(127), nullable=False)
    asset_upload_type_id: int = Column(
        TINYINT(unsigned=True),
        ForeignKey("asset_upload_type.asset_upload_type_id"),
        nullable=False,
    )
    token = Column(LargeBinary, nullable=True)
    filename = Column(String(64), nullable=False)
    api_version = Column(Integer, default=1)
    product_id = Column(Integer, default=0)
    upc = Column(String(50), default=None, nullable=True)
    track_unique_id = Column(INTEGER(unsigned=True), default=0)
    bucket_name = Column(String(255), nullable=True)
    original_filename = Column(String(255), nullable=True)
    deleted = Column(Boolean, default=False)
    is_correction = Column(Boolean, default=False)
    updated_timestamp = Column(
        TIMESTAMP, nullable=False, server_default=text("CURRENT_TIMESTAMP")
    )
    __table_args__ = (Index("filename_idx", "filename", "api_version", "product_id"),)

    def as_dict(self, get_updated_timestamp: bool = False) -> dict[str, Any]:
        """Return object as dict.

        Returns:
            dict: Dictionary representation of object
        """
        token = None
        if self.token:
            token = self.token.decode()

        asset_upload_dict: dict[str, Any] = {
            "id": self.asset_upload_id,
            "user_id": self.user_id,
            "asset_upload_type_id": self.asset_upload_type_id,
            "token": token,
            "filename": self.filename,
            "api_version": self.api_version,
            "product_id": self.product_id,
            "upc": self.upc,
            "track_unique_id": self.track_unique_id,
            "bucket_name": self.bucket_name,
            "original_filename": self.original_filename,
            "deleted": self.deleted,
            "is_correction": self.is_correction,
        }
        if get_updated_timestamp:
            asset_upload_dict["updated_timestamp"] = self.updated_timestamp

        return asset_upload_dict


def create_asset_upload_with_context(
    user_id: str,
    asset_upload_type_id: int,
    token: str,
    filename: str,
    product_id: int,
    upc: int,
    track_unique_id: int,
    is_correction: bool,
    original_filename: str,
) -> dict[str, Any]:
    """Create asset_upload record with context.

    Args:
        user_id (str): user id
        asset_upload_type_id (int): asset upload type id
        token (str): AWS S3 MultiPartUpload UploadId
        filename (str): unique filename
        product_id (int): product id
        upc (int): product upc
        track_unique_id (int): track unique id
        is_correction (bool): is correction flag
        original_filename (str): original asset filename

    Returns:
        dict: Contains asset_upload, asset_status and
            deleted_asset_uploads.
    """
    with mysql.au_db_session() as session:
        asset_upload = AssetUpload(
            user_id=user_id,
            asset_upload_type_id=asset_upload_type_id,
            token=token.encode(),
            filename=filename,
            api_version=api.API_VERSION_V2,
            product_id=product_id,
            upc=str(upc),
            track_unique_id=track_unique_id,
            bucket_name=config.RAW_ASSETS_BUCKET_NAME,
            original_filename=original_filename,
            is_correction=is_correction,
        )
        session.add(asset_upload)
        session.flush()

        asset_status = AssetStatus(
            asset_upload_id=asset_upload.asset_upload_id,
            status=asset_status_constants.STATUS_UPLOADING,
            status_time=datetime.now(timezone.utc),
        )
        session.add(asset_status)
        session.flush()

        deleted_asset_uploads = mark_previous_asset_deleted(
            api_version=api.API_VERSION_V2,
            product_id=product_id,
            track_unique_id=track_unique_id,
            is_correction=is_correction,
            asset_upload_type_id=asset_upload_type_id,
            new_asset_upload_id=asset_upload.asset_upload_id,
            session=session,
        )

        return {
            "asset_upload": asset_upload.as_dict(),
            "asset_status": asset_status.as_dict(),
            "deleted_asset_uploads": deleted_asset_uploads,
        }


def create_asset_upload(
    user_id: str,
    filename: str,
    token: str,
    asset_upload_type_id: int,
    api_version: int = api.API_VERSION_V1,
) -> dict[str, Any]:
    """Put new raw asset item into asset_upload table.

    Args:
        user_id (str): user id
        filename (str): unique filename
        token (str): sts token
        asset_upload_type_id (int): asset upload type id
        api_version (int): Workflow API version
    Returns:
        dict: AssetUpload.as_dict()
    """
    asset_upload = AssetUpload(
        user_id=user_id,
        asset_upload_type_id=asset_upload_type_id,
        filename=filename,
        token=token.encode(),
        api_version=api_version,
    )
    with mysql.au_db_session() as session:
        session.add(asset_upload)
        session.flush()

        return asset_upload.as_dict()


def create_asset_upload_record_for_copy(
    asset_upload_record: dict[str, Any],
) -> dict[str, Any]:
    """Create copy raw asset item into asset_upload table.

    Args:
        asset_upload_record (dict): Asset upload record to create
    Returns:
        dict: AssetUpload.as_dict()
    """
    with mysql.au_db_session() as session:
        asset_upload = AssetUpload(**asset_upload_record)
        session.add(asset_upload)
        session.commit()
        session.refresh(asset_upload)

        return asset_upload.as_dict()


def get_asset_upload(
    filename: str,
    token: str | None = None,
    user_id: str | None = None,
    api_version: int = api.API_VERSION_V1,
) -> dict[str, Any]:
    """Check if asset_upload exists in database.

    Args:
        filename (str): filename
        token (str): sts token
        user_id (str): user id
        api_version (int): Workflow API version

    Returns:
        dict: AssetUpload.as_dict()
    """
    with mysql.au_db_session(read_only=True) as session:
        filters = [
            AssetUpload.filename == filename,
            AssetUpload.api_version == api_version,
            AssetUpload.deleted == 0,
        ]
        if user_id:
            filters.append(AssetUpload.user_id == user_id)
        if token:
            filters.append(AssetUpload.token == token.encode())

        asset_upload = (
            session.query(AssetUpload)
            .filter(*filters)
            .order_by(AssetUpload.asset_upload_id.desc())
            .first()
        )
        if asset_upload is None:
            raise AssetUploadNotFound(error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND)

        return asset_upload.as_dict()


def mark_previous_asset_deleted(
    api_version: int,
    product_id: int,
    track_unique_id: int,
    is_correction: bool,
    asset_upload_type_id: int,
    new_asset_upload_id: int,
    session: Session,
) -> list[dict[str, Any]]:
    """Mark previous asset_upload as deleted if it existed.

    Args:
        api_version (int): Workflow API version.
        product_id (int): Product id.
        track_unique_id (int): Track unique id.
        is_correction (bool): Is correction flag.
        asset_upload_type_id (int): Asset upload type id.
        new_asset_upload_id (int): New replacement asset_upload id.
        session (object): SQLAlchemy database session.
    """
    filters = [
        AssetUpload.api_version == api_version,
        AssetUpload.product_id == product_id,
        AssetUpload.track_unique_id == track_unique_id,
        AssetUpload.is_correction == is_correction,
        AssetUpload.asset_upload_type_id == asset_upload_type_id,
        AssetUpload.asset_upload_id != new_asset_upload_id,
        AssetUpload.deleted == false(),
    ]

    asset_uploads_to_mark_as_deleted = session.query(AssetUpload).filter(*filters).all()

    if asset_uploads_to_mark_as_deleted:
        affected_asset_upload_ids = [
            asset_upload.asset_upload_id
            for asset_upload in asset_uploads_to_mark_as_deleted
        ]
        session.query(AssetUpload).filter(
            AssetUpload.asset_upload_id.in_(affected_asset_upload_ids)
        ).update({"deleted": True})

    return [asset_upload.as_dict() for asset_upload in asset_uploads_to_mark_as_deleted]


def _asset_upload_lock_name(
    product_id: int | None, track_unique_id: int | None, is_correction: bool = False
) -> str | None:
    if not product_id:
        return None
    return f"{product_id}:{track_unique_id}:{is_correction}"


def update_asset_upload(
    filename: str,
    upc: int | None,
    track_unique_id: int | None,
    product_id: int | None,
    original_filename: str | None,
    is_correction: bool = False,
    token: str | None = None,
    user_id: str | None = None,
    api_version: int = api.API_VERSION_V1,
) -> dict[str, Any]:
    """Check if asset_upload exists in database and update with context.

    Args:
        filename (str): Unique filename.
        upc (int): Product UPC.
        track_unique_id (int): Track unique id.
        product_id (int): Product id.
        original_filename (str): Original asset filename.
        is_correction (bool): Is correction flag.
        token (str): STS token.
        user_id (str): User id.
        api_version (int): Workflow API version.

    Returns:
        dict: AssetUpload.as_dict()
    """
    with mysql.au_db_session() as session:
        lock_name = (
            _asset_upload_lock_name(product_id, track_unique_id, is_correction)
            or filename
        )
        with mysql.db_lock(session, lock_name, config.ASSET_UPLOAD_UPDATE_LOCK_TIMEOUT):
            filters = [
                (AssetUpload.filename == filename),
                (AssetUpload.api_version == api_version),
                (AssetUpload.deleted == false()),
            ]

            if user_id and token:
                filters.extend(
                    [
                        (AssetUpload.user_id == user_id),
                        (AssetUpload.token == token.encode()),
                    ]
                )

            num_records_with_filename = (
                session.query(func.count(AssetUpload.asset_upload_id))
                .filter(
                    AssetUpload.filename == filename,
                    AssetUpload.api_version == api_version,
                )
                .scalar()
            )

            if num_records_with_filename > 1:
                raise DuplicateAssetUploadFilename(
                    "This filename is being used by multiple asset_upload records "
                    "which is the result of copying a product. Once multiple "
                    "asset_upload records exist for a filename, the "
                    "ENV-assets-transcoding-v2 sfn can't be rerun for that filename "
                    "since we don't know which asset_upload record to update. Use "
                    "this script to recreate asset finals: "
                    "https://github.com/theorchard/scripts-assets/tree/master/create-asset-finals"
                )

            asset_upload_to_update = (
                session.query(AssetUpload).filter(*filters).one_or_none()
            )

            if not asset_upload_to_update:
                error_message = "No asset_upload record to update"
                sentry_sdk.capture_message(message="not_found_error")
                sentry_sdk.set_context(
                    "Payload",
                    {
                        "message": error_message,
                        "errors": {
                            "code": "not_found_error",
                            "message": error_message,
                        },
                        "status": 404,
                    },
                )
                raise NoAssetsFound(error_message)

            if not product_id:
                if not asset_upload_to_update.product_id:
                    raise MissingContext(
                        "asset_upload is missing context fields "
                        "(product_id, track_unique_id, etc.) and none were "
                        "provided in the request."
                    )

                # The new upload flow was used so there is nothing to do. Metadata is already set.
                return {}

            if asset_upload_to_update.product_id:
                raise MissingContext(
                    "asset_upload has context fields "
                    "(product_id, track_unique_id, etc.) set but they were "
                    "also provided in the request."
                )

            mark_previous_asset_deleted(
                api_version=api_version,
                product_id=product_id,
                track_unique_id=track_unique_id or 0,
                is_correction=is_correction,
                asset_upload_type_id=asset_upload_to_update.asset_upload_type_id,
                new_asset_upload_id=asset_upload_to_update.asset_upload_id,
                session=session,
            )

            session.execute(
                update(AssetUpload)
                .filter(
                    AssetUpload.asset_upload_id
                    == asset_upload_to_update.asset_upload_id
                )
                .values(
                    upc=upc,
                    track_unique_id=track_unique_id,
                    product_id=product_id,
                    original_filename=original_filename,
                    is_correction=is_correction,
                )
            )

            result = session.execute(
                select(AssetUpload).filter(
                    AssetUpload.asset_upload_id
                    == asset_upload_to_update.asset_upload_id
                )
            )
            row = result.scalar_one_or_none()
            if row:
                return row.as_dict()
            else:
                return {}


def apply_asset_corrections(product_id: int) -> list[dict[str, Any]]:
    with mysql.au_db_session() as session:
        correction_assets = get_assets_to_apply_correction(session, product_id)
        if not correction_assets:
            return []

        num_affected_rows = update_asset_correction_to_false(session, correction_assets)
        if not num_affected_rows:
            return []

        _mark_previous_correction_assets_as_deleted(
            session, product_id, correction_assets
        )

        session.commit()

        return correction_assets


def update_asset_correction_to_false(
    session: Session, correction_assets: list[dict[str, Any]]
) -> int:
    """
    Mark is_correction=False for a given assets.

    Args:
        session (Session): SQLAlchemy database session.
        correction_assets (list): List of AssetUpload.as_dict() objects.

    Returns:
        int: number of updated records
    """
    correction_assets_ids = [
        correction_asset["id"] for correction_asset in correction_assets
    ]

    # Unset asset correction
    num_affected_rows = (
        session.query(AssetUpload)
        .filter(AssetUpload.asset_upload_id.in_(correction_assets_ids))
        .update({"is_correction": false()}, synchronize_session="fetch")
    )

    # Update in-memory list of assets with is_correction = false
    # to avoid additional db query for updated records
    if num_affected_rows == len(correction_assets):
        [item.update({"is_correction": False}) for item in correction_assets]

    return num_affected_rows


def get_assets_to_apply_correction(
    session: Session, product_id: int, api_version: int = api.API_VERSION_V2
) -> list[dict[str, Any]]:
    """
    Return corrected assets of the given product.

    Args:
        session (Session): SQLAlchemy database session.
        product_id (int): Product id.
        api_version (int): API Version of the given product.

    Returns:
        list: a list of AssetUpload.as_dict()
    """
    sub_query = (
        session.query(func.max(AssetUpload.asset_upload_id).label("latest_upload_id"))
        .filter(
            AssetUpload.product_id == product_id,
            AssetUpload.api_version == api_version,
            AssetUpload.is_correction == true(),
            AssetUpload.deleted == false(),
        )
        .group_by(AssetUpload.track_unique_id, AssetUpload.asset_upload_type_id)
        .subquery()
    )

    asset_uploads = (
        session.query(AssetUpload)
        .join(sub_query, AssetUpload.asset_upload_id == sub_query.c.latest_upload_id)
        .all()
    )

    return [asset_upload.as_dict() for asset_upload in asset_uploads]


def _mark_previous_correction_assets_as_deleted(
    session: Session,
    product_id: int,
    correction_assets: list[dict[str, Any]],
    api_version: int = api.API_VERSION_V2,
) -> None:
    correction_keys = [
        (c["asset_upload_type_id"], c["track_unique_id"]) for c in correction_assets
    ]
    correction_asset_upload_ids = [c["id"] for c in correction_assets]

    session.query(AssetUpload).filter(
        AssetUpload.product_id == product_id,
        AssetUpload.api_version == api_version,
        tuple_(AssetUpload.asset_upload_type_id, AssetUpload.track_unique_id).in_(
            correction_keys
        ),
        AssetUpload.asset_upload_id.notin_(correction_asset_upload_ids),
        AssetUpload.deleted == false(),
    ).update({"deleted": True}, synchronize_session="fetch")


def get_asset_upload_by_id(asset_upload_id: int) -> dict[str, Any]:
    """Return asset_upload by id.

    Args:
        asset_upload_id (int): Unique asset_upload id.

    Returns:
        dict: AssetUpload.as_dict()
    """
    with mysql.au_db_session(read_only=True) as session:
        asset_upload = (
            session.query(AssetUpload)
            .filter(AssetUpload.asset_upload_id == asset_upload_id)
            .first()
        )
        if asset_upload is None:
            raise AssetUploadNotFound(error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND)
        return asset_upload.as_dict()


def get_asset_uploads_by_product_id(
    product_id: int, api_version: int = api.API_VERSION_V1
) -> list[dict[str, Any]]:
    """Return asset_upload records by product_id.

    Args:
        product_id (int): Product id.
        api_version (int): Workflow API version.
    Returns:
        list: Asset list
    """
    with mysql.au_db_session(read_only=True) as session:
        sub_query = (
            session.query(
                func.max(AssetUpload.asset_upload_id).label(
                    "latest_undeleted_upload_id"
                )
            )
            .filter(
                AssetUpload.product_id == product_id,
                AssetUpload.api_version == api_version,
                AssetUpload.deleted == 0,
            )
            .group_by(
                AssetUpload.product_id,
                AssetUpload.track_unique_id,
                AssetUpload.asset_upload_type_id,
            )
            .subquery("sub_query")
        )

        asset_uploads = (
            session.query(AssetUpload)
            .filter(
                AssetUpload.asset_upload_id == sub_query.c.latest_undeleted_upload_id
            )
            .all()
        )

        return [asset_upload.as_dict() for asset_upload in asset_uploads]


def get_asset_uploads(
    api_version: int = api.API_VERSION_V1,
    track_id: int | None = None,
    product_id: int | None = None,
    order: str | None = None,
    limit: int | None = None,
    is_correction: int | None = None,
    asset_upload_type_id: int | None = None,
) -> list[dict[str, Any]]:
    """Return asset_uploads by track_id.

    Args:
        api_version (int): Workflow API version.
        track_id (int): Track unique id.
        product_id (int): Optional Product id, used for images.
        order: (str): Optional sort order (sorting by PK).
        limit: (int) Optional limit
        is_correction: (int) Optional correction flag
        asset_upload_type_id: (int) Optional asset upload type id to filter by
    Returns:
        list: list of AssetUpload.as_dict()
    """
    with mysql.au_db_session(read_only=True) as session:
        return get_asset_uploads_records(
            session=session,
            api_version=api_version,
            product_id=product_id,
            track_id=track_id,
            order=order,
            limit=limit,
            is_correction=is_correction,
            asset_upload_type_id=asset_upload_type_id,
        )


def get_latest_asset_upload(
    api_version: int = api.API_VERSION_V2,
    product_id: int | None = None,
    track_id: int | None = None,
    order: str = ORDER_DESC,
    limit: int = 1,
    is_correction_mode: bool = False,
    get_updated_timestamp: bool = False,
    asset_upload_type_id: int | None = None,
) -> list[dict[str, Any]]:
    """Return asset_uploads by track_id.

    Args:
        api_version (int): Workflow API version.
        track_id (int): Track unique id.
        product_id (int): Optional Product id, used for images.
        order (str): Optional sort order (sorting by PK).
        limit (int): Optional limit
        is_correction_mode (bool): Checks if track is in correction mode or not
        get_updated_timestamp (bool): return updated timestamp if true
    Returns:
        list: list of AssetUpload.as_dict()
    """
    with mysql.au_db_session(read_only=True) as session:
        if not is_correction_mode:
            result = get_asset_uploads_records(
                session=session,
                api_version=api_version,
                product_id=product_id,
                track_id=track_id,
                order=order,
                limit=limit,
                is_correction=0,
                get_updated_timestamp=get_updated_timestamp,
                asset_upload_type_id=asset_upload_type_id,
            )
        else:
            result = get_asset_uploads_records(
                session=session,
                api_version=api_version,
                product_id=product_id,
                track_id=track_id,
                order=order,
                limit=limit,
                get_updated_timestamp=get_updated_timestamp,
                asset_upload_type_id=asset_upload_type_id,
            )

        return result


def get_asset_uploads_records(
    session: Session,
    api_version: int = api.API_VERSION_V1,
    track_id: int | None = None,
    product_id: int | None = None,
    order: str | None = None,
    limit: int | None = None,
    is_correction: int | None = None,
    get_updated_timestamp: bool = False,
    asset_upload_type_id: int | None = None,
) -> list[dict[str, Any]]:
    """Reusable method that returns asset_uploads records.

    Args:
        session (object): SQLAlchemy database session.
        api_version (int): Workflow API version.
        track_id (int): Track unique id.
        product_id (int): Optional Product id, used for images.
        order (str): Optional sort order (sorting by PK).
        limit (int): Optional limit
        is_correction (bool): Optional correction flag
        get_updated_timestamp (bool): return updated timestamp if true
        asset_upload_type_id (int): Optional asset upload type id to filter by
    Returns:
        list: a list of AssetUpload.as_dict()
    """
    filters = [AssetUpload.api_version == api_version, AssetUpload.deleted == 0]
    if track_id is not None:
        filters.append(AssetUpload.track_unique_id == track_id)
    if product_id:
        filters.append(AssetUpload.product_id == product_id)
    if is_correction is not None:
        filters.append(AssetUpload.is_correction == is_correction)
    if asset_upload_type_id is not None:
        filters.append(AssetUpload.asset_upload_type_id == asset_upload_type_id)
    query = session.query(AssetUpload).filter(*filters)

    if order:
        query = query.order_by(
            AssetUpload.asset_upload_id.desc()
            if order == ORDER_DESC
            else AssetUpload.asset_upload_id.asc()
        )
    if limit:
        query = query.limit(limit)

    asset_uploads = query.all()

    if not asset_uploads:
        raise AssetUploadNotFound(error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND)

    return [
        asset_upload.as_dict(get_updated_timestamp) for asset_upload in asset_uploads
    ]


def get_asset_records_to_copy(
    product_id: int,
    tuid: int,
    asset_upload_type: str,
    ignore_if_not_exist: bool = False,
) -> dict[str, Any] | None:
    """Returns asset_upload, asset_status and asset_final records for a product.

    Args:
        product_id (int): Product id, used for to copy asset from.
        tuid (int): Track unique id, used to copy asset from (0 for product level assets).
        asset_upload_type (str): Type of the asset.
        ignore_if_not_exist (bool): Return null if ignore_if_not_exist flag is True.
    Returns:
        dict | None: AssetUpload.as_dict(); returns None when ignore_if_not_exist is True
        and no matching records are found.
    """
    copy_type_id = resolve_asset_upload_type_id(asset_upload_type)

    with mysql.au_db_session(read_only=True) as session:
        result = session.execute(
            text(raw_queries.GET_V2_ASSETS_TO_COPY_SQL),
            {
                "product_id": product_id,
                "track_unique_id": tuid,
                "asset_upload_type_id": copy_type_id,
            },
        )

        rows = result.mappings().all()

        if not rows:
            if ignore_if_not_exist:
                return None
            raise AssetUploadNotFound(error_constants.ERROR_ASSET_UPLOAD_NOT_FOUND)

        asset_upload_bucket_name = rows[0]["asset_upload_bucket_name"]
        asset_upload_filename = rows[0]["asset_upload_filename"]
        asset_upload_original_filename = rows[0]["original_filename"]
        asset_upload_type_id = rows[0]["asset_upload_type_id"]

        result = session.execute(
            text(raw_queries.GET_V2_ASSETS_STATUS_TO_COPY_SQL),
            {
                "product_id": product_id,
                "track_unique_id": tuid,
                "asset_upload_type_id": copy_type_id,
            },
        )
        asset_status_rows = result.mappings().all()

        asset_statuses = [
            {
                "status": row["status"],
                "description": row["description"],
                "message": row["message"],
                "status_time": row["status_time"],
            }
            for row in asset_status_rows
        ]

        return {
            "asset_upload": {
                "bucket_name": asset_upload_bucket_name,
                "filename": asset_upload_filename,
                "original_filename": asset_upload_original_filename,
                "asset_upload_type_id": asset_upload_type_id,
            },
            "asset_status": asset_statuses,
            "asset_finals": [
                {
                    "asset_type": row["asset_final_type"],
                    "asset_subtype": row["asset_subtype"],
                    "bucket": row["bucket"],
                    "filename": row["asset_final_filename"],
                    "duration": row["duration"],
                    "channels": row["channels"],
                    "codec": row["codec"],
                    "sample_rate": row["sample_rate"],
                    "bit_rate": row["bit_rate"],
                    "bit_depth": row["bit_depth"],
                }
                for row in rows
            ],
        }


def mark_asset_upload_deleted(
    asset_upload_ids: list[int] | int,
) -> list[dict[str, Any]]:
    """Mark asset uploads as deleted.

    Args:
        asset_upload_ids (list): Asset upload ids.

    Returns:
        list
    """
    if isinstance(asset_upload_ids, int):
        asset_upload_ids = [asset_upload_ids]

    with mysql.au_db_session() as session:
        updated_items = mark_asset_upload_records_deleted(
            session=session, asset_upload_ids=asset_upload_ids
        )
        if updated_items != len(asset_upload_ids):
            raise Exception(
                "{updated} records marked, {expected} expected".format(
                    updated=updated_items, expected=len(asset_upload_ids)
                )
            )

        session.commit()
        filters = [
            AssetUpload.asset_upload_id.in_(asset_upload_ids),
        ]
        asset_uploads = session.query(AssetUpload).filter(*filters).all()
        result = []
        for asset_upload in asset_uploads:
            result.append(asset_upload.as_dict())
        return result


def mark_asset_upload_records_deleted(
    session: Session, asset_upload_ids: list[int]
) -> int:
    """Reusable method that marks asset uploads as deleted.

    Args:
        session (Session): SQLAlchemy database session.
        asset_upload_ids (list): Asset upload ids.

    Returns:
        int: number of updated records
    """
    filters = [
        AssetUpload.asset_upload_id.in_(asset_upload_ids),
    ]
    return (
        session.query(AssetUpload)
        .filter(*filters)
        .update({"deleted": 1}, synchronize_session=False)
    )


def get_asset_info_for_track_duration_update(
    asset_upload_id: int,
) -> dict[str, Any]:
    """Get asset_upload info joined with asset_final info."""
    with mysql.au_db_session(read_only=True) as session:
        query = raw_queries.GET_ASSET_INFO_FOR_TRACK_DURATION_UPDATE.format(
            asset_upload_id=asset_upload_id
        )
        result = session.execute(text(query))
        row = result.mappings().first()

        if not row or row["deleted"]:
            raise NoAssetsFound(error_constants.ERROR_ASSET_NOT_FOUND)

        return {
            "upc": row["upc"],
            "product_id": row["product_id"],
            "track_unique_id": row["track_unique_id"],
            "duration": row["duration"],
            "is_correction": bool(int(row["is_correction"])),
        }