from pathlib import Path

import gfhub
import pandas as pd
from gfhub import nodes

client = gfhub.Client()

# Function 1: CSV to Parquet
def csv2parquet(input_file: Path, /) -> dict[str, Path]:
    """Convert a CSV file to Parquet format.

    :param input_file: Path to the input CSV file.
    :return: Dict with 'parquet' key containing the output Parquet file path.
    """
    # Read the CSV file
    df = pd.read_csv(input_file)

    # Define the output path with .parquet extension
    parquet_path = input_file.with_suffix(".parquet")

    # Write the DataFrame to a Parquet file
    df.to_parquet(parquet_path, index=False)

    return {"parquet": parquet_path}


func1 = client.add_function(
    gfhub.Function(
        csv2parquet,
        dependencies={"pandas[pyarrow]": "import pandas as pd"},
    )
)
print("Function 1 uploaded:")
print(func1)

# Function 2: Parquet to JSON

def parquet2json(path: Path, /, *, orient: str = "records", indent: int = 2) -> Path:
    """Convert Parquet to JSON."""
    df = pd.read_parquet(path)
    output = path.with_suffix(".json")
    df.to_json(output, orient=orient, indent=indent)
    return output

func2 = client.add_function(
    gfhub.Function(
        parquet2json,
        dependencies={"pandas[pyarrow]": "import pandas as pd"},
    )
)
print("Function 2 uploaded:")
print(func2)

p = gfhub.Pipeline()
p.trigger = nodes.on_file_upload(tags=[".csv"])
p.on_manual_trigger = nodes.on_manual_trigger()
p.load = nodes.load()
p.to_parquet = nodes.function(function="csv2parquet")
p.to_json = nodes.function(function="parquet2json")
p.save = nodes.save()
p += p.trigger >> p.load
p += p.on_manual_trigger >> p.load
p += p.load >> p.to_parquet
p += p.to_parquet >> p.to_json
p += p.to_json >> p.save

# Convert to JSON string and pass to add_pipeline
# update defaults to True
pipeline = client.add_pipeline("csv2json", p)
pipeline["updated_at"]