<commit_before>import logging
from typing import Optional, Tuple, Dict

import requests

from importer import JSON
from importer.models import CachedObject

logger = logging.getLogger(__name__)


class BaseLoader:
    """ Provides a json and file download function.

    This class can be overwritten for vendor specific fixups
    """

    def __init__(self, system: JSON) -> None:
        self.system = system

    def load(self, url: str, query: Optional[Dict[str, str]] = None) -> JSON:
        logger.debug("Loader is loading {}".format(url))
        if query is None:
            query = dict()
        response = requests.get(url, params=query)
        response.raise_for_status()
        return response.json()

    def load_file(self, url: str) -> Tuple[bytes, Optional[str]]:
        """ Returns the content and the content type """
        response = requests.get(url)
        response.raise_for_status()
        content = response.content
        content_type = response.headers.get("Content-Type")
        return content, content_type


class SternbergLoader(BaseLoader):
    empty_list_error = {
        "error": "Die angeforderte Ressource wurde nicht gefunden.",
        "code": 802,
        "type": "SD.NET RIM Webservice",
    }

    empty_page = {"data": [], "links": {}, "pagination": {}}

    def load(self, url: str, query: Optional[Dict[str, str]] = None) -> JSON:
        logger.debug("Loader is loading {}".format(url))
        if query is None:
            query = dict()

        # TODO: Better naming
        response = requests.get(url, params=query)
        response_json = response.json()

        # An error is returned when the list would have been empty
        if (
            response.status_code == 404
            and "modified_since" in query
            and response_json == self.empty_list_error
        ):
            response_json = self.empty_page
        else:
            response.raise_for_status()

        if "/body" in url:
            # Add missing "type"-attributes in body-lists
            if "data" in response_json:
                for data in response_json["data"]:
                    if "location" in data.keys() and isinstance(data["location"], dict):
                        data["location"][
                            "type"
                        ] = "https://schema.oparl.org/1.0/Location"

                # There are deleted entries in unfiltered external lists (which they shouldn't) and then
                # they don't even have type attributes (which are mandatory)
                for entry in response_json["data"][:]:
                    if entry.get("deleted") and not "type" in entry:
                        response_json["data"].remove(entry)

            # Add missing "type"-attributes in single bodies
            if "location" in response_json.keys() and isinstance(response_json["location"], dict):
                response_json["location"]["type"] = "https://schema.oparl.org/1.0/Location"

            # Location in Person must be a url, not an object
            if "/person" in url and "data" in response_json:
                for data in response_json["data"]:
                    if "location" in data and isinstance(data["location"], dict):
                        data["location"] = data["location"]["id"]

            if "/organization" in url and "data" in response_json:
                for data in response_json["data"]:
                    if "id" in data and "type" not in data:
                        data["type"] = "https://schema.oparl.org/1.0/Organization"

        if "/membership" in url:
            # If an array is returned instead of an object, we just skip all list entries except for the last one
            if isinstance(response_json, list):
                response_json = response_json[0]

        if "/person" in url:
            if "location" in response_json and not isinstance(response_json["location"], str):
                response_json["location"] = response_json["location"]["id"]

        if "/meeting" in url:
            if "location" in response_json and not isinstance(response_json["location"], str):
                response_json["location"]["type"] = "https://schema.oparl.org/1.0/Location"

        if response_json.get("type") == "https://schema.oparl.org/1.0/File":
            if "accessUrl" in response_json:
                response_json["accessUrl"] = response_json["accessUrl"].replace(
                    r"files//rim", r"files/rim"
                )
            if "downloadUrl" in response_json:
                response_json["downloadUrl"] = response_json["downloadUrl"].replace(
                    r"files//rim", r"files/rim"
                )

        return response_json

    def load_file(self, url: str) -> Tuple[bytes, Optional[str]]:
        content, content_type = super().load_file(url)
        if content_type == "application/octetstream; charset=UTF-8":
            content_type = None
        return content, content_type


class CCEgovLoader(BaseLoader):
    def visit(self, data: JSON):
        """ Removes quirks like `"streetAddress": " "` in Location """
        for key, value in data.copy().items():
            if isinstance(value, dict):
                self.visit(value)
            elif isinstance(value, str):
                if value == "N/A" or not value.strip():
                    del data[key]

    def load(self, url: str, query: Optional[dict] = None) -> JSON:
        response = super(CCEgovLoader, self).load(url, query)
        self.visit(response)
        return response


def get_loader_from_system(entrypoint: str) -> BaseLoader:
    system = requests.get(entrypoint).json()
    if system.get("contactName") == "STERNBERG Software GmbH & Co. KG":
        logger.info("Using Sternberg patches")<commit_msg>.... and back again<commit_after>import logging
from typing import Optional, Tuple, Dict

import requests

from importer import JSON
from importer.models import CachedObject

logger = logging.getLogger(__name__)


class BaseLoader:
    """ Provides a json and file download function.

    This class can be overwritten for vendor specific fixups
    """

    def __init__(self, system: JSON) -> None:
        self.system = system

    def load(self, url: str, query: Optional[Dict[str, str]] = None) -> JSON:
        logger.debug("Loader is loading {}".format(url))
        if query is None:
            query = dict()
        response = requests.get(url, params=query)
        response.raise_for_status()
        return response.json()

    def load_file(self, url: str) -> Tuple[bytes, Optional[str]]:
        """ Returns the content and the content type """
        response = requests.get(url)
        response.raise_for_status()
        content = response.content
        content_type = response.headers.get("Content-Type")
        return content, content_type


class SternbergLoader(BaseLoader):
    empty_list_error = {
        "error": "Die angeforderte Ressource wurde nicht gefunden.",
        "code": 802,
        "type": "SD.NET RIM Webservice",
    }

    empty_page = {"data": [], "links": {}, "pagination": {}}

    def load(self, url: str, query: Optional[Dict[str, str]] = None) -> JSON:
        logger.debug("Loader is loading {}".format(url))
        if query is None:
            query = dict()

        # TODO: Better naming
        request_response = requests.get(url, params=query)
        response = request_response.json()

        # An error is returned when the list would have been empty
        if (
            request_response.status_code == 404
            and "modified_since" in query
            and response == self.empty_list_error
        ):
            response = self.empty_page
        else:
            request_response.raise_for_status()

        # TODO: Temporary for weirdness
        logger.info("{}".format(response))

        if "/body" in url:
            # Add missing "type"-attributes in body-lists
            if "data" in response:
                for data in response["data"]:
                    if "location" in data.keys() and isinstance(data["location"], dict):
                        data["location"][
                            "type"
                        ] = "https://schema.oparl.org/1.0/Location"

                # There are deleted entries in unfiltered external lists (which they shouldn't) and then
                # they don't even have type attributes (which are mandatory)
                for entry in response["data"][:]:
                    if entry.get("deleted") and not "type" in entry:
                        response["data"].remove(entry)

            # Add missing "type"-attributes in single bodies
            if "location" in response.keys() and isinstance(response["location"], dict):
                response["location"]["type"] = "https://schema.oparl.org/1.0/Location"

            # Location in Person must be a url, not an object
            if "/person" in url and "data" in response:
                for data in response["data"]:
                    if "location" in data and isinstance(data["location"], dict):
                        data["location"] = data["location"]["id"]

            if "/organization" in url and "data" in response:
                for data in response["data"]:
                    if "id" in data and "type" not in data:
                        data["type"] = "https://schema.oparl.org/1.0/Organization"

        if "/membership" in url:
            # If an array is returned instead of an object, we just skip all list entries except for the last one
            if isinstance(response, list):
                response = response[0]

        if "/person" in url:
            if "location" in response and not isinstance(response["location"], str):
                response["location"] = response["location"]["id"]

        if "/meeting" in url:
            if "location" in response and not isinstance(response["location"], str):
                response["location"]["type"] = "https://schema.oparl.org/1.0/Location"

        if response.get("type") == "https://schema.oparl.org/1.0/File":
            if "accessUrl" in response:
                response["accessUrl"] = response["accessUrl"].replace(
                    r"files//rim", r"files/rim"
                )
            if "downloadUrl" in response:
                response["downloadUrl"] = response["downloadUrl"].replace(
                    r"files//rim", r"files/rim"
                )

        return response

    def load_file(self, url: str) -> Tuple[bytes, Optional[str]]:
        content, content_type = super().load_file(url)
        if content_type == "application/octetstream; charset=UTF-8":
            content_type = None
        return content, content_type


class CCEgovLoader(BaseLoader):
    def visit(self, data: JSON):
        """ Removes quirks like `"streetAddress": " "` in Location """
        for key, value in data.copy().items():
            if isinstance(value, dict):
                self.visit(value)
            elif isinstance(value, str):
                if value == "N/A" or not value.strip():
                    del data[key]

    def load(self, url: str, query: Optional[dict] = None) -> JSON:
        response = super(CCEgovLoader, self).load(url, query)
        self.visit(response)
        return response


def get_loader_from_system(entrypoint: str) -> BaseLoader:
    system = requests.get(entrypoint).json()
    if system.get("contactName") == "STERNBERG Software GmbH & Co. KG":
        logger.info("Using Sternberg patches")
{'fraction_non_alphanumeric': 0.1158240997229917, 'fraction_numerical': 0.0031163434903047093, 'mean_word_length': 2.096488876976682, 'pattern_counts': {'":': 20, '<': 3, '<?xml version=': 0, '>': 17, 'https://': 10, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4074707', 'n_tokens_mistral': 3119, 'n_tokens_neox': 2993, 'n_words': 915}