<commit_before>from dateutil.relativedelta import relativedelta

from custom.icds_reports.const import AGG_LS_VHND_TABLE, AGG_LS_AWC_VISIT_TABLE, AGG_LS_BENEFICIARY_TABLE
from custom.icds_reports.utils.aggregation_helpers import transform_day_to_month, month_formatter
from custom.icds_reports.utils.aggregation_helpers.distributed.base import BaseICDSAggregationDistributedHelper


class AggLsHelper(BaseICDSAggregationDistributedHelper):
    helper_key = 'agg-ls'
    base_tablename = 'agg_ls'

    def __init__(self, month):
        self.month_start = transform_day_to_month(month)
        self.next_month_start = self.month_start + relativedelta(months=1)

    def aggregate(self, cursor):
        drop_table_queries = [self.drop_table_if_exists(i) for i in range(4, 0, -1)]
        create_table_queries = [self.create_child_table(i) for i in range(4, 0, -1)]

        agg_query, agg_params = self.aggregate_query()
        rollup_queries = [self.rollup_query(i) for i in range(3, 0, -1)]
        index_queries = [self.indexes(i) for i in range(4, 0, -1)]
        index_queries = [query for index_list in index_queries for query in index_list]

        for drop_table_query in drop_table_queries:
            cursor.execute(drop_table_query)
        for create_table_query, create_params in create_table_queries:
            cursor.execute(create_table_query, create_params)

        cursor.execute(agg_query, agg_params)

        for rollup_query in rollup_queries:
            cursor.execute(rollup_query)

        for index_query in index_queries:
            cursor.execute(index_query)

    def _tablename_func(self, agg_level):
        return "{}_{}_{}".format(self.base_tablename, self.month_start.strftime("%Y-%m-%d"), agg_level)

    def drop_table_if_exists(self, agg_level):
        return """
        DROP TABLE IF EXISTS "{table_name}"
        """.format(table_name=self._tablename_func(agg_level))

    def create_child_table(self, agg_level):
        return """
        CREATE TABLE "{table_name}" (
        CHECK (month=DATE %(start_date)s AND aggregation_level={agg_level})
        ) INHERITS ({base_tablename})
        """.format(
            table_name=self._tablename_func(agg_level),
            base_tablename=self.base_tablename,
            agg_level=agg_level
        ), {
            "start_date": self.month_start
        }

    @property
    def tablename(self):
        return self._tablename_func(4)

    def aggregate_query(self):
        """
        Returns the base aggregate query which is used to insert all the locations
        into the LS data table.
        """

        columns = (
            ('state_id', 'location.state_id'),
            ('district_id', 'location.district_id'),
            ('block_id', 'location.block_id'),
            ('supervisor_id', 'location.supervisor_id'),
            ('month', "'{}'".format(month_formatter(self.month_start))),
            ('awc_visits', 'COALESCE(awc_table.awc_visits, 0)'),
            ('vhnd_observed', 'COALESCE(vhnd_table.vhnd_observed, 0)'),
            ('beneficiary_vists', 'COALESCE(beneficiary_table.beneficiary_vists, 0)'),
            ('aggregation_level', '4'),
            ('num_supervisor_launched', 'CASE WHEN GREATEST(awc_table.form_count, vhnd_table.form_count, beneficiary_table.form_count)>0 THEN 1 ELSE 0 END')
        )
        return """
        INSERT INTO "{tablename}" (
            {columns}
        )
        (
        SELECT
        {calculations}
        from (
            select distinct state_id, district_id, block_id, supervisor_id
            from awc_location_local where (
                state_is_test=0 AND
                district_is_test=0 AND
                block_is_test=0 AND
                supervisor_is_test=0 AND
                aggregation_level = 4
            )
        ) location
        LEFT  JOIN "{awc_table}" awc_table on (
            location.supervisor_id=awc_table.supervisor_id AND
            awc_table.month = %(start_date)s
        )
        LEFT  JOIN "{vhnd_table}" vhnd_table on (
            location.supervisor_id = vhnd_table.supervisor_id AND
            vhnd_table.month = %(start_date)s
        )
        LEFT  JOIN "{beneficiary_table}" beneficiary_table on (
            location.supervisor_id = beneficiary_table.supervisor_id AND
            beneficiary_table.month = %(start_date)s
        )
        )
        """.format(
            tablename=self.tablename,
            columns=", ".join([col[0] for col in columns]),
            calculations=", ".join([col[1] for col in columns]),
            awc_table=AGG_LS_AWC_VISIT_TABLE,
            vhnd_table=AGG_LS_VHND_TABLE,
            beneficiary_table=AGG_LS_BENEFICIARY_TABLE
        ), {
            'start_date': self.month_start
        }

    def indexes(self, aggregation_level):
        """
        Returns queries to create indices  for columns
        district_id, block_id, supervisor_id and state_id based on
        aggregation level
        """
        indexes = []
        agg_locations = ['state_id']<commit_msg>Roll up to next month<commit_after>from dateutil.relativedelta import relativedelta

from custom.icds_reports.const import AGG_LS_VHND_TABLE, AGG_LS_AWC_VISIT_TABLE, AGG_LS_BENEFICIARY_TABLE
from custom.icds_reports.utils.aggregation_helpers import transform_day_to_month, month_formatter
from custom.icds_reports.utils.aggregation_helpers.distributed.base import BaseICDSAggregationDistributedHelper


class AggLsHelper(BaseICDSAggregationDistributedHelper):
    helper_key = 'agg-ls'
    base_tablename = 'agg_ls'

    def __init__(self, month):
        self.month_start = transform_day_to_month(month)
        self.next_month_start = self.month_start + relativedelta(months=1)

    def aggregate(self, cursor):
        drop_table_queries = [self.drop_table_if_exists(i) for i in range(4, 0, -1)]
        create_table_queries = [self.create_child_table(i) for i in range(4, 0, -1)]

        agg_query, agg_params = self.aggregate_query()
        rollup_queries = [self.rollup_query(i) for i in range(3, 0, -1)]
        index_queries = [self.indexes(i) for i in range(4, 0, -1)]
        index_queries = [query for index_list in index_queries for query in index_list]

        for drop_table_query in drop_table_queries:
            cursor.execute(drop_table_query)
        for create_table_query, create_params in create_table_queries:
            cursor.execute(create_table_query, create_params)

        cursor.execute(agg_query, agg_params)

        for rollup_query in rollup_queries:
            cursor.execute(rollup_query)

        for index_query in index_queries:
            cursor.execute(index_query)

    def _tablename_func(self, month, agg_level):
        return "{}_{}_{}".format(self.base_tablename, month.strftime("%Y-%m-%d"), agg_level)

    def drop_table_if_exists(self, agg_level):
        return """
        DROP TABLE IF EXISTS "{table_name}"
        """.format(table_name=self._tablename_func(agg_level))

    def create_child_table(self, agg_level):
        return """
        CREATE TABLE "{table_name}" (
        CHECK (month=DATE %(start_date)s AND aggregation_level={agg_level})
        ) INHERITS ({base_tablename})
        """.format(
            table_name=self._tablename_func(agg_level),
            base_tablename=self.base_tablename,
            agg_level=agg_level
        ), {
            "start_date": self.month_start
        }

    @property
    def tablename(self):
        return self._tablename_func(self.month_start, 4)

    @property
    def prev_month_table(self):
        return self._tablename_func(self.month_start - relativedelta(months=1), 4)

    def aggregate_query(self):
        """
        Returns the base aggregate query which is used to insert all the locations
        into the LS data table.
        """

        columns = (
            ('state_id', 'location.state_id'),
            ('district_id', 'location.district_id'),
            ('block_id', 'location.block_id'),
            ('supervisor_id', 'location.supervisor_id'),
            ('month', "'{}'".format(month_formatter(self.month_start))),
            ('awc_visits', 'COALESCE(awc_table.awc_visits, 0)'),
            ('vhnd_observed', 'COALESCE(vhnd_table.vhnd_observed, 0)'),
            ('beneficiary_vists', 'COALESCE(beneficiary_table.beneficiary_vists, 0)'),
            ('aggregation_level', '4'),
            ('num_supervisor_launched', 'CASE WHEN GREATEST(awc_table.form_count, vhnd_table.form_count, beneficiary_table.form_count)>0 THEN 1 ELSE 0 END')
        )
        return """
        INSERT INTO "{tablename}" (
            {columns}
        )
        (
        SELECT
        {calculations}
        from (
            select distinct state_id, district_id, block_id, supervisor_id
            from awc_location_local where (
                state_is_test=0 AND
                district_is_test=0 AND
                block_is_test=0 AND
                supervisor_is_test=0 AND
                aggregation_level = 4
            )
        ) location
        LEFT  JOIN "{awc_table}" awc_table on (
            location.supervisor_id=awc_table.supervisor_id AND
            awc_table.month = %(start_date)s
        )
        LEFT  JOIN "{vhnd_table}" vhnd_table on (
            location.supervisor_id = vhnd_table.supervisor_id AND
            vhnd_table.month = %(start_date)s
        )
        LEFT  JOIN "{beneficiary_table}" beneficiary_table on (
            location.supervisor_id = beneficiary_table.supervisor_id AND
            beneficiary_table.month = %(start_date)s
        )
        );

        -- if launched in prev month make it launched in this month
        UPDATE "{tablename}" agg_ls SET
            num_supervisor_launched = ut.num_supervisor_launched
        FROM (
            SELECT supervisor_id, num_supervisor_launched
            FROM "{prev_month_table}" WHERE num_supervisor_launched=1
        ) ut
        agg_ls.supervisor_id = ut.supervisor_id;
        """.format(
            tablename=self.tablename,
            prev_month_table = self.prev_month_table,
            columns=", ".join([col[0] for col in columns]),
            calculations=", ".join([col[1] for col in columns]),
            awc_table=AGG_LS_AWC_VISIT_TABLE,
            vhnd_table=AGG_LS_VHND_TABLE,
            beneficiary_table=AGG_LS_BENEFICIARY_TABLE

        ), {
            'start_date': self.month_start
        }

    def indexes(self, aggregation_level):
        """
        Returns queries to create indices  for columns
        district_id, block_id, supervisor_id and state_id based on
        aggregation level
        """
        indexes = []
        agg_locations = ['state_id']
{'fraction_non_alphanumeric': 0.09441377374380087, 'fraction_numerical': 0.005520726115841677, 'mean_word_length': 2.528557279630241, 'pattern_counts': {'":': 2, '<': 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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2771156', 'n_tokens_mistral': 3487, 'n_tokens_neox': 3288, 'n_words': 638}