jseabold · December 13, 2011 18:58 · mouuff · Dec 16, 2013 · not7cd · Jul 25, 2017
diff --git a/translate.py b/translate.py
 # -*- coding: utf-8 -*-
 """
 You need to fill in your API key from google below. Note that querying
 supported languages is not implemented.

 Language Code
 -------- ----
 Afrikaans 	af
 Albanian 	sq
 Arabic 	ar
 Belarusian 	be
 Bulgarian 	bg
 Catalan 	ca
 Chinese Simplified 	zh-CN
 Chinese Traditional 	zh-TW
 Croatian 	hr
 Czech 	cs
 Danish 	da
 Dutch 	nl
 English 	en
 Estonian 	et
 Filipino 	tl
 Finnish 	fi
 French 	fr
 Galician 	gl
 German 	de
 Greek 	el
 Hebrew 	iw
 Hindi 	hi
 Hungarian 	hu
 Icelandic 	is
 Indonesian 	id
 Irish 	ga
 Italian 	it
 Japanese 	ja
 Korean 	ko
 Latvian 	lv
 Lithuanian 	lt
 Macedonian 	mk
 Malay 	ms
 Maltese 	mt
 Norwegian 	no
 Persian 	fa
 Polish 	pl
 Portuguese 	pt
 Romanian 	ro
 Russian 	ru
 Serbian 	sr
 Slovak 	sk
 Slovenian 	sl
 Spanish 	es
 Swahili 	sw
 Swedish 	sv
 Thai 	th
 Turkish 	tr
 Ukrainian 	uk
 Vietnamese 	vi
 Welsh 	cy
 Yiddish 	yi
 """
 import os
 import urlparse
 import urllib
 import urllib2
 import httplib2
 import gzip
 import json
 from httplib2 import FileCache
 from urllib2 import HTTPRedirectHandler, HTTPDefaultErrorHandler, HTTPError

 ### Hard-coded variables ###

 api = 'YOUR-API-KEY-GOES-HERE'

 languages = ["af", "sq", "ar","be", "bg", "ca", "zh-CN", "zh-TW", "hr",
             "cs", "da", "nl", "en", "et", "tl", "fi", "fr", "gl", "de",
             "el", "iw", "hi", "hu", "is", "id", "ga", "it", "ja", "ko",
             "lv", "lt", "mk", "ms", "mt", "no", "fa", "pl", "pt", "ro",
             "ru", "sr", "sk", "sl", "es", "sw", "sv", "th", "tr", "uk",
             "vi", "cy", "yi"]

 def _validate_language(lang):
    if lang in languages:
        return True
    return False

 ### Custom G-Zipped Cache ###

 def save_cached_key(path, value):
    f = gzip.open(path, 'wb')
    f.write(value)
    f.close()

 def load_cached_key(key):
    f = gzip.open(key)
    retval = f.read()
    f.close()
    return retval

 class ZipCache(FileCache):
    def __init__(self, cache='.cache'): #TODO: allow user configurable?
        super(ZipCache, self).__init__(cache)

    def get(self, key):
        cacheFullPath = os.path.join(self.cache, self.safe(key))
        retval = None
        try:
            retval = load_cached_key(cacheFullPath)
        except IOError:
            pass
        return retval

    def set(self, key, value):
        retval = None
        cacheFullPath = os.path.join(self.cache, self.safe(key))
        save_cached_key(cacheFullPath, value)

 ### Error Handlers ###

 class DefaultErrorHandler(HTTPDefaultErrorHandler):
    def http_error_default(self, req, fp, code, msg, headers):
        result = HTTPError(req.get_full_url(), code, msg, headers, fp)
        result.status = code
        return result


 class RedirectHandler(HTTPRedirectHandler):
    def http_error_301(self, req, fp, code, msg, headers):
        result = HTTPRedirectHandler.http_error_301(self, req, fp, code,
                        msg, headers)
        result.status = code
        return result

    def http_error_302(self, req, fp, code, msg, headers):
        results = HTTPRedirectHandler.http_error_302(self, req, fp, code,
                        msg, headers)
        results.status = code
        return result

 ### Translator Class ###

 class GoogleTranslator(object):
    """
    Google Translator object.

    Examples
    --------
    translator = GoogleTranslator()

    results1 = translator.translate("Einen schönen Tag allerseits")

    # try 2 at a time
    results2 = translator.translate(["Einen schönen Tag allerseits",
                                     "Ich nehme an"])

    # try detect
    results3 = translator.detect("Einen schönen Tag allerseits")

    # try to detect 2 at a time
    results4 = translator.detect(["Einen schönen Tag allerseits",
                                     "Ich nehme an"])
    """
    def __init__(self):
        #NOTE: caching is done on etag not expiry
        self.cache_control = 'max-age='+str(7 * 24 * 60 * 60)
        self.connection = httplib2.Http(ZipCache())
        self._opener = urllib2.build_opener(DefaultErrorHandler,
                                            RedirectHandler)
        self.base_url = "https://www.googleapis.com/language/translate/v2/"

    def _urlencode(self, params):
        """
        Rewrite urllib.urlencode to handle string input verbatim
        """
        params = "&".join(map("=".join,params))
        return params

    def _build_uri(self, extra_url, params):
        params = [('key', api)] + params
        params = self._urlencode(params)
        url = "%s?%s" % (urlparse.urljoin(self.base_url, extra_url), params)
        if len(url) > 2000: # for GET requests only, POST is 5K
            raise ValueError("Query is too long. URL can only be 2000 "
                             "characters")
        return url

    def _fetch_data(self, url):
        connection = self.connection
        resp, content = connection.request(url, headers={'user-agent' : api,
                            'cache-control' : self.cache_control})
        #DEBUG
        #if resp.fromcache:
        #   print "Using from the cache"
        return content

    def _sanitize_query(self, query):
        if isinstance(query, (list,tuple)):
            query = zip('q' * len(query), map(urllib.quote,query))
        else:
            query = [('q',urllib.quote(query))]
        return query

    def _decode_json(self, response):
        """
        Assumes that response only holds one result
        """
        json_data = json.loads(response)
        try:
            data = json_data["data"]
            if 'translations' in data:
                return data['translations']
            elif 'detections' in data:
                return data['detections']
        except:
            if 'error' in json_data:
                return json_data["error"]


    def detect(self, query):
        """
        Try to detect the language of a word, phrase, or list of either.

        Parameters
        ----------
        query : str or iterable
            Query or list of queries to translate

        Returns
        -------
        List of dictionaries for each query
        """
        query = self._sanitize_query(query)
        url = self._build_uri(extra_url='detect/', params=query)
        content = self._fetch_data(url)
        # going to have json, decode it first
        return self._decode_json(content)

    def translate(self, query, target="en", source="", _dirty=False):
        """
        Translate a query.

        Parameters
        ----------
        query : str or iterable
            Query or list of queries to translate
        target : str
            Language to translate into.
        source : str, optional
            Language of the source text, if known. Will be auto-detected
            if an empty string is passed.
        dirty : bool
            This is not intended to be used by users. It is here to avoid
            infinite recursion if the query returns an error because the
            language can't be detected.

        Returns
        -------
        List of dictionaries for each query

        Notes
        -----
        If the language can't be detected for a word an attempt is made
        to detect the language of the word and resubmit the query. If a
        list of words to translate is given and an error is encountered,
        it is assumed that the list of words all have the same source language
        when resubmitted.
        """
        try:
            assert _validate_language(target)
        except:
            raise ValueError("target language %s is not valid" % target)
        newquery = self._sanitize_query(query)
        params = [('key', api), ('target' , target)]
        if source:
            try:
                assert _validate_language(target)
            except:
                raise ValueError("source language %s is not valid" % target)
            params += ["source", source]
        params += newquery
        url = self._build_uri("", params)
        content = self._fetch_data(url)
        results = self._decode_json(content)

        if "errors" in results and not _dirty:
            if results['message'] == 'Bad language pair: {0}':
                # try to detect language and resubmit query
                source = self.detect(query)
                source = source[0]['language']
                return self.translate(query, target, source, True)

        return results

 if __name__ == "__main__":
    do_run = raw_input("Running this costs money. Key 1 to continue: ")
    if int(do_run) != 1:
        raise ValueError("Exiting")
    translator = GoogleTranslator()

    results1 = translator.translate("Einen schönen Tag allerseits")

    # try 2 at a time
    #results2 = translator.translate(["Einen schönen Tag allerseits",
    #                                 "Ich nehme an"])

    # try detect
    #results3 = translator.detect("Einen schönen Tag allerseits")

    # try 2 at a time
    #results4 = translator.detect(["Einen schönen Tag allerseits",
    #                                 "Ich nehme an"])
	# -- coding: utf-8 --
	"""
	You need to fill in your API key from google below. Note that querying
	supported languages is not implemented.

	Language Code
	-------- ----
	Afrikaans af
	Albanian sq
	Arabic ar
	Belarusian be
	Bulgarian bg
	Catalan ca
	Chinese Simplified zh-CN
	Chinese Traditional zh-TW
	Croatian hr
	Czech cs
	Danish da
	Dutch nl
	English en
	Estonian et
	Filipino tl
	Finnish fi
	French fr
	Galician gl
	German de
	Greek el
	Hebrew iw
	Hindi hi
	Hungarian hu
	Icelandic is
	Indonesian id
	Irish ga
	Italian it
	Japanese ja
	Korean ko
	Latvian lv
	Lithuanian lt
	Macedonian mk
	Malay ms
	Maltese mt
	Norwegian no
	Persian fa
	Polish pl
	Portuguese pt
	Romanian ro
	Russian ru
	Serbian sr
	Slovak sk
	Slovenian sl
	Spanish es
	Swahili sw
	Swedish sv
	Thai th
	Turkish tr
	Ukrainian uk
	Vietnamese vi
	Welsh cy
	Yiddish yi
	"""
	import os
	import urlparse
	import urllib
	import urllib2
	import httplib2
	import gzip
	import json
	from httplib2 import FileCache
	from urllib2 import HTTPRedirectHandler, HTTPDefaultErrorHandler, HTTPError

	### Hard-coded variables ###

	api = 'YOUR-API-KEY-GOES-HERE'

	languages = ["af", "sq", "ar","be", "bg", "ca", "zh-CN", "zh-TW", "hr",
	"cs", "da", "nl", "en", "et", "tl", "fi", "fr", "gl", "de",
	"el", "iw", "hi", "hu", "is", "id", "ga", "it", "ja", "ko",
	"lv", "lt", "mk", "ms", "mt", "no", "fa", "pl", "pt", "ro",
	"ru", "sr", "sk", "sl", "es", "sw", "sv", "th", "tr", "uk",
	"vi", "cy", "yi"]

	def _validate_language(lang):
	if lang in languages:
	return True
	return False

	### Custom G-Zipped Cache ###

	def save_cached_key(path, value):
	f = gzip.open(path, 'wb')
	f.write(value)
	f.close()

	def load_cached_key(key):
	f = gzip.open(key)
	retval = f.read()
	f.close()
	return retval

	class ZipCache(FileCache):
	def __init__(self, cache='.cache'): #TODO: allow user configurable?
	super(ZipCache, self).__init__(cache)

	def get(self, key):
	cacheFullPath = os.path.join(self.cache, self.safe(key))
	retval = None
	try:
	retval = load_cached_key(cacheFullPath)
	except IOError:
	pass
	return retval

	def set(self, key, value):
	retval = None
	cacheFullPath = os.path.join(self.cache, self.safe(key))
	save_cached_key(cacheFullPath, value)

	### Error Handlers ###

	class DefaultErrorHandler(HTTPDefaultErrorHandler):
	def http_error_default(self, req, fp, code, msg, headers):
	result = HTTPError(req.get_full_url(), code, msg, headers, fp)
	result.status = code
	return result


	class RedirectHandler(HTTPRedirectHandler):
	def http_error_301(self, req, fp, code, msg, headers):
	result = HTTPRedirectHandler.http_error_301(self, req, fp, code,
	msg, headers)
	result.status = code
	return result

	def http_error_302(self, req, fp, code, msg, headers):
	results = HTTPRedirectHandler.http_error_302(self, req, fp, code,
	msg, headers)
	results.status = code
	return result

	### Translator Class ###

	class GoogleTranslator(object):
	"""
	Google Translator object.

	Examples
	--------
	translator = GoogleTranslator()

	results1 = translator.translate("Einen schönen Tag allerseits")

	# try 2 at a time
	results2 = translator.translate(["Einen schönen Tag allerseits",
	"Ich nehme an"])

	# try detect
	results3 = translator.detect("Einen schönen Tag allerseits")

	# try to detect 2 at a time
	results4 = translator.detect(["Einen schönen Tag allerseits",
	"Ich nehme an"])
	"""
	def __init__(self):
	#NOTE: caching is done on etag not expiry
	self.cache_control = 'max-age='+str(7 * 24 * 60 * 60)
	self.connection = httplib2.Http(ZipCache())
	self._opener = urllib2.build_opener(DefaultErrorHandler,
	RedirectHandler)
	self.base_url = "https://www.googleapis.com/language/translate/v2/"

	def _urlencode(self, params):
	"""
	Rewrite urllib.urlencode to handle string input verbatim
	"""
	params = "&".join(map("=".join,params))
	return params

	def _build_uri(self, extra_url, params):
	params = [('key', api)] + params
	params = self._urlencode(params)
	url = "%s?%s" % (urlparse.urljoin(self.base_url, extra_url), params)
	if len(url) > 2000: # for GET requests only, POST is 5K
	raise ValueError("Query is too long. URL can only be 2000 "
	"characters")
	return url

	def _fetch_data(self, url):
	connection = self.connection
	resp, content = connection.request(url, headers={'user-agent' : api,
	'cache-control' : self.cache_control})
	#DEBUG
	#if resp.fromcache:
	# print "Using from the cache"
	return content

	def _sanitize_query(self, query):
	if isinstance(query, (list,tuple)):
	query = zip('q' * len(query), map(urllib.quote,query))
	else:
	query = [('q',urllib.quote(query))]
	return query

	def _decode_json(self, response):
	"""
	Assumes that response only holds one result
	"""
	json_data = json.loads(response)
	try:
	data = json_data["data"]
	if 'translations' in data:
	return data['translations']
	elif 'detections' in data:
	return data['detections']
	except:
	if 'error' in json_data:
	return json_data["error"]


	def detect(self, query):
	"""
	Try to detect the language of a word, phrase, or list of either.

	Parameters
	----------
	query : str or iterable
	Query or list of queries to translate

	Returns
	-------
	List of dictionaries for each query
	"""
	query = self._sanitize_query(query)
	url = self._build_uri(extra_url='detect/', params=query)
	content = self._fetch_data(url)
	# going to have json, decode it first
	return self._decode_json(content)

	def translate(self, query, target="en", source="", _dirty=False):
	"""
	Translate a query.

	Parameters
	----------
	query : str or iterable
	Query or list of queries to translate
	target : str
	Language to translate into.
	source : str, optional
	Language of the source text, if known. Will be auto-detected
	if an empty string is passed.
	dirty : bool
	This is not intended to be used by users. It is here to avoid
	infinite recursion if the query returns an error because the
	language can't be detected.

	Returns
	-------
	List of dictionaries for each query

	Notes
	-----
	If the language can't be detected for a word an attempt is made
	to detect the language of the word and resubmit the query. If a
	list of words to translate is given and an error is encountered,
	it is assumed that the list of words all have the same source language
	when resubmitted.
	"""
	try:
	assert _validate_language(target)
	except:
	raise ValueError("target language %s is not valid" % target)
	newquery = self._sanitize_query(query)
	params = [('key', api), ('target' , target)]
	if source:
	try:
	assert _validate_language(target)
	except:
	raise ValueError("source language %s is not valid" % target)
	params += ["source", source]
	params += newquery
	url = self._build_uri("", params)
	content = self._fetch_data(url)
	results = self._decode_json(content)

	if "errors" in results and not _dirty:
	if results['message'] == 'Bad language pair: {0}':
	# try to detect language and resubmit query
	source = self.detect(query)
	source = source[0]['language']
	return self.translate(query, target, source, True)

	return results

	if __name__ == "__main__":
	do_run = raw_input("Running this costs money. Key 1 to continue: ")
	if int(do_run) != 1:
	raise ValueError("Exiting")
	translator = GoogleTranslator()

	results1 = translator.translate("Einen schönen Tag allerseits")

	# try 2 at a time
	#results2 = translator.translate(["Einen schönen Tag allerseits",
	# "Ich nehme an"])

	# try detect
	#results3 = translator.detect("Einen schönen Tag allerseits")

	# try 2 at a time
	#results4 = translator.detect(["Einen schönen Tag allerseits",
	# "Ich nehme an"])