"""Cross-module reporting: the live dashboard, trend series, and exportable tables.

Nothing here writes, and nothing here re-implements a rule that already lives in a
module service — the dashboard reuses `applications`, `equipment`, and `complaints`
summaries as-is and only adds what spans them.

**Buckets are Asia/Dhaka days, not UTC days.** Every trend groups on
`date_trunc(<unit>, ts AT TIME ZONE 'Asia/Dhaka')`, so a ticket logged at 11pm Dhaka
lands on the day the operator would say it happened. Grouping on the raw `timestamptz`
would silently shift six hours of every day into the previous bucket.

**Gaps are filled in Python.** A day with no activity must appear as a zero, otherwise a
line chart joins across the hole and reads as though the quiet day never existed.
"""
from datetime import date, datetime, timedelta

from sqlalchemy import Select, case, func, select
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.time import range_bounds_utc, today_local, utcnow
from app.models.application import ApplicationStatusHistory, ConnectionApplication
from app.models.complaint import Complaint
from app.models.customer import Connection, Customer
from app.models.enums import (
    ApplicationStatus,
    ConnectionStatus,
    StockMovementType,
)
from app.models.equipment import (
    EquipmentCollection,
    EquipmentDistribution,
    EquipmentItem,
    EquipmentStockMovement,
)
from app.models.reference import Area, EquipmentType
from app.models.user import User
from app.services import complaints as complaint_service
from app.services import equipment as equipment_service

#: Trend buckets, mapped to the `date_trunc` unit that produces them.
GRANULARITIES = {"day": "day", "week": "week", "month": "month"}

#: Guard against a chart request that would return thousands of points.
MAX_BUCKETS = 400

#: How far back a trend looks when the caller gives no range.
DEFAULT_TREND_DAYS = 30

#: Export tables are capped — a report is something a person reads, and anything larger
#: belongs in a database query, not a spreadsheet.
EXPORT_LIMIT = 10000


# --- bucketing ---


def _local(column):
    """A stored UTC `timestamptz` rendered as a naive Asia/Dhaka timestamp."""
    return func.timezone(settings.TIMEZONE, column)


def _bucket(column, granularity: str):
    return func.date_trunc(GRANULARITIES[granularity], _local(column))


def _bucket_start(day: date, granularity: str) -> date:
    """The first day of the bucket `day` falls in — mirrors `date_trunc` in Python."""
    if granularity == "week":
        return day - timedelta(days=day.weekday())  # Postgres weeks start Monday
    if granularity == "month":
        return day.replace(day=1)
    return day


def _next_bucket(day: date, granularity: str) -> date:
    if granularity == "week":
        return day + timedelta(days=7)
    if granularity == "month":
        return (day.replace(day=28) + timedelta(days=4)).replace(day=1)
    return day + timedelta(days=1)


def resolve_range(date_from: date | None, date_to: date | None) -> tuple[date, date]:
    """Default to the last 30 Dhaka days, and never let `to` precede `from`."""
    end = date_to or today_local()
    start = date_from or (end - timedelta(days=DEFAULT_TREND_DAYS - 1))
    return (end, end) if start > end else (start, end)


def _series(rows: dict[date, dict], start: date, end: date, granularity: str, blank: dict) -> list[dict]:
    """Emit one entry per bucket across the whole range, zero-filling the quiet ones."""
    out: list[dict] = []
    cursor = _bucket_start(start, granularity)
    last = _bucket_start(end, granularity)
    while cursor <= last and len(out) < MAX_BUCKETS:
        out.append({"bucket": cursor, **blank, **rows.get(cursor, {})})
        cursor = _next_bucket(cursor, granularity)
    return out


def _as_date(value: datetime | date) -> date:
    return value.date() if isinstance(value, datetime) else value


def _tally(condition):
    """Count only the rows matching `condition`, inside a grouped select."""
    return func.count(case((condition, 1)))


# --- live dashboard ---


def dashboard(db: Session) -> dict:
    """The auto-refreshing snapshot: what is happening now, across every module.

    Composed from each module's own summary so the numbers on this screen can never
    disagree with the numbers on that module's screen.
    """
    today = today_local()
    today_start, today_end = range_bounds_utc(today, today)
    active_app = ConnectionApplication.is_deleted.is_(False)

    def apps(*conditions) -> int:
        return db.scalar(
            select(func.count()).select_from(ConnectionApplication).where(active_app, *conditions)
        ) or 0

    installed_today = db.scalar(
        select(func.count())
        .select_from(ApplicationStatusHistory)
        .join(ApplicationStatusHistory.application)
        .where(
            active_app,
            ApplicationStatusHistory.to_status == ApplicationStatus.INSTALLED,
            ApplicationStatusHistory.changed_at >= today_start,
            ApplicationStatusHistory.changed_at < today_end,
        )
    ) or 0

    issued_today = db.scalar(
        select(func.count())
        .select_from(EquipmentDistribution)
        .where(
            EquipmentDistribution.is_deleted.is_(False),
            EquipmentDistribution.issued_at >= today_start,
            EquipmentDistribution.issued_at < today_end,
        )
    ) or 0
    collected_today = db.scalar(
        select(func.count())
        .select_from(EquipmentCollection)
        .where(
            EquipmentCollection.is_deleted.is_(False),
            EquipmentCollection.collected_at >= today_start,
            EquipmentCollection.collected_at < today_end,
        )
    ) or 0

    active_connections = db.scalar(
        select(func.count())
        .select_from(Connection)
        .where(Connection.is_deleted.is_(False), Connection.status == ConnectionStatus.ACTIVE)
    ) or 0

    tickets = complaint_service.summary(db)
    stock = equipment_service.summary(db)

    return {
        "generated_at": utcnow(),
        "day": today,
        "applications": {
            "created_today": apps(
                ConnectionApplication.created_at >= today_start,
                ConnectionApplication.created_at < today_end,
            ),
            "installed_today": installed_today,
            "pending": apps(ConnectionApplication.status == ApplicationStatus.PENDING),
            "in_survey": apps(ConnectionApplication.status == ApplicationStatus.SURVEY),
            "pending_installs": apps(ConnectionApplication.status == ApplicationStatus.APPROVED),
            "total": apps(),
        },
        "complaints": {
            "open": tickets["open"],
            "in_progress": tickets["in_progress"],
            "unassigned": tickets["unassigned"],
            "reported_today": tickets["reported_today"],
            "resolved_today": tickets["resolved_today"],
        },
        "equipment": {
            "in_stock": stock["in_stock"],
            "issued_out": stock["issued_out"],
            "quarantined": stock["quarantined"],
            "overdue_returns": stock["overdue_distributions"],
            "types_below_reorder": stock["types_below_reorder"],
            "issued_today": issued_today,
            "collected_today": collected_today,
        },
        "active_connections": active_connections,
        "stock_by_type": equipment_service.stock_by_type(db),
    }


# --- trends ---


def applications_trend(
    db: Session, *, date_from: date | None, date_to: date | None, granularity: str
) -> list[dict]:
    """Intake against throughput: created, installed, and rejected per bucket.

    `created` is bucketed by `created_at`; the outcomes are bucketed by *when the move
    happened*, not by the application's own date — an application logged in March and
    installed in April is April's install.
    """
    start, end = resolve_range(date_from, date_to)
    lo, hi = range_bounds_utc(start, end)

    created = db.execute(
        select(_bucket(ConnectionApplication.created_at, granularity).label("bucket"), func.count())
        .where(
            ConnectionApplication.is_deleted.is_(False),
            ConnectionApplication.created_at >= lo,
            ConnectionApplication.created_at < hi,
        )
        .group_by("bucket")
    ).all()

    moves = db.execute(
        select(
            _bucket(ApplicationStatusHistory.changed_at, granularity).label("bucket"),
            _tally(ApplicationStatusHistory.to_status == ApplicationStatus.INSTALLED),
            _tally(ApplicationStatusHistory.to_status == ApplicationStatus.REJECTED),
        )
        .join(ApplicationStatusHistory.application)
        .where(
            ConnectionApplication.is_deleted.is_(False),
            ApplicationStatusHistory.changed_at >= lo,
            ApplicationStatusHistory.changed_at < hi,
        )
        .group_by("bucket")
    ).all()

    rows: dict[date, dict] = {}
    for bucket, count in created:
        rows.setdefault(_as_date(bucket), {})["created"] = count
    for bucket, installed, rejected in moves:
        entry = rows.setdefault(_as_date(bucket), {})
        entry["installed"] = installed
        entry["rejected"] = rejected

    return _series(rows, start, end, granularity, {"created": 0, "installed": 0, "rejected": 0})


def complaints_trend(
    db: Session, *, date_from: date | None, date_to: date | None, granularity: str
) -> list[dict]:
    """Volume reported and volume resolved per bucket.

    `resolved` is bucketed by `resolved_at`, not `reported_at`: a ticket resolved in a
    later bucket counts there, which is when the work actually finished.
    """
    start, end = resolve_range(date_from, date_to)
    lo, hi = range_bounds_utc(start, end)
    active = Complaint.is_deleted.is_(False)

    reported = db.execute(
        select(_bucket(Complaint.reported_at, granularity).label("bucket"), func.count())
        .where(active, Complaint.reported_at >= lo, Complaint.reported_at < hi)
        .group_by("bucket")
    ).all()

    resolved = db.execute(
        select(_bucket(Complaint.resolved_at, granularity).label("bucket"), func.count())
        .where(active, Complaint.resolved_at >= lo, Complaint.resolved_at < hi)
        .group_by("bucket")
    ).all()

    rows: dict[date, dict] = {}
    for bucket, count in reported:
        rows.setdefault(_as_date(bucket), {})["reported"] = count
    for bucket, count in resolved:
        rows.setdefault(_as_date(bucket), {})["resolved"] = count

    return _series(
        rows,
        start,
        end,
        granularity,
        {"reported": 0, "resolved": 0},
    )


def stock_movement_trend(
    db: Session,
    *,
    date_from: date | None,
    date_to: date | None,
    granularity: str,
    equipment_type_id: int | None = None,
) -> list[dict]:
    """Units moving through the warehouse, read off the append-only movement ledger.

    `quantity_delta` is signed against on-hand stock, so an issue is negative and a
    return is positive. The per-type columns report magnitudes — what a storekeeper
    counts — while `net` keeps the sign, because that is the number that reconciles.
    """
    start, end = resolve_range(date_from, date_to)
    lo, hi = range_bounds_utc(start, end)
    delta = EquipmentStockMovement.quantity_delta

    def moved(movement_type: StockMovementType):
        return func.coalesce(
            func.sum(case((EquipmentStockMovement.movement_type == movement_type, func.abs(delta)))),
            0,
        )

    stmt = (
        select(
            _bucket(EquipmentStockMovement.moved_at, granularity).label("bucket"),
            moved(StockMovementType.INBOUND),
            moved(StockMovementType.ISSUE),
            moved(StockMovementType.RETURN),
            moved(StockMovementType.ADJUSTMENT),
            moved(StockMovementType.WRITE_OFF),
            func.coalesce(func.sum(delta), 0),
        )
        .where(EquipmentStockMovement.moved_at >= lo, EquipmentStockMovement.moved_at < hi)
        .group_by("bucket")
    )
    if equipment_type_id is not None:
        stmt = stmt.where(EquipmentStockMovement.equipment_type_id == equipment_type_id)

    rows: dict[date, dict] = {}
    for bucket, inbound, issued, returned, adjusted, written_off, net in db.execute(stmt).all():
        rows[_as_date(bucket)] = {
            "inbound": int(inbound),
            "issued": int(issued),
            "returned": int(returned),
            "adjusted": int(adjusted),
            "written_off": int(written_off),
            "net": int(net),
        }

    return _series(
        rows,
        start,
        end,
        granularity,
        {"inbound": 0, "issued": 0, "returned": 0, "adjusted": 0, "written_off": 0, "net": 0},
    )


# --- tabular reports (also what the exports serialize) ---


def _limited(db: Session, stmt: Select) -> list:
    return db.execute(stmt.limit(EXPORT_LIMIT)).all()


def applications_report(db: Session, *, date_from: date | None, date_to: date | None) -> list[dict]:
    """One row per application created in the range."""
    start, end = resolve_range(date_from, date_to)
    lo, hi = range_bounds_utc(start, end)
    technician = User.__table__.alias("technician")

    stmt = (
        select(
            ConnectionApplication.application_no,
            ConnectionApplication.created_at,
            ConnectionApplication.applicant_name,
            ConnectionApplication.phone,
            ConnectionApplication.address,
            Area.name.label("area"),
            ConnectionApplication.status,
            ConnectionApplication.requested_date,
            technician.c.full_name.label("technician"),
            ConnectionApplication.installed_at,
        )
        .join(Area, Area.id == ConnectionApplication.area_id)
        .outerjoin(technician, technician.c.id == ConnectionApplication.assigned_technician_id)
        .where(
            ConnectionApplication.is_deleted.is_(False),
            ConnectionApplication.created_at >= lo,
            ConnectionApplication.created_at < hi,
        )
        .order_by(ConnectionApplication.created_at.desc())
    )
    return [dict(row._mapping) for row in _limited(db, stmt)]


def complaints_report(db: Session, *, date_from: date | None, date_to: date | None) -> list[dict]:
    """One row per ticket reported in the range."""
    start, end = resolve_range(date_from, date_to)
    lo, hi = range_bounds_utc(start, end)
    technician = User.__table__.alias("technician")

    stmt = (
        select(
            Complaint.ticket_no,
            Complaint.reported_at,
            func.coalesce(Customer.customer_code, "").label("customer_code"),
            func.coalesce(Customer.full_name, Complaint.customer_name, "").label("customer"),
            func.coalesce(Connection.connection_code, "").label("connection_code"),
            Area.name.label("area"),
            Complaint.complaint_type,
            Complaint.priority,
            Complaint.status,
            technician.c.full_name.label("technician"),
            Complaint.resolved_at,
            Complaint.description,
        )
        .outerjoin(Connection, Connection.id == Complaint.connection_id)
        .outerjoin(Customer, Customer.id == Connection.customer_id)
        .outerjoin(Area, Area.id == Connection.area_id)
        .outerjoin(technician, technician.c.id == Complaint.assigned_technician_id)
        .where(
            Complaint.is_deleted.is_(False),
            Complaint.reported_at >= lo,
            Complaint.reported_at < hi,
        )
        .order_by(Complaint.reported_at.desc())
    )
    return [dict(row._mapping) for row in _limited(db, stmt)]


def equipment_report(db: Session, *, date_from: date | None, date_to: date | None) -> list[dict]:
    """Issues and returns in the range, as one interleaved movement list.

    Two tables in one report because a storekeeper reconciles them together: an issue
    and its collection are the same loan seen from either end.
    """
    start, end = resolve_range(date_from, date_to)
    lo, hi = range_bounds_utc(start, end)
    holder = User.__table__.alias("holder")
    issuer = User.__table__.alias("issuer")
    collector = User.__table__.alias("collector")

    # Two selects merged in Python rather than one UNION: the sides carry different
    # columns (a return has a condition, an issue has a due date), so a UNION would mean
    # padding both with nulls for the other's fields just to satisfy SQL.
    issued = _limited(
        db,
        select(
            EquipmentDistribution.issued_at.label("at"),
            EquipmentType.name.label("equipment_type"),
            EquipmentItem.serial_number,
            EquipmentDistribution.quantity,
            holder.c.full_name.label("to_user"),
            Customer.full_name.label("to_customer"),
            EquipmentDistribution.expected_return_date,
            EquipmentDistribution.status,
            issuer.c.full_name.label("by"),
        )
        .join(EquipmentItem, EquipmentItem.id == EquipmentDistribution.equipment_item_id)
        .join(EquipmentType, EquipmentType.id == EquipmentItem.equipment_type_id)
        .outerjoin(holder, holder.c.id == EquipmentDistribution.issued_to_user_id)
        .outerjoin(Customer, Customer.id == EquipmentDistribution.issued_to_customer_id)
        .outerjoin(issuer, issuer.c.id == EquipmentDistribution.issued_by_id)
        .where(
            EquipmentDistribution.is_deleted.is_(False),
            EquipmentDistribution.issued_at >= lo,
            EquipmentDistribution.issued_at < hi,
        )
        .order_by(EquipmentDistribution.issued_at.desc()),
    )

    rows = [
        {
            "movement": "Issued",
            "at": row.at,
            "equipment_type": row.equipment_type,
            "serial_number": row.serial_number,
            "quantity": row.quantity,
            "counterparty": row.to_user or row.to_customer,
            "expected_return_date": row.expected_return_date,
            "condition_on_return": None,
            "status": row.status,
            "by": row.by,
        }
        for row in issued
    ]

    collected = _limited(
        db,
        select(
            EquipmentCollection.collected_at.label("at"),
            EquipmentType.name.label("equipment_type"),
            EquipmentItem.serial_number,
            EquipmentCollection.quantity,
            holder.c.full_name.label("from_user"),
            Customer.full_name.label("from_customer"),
            EquipmentCollection.condition_on_return,
            collector.c.full_name.label("by"),
        )
        .join(EquipmentItem, EquipmentItem.id == EquipmentCollection.equipment_item_id)
        .join(EquipmentType, EquipmentType.id == EquipmentItem.equipment_type_id)
        .outerjoin(holder, holder.c.id == EquipmentCollection.collected_from_user_id)
        .outerjoin(Customer, Customer.id == EquipmentCollection.collected_from_customer_id)
        .outerjoin(collector, collector.c.id == EquipmentCollection.collected_by_id)
        .where(
            EquipmentCollection.is_deleted.is_(False),
            EquipmentCollection.collected_at >= lo,
            EquipmentCollection.collected_at < hi,
        )
        .order_by(EquipmentCollection.collected_at.desc()),
    )
    rows.extend(
        {
            "movement": "Returned",
            "at": row.at,
            "equipment_type": row.equipment_type,
            "serial_number": row.serial_number,
            "quantity": row.quantity,
            "counterparty": row.from_user or row.from_customer,
            "expected_return_date": None,
            "condition_on_return": row.condition_on_return,
            "status": None,
            "by": row.by,
        }
        for row in collected
    )

    rows.sort(key=lambda r: r["at"], reverse=True)
    return rows[:EXPORT_LIMIT]


def combined_report(
    db: Session, *, date_from: date | None, date_to: date | None, granularity: str = "day"
) -> list[dict]:
    """Every module's daily totals side by side — the one-page "how did the month go".

    Assembled from the three trend series so a number here can never disagree with the
    same number on its own chart.
    """
    applications = {r["bucket"]: r for r in applications_trend(
        db, date_from=date_from, date_to=date_to, granularity=granularity
    )}
    tickets = {r["bucket"]: r for r in complaints_trend(
        db, date_from=date_from, date_to=date_to, granularity=granularity
    )}
    stock = {r["bucket"]: r for r in stock_movement_trend(
        db, date_from=date_from, date_to=date_to, granularity=granularity
    )}

    return [
        {
            "bucket": bucket,
            "applications_created": applications[bucket]["created"],
            "applications_installed": applications[bucket]["installed"],
            "applications_rejected": applications[bucket]["rejected"],
            "complaints_reported": tickets.get(bucket, {}).get("reported", 0),
            "complaints_resolved": tickets.get(bucket, {}).get("resolved", 0),
            "equipment_issued": stock.get(bucket, {}).get("issued", 0),
            "equipment_returned": stock.get(bucket, {}).get("returned", 0),
        }
        for bucket in applications
    ]


# --- dataset registry ---
#
# One place that knows, per report: the human title, the query, and the column order the
# exports use. Adding a report means adding an entry here — the API layer and both
# exporters pick it up without changes, so a CSV can never drift from its JSON.

DATASETS: dict[str, dict] = {
    "applications": {
        "title": "Connection applications",
        "rows": applications_report,
        "columns": [
            ("application_no", "Application No"),
            ("created_at", "Created"),
            ("applicant_name", "Applicant"),
            ("phone", "Phone"),
            ("address", "Address"),
            ("area", "Area"),
            ("status", "Status"),
            ("requested_date", "Requested"),
            ("technician", "Technician"),
            ("installed_at", "Installed"),
        ],
    },
    "complaints": {
        "title": "Complaints",
        "rows": complaints_report,
        "columns": [
            ("ticket_no", "Ticket No"),
            ("reported_at", "Reported"),
            ("customer_code", "Customer Code"),
            ("customer", "Customer"),
            ("connection_code", "Connection"),
            ("area", "Area"),
            ("complaint_type", "Type"),
            ("priority", "Priority"),
            ("status", "Status"),
            ("technician", "Technician"),
            ("resolved_at", "Resolved"),
            ("description", "Description"),
        ],
    },
    "equipment": {
        "title": "Equipment movements",
        "rows": equipment_report,
        "columns": [
            ("at", "When"),
            ("movement", "Movement"),
            ("equipment_type", "Equipment"),
            ("serial_number", "Serial"),
            ("quantity", "Qty"),
            ("counterparty", "Holder"),
            ("expected_return_date", "Due"),
            ("condition_on_return", "Condition"),
            ("status", "Status"),
            ("by", "Recorded By"),
        ],
    },
    "combined": {
        "title": "Combined daily summary",
        "rows": combined_report,
        "columns": [
            ("bucket", "Date"),
            ("applications_created", "Apps Created"),
            ("applications_installed", "Apps Installed"),
            ("applications_rejected", "Apps Rejected"),
            ("complaints_reported", "Complaints Reported"),
            ("complaints_resolved", "Complaints Resolved"),
            ("equipment_issued", "Equipment Issued"),
            ("equipment_returned", "Equipment Returned"),
        ],
    },
}


def dataset_rows(db: Session, dataset: str, *, date_from: date | None, date_to: date | None) -> list[dict]:
    return DATASETS[dataset]["rows"](db, date_from=date_from, date_to=date_to)