from io import BytesIO
import re

from flask import Flask
from openpyxl import load_workbook
import pandas as pd
from sqlalchemy import create_engine, event, text
from sqlalchemy.dialects import postgresql

import routes.sales_routes as sales_routes
from services.sales_analysis_export_service import (
    combine_sales_marketing_export_frames,
    query_sales_marketing_export_frames,
    query_sales_vendor_export_frame,
    sanitize_excel_dataframe,
)
from services.sales_analysis_query_service import (
    SALES_TIME_TEXT_PATTERN,
    build_sales_where_clause,
    normalize_sales_query_args,
    prepare_sales_query_context,
    resolve_sales_query_columns,
)


def _build_sales_engine():
    engine = create_engine("sqlite:///:memory:")
    with engine.begin() as conn:
        conn.exec_driver_sql(
            '''
            CREATE TABLE realtime_sales_monthly (
                "日期" TEXT,
                "時間" TEXT,
                "商品ID" TEXT,
                "商品名稱" TEXT,
                "商品館" TEXT,
                "品牌" TEXT,
                "廠商名稱" TEXT,
                "總業績" REAL,
                "數量" REAL,
                "總成本" REAL,
                "折扣活動名稱" TEXT,
                "折價券活動名稱" TEXT
            )
            '''
        )
        conn.exec_driver_sql(
            '''
            INSERT INTO realtime_sales_monthly
                ("日期", "時間", "商品ID", "商品名稱", "商品館", "品牌",
                 "廠商名稱", "總業績", "數量", "總成本", "折扣活動名稱", "折價券活動名稱")
            VALUES
                ('2025/04/07', '10:00:00', 'P1', '商品 A', '美妝', '品牌 A', '廠商 A', 100, 2, 60, '母親節', ''),
                ('2025/04/07', '20:00:00', 'P2', '商品 B', '美妝', '品牌 B', '廠商 B', 300, 3, 200, '', '折價券'),
                ('2025/04/08', '10:00:00', 'P3', '商品 C', '美妝', '品牌 C', '廠商 C', 600, 4, 300, '會員日', ''),
                ('2026/04/06', '10:00:00', 'P1', '商品 A', '美妝', '品牌 A', '廠商 A', 150, 3, 80, '母親節', ''),
                ('2026/04/06', '20:00:00', 'P2', '商品 B', '美妝', '品牌 B', '廠商 B', 400, 4, 250, '', '折價券'),
                ('2026/04/07', '10:00:00', 'P3', '商品 C', '美妝', '品牌 C', '廠商 C', 800, 5, 400, '會員日', ''),
                ('2026/04/06', '10:00:00', 'P4', '商品 D', '3C', '品牌 D', '廠商 D', 9999, 1, 1, '清倉', '')
            '''
        )
    return engine


def test_query_filter_values_are_bound_instead_of_interpolated():
    engine = _build_sales_engine()
    attack = "美妝' OR 1=1 --"
    filters = normalize_sales_query_args({
        "data_range": "0",
        "category": attack,
        "start_date": "2026-04-30",
        "end_date": "2026-04-01",
    })
    columns = resolve_sales_query_columns(engine, "realtime_sales_monthly")

    clause, params = build_sales_where_clause(engine, columns, filters)

    assert attack not in clause
    assert params["filter_category"] == attack
    assert params["filter_start_date"] == "2026-04-01"
    assert params["filter_end_date"] == "2026-04-30"


def test_keyword_search_is_case_insensitive_and_treats_wildcards_as_text():
    engine = _build_sales_engine()
    with engine.begin() as connection:
        connection.exec_driver_sql(
            "UPDATE realtime_sales_monthly SET \"商品名稱\" = 'Alpha%_Beta' "
            "WHERE \"商品ID\" = 'P1'"
        )
    filters = normalize_sales_query_args({
        "data_range": "0",
        "keyword": "alpha%_beta",
    })
    columns = resolve_sales_query_columns(engine, "realtime_sales_monthly")
    where_sql, params = build_sales_where_clause(engine, columns, filters)

    with engine.connect() as connection:
        product_ids = connection.execute(
            text(f'SELECT "商品ID" FROM realtime_sales_monthly WHERE {where_sql}'),
            params,
        ).scalars().all()

    assert product_ids == ["P1", "P1"]
    assert params["filter_keyword"] == "%alpha\\%\\_beta%"


def test_postgres_filter_contract_uses_bound_temporal_dimensions():
    class PostgresEngine:
        dialect = postgresql.dialect()

    filters = normalize_sales_query_args({
        "data_range": "0",
        "start_date": "2026-04-01",
        "end_date": "2026-04-30",
        "dow": "0",
        "hour": "10",
        "keyword": "Alpha",
    })
    columns = {
        "date": "日期",
        "time": "時間",
        "name": "商品名稱",
        "pid": "商品ID",
        "brand": "品牌",
        "vendor": "廠商名稱",
    }

    where_sql, params = build_sales_where_clause(PostgresEngine(), columns, filters)

    assert "EXTRACT(DOW FROM TO_DATE" in where_sql
    assert "SUBSTRING(TRIM(CAST" in where_sql
    assert "(\\.[0-9]+)?)?$" in where_sql
    assert "LOWER(CAST" in where_sql
    assert params["filter_db_dow"] == 1
    assert params["filter_hour"] == 10
    assert re.fullmatch(SALES_TIME_TEXT_PATTERN, "10:30:59")
    assert not re.fullmatch(SALES_TIME_TEXT_PATTERN, "10:30junk")


def test_invalid_time_text_never_matches_midnight_filter():
    engine = _build_sales_engine()
    with engine.begin() as connection:
        connection.exec_driver_sql(
            "UPDATE realtime_sales_monthly SET \"時間\" = '' WHERE \"商品ID\" = 'P1'"
        )
    filters = normalize_sales_query_args({"data_range": "0", "hour": "0"})
    columns = resolve_sales_query_columns(engine, "realtime_sales_monthly")
    where_sql, params = build_sales_where_clause(engine, columns, filters)

    with engine.connect() as connection:
        count = connection.execute(
            text(f"SELECT COUNT(*) FROM realtime_sales_monthly WHERE {where_sql}"),
            params,
        ).scalar_one()

    assert count == 0


def test_top_detail_rejects_injection_as_data_and_keeps_the_table_intact():
    engine = _build_sales_engine()
    frame, _meta = sales_routes._query_sales_top_detail_frame(engine, {
        "data_range": "0",
        "start_date": "2026-04-01",
        "end_date": "2026-04-30",
        "category": "美妝' OR 1=1 --",
        "metric": "amount",
        "view": "product",
        "type": "revenue",
    })

    assert frame.empty
    with engine.connect() as conn:
        count = conn.exec_driver_sql("SELECT COUNT(*) FROM realtime_sales_monthly").scalar_one()
    assert count == 7


def test_top_detail_rejects_a_mismatched_business_type_and_metric():
    engine = _build_sales_engine()

    try:
        sales_routes._query_sales_top_detail_frame(engine, {
            "data_range": "0",
            "type": "margin",
            "metric": "amount",
            "view": "product",
        })
    except ValueError as exc:
        assert "type 與 metric" in str(exc)
    else:
        raise AssertionError("mismatched type/metric must be rejected")


def test_yoy_applies_month_weekday_hour_and_category_together(monkeypatch):
    engine = _build_sales_engine()

    class FakeDatabaseManager:
        def __init__(self):
            self.engine = engine

    monkeypatch.setattr(sales_routes, "DatabaseManager", FakeDatabaseManager)
    app = Flask(__name__)
    with app.test_request_context(
        "/api/sales_analysis/yoy_comparison"
        "?year1=2025&year2=2026&metric=revenue&data_range=0"
        "&month=2026-04&dow=0&hour=10&category=美妝"
    ):
        response = sales_routes.api_yoy_comparison.__wrapped__()

    payload = response.get_json()
    assert payload["year1"]["total"] == 100
    assert payload["year2"]["total"] == 150
    assert payload["growth_rate"] == 50


def test_table_api_normalizes_reverse_dates_and_preserves_metric_order(monkeypatch):
    engine = _build_sales_engine()

    class FakeDatabaseManager:
        def __init__(self):
            self.engine = engine

    monkeypatch.setattr(sales_routes, "DatabaseManager", FakeDatabaseManager)
    sales_routes._TABLE_DATA_CACHE.clear()
    app = Flask(__name__)
    with app.test_request_context(
        "/api/sales_analysis/table_data?metric=qty&data_range=0"
        "&start_date=2026-04-30&end_date=2026-04-01&category=美妝"
    ):
        response = sales_routes.api_sales_table_data.__wrapped__()

    payload = response.get_json()
    assert [item["product_id"] for item in payload["data"]] == ["P3", "P2", "P1"]
    assert [item["qty"] for item in payload["data"]] == [5, 4, 3]


def test_sales_page_redirects_reverse_dates_to_one_canonical_url(monkeypatch):
    engine = _build_sales_engine()

    class FakeDatabaseManager:
        def __init__(self):
            self.engine = engine

    monkeypatch.setattr(sales_routes, "DatabaseManager", FakeDatabaseManager)
    sales_routes._SALES_OPTIONS_CACHE.clear()
    app = Flask(__name__)
    app.register_blueprint(sales_routes.sales_bp)
    with app.test_request_context(
        "/sales_analysis?data_range=0&start_date=2026-04-30&end_date=2026-04-01"
    ):
        response = sales_routes.sales_analysis.__wrapped__()

    assert response.status_code == 302
    assert "start_date=2026-04-01" in response.location
    assert "end_date=2026-04-30" in response.location


def test_sales_page_closes_one_sided_dates_with_live_source_bounds(monkeypatch):
    engine = _build_sales_engine()

    class FakeDatabaseManager:
        def __init__(self):
            self.engine = engine

    monkeypatch.setattr(sales_routes, "DatabaseManager", FakeDatabaseManager)
    sales_routes._SALES_OPTIONS_CACHE.clear()
    app = Flask(__name__)
    app.register_blueprint(sales_routes.sales_bp)

    with app.test_request_context(
        "/sales_analysis?data_range=0&end_date=2026-04-10"
    ):
        response = sales_routes.sales_analysis.__wrapped__()

    assert response.status_code == 302
    assert "start_date=2025-04-07" in response.location
    assert "end_date=2026-04-10" in response.location


def test_other_category_uses_the_same_filtered_top_twelve_contract():
    engine = create_engine("sqlite:///:memory:")
    with engine.begin() as connection:
        connection.exec_driver_sql(
            'CREATE TABLE realtime_sales_monthly ('
            '"日期" TEXT, "商品名稱" TEXT, "商品館" TEXT, "總業績" REAL)'
        )
        connection.execute(
            text(
                'INSERT INTO realtime_sales_monthly '
                '("日期", "商品名稱", "商品館", "總業績") '
                'VALUES (:sale_date, :name, :category, :amount)'
            ),
            [
                {
                    "sale_date": "2026/04/01",
                    "name": f"商品 {index}",
                    "category": f"分類 {index:02d}",
                    "amount": 100 - index if index <= 10 else 1,
                }
                for index in range(14)
            ] + [{
                "sale_date": "2026/04/01",
                "name": "未分類商品",
                "category": None,
                "amount": 1,
            }],
        )

    filters, columns = prepare_sales_query_context(
        engine,
        "realtime_sales_monthly",
        {"data_range": "0", "category": "其他"},
    )
    where_sql, params = build_sales_where_clause(engine, columns, filters)
    with engine.connect() as connection:
        categories = connection.execute(
            text(
                'SELECT "商品館" FROM realtime_sales_monthly '
                f'WHERE {where_sql} ORDER BY "商品館"'
            ),
            params,
        ).scalars().all()

    assert categories == ["分類 12", "分類 13"]

    sales_routes._SALES_PROCESSED_CACHE["other-contract"] = {
        "df": pd.read_sql(text("SELECT * FROM realtime_sales_monthly"), engine),
        "cols": {
            "name": "商品名稱",
            "category": "商品館",
            "amount": "總業績",
        },
    }
    app = Flask(__name__)
    with app.test_request_context("/sales_analysis?category=其他"):
        page_frame, _columns, error = sales_routes._get_filtered_sales_data(
            "other-contract"
        )
    sales_routes._SALES_PROCESSED_CACHE.pop("other-contract", None)

    assert error is None
    assert page_frame["商品館"].tolist() == ["分類 12", "分類 13"]


def test_vendor_and_marketing_exports_apply_the_page_filters():
    engine = _build_sales_engine()
    statements = []

    def capture_statement(_conn, _cursor, statement, _params, _context, _many):
        statements.append(statement)

    event.listen(engine, "before_cursor_execute", capture_statement)
    args = {
        "metric": "amount",
        "data_range": "0",
        "start_date": "2026-04-01",
        "end_date": "2026-04-30",
        "category": "美妝",
        "hour": "10",
    }

    vendor_frame = query_sales_vendor_export_frame(engine, args)
    marketing_frames = query_sales_marketing_export_frames(engine, args)

    assert vendor_frame[["廠商", "銷售金額"]].to_dict("records") == [
        {"廠商": "廠商 C", "銷售金額": 800.0},
        {"廠商": "廠商 A", "銷售金額": 150.0},
    ]
    assert marketing_frames["折扣活動"][["活動名稱", "銷售金額"]].to_dict("records") == [
        {"活動名稱": "會員日", "銷售金額": 800.0},
        {"活動名稱": "母親節", "銷售金額": 150.0},
    ]
    assert "折價券活動" not in marketing_frames
    assert sum("UNION ALL" in statement for statement in statements) == 1


def test_excel_export_sanitizer_blocks_formula_cells():
    frame = sanitize_excel_dataframe(pd.DataFrame({"活動名稱": ["=2+2", "一般活動"]}))

    assert frame["活動名稱"].tolist() == ["'=2+2", "一般活動"]


def test_marketing_combined_sheet_follows_the_active_metric():
    frames = {
        "折扣活動": pd.DataFrame([
            {"活動名稱": "高營收", "銷售金額": 1000, "銷售數量": 1, "毛利金額": 20},
        ]),
        "折價券活動": pd.DataFrame([
            {"活動名稱": "高銷量", "銷售金額": 100, "銷售數量": 10, "毛利金額": 50},
        ]),
    }

    combined = combine_sales_marketing_export_frames(frames, "qty")

    assert combined["活動名稱"].tolist() == ["高銷量", "高營收"]


def test_sales_export_routes_return_valid_period_linked_workbooks(monkeypatch):
    engine = _build_sales_engine()

    class FakeDatabaseManager:
        def __init__(self):
            self.engine = engine

    monkeypatch.setattr(sales_routes, "DatabaseManager", FakeDatabaseManager)
    app = Flask(__name__)
    query = (
        "?metric=amount&data_range=0&start_date=2026-04-01&end_date=2026-04-30"
        "&category=美妝&hour=10"
    )
    routes = (
        (sales_routes.api_export_sales_vendor, "/api/sales_analysis/export_vendor", "廠商分析"),
        (sales_routes.api_export_sales_marketing, "/api/sales_analysis/export_marketing", "折扣活動"),
    )

    for route, path, expected_sheet in routes:
        with app.test_request_context(f"{path}{query}"):
            response = route.__wrapped__()
        response.direct_passthrough = False
        workbook = load_workbook(BytesIO(response.get_data()), read_only=True)

        assert response.status_code == 200
        assert expected_sheet in workbook.sheetnames