"""

[MGI Alliance Homology, HGNC and CCDS IDs report](http://www.informatics.jax.org/downloads/reports/HGNC_AllianceHomology.rpt) extraction Task.
This report contains the list of Gene Phenotype Associations reported in MGI.

### From [MGI Reports / Vertebrate Homology section](http://www.informatics.jax.org/downloads/reports/index.html#homology):

#### 3. Mouse Genes with Alliance Homology and HGNC and CCDS IDs.

(File contains header, so no example is provided. See full report [here](http://www.informatics.jax.org/downloads/reports/HGNC_AllianceHomology.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 MGIHomologyReportExtractor(PySparkTask):
    """
    PySpark Task class to extract the information from the
    [Alliance Homology, HGNC and CCDS IDs report](http://www.informatics.jax.org/downloads/reports/HGNC_AllianceHomology.rpt).
    """

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

    #: Path of the MGI Homology report *.rpt file.
    mgi_homology_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_homology_parquet)
        """
        return ImpcConfig().get_target(f"{self.output_path}mgi_homology_parquet")

    def app_options(self):
        """
        Generates the options pass to the PySpark job
        """
        return [
            self.mgi_homology_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_homology_report_path = args[0]
        output_path = args[1]

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