"""Admin advertisement service layer.

All admin-side advertisement business logic: status derivation, CTR, filtered/
sorted/paginated listing, secure image handling, CRUD, duplication, bulk actions
and analytics aggregation.

This module complements — and does not duplicate — :mod:`app.advertisements`
(which owns active-ad fetching for display and click/impression tracking). The
single-record lookup reuses ``get_advertisement_or_404`` from that module.
"""

from __future__ import annotations

import os
import time
import uuid

from flask import current_app
from sqlalchemy import case, func
from werkzeug.datastructures import FileStorage

from app.extensions import db
from app.models import Advertisement, RewardedAdEvent
from app.models.base import utcnow
from app.models.enums import (
    AdvertisementMediaType,
    AdvertisementPlacement,
    AdvertisementType,
)

# Derived status values (never rely on the enabled flag alone).
STATUS_ACTIVE = "active"
STATUS_SCHEDULED = "scheduled"
STATUS_EXPIRED = "expired"
STATUS_DISABLED = "disabled"

# Bootstrap contextual class per status, for badges.
STATUS_BADGE_CLASS = {
    STATUS_ACTIVE: "success",
    STATUS_SCHEDULED: "info",
    STATUS_EXPIRED: "warning",
    STATUS_DISABLED: "secondary",
}

_SORTS = {
    "newest": Advertisement.created_at.desc(),
    "oldest": Advertisement.created_at.asc(),
    "priority_high": Advertisement.priority.desc(),
    "priority_low": Advertisement.priority.asc(),
    "title": Advertisement.title.asc(),
    "impressions": Advertisement.impression_count.desc(),
    "clicks": Advertisement.click_count.desc(),
}


# ---------------------------------------------------------------------------
# Derived values
# ---------------------------------------------------------------------------
def advertisement_status(ad: Advertisement, now=None) -> str:
    """Derive the effective status from the enabled flag and the schedule."""
    if not ad.is_active:
        return STATUS_DISABLED
    now = now or utcnow()
    if ad.start_at is not None and now < ad.start_at:
        return STATUS_SCHEDULED
    if ad.end_at is not None and now > ad.end_at:
        return STATUS_EXPIRED
    return STATUS_ACTIVE


def compute_ctr(impressions: int, clicks: int) -> float:
    """Click-through rate as a percentage; 0.0 when there are no impressions."""
    if not impressions:
        return 0.0
    return round((clicks / impressions) * 100, 2)


# ---------------------------------------------------------------------------
# Rewarded-advertisement analytics (backed by the rewarded_ad_events stream)
# ---------------------------------------------------------------------------
# These aggregate the append-only event stream written by
# ``app.advertisements.reward_analytics.record_rewarded_event``. Two grouped
# queries for the overview and one for the per-ad breakdown — no N+1, negligible
# cost, and no duplication of the shared impression_count/click_count columns.

def rewarded_analytics() -> dict:
    """Global rewarded-ad metrics: impressions, video, completions, rejections.

    Completion rate = completed rewarded views (rewards granted) ÷ impressions.
    """
    counts = {
        t: int(n)
        for t, n in db.session.execute(
            db.select(RewardedAdEvent.event_type, func.count()).group_by(RewardedAdEvent.event_type)
        ).all()
    }
    impressions = counts.get("impression", 0)
    granted = counts.get("reward_granted", 0)
    rejected = counts.get("reward_rejected", 0)

    reasons = {
        (reason or "unknown"): int(n)
        for reason, n in db.session.execute(
            db.select(RewardedAdEvent.reason, func.count())
            .where(RewardedAdEvent.event_type == "reward_rejected")
            .group_by(RewardedAdEvent.reason)
        ).all()
    }

    return {
        "impressions": impressions,
        "video_starts": counts.get("video_start", 0),
        "completed_views": granted,
        "rewards_granted": granted,
        "completion_rate": round(granted / impressions * 100, 2) if impressions else 0.0,
        "rejections": rejected,
        "rejections_by_reason": reasons,
        "video_success": counts.get("video_success", 0),
        "image_fallbacks": counts.get("video_fallback", 0),
        "video_failures": counts.get("video_error", 0),
    }


# Stable display order for rejection reasons in the chart/table.
_REJECT_REASON_ORDER = ["expired", "replay", "tampered", "wrong_user", "inactive", "rate_limit", "missing"]


def rewarded_chart_data() -> dict:
    """Chart.js-ready series for the rewarded-ad analytics section."""
    a = rewarded_analytics()
    by_reason = a["rejections_by_reason"]
    ordered = [r for r in _REJECT_REASON_ORDER if by_reason.get(r)]
    ordered += [r for r in by_reason if r not in _REJECT_REASON_ORDER]
    return {
        "funnel_labels": ["Impressions", "Video starts", "Rewards granted"],
        "funnel_counts": [a["impressions"], a["video_starts"], a["rewards_granted"]],
        "reject_labels": [r.replace("_", " ").title() for r in ordered],
        "reject_counts": [by_reason[r] for r in ordered],
        "video_labels": ["Successful", "Image fallback", "Load failure"],
        "video_counts": [a["video_success"], a["image_fallbacks"], a["video_failures"]],
    }


def rewarded_ad_breakdown(limit: int = 10) -> list[dict]:
    """Per-advertisement rewarded performance (top by impressions). One query."""
    E = RewardedAdEvent
    impressions = func.sum(case((E.event_type == "impression", 1), else_=0))
    granted = func.sum(case((E.event_type == "reward_granted", 1), else_=0))
    rows = db.session.execute(
        db.select(Advertisement.public_id, Advertisement.title, impressions, granted)
        .join(Advertisement, Advertisement.id == E.advertisement_id)
        .group_by(Advertisement.id, Advertisement.public_id, Advertisement.title)
        .order_by(impressions.desc())
        .limit(limit)
    ).all()
    result = []
    for public_id, title, imp, grant in rows:
        imp, grant = int(imp or 0), int(grant or 0)
        result.append({
            "public_id": public_id,
            "title": title,
            "impressions": imp,
            "rewards": grant,
            "completion_rate": round(grant / imp * 100, 2) if imp else 0.0,
        })
    return result


def _rewarded_stat_row(counts: dict) -> dict:
    """Shape a per-ad rewarded metric dict from a {event_type: count} map."""
    impressions = counts.get("impression", 0)
    granted = counts.get("reward_granted", 0)
    return {
        "impressions": impressions,
        "video_starts": counts.get("video_start", 0),
        "rewarded_views": granted,
        "rewards_granted": granted,
        "completion_rate": round(granted / impressions * 100, 1) if impressions else 0.0,
        "rejections": counts.get("reward_rejected", 0),
        "video_failures": counts.get("video_error", 0),
        "image_fallbacks": counts.get("video_fallback", 0),
        "has_activity": bool(impressions or granted or counts.get("reward_rejected", 0)),
    }


def rewarded_stats_map(ad_ids: list[int]) -> dict[int, dict]:
    """Per-advertisement rewarded metrics for ``ad_ids`` in ONE grouped query.

    Reuses the rewarded event stream (see ``app.advertisements.reward_analytics``)
    so the list page can show completion rate / rewards / rejections per ad with
    no N+1. Ids with no events get a zeroed row.
    """
    if not ad_ids:
        return {}
    E = RewardedAdEvent
    grouped: dict[int, dict] = {aid: {} for aid in ad_ids}
    rows = db.session.execute(
        db.select(E.advertisement_id, E.event_type, func.count())
        .where(E.advertisement_id.in_(ad_ids))
        .group_by(E.advertisement_id, E.event_type)
    ).all()
    for aid, etype, n in rows:
        grouped.setdefault(aid, {})[etype] = int(n)
    return {aid: _rewarded_stat_row(grouped.get(aid, {})) for aid in ad_ids}


def rewarded_stats_for_ad(ad_id: int) -> dict:
    """Rewarded metrics for a single advertisement (per-ad analytics page)."""
    return rewarded_stats_map([ad_id])[ad_id]


# ---------------------------------------------------------------------------
# Advertisement storage statistics
# ---------------------------------------------------------------------------
# A tiny TTL cache keeps the (bounded) filesystem stat off the hot path so it
# does not run on every list render. Mutations invalidate it explicitly.
_STORAGE_TTL_SECONDS = 60
_storage_cache: dict = {"at": 0.0, "data": None}


def invalidate_storage_stats() -> None:
    """Drop the cached storage summary (called after any media mutation)."""
    _storage_cache["data"] = None


def _human_bytes(num) -> str:
    value = float(num or 0)
    for unit in ("B", "KB", "MB", "GB", "TB"):
        if value < 1024 or unit == "TB":
            return f"{value:.0f} {unit}" if unit in ("B", "KB") else f"{value:.1f} {unit}"
        value /= 1024
    return "0 B"


def _local_asset_name(path: str | None) -> str | None:
    """Basename of a locally-stored ad asset, or None for external/legacy URLs."""
    if not path:
        return None
    prefix = f"{current_app.static_url_path}/{current_app.config['ADVERTISEMENT_UPLOAD_URL']}"
    return os.path.basename(path) if path.startswith(prefix) else None


def advertisement_storage_stats(*, force: bool = False) -> dict:
    """Summary of advertisement media storage (total / count / average / largest
    video / unused files).

    Only referenced local files are stat-ed (bounded by ad count) plus a single
    directory listing for orphans — never a recursive scan. Cached for
    ``_STORAGE_TTL_SECONDS`` to avoid filesystem access on every page load.
    """
    now = time.time()
    if not force and _storage_cache["data"] is not None and now - _storage_cache["at"] < _STORAGE_TTL_SECONDS:
        return _storage_cache["data"]

    folder = current_app.config["ADVERTISEMENT_UPLOAD_FOLDER"]
    rows = db.session.execute(
        db.select(Advertisement.image_path, Advertisement.media_url)
    ).all()
    referenced: set[str] = set()
    video_names: set[str] = set()
    for image_path, media_url in rows:
        img = _local_asset_name(image_path)
        if img:
            referenced.add(img)
        vid = _local_asset_name(media_url)
        if vid:
            referenced.add(vid)
            video_names.add(vid)

    disk_names: set[str] = set()
    try:
        for entry in os.scandir(folder):
            if entry.is_file():
                disk_names.add(entry.name)
    except OSError:
        disk_names = set()

    total_bytes = 0
    largest_video = 0
    file_count = 0
    for name in referenced:
        try:
            size = (folder / name).stat().st_size
        except OSError:
            continue
        total_bytes += size
        file_count += 1
        if name in video_names and size > largest_video:
            largest_video = size

    ad_count = int(db.session.scalar(db.select(func.count(Advertisement.id))) or 0)
    average = total_bytes / file_count if file_count else 0
    data = {
        "total_bytes": total_bytes,
        "total_human": _human_bytes(total_bytes),
        "ad_count": ad_count,
        "media_file_count": file_count,
        "average_bytes": average,
        "average_human": _human_bytes(average),
        "largest_video_bytes": largest_video,
        "largest_video_human": _human_bytes(largest_video),
        "unused_files": len(disk_names - referenced),
    }
    _storage_cache["at"] = now
    _storage_cache["data"] = data
    return data


# ---------------------------------------------------------------------------
# Listing (search + filters + sort + pagination)
# ---------------------------------------------------------------------------
def _status_predicate(status: str, now):
    """Return a SQL predicate for a derived status (keeps pagination correct)."""
    active_window = db.and_(
        Advertisement.is_active.is_(True),
        db.or_(Advertisement.start_at.is_(None), Advertisement.start_at <= now),
        db.or_(Advertisement.end_at.is_(None), Advertisement.end_at >= now),
    )
    if status == STATUS_DISABLED:
        return Advertisement.is_active.is_(False)
    if status == STATUS_SCHEDULED:
        return db.and_(
            Advertisement.is_active.is_(True),
            Advertisement.start_at.is_not(None),
            Advertisement.start_at > now,
        )
    if status == STATUS_EXPIRED:
        return db.and_(
            Advertisement.is_active.is_(True),
            Advertisement.end_at.is_not(None),
            Advertisement.end_at < now,
        )
    if status == STATUS_ACTIVE:
        return active_window
    return None


def list_advertisements(
    page: int,
    *,
    q: str | None = None,
    placement: str | None = None,
    ad_type: str | None = None,
    media_type: str | None = None,
    status: str | None = None,
    priority: int | None = None,
    date_from=None,
    date_to=None,
    sort: str = "newest",
):
    """Return a pagination of advertisements matching the admin filters.

    Filters combine (AND). All filtering happens server-side so pagination
    counts stay correct.
    """
    per_page = current_app.config.get("DEFAULT_PAGE_SIZE", 20)
    now = utcnow()
    query = db.select(Advertisement)

    if q:
        escaped = q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
        query = query.where(Advertisement.title.ilike(f"%{escaped}%", escape="\\"))
    if placement in {p.value for p in AdvertisementPlacement}:
        query = query.where(Advertisement.placement == AdvertisementPlacement(placement))
    if ad_type in {t.value for t in AdvertisementType}:
        query = query.where(Advertisement.advertisement_type == AdvertisementType(ad_type))
    if media_type in {m.value for m in AdvertisementMediaType}:
        query = query.where(Advertisement.media_type == AdvertisementMediaType(media_type))
    if priority is not None:
        query = query.where(Advertisement.priority == priority)
    if date_from is not None:
        query = query.where(Advertisement.created_at >= date_from)
    if date_to is not None:
        query = query.where(Advertisement.created_at <= date_to)
    predicate = _status_predicate(status, now) if status else None
    if predicate is not None:
        query = query.where(predicate)

    query = query.order_by(_SORTS.get(sort, _SORTS["newest"]))
    return db.paginate(query, page=page, per_page=per_page, error_out=False)


def get_advertisements_by_ids(ids: list[int]) -> list[Advertisement]:
    """Return advertisements for the given internal ids (bulk operations)."""
    if not ids:
        return []
    return list(
        db.session.scalars(db.select(Advertisement).where(Advertisement.id.in_(ids)))
    )


# ---------------------------------------------------------------------------
# Secure image handling
# ---------------------------------------------------------------------------
def _allowed_ext(filename: str, config_key: str) -> bool:
    exts = current_app.config[config_key]
    return "." in filename and filename.rsplit(".", 1)[1].lower() in exts


def _allowed_image(filename: str) -> bool:
    return _allowed_ext(filename, "ALLOWED_AD_IMAGE_EXTENSIONS")


def _allowed_video(filename: str) -> bool:
    return _allowed_ext(filename, "ALLOWED_AD_VIDEO_EXTENSIONS")


def _file_size(file: FileStorage) -> int:
    file.seek(0, os.SEEK_END)
    size = file.tell()
    file.seek(0)
    return size


def _image_content_ok(file: FileStorage) -> bool:
    """True when the upload decodes as a genuine image.

    Defends against a non-image file merely renamed with an image extension (or a
    corrupted/truncated image). Reads the already size-capped bytes once, runs
    Pillow's ``verify()`` and rewinds the stream so the existing save helper still
    works — reusing the project's established ``Image.open(...).verify()`` idiom
    (see :mod:`app.marketplace.validators` / premium receipt validation)."""
    import io as _io

    from PIL import Image

    file.stream.seek(0)
    data = file.read()
    file.stream.seek(0)  # rewind for the subsequent save
    try:
        Image.open(_io.BytesIO(data)).verify()
        return True
    except Exception:  # noqa: BLE001 - any Pillow failure ⇒ not a valid image
        return False


def image_validation_error(file: FileStorage) -> str | None:
    """Return a human message if the uploaded image is invalid, else None.

    Validates extension, size **and** actual image content (Pillow), so a file
    that merely has an image extension is rejected. This is the single shared
    image validator — advertisement images and the Premium QR upload both benefit.
    """
    if not file or not file.filename:
        return None  # No file supplied is valid (image is optional).
    if not _allowed_image(file.filename):
        allowed = ", ".join(sorted(current_app.config["ALLOWED_AD_IMAGE_EXTENSIONS"]))
        return f"Unsupported image type. Allowed formats: {allowed}."
    max_size = current_app.config["MAX_AD_IMAGE_SIZE"]
    if _file_size(file) > max_size:
        return f"Image is too large. Maximum size is {max_size // (1024 * 1024)} MB."
    if not _image_content_ok(file):
        return "The uploaded file is not a valid image."
    return None


# Video container magic bytes — content is sniffed so a mislabelled/executable
# file renamed to .mp4 is rejected (never trust the extension alone).
def _looks_like_video(file: FileStorage) -> bool:
    file.seek(0)
    head = file.read(16) or b""
    file.seek(0)
    if len(head) < 12:
        return False
    if head[4:8] == b"ftyp":                 # MP4 / MOV / M4V (ISO base media)
        return True
    if head[:4] == b"\x1a\x45\xdf\xa3":       # WebM / Matroska (EBML header)
        return True
    if head[:4] == b"OggS":                   # Ogg
        return True
    return False


def video_validation_error(file: FileStorage) -> str | None:
    """Return a human message if the uploaded video is invalid, else None.

    Validates extension, size (``MAX_AD_VIDEO_SIZE``) and — crucially — the file
    content (magic bytes), so the extension is never trusted on its own.
    """
    if not file or not file.filename:
        return None
    if not _allowed_video(file.filename):
        allowed = ", ".join(sorted(current_app.config["ALLOWED_AD_VIDEO_EXTENSIONS"]))
        return f"Unsupported video format. Allowed formats: {allowed}."
    max_size = current_app.config["MAX_AD_VIDEO_SIZE"]
    if _file_size(file) > max_size:
        return f"Video is too large. Maximum size is {max_size // (1024 * 1024)} MB."
    if not _looks_like_video(file):
        return "That file does not look like a valid MP4 or WebM video."
    return None


def _save_ad_upload(
    file: FileStorage,
    folder_key: str = "ADVERTISEMENT_UPLOAD_FOLDER",
    url_key: str = "ADVERTISEMENT_UPLOAD_URL",
) -> str:
    """Persist an uploaded image/asset under a generated UUID name in a public
    static folder; return its public ``/static/...`` URL.

    A UUID filename (never the client-supplied name) prevents path traversal and
    collisions. ``folder_key`` / ``url_key`` name the config entries for the
    destination folder and its public URL prefix, so the same secure save logic is
    reused for other public uploads (e.g. the Premium UPI QR) without duplication;
    they default to the advertisement upload location for existing callers. The
    returned value is directly usable as an ``<img src>`` / ``<video src>``.
    """
    ext = file.filename.rsplit(".", 1)[1].lower()
    stored_name = f"{uuid.uuid4().hex}.{ext}"
    destination = current_app.config[folder_key] / stored_name
    file.save(destination)
    url_path = f"{current_app.config[url_key]}/{stored_name}"
    # Build from static_url_path (works in any app context, no request needed).
    return f"{current_app.static_url_path}/{url_path}"


def save_ad_image(file: FileStorage) -> str:
    """Persist an uploaded advertisement image; return its public URL."""
    return _save_ad_upload(file)


def save_ad_video(file: FileStorage) -> str:
    """Persist an uploaded advertisement video; return its public URL."""
    return _save_ad_upload(file)


def remove_ad_asset(
    path: str | None,
    folder_key: str = "ADVERTISEMENT_UPLOAD_FOLDER",
    url_key: str = "ADVERTISEMENT_UPLOAD_URL",
) -> None:
    """Delete a locally-stored public upload (image/video/QR). External URLs and
    legacy advertiser-hosted links are left untouched. ``folder_key`` / ``url_key``
    default to the advertisement location so existing callers are unchanged; other
    public uploads (e.g. the Premium QR) pass their own config keys."""
    if not path:
        return
    prefix = f"{current_app.static_url_path}/{current_app.config[url_key]}"
    if not path.startswith(prefix):
        return  # External asset (e.g. advertiser-hosted URL): nothing to delete.
    stored_name = os.path.basename(path)
    # Guard against traversal: only ever touch our upload folder + a bare name.
    if stored_name in ("", ".", ".."):
        return
    file_path = current_app.config[folder_key] / stored_name
    try:
        if file_path.is_file():
            file_path.unlink()
    except OSError:  # pragma: no cover - best effort cleanup
        pass


# Backwards-compatible alias (kept for callers referring to the image-specific name).
remove_ad_image = remove_ad_asset


# ---------------------------------------------------------------------------
# CRUD + bulk operations
# ---------------------------------------------------------------------------
def video_upload_error(form, ad: Advertisement | None) -> str | None:
    """Validate the video upload for a submitted form given the current ad.

    Returns a human message or None. Enforces the business rule that only
    Rewarded advertisements may use video, and that a video advertisement must
    end up with a video (a freshly uploaded valid file, or a kept existing one).
    Non-video ads never fail here. This is the server-side guard against manually
    crafted Banner + Video requests — never relies on the JavaScript form logic.
    """
    if form.media_type.data != AdvertisementMediaType.VIDEO.value:
        return None
    if form.advertisement_type.data != AdvertisementType.REWARDED.value:
        return "Video advertisements are only supported for Rewarded advertisements."
    file = form.video.data
    if file and file.filename:
        return video_validation_error(file)
    removing = bool(getattr(form, "remove_video", None) and form.remove_video.data)
    existing = ad.media_url if ad else None
    if removing or not existing:
        return "A video file is required for video advertisements."
    return None


def _apply_form(ad: Advertisement, form) -> None:
    """Copy validated scalar form fields onto an advertisement.

    Media *files* (image_path / media_url video) are handled separately in
    create/update so file IO stays out of this pure field-copy step.
    """
    ad.title = form.title.data.strip()
    ad.description = (form.description.data or "").strip() or None
    ad.redirect_url = (form.redirect_url.data or "").strip() or None
    ad.advertisement_type = AdvertisementType(form.advertisement_type.data)
    ad.placement = AdvertisementPlacement(form.placement.data)
    # Rewarded advertisements always live in the CHAT placement — the single home of
    # the "watch ad to unlock messages" flow. The form hides the placement dropdown
    # for Rewarded and assigns it automatically; this normalises any crafted request
    # too. Conversely CHAT is reserved for rewarded ads, so a banner that somehow
    # arrives with CHAT (crafted request) is pushed back to the default LANDING.
    if ad.advertisement_type == AdvertisementType.REWARDED:
        ad.placement = AdvertisementPlacement.CHAT
    elif ad.placement == AdvertisementPlacement.CHAT:
        ad.placement = AdvertisementPlacement.LANDING
    # Display Weight (stored in ``priority``). Preserve an explicit 0 (valid — it
    # means "don't show unless it's the only option"); a blank field defaults.
    ad.priority = (
        form.priority.data
        if form.priority.data is not None
        else current_app.config.get("DEFAULT_AD_PRIORITY", 100)
    )
    ad.icon = (form.icon.data or "").strip() or None
    ad.start_at = form.start_at.data
    ad.end_at = form.end_at.data
    ad.is_active = bool(form.is_active.data)

    ad.media_type = AdvertisementMediaType(form.media_type.data)
    # Business rule: only Rewarded advertisements may be video. Banner is always
    # image (belt-and-suspenders — validation rejects Banner+Video before this,
    # but normalising here keeps the invariant even if that guard is bypassed;
    # _apply_video_media then clears any stored video for the non-video type).
    if ad.advertisement_type != AdvertisementType.REWARDED:
        ad.media_type = AdvertisementMediaType.IMAGE
    ad.video_duration = (
        (form.video_duration.data or current_app.config.get("REWARDED_AD_DURATION_SECONDS", 20))
        if ad.media_type == AdvertisementMediaType.VIDEO
        else None
    )


def _apply_video_media(ad: Advertisement, form, *, is_update: bool) -> None:
    """Persist the video file for a video ad, or clear the video for a non-video ad.

    Mirrors the image handling: a new upload replaces the old file (deleting the
    previous local asset on update); switching away from video removes the stored
    video file. A kept existing video (no new upload) is left untouched.
    """
    if ad.media_type == AdvertisementMediaType.VIDEO:
        file = form.video.data
        if file and file.filename:
            old = ad.media_url
            ad.media_url = save_ad_video(file)
            if is_update:
                remove_ad_asset(old)
        # else: keep the existing ad.media_url (guaranteed present by validation).
    else:
        old = ad.media_url
        ad.media_url = None
        if is_update and old:
            remove_ad_asset(old)


def create_advertisement(form) -> Advertisement:
    """Create an advertisement from a validated form, saving any uploaded media."""
    ad = Advertisement()
    _apply_form(ad, form)
    file = form.image.data
    if file and file.filename:
        ad.image_path = save_ad_image(file)
    _apply_video_media(ad, form, is_update=False)
    db.session.add(ad)
    db.session.commit()
    invalidate_storage_stats()
    return ad


def update_advertisement(ad: Advertisement, form) -> Advertisement:
    """Update an advertisement. A new upload replaces the old asset; otherwise the
    current image/video is kept. Old local files are cleaned up on replacement."""
    _apply_form(ad, form)
    file = form.image.data
    if file and file.filename:
        old = ad.image_path
        ad.image_path = save_ad_image(file)
        remove_ad_asset(old)
    _apply_video_media(ad, form, is_update=True)
    db.session.commit()
    invalidate_storage_stats()
    return ad


def duplicate_advertisement(ad: Advertisement) -> Advertisement:
    """Create a disabled copy of an advertisement named "<title> (Copy)"."""
    copy = Advertisement(
        title=f"{ad.title} (Copy)",
        description=ad.description,
        advertisement_type=ad.advertisement_type,
        placement=ad.placement,
        media_type=ad.media_type,
        media_url=ad.media_url,
        video_duration=ad.video_duration,
        image_path=ad.image_path,  # reuse the same image reference
        icon=ad.icon,
        redirect_url=ad.redirect_url,
        priority=ad.priority,
        start_at=ad.start_at,
        end_at=ad.end_at,
        is_active=False,  # copies start disabled so they can be reviewed first
    )
    db.session.add(copy)
    db.session.commit()
    return copy


def _cleanup_asset_if_unused(column, path: str | None) -> None:
    """Remove a stored asset file only if no advertisement still references it
    (duplicates share the same file reference)."""
    if not path:
        return
    still_used = db.session.scalar(
        db.select(db.func.count()).select_from(Advertisement).where(column == path)
    )
    if not still_used:
        remove_ad_asset(path)


def delete_advertisement(ad: Advertisement) -> None:
    """Delete an advertisement and clean up its uploaded image/video if unused."""
    image_path = ad.image_path
    media_url = ad.media_url
    db.session.delete(ad)
    db.session.commit()
    _cleanup_asset_if_unused(Advertisement.image_path, image_path)
    _cleanup_asset_if_unused(Advertisement.media_url, media_url)
    invalidate_storage_stats()


def set_active(ad: Advertisement, active: bool) -> None:
    """Enable or disable a single advertisement."""
    ad.is_active = active
    db.session.commit()


def bulk_set_active(ids: list[int], active: bool) -> int:
    ads = get_advertisements_by_ids(ids)
    for ad in ads:
        ad.is_active = active
    db.session.commit()
    return len(ads)


def bulk_delete(ids: list[int]) -> int:
    ads = get_advertisements_by_ids(ids)
    image_paths = [ad.image_path for ad in ads if ad.image_path]
    media_urls = [ad.media_url for ad in ads if ad.media_url]
    for ad in ads:
        db.session.delete(ad)
    db.session.commit()
    for image_path in image_paths:
        _cleanup_asset_if_unused(Advertisement.image_path, image_path)
    for media_url in media_urls:
        _cleanup_asset_if_unused(Advertisement.media_url, media_url)
    invalidate_storage_stats()
    return len(ads)


def bulk_set_priority(ids: list[int], priority: int) -> int:
    ads = get_advertisements_by_ids(ids)
    for ad in ads:
        ad.priority = priority
    db.session.commit()
    return len(ads)


def bulk_set_placement(ids: list[int], placement: str) -> int:
    if placement not in {p.value for p in AdvertisementPlacement}:
        return 0
    ads = get_advertisements_by_ids(ids)
    for ad in ads:
        ad.placement = AdvertisementPlacement(placement)
    db.session.commit()
    return len(ads)


def bulk_duplicate(ids: list[int]) -> int:
    """Duplicate each selected advertisement (copies start disabled)."""
    ads = get_advertisements_by_ids(ids)
    for ad in ads:
        duplicate_advertisement(ad)
    return len(ads)


# ---------------------------------------------------------------------------
# Analytics
# ---------------------------------------------------------------------------
def analytics_overview() -> dict:
    """Aggregate headline metrics for the analytics dashboard."""
    ads = list(db.session.scalars(db.select(Advertisement)))
    now = utcnow()
    status_counts = {STATUS_ACTIVE: 0, STATUS_SCHEDULED: 0, STATUS_EXPIRED: 0, STATUS_DISABLED: 0}
    total_impressions = total_clicks = 0
    most_clicked = None
    highest_ctr = None
    highest_ctr_value = -1.0

    for ad in ads:
        status_counts[advertisement_status(ad, now)] += 1
        total_impressions += ad.impression_count
        total_clicks += ad.click_count
        if most_clicked is None or ad.click_count > most_clicked.click_count:
            most_clicked = ad
        ctr = compute_ctr(ad.impression_count, ad.click_count)
        if ad.impression_count and ctr > highest_ctr_value:
            highest_ctr_value = ctr
            highest_ctr = ad

    return {
        "total": len(ads),
        "status_counts": status_counts,
        "total_impressions": total_impressions,
        "total_clicks": total_clicks,
        "average_ctr": compute_ctr(total_impressions, total_clicks),
        "most_clicked": most_clicked,
        "highest_ctr": highest_ctr,
        "highest_ctr_value": max(highest_ctr_value, 0.0),
    }


def analytics_chart_data(top_n: int = 8) -> dict:
    """Build Chart.js-ready datasets (labels + numeric series)."""
    now = utcnow()
    ads = list(
        db.session.scalars(
            db.select(Advertisement).order_by(Advertisement.impression_count.desc())
        )
    )
    top = ads[:top_n]
    labels = [ad.title for ad in top]
    impressions = [ad.impression_count for ad in top]
    clicks = [ad.click_count for ad in top]
    ctrs = [compute_ctr(ad.impression_count, ad.click_count) for ad in top]

    placement_counts = {p.value: 0 for p in AdvertisementPlacement}
    active_expired = {"active": 0, "expired": 0}
    for ad in ads:
        placement_counts[ad.placement.value] += 1
        status = advertisement_status(ad, now)
        if status == STATUS_ACTIVE:
            active_expired["active"] += 1
        elif status == STATUS_EXPIRED:
            active_expired["expired"] += 1

    # Last 30 days: advertisements created per day (real timestamps available).
    created_labels: list[str] = []
    created_series: list[int] = []
    buckets: dict[str, int] = {}
    for ad in ads:
        key = ad.created_at.strftime("%Y-%m-%d")
        buckets[key] = buckets.get(key, 0) + 1
    for key in sorted(buckets)[-30:]:
        created_labels.append(key)
        created_series.append(buckets[key])

    return {
        "labels": labels,
        "impressions": impressions,
        "clicks": clicks,
        "ctrs": ctrs,
        "placement_labels": [AdvertisementPlacement(v).display_name for v in placement_counts],
        "placement_counts": list(placement_counts.values()),
        "active_expired_labels": ["Active", "Expired"],
        "active_expired_counts": [active_expired["active"], active_expired["expired"]],
        "created_labels": created_labels,
        "created_series": created_series,
    }


def placement_summary() -> list[dict]:
    """Per-placement rollup (campaigns, impressions, clicks, CTR) in one scan.

    Read-only presentation helper for the dashboard placement cards. Ordered to
    match the :class:`AdvertisementPlacement` enum so every placement — including
    those with no campaigns yet — always renders.
    """
    rows = {
        p: {"value": p.value, "label": p.display_name, "count": 0, "impressions": 0, "clicks": 0}
        for p in AdvertisementPlacement
    }
    for ad in db.session.scalars(db.select(Advertisement)):
        row = rows[ad.placement]
        row["count"] += 1
        row["impressions"] += ad.impression_count
        row["clicks"] += ad.click_count
    summary = []
    for p in AdvertisementPlacement:
        row = rows[p]
        row["ctr"] = compute_ctr(row["impressions"], row["clicks"])
        summary.append(row)
    return summary