import marimo

__generated_with = "0.13.15"
app = marimo.App()


@app.cell
def init():

    import sys
    sys.path.append('/opt/spark/work-dir/')
    from workflow_templates.spark.udf_manager import bootstrap_udfs 
    from pyspark.sql.functions import udf 
    from pyspark.sql.functions import lit
    from pyspark.sql.types import StringType, IntegerType
    import uuid
    from pathlib import Path
    from pyspark import SparkConf, Row
    from pyspark.sql import SparkSession
    import os
    import pandas as pd
    import polars as pl
    import pyarrow as pa
    from pyspark.sql.functions import expr,to_json,col,struct
    from functools import reduce
    from handle_structs_or_arrays import preprocess_then_expand
    import requests
    from jinja2 import Template
    import json


    from secrets_manager import SecretsManager

    from WorkflowManager import WorkflowDSL, WorkflowManager
    from KnowledgebaseManager import KnowledgebaseManager
    from gitea_client import GiteaClient, WorkspaceVersionedContent

    from dremio.flight.endpoint import DremioFlightEndpoint
    from dremio.flight.query import DremioFlightEndpointQuery


    alias_str='abcdefghijklmnopqrstuvwxyz'
    workspace = os.getenv('WORKSPACE') or 'exp360cust'

    job_id = os.getenv("EXECUTION_ID") or str(uuid.uuid4())

    sm = SecretsManager(os.getenv('SECRET_MANAGER_URL'), os.getenv('SECRET_MANAGER_NAMESPACE'), os.getenv('SECRET_MANAGER_ENV'), os.getenv('SECRET_MANAGER_TOKEN'))
    secrets = sm.list_secrets(workspace)

    gitea_client=GiteaClient(os.getenv('GITEA_HOST'), os.getenv('GITEA_TOKEN'), os.getenv('GITEA_OWNER') or 'gitea_admin', os.getenv('GITEA_REPO') or 'tenant1')
    workspaceVersionedContent=WorkspaceVersionedContent(gitea_client)
    conf = SparkConf()
    params = {
     "spark.hadoop.fs.s3a.access.key": secrets.get('S3_ACCESS_KEY'),
     "spark.hadoop.fs.s3a.secret.key": secrets.get('S3_SECRET_KEY'),
     "spark.hadoop.fs.s3a.aws.region": "us-west-1",
     "spark.sql.catalog.dremio.warehouse" : 's3://'+ (secrets.get('LAKEHOUSE_BUCKET') or ''),
     "spark.sql.catalog.dremio" : "org.apache.iceberg.spark.SparkCatalog",
     "spark.sql.catalog.dremio.type" : "hadoop",
     "spark.hadoop.fs.s3a.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
     "spark.sql.extensions": "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
     "spark.hadoop.fs.s3.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
     "spark.jars.packages": "com.amazonaws:aws-java-sdk-bundle:1.12.262,com.github.ben-manes.caffeine:caffeine:3.2.0,org.apache.iceberg:iceberg-aws-bundle:1.8.1,org.apache.iceberg:iceberg-common:1.8.1,org.apache.iceberg:iceberg-core:1.8.1,org.apache.iceberg:iceberg-spark:1.8.1,org.apache.hadoop:hadoop-aws:3.3.4,com.amazonaws:aws-java-sdk-bundle:1.11.901,org.apache.hadoop:hadoop-common:3.3.4,org.apache.hadoop:hadoop-cloud-storage:3.3.4,org.apache.hadoop:hadoop-client-runtime:3.3.4,org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.1,org.projectnessie.nessie-integrations:nessie-spark-extensions-3.5_2.12:0.103.2,org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.2"
    }



    conf.setAll(list(params.items()))

    spark = SparkSession.builder.appName(workspace).config(conf=conf).getOrCreate()
    bootstrap_udfs(spark)
    return (spark,)


@app.cell
def data_reader__1(spark):



    data_reader__1_df = spark.read.table('dremio.payments')
    data_reader__1_df.createOrReplaceTempView('data_reader__1_df')
    return


if __name__ == "__main__":
    app.run()