from datetime import date, datetime
from types import SimpleNamespace

from sqlalchemy import BigInteger, create_engine, select
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import sessionmaker

from app.database import Base
from app.models import ProductionReport, ProductionReportAllocation, ProductionReportItem
from app.services.metrics import calculate_report_metrics
from app.services.report_allocations import (
    allocation_summary_text,
    build_report_allocation_drafts,
    refresh_report_allocations,
)
from app.services.work_schedule import WorkScheduleConfig


@compiles(BigInteger, "sqlite")
def _compile_big_integer_for_sqlite(type_, compiler, **kw):
    return "INTEGER"


def _item(
    *,
    id: int = 1,
    good_qty: float = 0,
    defect_qty: float = 0,
    scrap_qty: float = 0,
    changeover_count: float = 0,
    process_unit_price_yuan: float = 0,
    started_at: datetime | None = None,
    device_no: str = "23#",
    product_name: str | None = None,
    process_name: str | None = "冲压",
):
    return SimpleNamespace(
        id=id,
        device_no=device_no,
        product_name=product_name or device_no,
        process_name=process_name,
        started_at=started_at,
        standard_beat=1,
        standard_workload=0,
        stamping_method=None,
        good_qty=good_qty,
        defect_qty=defect_qty,
        scrap_qty=scrap_qty,
        changeover_count=changeover_count,
        process_unit_price_yuan=process_unit_price_yuan,
        allocated_minutes=0,
    )


def _device(scanned_at: datetime, *, device_no: str = "23#", process_name: str = "冲压", sort_order: int = 0):
    return SimpleNamespace(
        device_no=device_no,
        process_name=process_name,
        scanned_at=scanned_at,
        released_at=None,
        release_reason=None,
        sort_order=sort_order,
    )


def _report(
    start_at: datetime,
    end_at: datetime,
    item,
    *,
    id: int = 1,
    report_date: date | None = None,
    devices: list | None = None,
):
    return SimpleNamespace(
        id=id,
        attendance_point_name="总厂",
        employee_phone="13800000000",
        report_date=report_date or start_at.date(),
        start_at=start_at,
        end_at=end_at,
        items=item if isinstance(item, list) else [item],
        session=SimpleNamespace(devices=devices or []),
    )


def test_cross_day_after_midnight_allocates_previous_night_and_current_overtime_day():
    item = _item(
        good_qty=650,
        defect_qty=65,
        scrap_qty=6.5,
        changeover_count=13,
        process_unit_price_yuan=1,
    )
    report = _report(datetime(2026, 7, 25, 0, 10), datetime(2026, 7, 25, 11, 0), item)

    rows = build_report_allocation_drafts(report)

    assert len(rows) == 2
    previous, current = rows
    assert previous.allocation_date == date(2026, 7, 24)
    assert previous.night_minutes == 350
    assert previous.day_minutes == 0
    assert previous.overtime_minutes == 0
    assert previous.effective_minutes == 350
    assert previous.good_qty == 350
    assert previous.defect_qty == 35
    assert previous.scrap_qty == 3.5
    assert previous.changeover_count == 7
    assert previous.reference_wage == 350

    assert current.allocation_date == date(2026, 7, 25)
    assert current.night_minutes == 0
    assert current.overtime_minutes == 120
    assert current.day_minutes == 180
    assert current.effective_minutes == 300
    assert current.good_qty == 300
    assert current.defect_qty == 30
    assert current.scrap_qty == 3
    assert current.changeover_count == 6
    assert current.reference_wage == 300


def test_one_minute_before_night_end_allocates_to_previous_date():
    item = _item(good_qty=301, process_unit_price_yuan=2)
    report = _report(datetime(2026, 7, 25, 5, 59), datetime(2026, 7, 25, 11, 0), item)

    rows = build_report_allocation_drafts(report)

    assert [(row.allocation_date, row.night_minutes, row.overtime_minutes, row.day_minutes) for row in rows] == [
        (date(2026, 7, 24), 1, 0, 0),
        (date(2026, 7, 25), 0, 120, 180),
    ]
    assert rows[0].good_qty == 1
    assert rows[0].reference_wage == 2
    assert rows[1].good_qty == 300
    assert rows[1].reference_wage == 600


def test_no_device_shared_full_report_splits_effective_minutes_between_items():
    first_item = _item(id=1, good_qty=10)
    second_item = _item(id=2, good_qty=20)
    report = _report(
        datetime(2026, 7, 25, 8, 0),
        datetime(2026, 7, 25, 11, 0),
        [first_item, second_item],
    )

    rows = build_report_allocation_drafts(report)

    assert [(row.report_item_id, row.effective_minutes, row.day_minutes) for row in rows] == [
        (1, 90, 90),
        (2, 90, 90),
    ]


def test_short_shared_report_rounding_residual_keeps_aggregate_night_total():
    start_at = datetime(2026, 5, 3, 1, 4, 50)
    end_at = datetime(2026, 5, 3, 1, 4, 53)
    first_item = _item(id=1, good_qty=1)
    second_item = _item(id=2, good_qty=1)
    report = _report(start_at, end_at, [first_item, second_item])

    metrics = calculate_report_metrics(start_at, end_at, [first_item, second_item])
    rows = build_report_allocation_drafts(report)

    assert metrics["effective_minutes"] == 0.05
    assert metrics["shift_night_minutes"] == 0.05
    assert [row.allocation_date for row in rows] == [date(2026, 5, 2), date(2026, 5, 2)]
    assert [row.effective_minutes for row in rows] == [0.03, 0.02]
    assert round(sum(row.effective_minutes for row in rows), 2) == 0.05
    assert round(sum(row.night_minutes for row in rows), 2) == 0.05
    assert round(sum(row.day_minutes for row in rows), 2) == 0
    assert round(sum(row.overtime_minutes for row in rows), 2) == 0
    assert all(row.effective_minutes == round(row.day_minutes + row.overtime_minutes + row.night_minutes, 2) for row in rows)


def test_items_sharing_same_started_at_split_the_segment_minutes():
    started_at = datetime(2026, 7, 25, 8, 0)
    first_item = _item(id=1, good_qty=10, started_at=started_at)
    second_item = _item(id=2, good_qty=20, started_at=started_at)
    report = _report(
        started_at,
        datetime(2026, 7, 25, 11, 0),
        [first_item, second_item],
        devices=[_device(started_at)],
    )

    rows = build_report_allocation_drafts(report)

    assert [(row.report_item_id, row.effective_minutes, row.day_minutes) for row in rows] == [
        (1, 90, 90),
        (2, 90, 90),
    ]


def test_device_matched_and_unmatched_items_match_metrics_remaining_minutes():
    start_at = datetime(2026, 7, 25, 8, 0)
    end_at = datetime(2026, 7, 25, 11, 0)
    matching_item = _item(id=1, good_qty=10, device_no="23#")
    unmatched_item = _item(id=2, good_qty=20, device_no="99#")
    report = _report(start_at, end_at, [matching_item, unmatched_item], devices=[_device(start_at)])

    calculate_report_metrics(start_at, end_at, [matching_item, unmatched_item], devices=[_device(start_at)])
    rows = build_report_allocation_drafts(report)

    assert {matching_item.id: matching_item.allocated_minutes, unmatched_item.id: unmatched_item.allocated_minutes} == {
        1: 180,
        2: 0,
    }
    assert {row.report_item_id: row.effective_minutes for row in rows} == {
        1: 180,
        2: 0,
    }


def test_unassigned_device_segment_is_allocated_when_all_items_already_matched():
    start_at = datetime(2026, 7, 25, 0, 0)
    switch_at = datetime(2026, 7, 25, 8, 0)
    end_at = datetime(2026, 7, 25, 10, 0)
    devices = [
        _device(start_at, device_no="23#", sort_order=0),
        _device(switch_at, device_no="99#", sort_order=1),
    ]
    item = _item(id=1, good_qty=600, device_no="23#", started_at=start_at)
    report = _report(start_at, end_at, item, devices=devices)

    metrics = calculate_report_metrics(start_at, end_at, [item], devices=devices)
    rows = build_report_allocation_drafts(report)

    assert metrics["effective_minutes"] == 600
    assert metrics["shift_day_minutes"] == 120
    assert metrics["shift_overtime_minutes"] == 120
    assert metrics["shift_night_minutes"] == 360
    assert round(sum(row.effective_minutes for row in rows), 2) == metrics["effective_minutes"]
    assert {
        "day": round(sum(row.day_minutes for row in rows), 2),
        "overtime": round(sum(row.overtime_minutes for row in rows), 2),
        "night": round(sum(row.night_minutes for row in rows), 2),
    } == {
        "day": metrics["shift_day_minutes"],
        "overtime": metrics["shift_overtime_minutes"],
        "night": metrics["shift_night_minutes"],
    }
    assert all(row.effective_minutes == round(row.day_minutes + row.overtime_minutes + row.night_minutes, 2) for row in rows)


def test_night_allocation_uses_meal_deducted_minutes_with_previous_date_rules():
    schedule = WorkScheduleConfig(
        day_start="08:00",
        day_end="17:20",
        lunch_start="11:40",
        lunch_end="12:40",
        dinner_start="21:00",
        dinner_end="21:30",
        overtime_start="18:00",
        overtime_end="20:00",
        night_start="20:00",
        night_end="06:00",
    )
    start_at = datetime(2026, 7, 25, 20, 0)
    end_at = datetime(2026, 7, 25, 22, 0)
    item = _item(id=1, good_qty=90)
    report = _report(start_at, end_at, item)

    metrics = calculate_report_metrics(start_at, end_at, [item], schedule=schedule)
    rows = build_report_allocation_drafts(report, schedule=schedule)

    assert metrics["effective_minutes"] == 90
    assert metrics["shift_night_minutes"] == 90
    assert [(row.allocation_date, row.night_minutes, row.effective_minutes) for row in rows] == [
        (date(2026, 7, 25), 90, 90),
    ]


def test_blank_overtime_meal_deduction_is_clipped_like_metrics():
    start_at = datetime(2026, 7, 25, 6, 0)
    end_at = datetime(2026, 7, 25, 19, 0)
    item = _item(id=1, good_qty=680)
    report = _report(start_at, end_at, item)

    metrics = calculate_report_metrics(start_at, end_at, [item])
    rows = build_report_allocation_drafts(report)

    assert metrics["effective_minutes"] == 680
    assert metrics["shift_day_minutes"] == 500
    assert metrics["shift_overtime_minutes"] == 180
    assert [(row.effective_minutes, row.day_minutes, row.overtime_minutes, row.night_minutes) for row in rows] == [
        (680, 500, 180, 0),
    ]


def test_same_device_item_split_inside_meal_is_rescaled_to_metrics_effective_minutes():
    start_at = datetime(2026, 7, 25, 8, 0)
    second_started_at = datetime(2026, 7, 25, 12, 0)
    end_at = datetime(2026, 7, 25, 13, 0)
    first_item = _item(id=1, good_qty=192, started_at=start_at)
    second_item = _item(id=2, good_qty=48, started_at=second_started_at)
    report = _report(start_at, end_at, [first_item, second_item], devices=[_device(start_at)])

    metrics = calculate_report_metrics(start_at, end_at, [first_item, second_item], devices=[_device(start_at)])
    rows = build_report_allocation_drafts(report)

    assert metrics["effective_minutes"] == 240
    assert metrics["shift_day_minutes"] == 240
    assert {first_item.id: first_item.allocated_minutes, second_item.id: second_item.allocated_minutes} == {
        1: 192,
        2: 48,
    }
    assert [(row.report_item_id, row.effective_minutes, row.day_minutes) for row in rows] == [
        (1, 192, 192),
        (2, 48, 48),
    ]
    assert sum(row.effective_minutes for row in rows) == 240


def test_matched_meal_split_and_unmatched_device_range_follow_metrics_order():
    start_at = datetime(2026, 7, 25, 8, 0)
    second_started_at = datetime(2026, 7, 25, 12, 0)
    switch_at = datetime(2026, 7, 25, 13, 0)
    end_at = datetime(2026, 7, 25, 20, 0)
    devices = [
        _device(start_at, device_no="23#", sort_order=0),
        _device(switch_at, device_no="77#", sort_order=1),
    ]
    first_item = _item(id=1, good_qty=240, device_no="23#", started_at=start_at)
    second_item = _item(id=2, good_qty=60, device_no="23#", started_at=second_started_at)
    unmatched_item = _item(id=3, good_qty=320, device_no="99#")
    items = [first_item, second_item, unmatched_item]
    report = _report(start_at, end_at, items, devices=devices)

    metrics = calculate_report_metrics(start_at, end_at, items, devices=devices)
    rows = build_report_allocation_drafts(report)

    assert metrics["effective_minutes"] == 620
    assert {item.id: item.allocated_minutes for item in items} == {
        1: 240,
        2: 60,
        3: 320,
    }
    assert {
        item_id: round(sum(row.effective_minutes for row in rows if row.report_item_id == item_id), 2)
        for item_id in [1, 2, 3]
    } == {
        1: 240,
        2: 60,
        3: 320,
    }
    assert round(sum(row.effective_minutes for row in rows), 2) == metrics["effective_minutes"]


def test_item_totals_anchor_to_metrics_when_unmatched_remaining_buckets_exceed_item_target():
    start_at = datetime(2026, 7, 25, 8, 0)
    second_started_at = datetime(2026, 7, 25, 12, 0)
    switch_at = datetime(2026, 7, 25, 18, 0)
    end_at = datetime(2026, 7, 25, 20, 0)
    devices = [
        _device(start_at, device_no="23#", sort_order=0),
        _device(switch_at, device_no="77#", sort_order=1),
    ]
    first_item = _item(id=1, good_qty=240, device_no="23#", started_at=start_at)
    second_item = _item(id=2, good_qty=360, device_no="23#", started_at=second_started_at)
    unmatched_item = _item(id=3, good_qty=60, device_no="99#")
    items = [first_item, second_item, unmatched_item]
    report = _report(start_at, end_at, items, devices=devices)

    metrics = calculate_report_metrics(start_at, end_at, items, devices=devices)
    rows = build_report_allocation_drafts(report)

    assert metrics["effective_minutes"] == 660
    assert {item.id: item.allocated_minutes for item in items} == {
        1: 240,
        2: 360,
        3: 60,
    }
    assert {
        item_id: round(sum(row.effective_minutes for row in rows if row.report_item_id == item_id), 2)
        for item_id in [1, 2, 3]
    } == {
        1: 240,
        2: 360,
        3: 60,
    }
    assert {
        "day": round(sum(row.day_minutes for row in rows), 2),
        "overtime": round(sum(row.overtime_minutes for row in rows), 2),
        "night": round(sum(row.night_minutes for row in rows), 2),
    } == {
        "day": metrics["shift_day_minutes"],
        "overtime": metrics["shift_overtime_minutes"],
        "night": metrics["shift_night_minutes"],
    }
    assert round(sum(row.effective_minutes for row in rows), 2) == 660
    assert all(row.effective_minutes == row.day_minutes + row.overtime_minutes + row.night_minutes for row in rows)


def test_zero_effective_minutes_falls_back_to_original_report_date():
    item = _item(good_qty=5, defect_qty=1, scrap_qty=0.5, changeover_count=2, process_unit_price_yuan=3)
    report = _report(
        datetime(2026, 7, 25, 9, 0),
        datetime(2026, 7, 25, 9, 0),
        item,
        report_date=date(2026, 7, 25),
    )

    rows = build_report_allocation_drafts(report)

    assert len(rows) == 1
    row = rows[0]
    assert row.allocation_date == date(2026, 7, 25)
    assert row.effective_minutes == 0
    assert row.day_minutes == 0
    assert row.overtime_minutes == 0
    assert row.night_minutes == 0
    assert row.good_qty == 5
    assert row.defect_qty == 1
    assert row.scrap_qty == 0.5
    assert row.changeover_count == 2
    assert row.reference_wage == 15


def test_allocation_summary_text_groups_dates_and_shift_kinds():
    item = _item(good_qty=650, process_unit_price_yuan=1)
    report = _report(datetime(2026, 7, 25, 0, 10), datetime(2026, 7, 25, 11, 0), item)

    rows = build_report_allocation_drafts(report)

    assert allocation_summary_text(rows) == "2026-07-24 夜班5.83小时；2026-07-25 白班3小时、加班2小时"


def test_refresh_report_allocations_replaces_existing_rows_and_flushes():
    engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
    Base.metadata.create_all(engine)
    SessionLocal = sessionmaker(bind=engine, future=True)
    db = SessionLocal()
    try:
        report = ProductionReport(
            id=1,
            session_id=1,
            attendance_point_name="总厂",
            employee_phone="13800000000",
            report_date=date(2026, 7, 25),
            start_at=datetime(2026, 7, 25, 5, 59),
            end_at=datetime(2026, 7, 25, 11, 0),
        )
        item = ProductionReportItem(
            id=1,
            report_id=1,
            attendance_point_name="总厂",
            device_no="23#",
            project_no="P1",
            product_name="23#",
            process_name="冲压",
            process_unit_price_yuan=2,
            good_qty=301,
            defect_qty=0,
            scrap_qty=0,
            changeover_count=0,
            standard_beat=1,
            standard_workload=0,
        )
        stale = ProductionReportAllocation(
            id=99,
            report_id=1,
            report_item_id=1,
            attendance_point_name="总厂",
            employee_phone="13800000000",
            allocation_date=date(2026, 7, 20),
            effective_minutes=1,
        )
        report.items.append(item)
        db.add_all([report, stale])
        db.flush()

        rows = refresh_report_allocations(db, report)

        assert len(rows) == 2
        assert all(row.id is not None for row in rows)
        stored = db.scalars(select(ProductionReportAllocation).order_by(ProductionReportAllocation.allocation_date)).all()
        assert [row.allocation_date for row in stored] == [date(2026, 7, 24), date(2026, 7, 25)]
        assert [row.good_qty for row in stored] == [1, 300]
    finally:
        db.close()