<commit_before>import os
import urllib

from abc import ABCMeta
from luigi import Task, Parameter, WrapperTask, LocalTarget
from collections import OrderedDict
from tasks.util import DownloadUnzipTask, shell, TableTask, TempTableTask, classpath, MetaWrapper, LOGGER
from tasks.meta import current_session
from tasks.ca.statcan.geo import (
    GEO_CT, GEO_PR, GEO_CD, GEO_CSD, GEO_CMA,
    GEOGRAPHY_CODES, GEOGRAPHIES, GeographyColumns, Geography)
from tasks.ca.statcan.util import StatCanParser
from tasks.ca.statcan.cols_census import CensusColumns
from tasks.ca.statcan.cols_nhs import NHSColumns


SURVEY_CEN = 'census'
SURVEY_NHS = 'nhs'

SURVEYS = (
    SURVEY_CEN,
    SURVEY_NHS,
)

SURVEY_CODES = {
    SURVEY_CEN: '98-316-XWE2011001',
    SURVEY_NHS: '99-004-XWE2011001',
}

SURVEY_URLS = {
    SURVEY_CEN: 'census-recensement',
    SURVEY_NHS: 'nhs-enm',
}

URL = 'http://www12.statcan.gc.ca/{survey_url}/2011/dp-pd/prof/details/download-telecharger/comprehensive/comp_download.cfm?CTLG={survey_code}&FMT=CSV{geo_code}'


class BaseParams:
    __metaclass__ = ABCMeta

    resolution = Parameter(default=GEO_PR)
    survey = Parameter(default=SURVEY_CEN)


class DownloadData(BaseParams, DownloadUnzipTask):
    def download(self):
        urllib.urlretrieve(url=URL.format(
                           survey_url=SURVEY_URLS[self.survey],
                           survey_code=SURVEY_CODES[self.survey],
                           geo_code=GEOGRAPHY_CODES[self.resolution],
                           ),
                           filename=self.output().path + '.zip')


class SplitAndTransposeData(BaseParams, Task):
    IGNORED_FILE_SUFFIXES = ('-DQ',)
    DIVISION_SPLITTED = {
        GEO_CT: None,
        GEO_PR: None,
        GEO_CD: None,
        GEO_CSD: None,
        GEO_CMA: ('cma_ca_name', (r'part\)$',))
    }

    def requires(self):
        return DownloadData(resolution=self.resolution, survey=self.survey)

    def run(self):
        infiles = shell('ls {input}/{survey_code}-{geo_code}*.[cC][sS][vV]'.format(
            input=self.input().path,
            survey_code=SURVEY_CODES[self.survey],
            geo_code=GEOGRAPHY_CODES[self.resolution]
        ))
        in_csv_files = []
        for in_csv_file in infiles.strip().split('\n'):
            if not self._is_ignored_suffix(in_csv_file):
                in_csv_files.append(in_csv_file)
            else:
                LOGGER.warning('Ignoring file %s' % in_csv_file)
        os.makedirs(self.output().path)
        StatCanParser(self.DIVISION_SPLITTED[self.resolution]).parse_csv_to_files(in_csv_files, self.output().path)

    def _is_ignored_suffix(self, csv_file):
        if os.path.splitext(csv_file)[0].endswith(self.IGNORED_FILE_SUFFIXES):
            return True
        return False

    def output(self):
        return LocalTarget(os.path.join('tmp', classpath(self), self.task_id))


#####################################
# IMPORT TO TEMP TABLES
#####################################
class CopyDataToTable(BaseParams, TempTableTask):<commit_msg>Added support for exceptions in both Census and NHS data<commit_after>import os
import urllib

from abc import ABCMeta
from luigi import Task, Parameter, WrapperTask, LocalTarget
from collections import OrderedDict
from tasks.util import DownloadUnzipTask, shell, TableTask, TempTableTask, classpath, MetaWrapper, LOGGER
from tasks.meta import current_session
from tasks.ca.statcan.geo import (
    GEO_CT, GEO_PR, GEO_CD, GEO_CSD, GEO_CMA,
    GEOGRAPHY_CODES, GEOGRAPHIES, GeographyColumns, Geography)
from tasks.ca.statcan.util import StatCanParser
from tasks.ca.statcan.cols_census import CensusColumns
from tasks.ca.statcan.cols_nhs import NHSColumns


SURVEY_CEN = 'census'
SURVEY_NHS = 'nhs'

SURVEYS = (
    SURVEY_CEN,
    SURVEY_NHS,
)

SURVEY_CODES = {
    SURVEY_CEN: '98-316-XWE2011001',
    SURVEY_NHS: '99-004-XWE2011001',
}

SURVEY_URLS = {
    SURVEY_CEN: 'census-recensement',
    SURVEY_NHS: 'nhs-enm',
}

URL = 'http://www12.statcan.gc.ca/{survey_url}/2011/dp-pd/prof/details/download-telecharger/comprehensive/comp_download.cfm?CTLG={survey_code}&FMT=CSV{geo_code}'


class BaseParams:
    __metaclass__ = ABCMeta

    resolution = Parameter(default=GEO_PR)
    survey = Parameter(default=SURVEY_CEN)


class DownloadData(BaseParams, DownloadUnzipTask):
    def download(self):
        urllib.urlretrieve(url=URL.format(
                           survey_url=SURVEY_URLS[self.survey],
                           survey_code=SURVEY_CODES[self.survey],
                           geo_code=GEOGRAPHY_CODES[self.resolution],
                           ),
                           filename=self.output().path + '.zip')


class SplitAndTransposeData(BaseParams, Task):
    IGNORED_FILE_SUFFIXES = ('-DQ',)
    DIVISION_SPLITTED = {
        SURVEY_CEN: {
            GEO_CT: None,
            GEO_PR: None,
            GEO_CD: None,
            GEO_CSD: None,
            GEO_CMA: ('cmaca_name', (r'part\)$',))
        },
        SURVEY_NHS: {
            GEO_CT: None,
            GEO_PR: None,
            GEO_CD: None,
            GEO_CSD: None,
            GEO_CMA: ('cma_ca_name', (r'part\)$',))
        }
    }

    def requires(self):
        return DownloadData(resolution=self.resolution, survey=self.survey)

    def run(self):
        infiles = shell('ls {input}/{survey_code}-{geo_code}*.[cC][sS][vV]'.format(
            input=self.input().path,
            survey_code=SURVEY_CODES[self.survey],
            geo_code=GEOGRAPHY_CODES[self.resolution]
        ))
        in_csv_files = []
        for in_csv_file in infiles.strip().split('\n'):
            if not self._is_ignored_suffix(in_csv_file):
                in_csv_files.append(in_csv_file)
            else:
                LOGGER.warning('Ignoring file %s' % in_csv_file)
        os.makedirs(self.output().path)
        StatCanParser(self.DIVISION_SPLITTED[self.survey][self.resolution]).parse_csv_to_files(in_csv_files, self.output().path)

    def _is_ignored_suffix(self, csv_file):
        if os.path.splitext(csv_file)[0].endswith(self.IGNORED_FILE_SUFFIXES):
            return True
        return False

    def output(self):
        return LocalTarget(os.path.join('tmp', classpath(self), self.task_id))


#####################################
# IMPORT TO TEMP TABLES
#####################################
class CopyDataToTable(BaseParams, TempTableTask):
{'fraction_non_alphanumeric': 0.13012950538305507, 'fraction_numerical': 0.009673896083632392, 'mean_word_length': 2.988799004355943, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6404619', 'n_tokens_mistral': 2347, 'n_tokens_neox': 2257, 'n_words': 349}