<commit_before>import logging
from collections import namedtuple
from django.db import connection
from councils.models import Council
from pollingstations.models import (PollingDistrict, ResidentialAddress,
                                    PollingStation)
from uk_geo_utils.helpers import Postcode
from addressbase.models import Address


AddressTuple = namedtuple('Address', [
    'address',
    'postcode',
    'council_id',
    'polling_station_id',
    'slug'])


class AddressSet(set):

    def save(self, batch_size):

        addresses_db = []
        for address in self:
            record = ResidentialAddress(
                address=address.address,
                postcode=Postcode(address.postcode).without_space,
                polling_station_id=address.polling_station_id,
                council_id=address.council_id,
                slug=address.slug,
            )
            addresses_db.append(record)

        ResidentialAddress.objects.bulk_create(
            addresses_db, batch_size=batch_size)


class EdgeCaseFixer:

    def __init__(self, target_council_id, logger):
        self.address_set = AddressSet()
        self.target_council_id = target_council_id
        self.logger = logger
        self.AddressRecord = namedtuple('AddressRecord', [
            'uprn',
            'address',
            'postcode',
            'district_id',
            'station_id',
            'council_id',
            'count',
            'location',
        ])

    def unpack_address(self, record):
        return self.AddressRecord(*record)

    def get_station_id(self, address):
        if not address.council_id:
            c = Council.objects\
                .defer("area")\
                .get(area__covers=address.location)
            council_id = c.council_id
        else:
            council_id = address.council_id

        if council_id != self.target_council_id:
            # treat addresses in other council areas as district not found
            raise PollingDistrict.DoesNotExist

        if address.count > 1:
            raise PollingDistrict.MultipleObjectsReturned

        if not address.district_id:
            raise PollingDistrict.DoesNotExist

        if address.station_id:
            polling_station = address.station_id
        else:
            """
            We do not know which station this district is served by (orphan district)

            Because we have no way of knowing what the correct station is, intentionally insert a record with an empty station id
            This allows us to *list* the address, but if the user *chooses* it, we will show "we don't know: call your council"
            """
            polling_station = ''

        return polling_station

    def make_addresses_for_postcode(self, postcode):
        cursor = connection.cursor()
        cursor.execute(
            """
            SELECT
                ab.uprn,
                ab.address,
                ab.postcode,
                pd.internal_council_id,
                ps.internal_council_id,
                os.lad,
                ct.count,
                ab.location
            FROM addressbase_address ab

            LEFT JOIN pollingstations_pollingdistrict pd
            ON ST_CONTAINS(pd.area, ab.location)

            LEFT JOIN uk_geo_utils_onsud os
            ON os.uprn=ab.uprn

            LEFT JOIN pollingstations_pollingstation ps
            ON (
                (pd.polling_station_id=ps.internal_council_id
                    AND pd.council_id=ps.council_id)
                OR
                (pd.internal_council_id=ps.polling_district_id
                    AND pd.council_id=ps.council_id)
            )

            JOIN (
                SELECT
                    ab.uprn,
                    COUNT(*) AS count
                FROM addressbase_address ab
                LEFT JOIN pollingstations_pollingdistrict pd
                ON ST_CONTAINS(pd.area, ab.location)
                WHERE ab.postcode=%s
                GROUP BY ab.uprn
            ) ct
            ON ab.uprn=ct.uprn

            WHERE ab.postcode=%s
            """, [postcode, postcode]
        )
        addresses = cursor.fetchall()<commit_msg>update table name in helper script<commit_after>import logging
from collections import namedtuple
from django.db import connection
from councils.models import Council
from pollingstations.models import (PollingDistrict, ResidentialAddress,
                                    PollingStation)
from uk_geo_utils.helpers import Postcode
from addressbase.models import Address


AddressTuple = namedtuple('Address', [
    'address',
    'postcode',
    'council_id',
    'polling_station_id',
    'slug'])


class AddressSet(set):

    def save(self, batch_size):

        addresses_db = []
        for address in self:
            record = ResidentialAddress(
                address=address.address,
                postcode=Postcode(address.postcode).without_space,
                polling_station_id=address.polling_station_id,
                council_id=address.council_id,
                slug=address.slug,
            )
            addresses_db.append(record)

        ResidentialAddress.objects.bulk_create(
            addresses_db, batch_size=batch_size)


class EdgeCaseFixer:

    def __init__(self, target_council_id, logger):
        self.address_set = AddressSet()
        self.target_council_id = target_council_id
        self.logger = logger
        self.AddressRecord = namedtuple('AddressRecord', [
            'uprn',
            'address',
            'postcode',
            'district_id',
            'station_id',
            'council_id',
            'count',
            'location',
        ])

    def unpack_address(self, record):
        return self.AddressRecord(*record)

    def get_station_id(self, address):
        if not address.council_id:
            c = Council.objects\
                .defer("area")\
                .get(area__covers=address.location)
            council_id = c.council_id
        else:
            council_id = address.council_id

        if council_id != self.target_council_id:
            # treat addresses in other council areas as district not found
            raise PollingDistrict.DoesNotExist

        if address.count > 1:
            raise PollingDistrict.MultipleObjectsReturned

        if not address.district_id:
            raise PollingDistrict.DoesNotExist

        if address.station_id:
            polling_station = address.station_id
        else:
            """
            We do not know which station this district is served by (orphan district)

            Because we have no way of knowing what the correct station is, intentionally insert a record with an empty station id
            This allows us to *list* the address, but if the user *chooses* it, we will show "we don't know: call your council"
            """
            polling_station = ''

        return polling_station

    def make_addresses_for_postcode(self, postcode):
        cursor = connection.cursor()
        cursor.execute(
            """
            SELECT
                ab.uprn,
                ab.address,
                ab.postcode,
                pd.internal_council_id,
                ps.internal_council_id,
                os.lad,
                ct.count,
                ab.location
            FROM addressbase_address ab

            LEFT JOIN pollingstations_pollingdistrict pd
            ON ST_CONTAINS(pd.area, ab.location)

            LEFT JOIN addressbase_onsud os
            ON os.uprn=ab.uprn

            LEFT JOIN pollingstations_pollingstation ps
            ON (
                (pd.polling_station_id=ps.internal_council_id
                    AND pd.council_id=ps.council_id)
                OR
                (pd.internal_council_id=ps.polling_district_id
                    AND pd.council_id=ps.council_id)
            )

            JOIN (
                SELECT
                    ab.uprn,
                    COUNT(*) AS count
                FROM addressbase_address ab
                LEFT JOIN pollingstations_pollingdistrict pd
                ON ST_CONTAINS(pd.area, ab.location)
                WHERE ab.postcode=%s
                GROUP BY ab.uprn
            ) ct
            ON ab.uprn=ct.uprn

            WHERE ab.postcode=%s
            """, [postcode, postcode]
        )
        addresses = cursor.fetchall()
{'fraction_non_alphanumeric': 0.06714116251482799, 'fraction_numerical': 0.00023724792408066428, 'mean_word_length': 1.659621451104101, 'pattern_counts': {'":': 0, '<': 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': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2263810', 'n_tokens_mistral': 2363, 'n_tokens_neox': 2178, 'n_words': 524}