<commit_before>import csv
import datetime
import glob
import logging
import time

from google.cloud import storage
from google.cloud import bigquery
from google.cloud.exceptions import NotFound
from google.cloud.bigquery.table import Table
from google.cloud.bigquery.dataset import Dataset

from django.conf import settings
from django.core.management.base import BaseCommand
from django.core.management.base import CommandError

from ebmdatalab.bigquery_old import copy_table_to_gcs
from ebmdatalab.bigquery_old import download_from_gcs
from ebmdatalab.bigquery_old import wait_for_job

from ebmdatalab.bigquery import Client

logger = logging.getLogger(__name__)

TEMP_DATASET = 'tmp_eu'
TEMP_SOURCE_NAME = 'raw_nhs_digital_data'


class Command(BaseCommand):
    """There are two kinds of source we use to generate data.


    The legacy source (the code paths for which, once the new source
    has successfully been imported a few times, can be removed) is
    published erratically; the new "detailed" data source is published
    regularly, each month, so we now prefer that.

    The "detailed" source format has one iine for each presentation
    *and pack size*, so prescriptions of 28 paracetamol will be on a
    separate line from prescriptions of 100 paracetamol.

    The destination format has one line for paracetamol of any pack
    size.

    """
    args = ''
    help = 'Converts HSCIC data files into the format needed for our SQL COPY '
    help += 'statement. We use COPY because it is much faster than INSERT.'

    def add_arguments(self, parser):
        parser.add_argument('--filename')

    def handle(self, *args, **options):
        self.IS_VERBOSE = (options['verbosity'] > 1)

        if 'is_test' in options:
            self.IS_TEST = True
        else:
            self.IS_TEST = False

        filename = options['filename']

        if self.IS_VERBOSE:
            print "--------- Converting %s -----------" % filename

        filename_for_output = self.create_filename_for_output_file(filename)

        filename = filename.split('/prescribing/')[1]
        uri = 'gs://ebmdatalab/hscic/prescribing/' + filename
        # Grab date from file path
        try:
            date = datetime.datetime.strptime(
                uri.split("/")[-2] + "_01", "%Y_%m_%d"
            ).strftime('%Y_%m_%d')
        except ValueError as e:
            message = ('The file path must have a YYYY_MM '
                       'date component in the containing directory: ')
            message += e.message
            raise CommandError(message)

        self.aggregate_nhs_digital_data(uri, filename_for_output, date)

    def create_filename_for_output_file(self, filename):
        if self.IS_TEST:
            return filename[:-4] + '_test.CSV'
        else:
            return filename[:-4] + '_formatted.CSV'

    def aggregate_nhs_digital_data(self, uri, local_path, date):
        """Given a GCS URI for "detailed" prescribing data, run a query to
        aggregate it into the format we use internally, and download
        the resulting data to a `*_formatted.CSV` file, ready for
        importing.
        """
        # Create table at raw_nhs_digital_data
        table_ref = create_temporary_data_source(uri)
        self.append_aggregated_data_to_prescribing_table(
            table_ref, date)
        temp_table = self.write_aggregated_data_to_temp_table(
            table_ref, date)
        converted_uri = uri[:-4] + '_formatted-*.csv.gz'
        copy_table_to_gcs(temp_table, converted_uri)
        download_from_gcs(converted_uri, local_path)

    def assert_latest_data_not_already_uploaded(self, date):
        client = Client(settings.BQ_HSCIC_DATASET)
        sql = """SELECT COUNT(*)
        FROM [ebmdatalab:hscic.prescribing]
        WHERE month = TIMESTAMP('%s')""" % date.replace('_', '-')
        results = client.query(sql)
        assert query.rows[0][0] == 0

    def append_aggregated_data_to_prescribing_table(
            self, source_table_ref, date):
        self.assert_latest_data_not_already_uploaded(date)

        client = Client(settings.BQ_HSCIC_DATASET)
        table = client.get_table_ref('prescribing')

        sql = """
         SELECT
          Area_Team_Code AS sha,
          LEFT(PCO_Code, 3) AS pct,
          Practice_Code AS practice,
          BNF_Code AS bnf_code,
          BNF_Description AS bnf_name,
          SUM(Items) AS items,
          SUM(NIC) AS net_cost,
          SUM(Actual_Cost) AS actual_cost,
          SUM(Quantity * Items) AS quantity,
          TIMESTAMP('%s') AS month,
         FROM %s
         WHERE Practice_Code NOT LIKE '%%998'  -- see issue #349
         GROUP BY
           bnf_code, bnf_name, pct,
           practice, sha
        """ % (date.replace('_', '-'), source_table_ref)
        table.insert_rows_from_query(sql, legacy=True, write_disposition='WRITE_APPEND')

    def write_aggregated_data_to_temp_table(
        self, source_table_ref, date):
        sql = """
         SELECT
          LEFT(PCO_Code, 3) AS pct_id,
          Practice_Code AS practice_code,
          BNF_Code AS presentation_code,
          SUM(Items) AS total_items,
          SUM(NIC) AS net_cost,
          SUM(Actual_Cost) AS actual_cost,
          SUM(Quantity * Items) AS quantity,
          '%s' AS processing_date,
         FROM %s
         WHERE Practice_Code NOT LIKE '%%998'  -- see issue #349
         GROUP BY
           presentation_code, pct_id, practice_code
        """ % (date, source_table_ref)

        client = Client(TEMP_DATASET)
        table = client.get_table_ref('formatted_prescribing_%s' % date)
        table.insert_rows_from_query(sql, legacy=True)
        return table.gcbq_table


def create_temporary_data_source(source_uri):
    """Create a temporary data source so BigQuery can query the CSV in
    Google Cloud Storage.

    Nothing like this is currently implemented in the
    google-cloud-python library.

    Returns a table reference suitable for using in a BigQuery SQL
    query (legacy format).

    """
    schema = [
        {"name": "Regional_Office_Name", "type": "string"},
        {"name": "Regional_Office_Code", "type": "string"},
        {"name": "Area_Team_Name", "type": "string"},
        {"name": "Area_Team_Code", "type": "string", "mode": "required"},
        {"name": "PCO_Name", "type": "string"},
        {"name": "PCO_Code", "type": "string"},
        {"name": "Practice_Name", "type": "string"},
        {"name": "Practice_Code", "type": "string", "mode": "required"},
        {"name": "BNF_Code", "type": "string", "mode": "required"},
        {"name": "BNF_Description", "type": "string", "mode": "required"},
        {"name": "Items", "type": "integer", "mode": "required"},
        {"name": "Quantity", "type": "integer", "mode": "required"},
        {"name": "ADQ_Usage", "type": "float"},
        {"name": "NIC", "type": "float", "mode": "required"},
        {"name": "Actual_Cost", "type": "float", "mode": "required"},
    ]
    client = Client(TEMP_DATASET)
    gcs_path = source_uri.split('ebmdatalab/')[1]
    table = client.get_or_create_storage_backed_table(TEMP_SOURCE_NAME, schema, gcs_path)
    return table.legacy_full_qualified_name<commit_msg>convert_hscic_prescribing: Return actual table from create_temporary_data_source<commit_after>import csv
import datetime
import glob
import logging
import time

from google.cloud import storage
from google.cloud import bigquery
from google.cloud.exceptions import NotFound
from google.cloud.bigquery.table import Table
from google.cloud.bigquery.dataset import Dataset

from django.conf import settings
from django.core.management.base import BaseCommand
from django.core.management.base import CommandError

from ebmdatalab.bigquery_old import copy_table_to_gcs
from ebmdatalab.bigquery_old import download_from_gcs
from ebmdatalab.bigquery_old import wait_for_job

from ebmdatalab.bigquery import Client

logger = logging.getLogger(__name__)

TEMP_DATASET = 'tmp_eu'
TEMP_SOURCE_NAME = 'raw_nhs_digital_data'


class Command(BaseCommand):
    """There are two kinds of source we use to generate data.


    The legacy source (the code paths for which, once the new source
    has successfully been imported a few times, can be removed) is
    published erratically; the new "detailed" data source is published
    regularly, each month, so we now prefer that.

    The "detailed" source format has one iine for each presentation
    *and pack size*, so prescriptions of 28 paracetamol will be on a
    separate line from prescriptions of 100 paracetamol.

    The destination format has one line for paracetamol of any pack
    size.

    """
    args = ''
    help = 'Converts HSCIC data files into the format needed for our SQL COPY '
    help += 'statement. We use COPY because it is much faster than INSERT.'

    def add_arguments(self, parser):
        parser.add_argument('--filename')

    def handle(self, *args, **options):
        self.IS_VERBOSE = (options['verbosity'] > 1)

        if 'is_test' in options:
            self.IS_TEST = True
        else:
            self.IS_TEST = False

        filename = options['filename']

        if self.IS_VERBOSE:
            print "--------- Converting %s -----------" % filename

        filename_for_output = self.create_filename_for_output_file(filename)

        filename = filename.split('/prescribing/')[1]
        uri = 'gs://ebmdatalab/hscic/prescribing/' + filename
        # Grab date from file path
        try:
            date = datetime.datetime.strptime(
                uri.split("/")[-2] + "_01", "%Y_%m_%d"
            ).strftime('%Y_%m_%d')
        except ValueError as e:
            message = ('The file path must have a YYYY_MM '
                       'date component in the containing directory: ')
            message += e.message
            raise CommandError(message)

        self.aggregate_nhs_digital_data(uri, filename_for_output, date)

    def create_filename_for_output_file(self, filename):
        if self.IS_TEST:
            return filename[:-4] + '_test.CSV'
        else:
            return filename[:-4] + '_formatted.CSV'

    def aggregate_nhs_digital_data(self, uri, local_path, date):
        """Given a GCS URI for "detailed" prescribing data, run a query to
        aggregate it into the format we use internally, and download
        the resulting data to a `*_formatted.CSV` file, ready for
        importing.
        """
        # Create table at raw_nhs_digital_data
        temp_table = get_or_create_raw_data_table(uri)
        self.append_aggregated_data_to_prescribing_table(
            temp_table.legacy_full_qualified_name, date)
        temp_table = self.write_aggregated_data_to_temp_table(
            temp_table.legacy_full_qualified_name, date)

        converted_uri = uri[:-4] + '_formatted-*.csv.gz'
        copy_table_to_gcs(temp_table, converted_uri)
        download_from_gcs(converted_uri, local_path)

    def assert_latest_data_not_already_uploaded(self, date):
        client = Client(settings.BQ_HSCIC_DATASET)
        sql = """SELECT COUNT(*)
        FROM [ebmdatalab:hscic.prescribing]
        WHERE month = TIMESTAMP('%s')""" % date.replace('_', '-')
        results = client.query(sql)
        assert query.rows[0][0] == 0

    def append_aggregated_data_to_prescribing_table(
            self, temp_table_name, date):
        self.assert_latest_data_not_already_uploaded(date)

        client = Client(settings.BQ_HSCIC_DATASET)
        table = client.get_table_ref('prescribing')

        sql = """
         SELECT
          Area_Team_Code AS sha,
          LEFT(PCO_Code, 3) AS pct,
          Practice_Code AS practice,
          BNF_Code AS bnf_code,
          BNF_Description AS bnf_name,
          SUM(Items) AS items,
          SUM(NIC) AS net_cost,
          SUM(Actual_Cost) AS actual_cost,
          SUM(Quantity * Items) AS quantity,
          TIMESTAMP('%s') AS month,
         FROM %s
         WHERE Practice_Code NOT LIKE '%%998'  -- see issue #349
         GROUP BY
           bnf_code, bnf_name, pct,
           practice, sha
        """ % (date.replace('_', '-'), temp_table_name)
        table.insert_rows_from_query(sql, legacy=True, write_disposition='WRITE_APPEND')

    def write_aggregated_data_to_temp_table(
        self, temp_table_name, date):
        sql = """
         SELECT
          LEFT(PCO_Code, 3) AS pct_id,
          Practice_Code AS practice_code,
          BNF_Code AS presentation_code,
          SUM(Items) AS total_items,
          SUM(NIC) AS net_cost,
          SUM(Actual_Cost) AS actual_cost,
          SUM(Quantity * Items) AS quantity,
          '%s' AS processing_date,
         FROM %s
         WHERE Practice_Code NOT LIKE '%%998'  -- see issue #349
         GROUP BY
           presentation_code, pct_id, practice_code
        """ % (date, temp_table_name)

        client = Client(TEMP_DATASET)
        table = client.get_table_ref('formatted_prescribing_%s' % date)
        table.insert_rows_from_query(sql, legacy=True)
        return table.gcbq_table


def get_or_create_raw_data_table(source_uri):
    """Create a temporary data source so BigQuery can query the CSV in
    Google Cloud Storage.

    Nothing like this is currently implemented in the
    google-cloud-python library.

    Returns a table reference suitable for using in a BigQuery SQL
    query (legacy format).

    """
    schema = [
        {"name": "Regional_Office_Name", "type": "string"},
        {"name": "Regional_Office_Code", "type": "string"},
        {"name": "Area_Team_Name", "type": "string"},
        {"name": "Area_Team_Code", "type": "string", "mode": "required"},
        {"name": "PCO_Name", "type": "string"},
        {"name": "PCO_Code", "type": "string"},
        {"name": "Practice_Name", "type": "string"},
        {"name": "Practice_Code", "type": "string", "mode": "required"},
        {"name": "BNF_Code", "type": "string", "mode": "required"},
        {"name": "BNF_Description", "type": "string", "mode": "required"},
        {"name": "Items", "type": "integer", "mode": "required"},
        {"name": "Quantity", "type": "integer", "mode": "required"},
        {"name": "ADQ_Usage", "type": "float"},
        {"name": "NIC", "type": "float", "mode": "required"},
        {"name": "Actual_Cost", "type": "float", "mode": "required"},
    ]
    client = Client(TEMP_DATASET)
    gcs_path = source_uri.split('ebmdatalab/')[1]
    table = client.get_or_create_storage_backed_table(TEMP_SOURCE_NAME, schema, gcs_path)
    return table
{'fraction_non_alphanumeric': 0.10862092671524817, 'fraction_numerical': 0.004262340162243916, 'mean_word_length': 2.7863092139510672, 'pattern_counts': {'":': 76, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1696230', 'n_tokens_mistral': 4544, 'n_tokens_neox': 4355, 'n_words': 1198}