<commit_before>#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Proxy models for augmenting our source data tables with methods useful for processing.
"""
import re
from datetime import date
from django.db import models
from django.utils import timezone
from django.db.models import Value
from django.db.models import CharField
from calaccess_processed import corrections
from django.db.models.functions import Concat
from calaccess_processed import special_elections
from opencivicdata.core.models import Organization
from calaccess_raw.models import FilerToFilerTypeCd
from opencivicdata.elections.models import Election
from calaccess_processed.models import Form501Filing
from calaccess_scraped.models import CandidateElection, IncumbentElection, Candidate


class OCDPartyManager(models.Manager):
    """
    Limited the OCD Organization model to politics parties.
    """
    def get_queryset(self):
        """
        Override the default manager to limit the results to political parties.
        """
        return super(OCDPartyManager, self).get_queryset().filter(classification='party')

    def get_by_name(self, name):
        """
        Helper for getting the OCD party object giving a raw name from CAL-ACCESS.

        If not found, return the "UNKNOWN" Organization object.
        """
        # First try a full name
        try:
            return self.get_queryset().get(name=name)
        except self.model.DoesNotExist:
            pass

        # If that doesn't work, try an alternate name
        try:
            return self.get_queryset().get(other_names__name=name)
        except self.model.DoesNotExist:
            pass

        # And if that doesn't work, just return the unknown party object
        return self.get_queryset().get(name='UNKNOWN')

    def get_by_filer_id(self, filer_id, election_date):
        """
        Lookup the party for the given filer_id, effective before election_date.

        If not found, return the "UNKNOWN" Organization object.
        """
        # Try to see if the record exists in the raw data with a party code
        try:
            party_code = FilerToFilerTypeCd.objects.filter(
                filer_id=filer_id,
                effect_dt__lte=election_date,
            ).latest('effect_dt').party_cd
        except FilerToFilerTypeCd.DoesNotExist:
            # If it doesn't hit just quit now
            return self.get_queryset().get(name='UNKNOWN')

        # IF we have a code, transform "INDEPENDENT" and "NON-PARTISAN" codes to "NO PARTY PREFERENCE"
        if party_code in [16007, 16009]:
            party_code = 16012

        # Try pulling out the party using the lookup code
        try:
            return self.get_queryset().get(identifiers__identifier=party_code)
        except self.model.DoesNotExist:
            pass

        # If that fails, just quit and return the unknown party object
        return self.get_queryset().get(name='UNKNOWN')


class OCDPartyProxy(Organization):
    """
    A proxy on the OCD Organization model with helper methods for interacting with party entities.
    """
    objects = OCDPartyManager()

    class Meta:
        """
        Make this a proxy model.
        """
        proxy = True

    def is_unknown(self):
        """
        Returns whether or not the provided party is unknown.
        """
        return self.name == 'UNKNOWN'


class ScrapedCandidateProxy(Candidate):
    """
    A proxy for the Candidate model in calaccess_scraped.
    """
    class Meta:
        """
        Make this a proxy model.
        """
        proxy = True

    @property
    def election_proxy(self):
        """
        Return the proxy model for the related election object.
        """
        return ScrapedCandidateElectionProxy.objects.get(id=self.election.id)

    def parse_office_name(self):
        """
        Parse string containg the name for an office.

        Expected format is "{TYPE NAME}[{DISTRICT NUMBER}]".

        Return a dict with two keys: type and district.
        """
        office_pattern = r'^(?P<type>[A-Z ]+)(?P<district>\d{2})?$'
        try:
            parsed = re.match(office_pattern, self.office_name.upper()).groupdict()
        except AttributeError:
            parsed = {'type': None, 'district': None}
        else:
            parsed['type'] = parsed['type'].strip()
            try:
                parsed['district'] = int(parsed['district'])
            except TypeError:
                pass
        return parsed

    def get_party(self):
        """
        Returns the party we believe the candidate was associated with this election.
        """
        # Pull the OCD election record so have it to inspect
        ocd_election = self.election_proxy

        # Check if this candidate has been manually corrected.
        if ocd_election.parsed_date:
            party = corrections.candidate_party(
                self.name,
                ocd_election.parsed_date.year,
                ocd_election.parsed_name['type'],
                self.office_name,
            )
            # If so, just pass that out right away
            if party:
                return party

        #
        # Otherwise, proceed with our standard fallback system for infering their party
        #

        # First, if the candidate is running for this office, it is by definition non-partisan
        if self.office_name == 'SUPERINTENDENT OF PUBLIC INSTRUCTION':
            return OCDPartyProxy.objects.get(name="NO PARTY PREFERENCE")

        # Next, if they have filed a 501 form, let's use that
        party = None
        form501 = self.get_form501_filing()
        if form501:
            party = OCDPartyProxy.objects.get_by_name(form501.party)
            # If they still don't have a party, try using their filer_id
            if party.is_unknown():
                party = OCDPartyProxy.objects.get_by_filer_id(int(form501.filer_id), ocd_election.parsed_date)
            # If we got a real one, return that and we're done.
            else:
                return party

        # If there's no 501, or if the 501 returned an unknown party ...
        # ... try one last stab at using the filer id (assuming it exists)
        if (party is None or party.is_unknown()) and self.scraped_id:
            return OCDPartyProxy.objects.get_by_filer_id(int(self.scraped_id), ocd_election.parsed_date)
        # Otherwise just give up and return the unknown party
        else:
            return OCDPartyProxy.objects.get(name="UNKNOWN")

    def get_form501_filing(self):
        """
        Return a Form501Filing that matches the scraped Candidate.

        By default, return the latest Form501FilingVersion, unless earliest<commit_msg>Work in a catch for records that don't have a party or a 501<commit_after>#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Proxy models for augmenting our source data tables with methods useful for processing.
"""
import re
from datetime import date
from django.db import models
from django.utils import timezone
from django.db.models import Value
from django.db.models import CharField
from calaccess_processed import corrections
from django.db.models.functions import Concat
from calaccess_processed import special_elections
from opencivicdata.core.models import Organization
from calaccess_raw.models import FilerToFilerTypeCd
from opencivicdata.elections.models import Election
from calaccess_processed.models import Form501Filing
from calaccess_scraped.models import CandidateElection, IncumbentElection, Candidate


class OCDPartyManager(models.Manager):
    """
    Limited the OCD Organization model to politics parties.
    """
    def get_queryset(self):
        """
        Override the default manager to limit the results to political parties.
        """
        return super(OCDPartyManager, self).get_queryset().filter(classification='party')

    def get_by_name(self, name):
        """
        Helper for getting the OCD party object giving a raw name from CAL-ACCESS.

        If not found, return the "UNKNOWN" Organization object.
        """
        # First try a full name
        try:
            return self.get_queryset().get(name=name)
        except self.model.DoesNotExist:
            pass

        # If that doesn't work, try an alternate name
        try:
            return self.get_queryset().get(other_names__name=name)
        except self.model.DoesNotExist:
            pass

        # And if that doesn't work, just return the unknown party object
        return self.get_queryset().get(name='UNKNOWN')

    def get_by_filer_id(self, filer_id, election_date):
        """
        Lookup the party for the given filer_id, effective before election_date.

        If not found, return the "UNKNOWN" Organization object.
        """
        # Try to see if the record exists in the raw data with a party code
        try:
            party_code = FilerToFilerTypeCd.objects.filter(
                filer_id=filer_id,
                effect_dt__lte=election_date,
            ).latest('effect_dt').party_cd
        except FilerToFilerTypeCd.DoesNotExist:
            # If it doesn't hit just quit now
            return self.get_queryset().get(name='UNKNOWN')

        # IF we have a code, transform "INDEPENDENT" and "NON-PARTISAN" codes to "NO PARTY PREFERENCE"
        if party_code in [16007, 16009]:
            party_code = 16012

        # Try pulling out the party using the lookup code
        try:
            return self.get_queryset().get(identifiers__identifier=party_code)
        except self.model.DoesNotExist:
            pass

        # If that fails, just quit and return the unknown party object
        return self.get_queryset().get(name='UNKNOWN')


class OCDPartyProxy(Organization):
    """
    A proxy on the OCD Organization model with helper methods for interacting with party entities.
    """
    objects = OCDPartyManager()

    class Meta:
        """
        Make this a proxy model.
        """
        proxy = True

    def is_unknown(self):
        """
        Returns whether or not the provided party is unknown.
        """
        return self.name == 'UNKNOWN'


class ScrapedCandidateProxy(Candidate):
    """
    A proxy for the Candidate model in calaccess_scraped.
    """
    class Meta:
        """
        Make this a proxy model.
        """
        proxy = True

    @property
    def election_proxy(self):
        """
        Return the proxy model for the related election object.
        """
        return ScrapedCandidateElectionProxy.objects.get(id=self.election.id)

    def parse_office_name(self):
        """
        Parse string containg the name for an office.

        Expected format is "{TYPE NAME}[{DISTRICT NUMBER}]".

        Return a dict with two keys: type and district.
        """
        office_pattern = r'^(?P<type>[A-Z ]+)(?P<district>\d{2})?$'
        try:
            parsed = re.match(office_pattern, self.office_name.upper()).groupdict()
        except AttributeError:
            parsed = {'type': None, 'district': None}
        else:
            parsed['type'] = parsed['type'].strip()
            try:
                parsed['district'] = int(parsed['district'])
            except TypeError:
                pass
        return parsed

    def get_party(self):
        """
        Returns the party we believe the candidate was associated with this election.
        """
        # First, if the candidate is running for this office, it is by definition non-partisan
        if self.office_name == 'SUPERINTENDENT OF PUBLIC INSTRUCTION':
            return OCDPartyProxy.objects.get(name="NO PARTY PREFERENCE")

        # Next pull the OCD election record so we have it to inspect
        ocd_election = self.election_proxy

        # Pull the Form 501 so we have that to inspect
        form501 = self.get_form501_filing()

        # If the record doesn't have either of those, we're not going to get an answer
        if not ocd_election.parsed_date and not form501:
            return OCDPartyProxy.objects.get(name="UNKNOWN")

        #
        # Otherwise, proceed with our standard fallback system for infering their party
        #

        # Check if this candidate has been manually corrected.
        if ocd_election.parsed_date:
            party = corrections.candidate_party(
                self.name,
                ocd_election.parsed_date.year,
                ocd_election.parsed_name['type'],
                self.office_name,
            )
            # If so, just pass that out right away
            if party:
                return party

        # Next, if they have filed a 501 form, let's use that
        party = None
        if form501:
            party = OCDPartyProxy.objects.get_by_name(form501.party)
            # If they still don't have a party, try using their filer_id
            if party.is_unknown():
                party = OCDPartyProxy.objects.get_by_filer_id(int(form501.filer_id), ocd_election.parsed_date)
            # If we got a real one, return that and we're done.
            else:
                return party

        # If there's no 501, or if the 501 returned an unknown party ...
        # ... try one last stab at using the filer id (assuming it exists)
        if (party is None or party.is_unknown()) and self.scraped_id:
            return OCDPartyProxy.objects.get_by_filer_id(int(self.scraped_id), ocd_election.parsed_date)
        # Otherwise just give up and return the unknown party
        else:
            return OCDPartyProxy.objects.get(name="UNKNOWN")

    def get_form501_filing(self):
        """
        Return a Form501Filing that matches the scraped Candidate.

        By default, return the latest Form501FilingVersion, unless earliest
{'fraction_non_alphanumeric': 0.07712437304644908, 'fraction_numerical': 0.008359380678927091, 'mean_word_length': 2.2866698518872433, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '2540343', 'n_tokens_mistral': 3856, 'n_tokens_neox': 3626, 'n_words': 1231}