import time
from prefect import flow, task, get_run_logger, unmapped
import datetime
import pandas as pd
import random
from collections import namedtuple


# flow
@flow(
    name="Minimal bug reproduce",
    description="Showing a bug.",
    version="0.1",
    log_prints=True,
)
def extract_geneva_data_pre_backtest_run(
    # safe_time_start: str = "",  # time is not json serializable.
    # safe_time_end: str = "",
    # type_of_pull: str = "",
    # accounts_date_df: Any = None,  # pd.DataFrame = None,
    # geneva_args: dict = {},
    knowledge_date: str = datetime.datetime.now().strftime("%Y/%m/%d:%H:%M:%S"),
):
    logger = get_run_logger()

    # Define the data as a list of dictionaries or a list of lists
    data = [
        ["2025-11-17", 20251117, "ASG Internal", 6005],
        ["2025-11-17", 20251117, "MFLASER Master", 7000],
        ["2025-11-17", 20251117, "MFMF Master", 7060],
        ["2025-11-17", 20251117, "USNGAM", 7085],
        ["2025-11-17", 20251117, "EUNGAM", 7090],
        ["2025-11-17", 20251117, "REA Alternatives", 7120],
        ["2025-11-17", 20251117, "AMMF Master", 7125],
        ["2025-11-17", 20251117, "UCMF", 7130],
        ["2025-11-17", 20251117, "CMAP", 7135],
        ["2025-11-17", 20251117, "LGTMF", 7145],
        ["2025-11-17", 20251117, "ATFLP", 7150],
        ["2025-11-17", 20251117, "CROMF", 7160],
        ["2025-11-17", 20251117, "CTNMF Master", 7165],
        ["2025-11-17", 20251117, "ATFCAY Offshore LTD", 7170],
        ["2025-11-17", 20251117, "ASMF", 7180],
        ["2025-11-17", 20251117, "ECMF", 7185],
        ["2025-11-17", 20251117, "ETFMF Master", 7190],
        ["2025-11-17", 20251117, "DELTA1SGUCMF", 7195],
        ["2025-11-17", 20251117, "EMAMF Master", 7200],
        ["2025-11-17", 20251117, "ETFGM Master", 7230],
        ["2025-11-16", 20251116, "REA Alternatives", 7120],
        ["2025-11-17", 20251117, "REA Alternatives", 7120],
        ["2025-11-15", 20251117, "ASG Internal", 6005],
        ["2025-11-15", 20251117, "MFLASER Master", 7000],
        ["2025-11-15", 20251117, "MFMF Master", 7060],
        ["2025-11-15", 20251117, "USNGAM", 7085],
        ["2025-11-15", 20251117, "EUNGAM", 7090],
        ["2025-11-15", 20251117, "REA Alternatives", 7120],
        ["2025-11-15", 20251117, "AMMF Master", 7125],
        ["2025-11-15", 20251117, "UCMF", 7130],
        ["2025-11-15", 20251117, "CMAP", 7135],
        ["2025-11-15", 20251117, "LGTMF", 7145],
        ["2025-11-15", 20251117, "ATFLP", 7150],
        ["2025-11-15", 20251117, "CROMF", 7160],
        ["2025-11-15", 20251117, "CTNMF Master", 7165],
        ["2025-11-15", 20251117, "ATFCAY Offshore LTD", 7170],
        ["2025-11-15", 20251117, "ASMF", 7180],
        ["2025-11-15", 20251117, "ECMF", 7185],
        ["2025-11-15", 20251117, "ETFMF Master", 7190],
        ["2025-11-15", 20251117, "DELTA1SGUCMF", 7195],
        ["2025-11-15", 20251117, "EMAMF Master", 7200],
        ["2025-11-15", 20251117, "ETFGM Master", 7230],
        ["2025-11-15", 20251116, "REA Alternatives", 7120],
        ["2025-11-15", 20251117, "REA Alternatives", 7120],
        ["2025-11-14", 20251117, "ASG Internal", 6005],
        ["2025-11-14", 20251117, "MFLASER Master", 7000],
        ["2025-11-14", 20251117, "MFMF Master", 7060],
        ["2025-11-14", 20251117, "USNGAM", 7085],
        ["2025-11-14", 20251117, "EUNGAM", 7090],
        ["2025-11-14", 20251117, "REA Alternatives", 7120],
        ["2025-11-14", 20251117, "AMMF Master", 7125],
        ["2025-11-14", 20251117, "UCMF", 7130],
        ["2025-11-14", 20251117, "CMAP", 7135],
        ["2025-11-14", 20251117, "LGTMF", 7145],
        ["2025-11-14", 20251117, "ATFLP", 7150],
        ["2025-11-14", 20251117, "CROMF", 7160],
        ["2025-11-14", 20251117, "CTNMF Master", 7165],
        ["2025-11-14", 20251117, "ATFCAY Offshore LTD", 7170],
        ["2025-11-14", 20251117, "ASMF", 7180],
        ["2025-11-14", 20251117, "ECMF", 7185],
        ["2025-11-14", 20251117, "ETFMF Master", 7190],
        ["2025-11-14", 20251117, "DELTA1SGUCMF", 7195],
        ["2025-11-14", 20251117, "EMAMF Master", 7200],
        ["2025-11-14", 20251117, "ETFGM Master", 7230],
        ["2025-11-14", 20251116, "REA Alternatives", 7120],
        ["2025-11-14", 20251117, "REA Alternatives", 7120],
    ]

    list_that_changes = {
        "unchanged_value": "SameSame"
    }  # <- Here's the list that changes

    # Define column names
    columns = ["Date", "DateKey", "GenevaAccountName", "FundId"]

    # Create the DataFrame
    accounts_date_df = pd.DataFrame(data, columns=columns)

    # Convert Date column to datetime
    accounts_date_df["Date"] = pd.to_datetime(accounts_date_df["Date"])

    print("Here are the accounts and dates to be loaded")
    print(accounts_date_df.to_string())
    knowledge_date = datetime.datetime.now().strftime("%Y/%m/%d:%H:%M:%S")
    logger.info(f"Using knowledge date of: {knowledge_date}")
    # for update_row in accounts_date_df.itertuples():
    geneva_data_dfs = extract_geneva_data_with_update_data_and_file.map(
        update_row=accounts_date_df.itertuples(),
        # geneva_tax_lot_args=unmapped(geneva_args),
        type_of_pull="test",
        list_that_changes_over_all_tasks=unmapped(list_that_changes),
        # knowledge_date=knowledge_date,
    )
    geneva_data_dfs.wait()


#  geneva task
@task(
    retries=0,
    retry_delay_seconds=10,  # TODO: Change later.
    timeout_seconds=1800,  # 60 * 30 = 1800 seconds.
    tags=["hits-geneva"],
    task_run_name="extract_{type_of_pull}_data-fund:{update_row.FundId}_date:{update_row.Date}",
)
def extract_geneva_data_with_update_data_and_file(
    update_row: namedtuple,
    # geneva_tax_lot_args,
    type_of_pull: str,
    list_that_changes_over_all_tasks,
    # knowledge_date: str = datetime.datetime.now().strftime("%Y/%m/%d:%H:%M:%S"),
) -> None:
    logger = get_run_logger()
    fund_id, current_date, account = process_update_row(update_row, logger)
    # Putting these steps in a function causes strange behavior
    logger.info(f"Update Row is : {update_row}")
    list_that_changes_over_all_tasks["changing_value"] = (
        update_row.GenevaAccountName  # <- Here's where the value gets changed.
    )
    try:
        # Putting these here will cause it to work fine.
        print(f"Here is the current row from data frame: {update_row}")
        fund_id_expected = update_row.FundId
        # fund_id = int(update_row[1])
        print(f"Here is the current fundid from data frame: {fund_id}")
        current_date_expected = update_row.Date.date()
        # date = update_row[2]
        print(f"Here is the current date from data frame: {current_date}")
        # Given a fundId (e.g., 7000) and a date, return a dataframe with
        # position/performance detail drawn straight from bisTaxLot within Geneva
        # account = fundId2GenevaAccountName(fundId)
        account_expected = update_row.GenevaAccountName
        print(f"Here is the current account from data frame: {account}")
        update_row = update_row._asdict()
        # End of set up.
        run_geneva_gsql_file_pull(
            fund_id=fund_id,
            account=account,
            current_date=current_date,
            fund_id_expected=fund_id_expected,
            account_expected=account_expected,
            current_date_expected=current_date_expected,
            update_row=update_row,
            list_that_changes_over_all_tasks=list_that_changes_over_all_tasks,
        )
    except Exception as e:
        logger.error(f"Couldn't process update row {update_row}. Exception: {e}")
    # End of putting these steps in a function causes strange behavior

    return


# This needs columns [ "FundId", "Date", "GenevaAccountName"] to exist in the data frame.
def process_update_row(update_row, logger):
    logger.info(f"Update Row is : {update_row}")
    try:
        print(f"Here is the current row: {update_row}")
        fund_id = update_row.FundId
        # fund_id = int(update_row[1])
        print(f"Here is the current fundid: {fund_id}")
        current_date = update_row.Date.date()
        # date = update_row[2]
        print(f"Here is the current date: {current_date}")
        # Given a fundId (e.g., 7000) and a date, return a dataframe with
        # position/performance detail drawn straight from bisTaxLot within Geneva
        # account = fundId2GenevaAccountName(fundId)
        account = update_row.GenevaAccountName
        print(f"Here is the current account: {account}")
    except Exception as e:
        logger.error(f"Couldn't process update row {update_row}. Exception: {e}")
    return fund_id, current_date, account


def run_geneva_gsql_file_pull(
    fund_id,
    account,
    current_date,
    fund_id_expected,
    account_expected,
    current_date_expected,
    update_row,
    list_that_changes_over_all_tasks,
):
    logger = get_run_logger()
    seconds = random.uniform(1.0, 5.0)
    print(f"Sleeping for {seconds:.2f} seconds...")
    time.sleep(seconds)
    logger.info("Values in the next call.")
    logger.info(f"{fund_id=}")
    logger.info(f"{account=}")
    logger.info(f"{current_date=}")
    if fund_id != fund_id_expected:
        logger.error(f"{fund_id} does not equal {fund_id_expected}!")
        raise ValueError(f"{fund_id} does not equal {fund_id_expected}!")
    if account != account_expected:
        logger.error(f"{account} does not equal {account_expected}!")
        raise ValueError(f"{account} does not equal {account_expected}!")
    if current_date != current_date_expected:
        logger.error(f"{current_date} does not equal {current_date_expected}!")
        raise ValueError(f"{current_date} does not equal {current_date_expected}!")
    update_row_dict_copy = update_row
    update_row_dict_copy["fund_id"] = fund_id
    update_row_dict_copy["accountid"] = account
    update_row_dict_copy["current_date"] = current_date
    seconds = random.uniform(1.0, 5.0)
    logger.info(f"Sleeping for {seconds:.2f} seconds...")
    if fund_id != fund_id_expected:
        logger.error(f"{fund_id} does not equal {fund_id_expected}!")
        raise ValueError(f"{fund_id} does not equal {fund_id_expected}!")
    if account != account_expected:
        logger.error(f"{account} does not equal {account_expected}!")
        raise ValueError(f"{account} does not equal {account_expected}!")
    if current_date != current_date_expected:
        logger.error(f"{current_date} does not equal {current_date_expected}!")
        raise ValueError(f"{current_date} does not equal {current_date_expected}!")
    logger.info("Checking for the value that is likely to change.")
    if (
        account_expected != list_that_changes_over_all_tasks["changing_value"]
    ):  # <- Here's where the issue should be found.
        list_account = list_that_changes_over_all_tasks["changing_value"]
        logger.error(f"{list_account} does not equal {account_expected}!")
        raise ValueError(f"{list_account} does not equal {account_expected}!")


if __name__ == "__main__":
    extract_geneva_data_pre_backtest_run()