iframe-proxy | Sunbelt Computer Software

This repository was archived by the owner on Jul 3, 2025. It is now read-only.
executable file
196 lines (163 loc) · 8.62 KB
import urllib.request, urllib.parse, urllib.error
from .oauth2 import OAuth2Request
from .json_import import simplejson
import hmac
from hashlib import sha256
re_path_template = re.compile('{\w+}')
def encode_string(value):
    return value.encode('utf-8') \
                        if isinstance(value, str) else str(value)
class InstagramClientError(Exception):
    def __init__(self, error_message, status_code=None):
        self.status_code = status_code
        self.error_message = error_message
    def __str__(self):
        if self.status_code:
            return "(%s) %s" % (self.status_code, self.error_message)
        else:
            return self.error_message
class InstagramAPIError(Exception):
    def __init__(self, status_code, error_type, error_message, *args, **kwargs):
        self.status_code = status_code
        self.error_type = error_type
        self.error_message = error_message
    def __str__(self):
        return "(%s) %s-%s" % (self.status_code, self.error_type, self.error_message)
def bind_method(**config):
    class InstagramAPIMethod(object):
        path = config['path']
        method = config.get('method', 'GET')
        accepts_parameters = config.get("accepts_parameters", [])
        signature = config.get("signature", False)
        requires_target_user = config.get('requires_target_user', False)
        paginates = config.get('paginates', False)
        root_class = config.get('root_class', None)
        response_type = config.get("response_type", "list")
        include_secret = config.get("include_secret", False)
        objectify_response = config.get("objectify_response", True)
        def __init__(self, api, *args, **kwargs):
            self.api = api
            self.as_generator = kwargs.pop("as_generator", False)
            if self.as_generator:
                self.pagination_format = 'next_url'
            else:
                self.pagination_format = kwargs.pop('pagination_format', 'next_url')
            self.return_json = kwargs.pop("return_json", False)
            self.max_pages = kwargs.pop("max_pages", 3)
            self.with_next_url = kwargs.pop("with_next_url", None)
            self.parameters = {}
            self._build_parameters(args, kwargs)
            self._build_path()
        def _build_parameters(self, args, kwargs):
            # via tweepy https://github.com/joshthecoder/tweepy/
            for index, value in enumerate(args):
                if value is None:
                    continue
                    self.parameters[self.accepts_parameters[index]] = encode_string(value)
                except IndexError:
                    raise InstagramClientError("Too many arguments supplied")
            for key, value in list(kwargs.items()):
                if value is None:
                    continue
                if key in self.parameters:
                    raise InstagramClientError("Parameter %s already supplied" % key)
                self.parameters[key] = encode_string(value)
            if 'user_id' in self.accepts_parameters and not 'user_id' in self.parameters \
               and not self.requires_target_user:
                self.parameters['user_id'] = 'self'
        def _build_path(self):
            for variable in re_path_template.findall(self.path):
                name = variable.strip('{}')
                    value = urllib.parse.quote(self.parameters[name])
                except KeyError:
                    raise Exception('No parameter value found for path variable: %s' % name)
                del self.parameters[name]
                self.path = self.path.replace(variable, value)
            self.path = self.path + '.%s' % self.api.format
        def _build_pagination_info(self, content_obj):
            """Extract pagination information in the desired format."""
            pagination = content_obj.get('pagination', {})
            if self.pagination_format == 'next_url':
                return pagination.get('next_url')
            if self.pagination_format == 'dict':
                return pagination
            raise Exception('Invalid value for pagination_format: %s' % self.pagination_format)
        def _do_api_request(self, url, method="GET", body=None, headers=None):
            headers = headers or {}
            if self.signature and self.api.client_ips != None and self.api.client_secret != None:
                secret = self.api.client_secret
                ips = self.api.client_ips
                signature = hmac.new(secret, ips, sha256).hexdigest()
                headers['X-Insta-Forwarded-For'] = '|'.join([ips, signature])
            response, content = OAuth2Request(self.api).make_request(url, method=method, body=body, headers=headers)
            if response['status'] == '503' or response['status'] == '429':
                raise InstagramAPIError(response['status'], "Rate limited", "Your client is making too many request per second")
            try:
                content_obj = simplejson.loads(content)
            except ValueError:
                raise InstagramClientError('Unable to parse response, not valid JSON.', status_code=response['status'])
            # Handle OAuthRateLimitExceeded from Instagram's Nginx which uses different format to documented api responses
            if 'meta' not in content_obj:
                if content_obj.get('code') == 420 or content_obj.get('code') == 429:
                    error_message = content_obj.get('error_message') or "Your client is making too many request per second"
                    raise InstagramAPIError(content_obj.get('code'), "Rate limited", error_message)
                raise InstagramAPIError('code' in content_obj, 'error_type' in content_obj, 'error_message' in content_obj)
            api_responses = []
            status_code = content_obj['meta']['code']
            self.api.x_ratelimit_remaining = response.get("x-ratelimit-remaining",None)
            self.api.x_ratelimit = response.get("x-ratelimit-limit",None)
            if status_code == 200:
                if not self.objectify_response:
                    return content_obj, None
                if self.response_type == 'list':
                    for entry in content_obj['data']:
                        if self.return_json:
                            api_responses.append(entry)
                            obj = self.root_class.object_from_dictionary(entry)
                            api_responses.append(obj)
                elif self.response_type == 'entry':
                    data = content_obj['data']
                    if self.return_json:
                        api_responses = data
                        api_responses = self.root_class.object_from_dictionary(data)
                elif self.response_type == 'empty':
                return api_responses, self._build_pagination_info(content_obj)
            else:
                raise InstagramAPIError(status_code, content_obj['meta']['error_type'], content_obj['meta']['error_message'])
        def _paginator_with_url(self, url, method="GET", body=None, headers=None):
            headers = headers or {}
            pages_read = 0
            while url and pages_read < self.max_pages:
                api_responses, url = self._do_api_request(url, method, body, headers)
                pages_read += 1
                yield api_responses, url
            return
        def _get_with_next_url(self, url, method="GET", body=None, headers=None):
            headers = headers or {}
            content, next = self._do_api_request(url, method, body, headers)
            return content, next
        def execute(self):
            url, method, body, headers = OAuth2Request(self.api).prepare_request(self.method,
                                                                                 include_secret=self.include_secret)
            if self.with_next_url:
                return self._get_with_next_url(self.with_next_url, method, body, headers)
            if self.as_generator:
                return self._paginator_with_url(url, method, body, headers)
            else:
                content, next = self._do_api_request(url, method, body, headers)
            if self.paginates:
                return content, next
            else:
                return content
    def _call(api, *args, **kwargs):
        method = InstagramAPIMethod(api, *args, **kwargs)
        return method.execute()
    return _call
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

FilesExpand file tree

bind.py

Latest commit

History

bind.py

File metadata and controls
