"""

[MGI Markers report](http://www.informatics.jax.org/downloads/reports/MRK_List1.rpt) extraction Task.
This report contains the list of valid MGI Markers and Identifiers.

### From [MGI Reports / Mouse Genetic Markers section](http://www.informatics.jax.org/downloads/reports/index.html#marker):

#### 1. List of Mouse Genetic Markers (sorted alphabetically by marker symbol, tab-delimited)
        - MRK_List1.rpt (including withdrawn marker symbols)
(File contains header, so no example is provided. See full report [here](http://www.informatics.jax.org/downloads/reports/MRK_List1.rpt))
"""
import luigi
from luigi.contrib.spark import PySparkTask
from pyspark import SparkContext
from pyspark.sql import SparkSession

from impc_etl.shared.utils import extract_tsv
from impc_etl.workflow.config import ImpcConfig


class MGIMarkerListReportExtractor(PySparkTask):
    """
    PySpark Task class to extract the information from the
    [MGI Markers report](http://www.informatics.jax.org/downloads/reports/MRK_List1.rpt).
    """

    #: Name of the Spark task
    name: str = "IMPC_MGI_Marker_List_Report_Extractor"

    #: Path of the MGI Marker List report *.rpt file.
    mgi_markers_report_path: luigi.Parameter = luigi.Parameter()

    #: Path of the output directory where ethe new parquet file will be generated.
    output_path: luigi.Parameter = luigi.Parameter()

    def output(self):
        """
        Returns the full parquet path as an output for the Luigi Task
        (e.g. impc/dr15.2/parquet/mgi_markers_parquet)
        """
        return ImpcConfig().get_target(f"{self.output_path}mgi_markers_parquet")

    def app_options(self):
        """
        Generates the options pass to the PySpark job
        """
        return [
            self.mgi_markers_report_path,
            self.output().path,
        ]

    def main(self, sc: SparkContext, *args):
        """
        Takes in a SparkContext and the list of arguments generated by `app_options` and executes the PySpark job.
        """
        mgi_markers_report_path = args[0]
        output_path = args[1]

        spark = SparkSession(sc)
        # TODO fix empty column on rpt
        mgi_mrk_list_df = extract_tsv(spark, mgi_markers_report_path, header=True)
        for col_name in mgi_mrk_list_df.columns:
            mgi_mrk_list_df = mgi_mrk_list_df.withColumnRenamed(
                col_name,
                col_name.lower().replace(" ", "_").replace("_(pipe-separated)", ""),
            )
        mgi_mrk_list_df.write.mode("overwrite").parquet(output_path)
