buildbot-transifex/buildbot_transifex/webhook.py

import sys

import os
import base64

import time
import json
import re
import hmac
import pprint
import hashlib
import requests
from subprocess import call
from base64 import b64encode

from buildbot.process.properties import Properties
from buildbot.util import bytes2unicode, unicode2bytes
from buildbot.www.hooks.base import BaseHookHandler

from twisted.internet import defer
from twisted.python import log

from dateutil.parser import parse as dateparse

HTTP_USER_AGENT = 'User-Agent'
_HEADER_SIGNATURE = 'X-TX-Signature'
_EVENT_KEY = 'event'
branch = 'v0.6'
transifexSecret = ""
transifexUsername = ""
transifexPassword = ""
transifex_webhook_url = 'https://buildbot.bitmessage.org/change_hook/transifex'
transifex_dict = {}
secret = ""
master = ""

from transifex.api import transifex_api

transifex_api.setup(auth='my_token')
org = transifex_api.Organization.get(slug='my_org')
proj = org.fetch('projects').get(slug='my_project')
lang = transifex_api.Language.get(code='<lang>')
resource = proj.fetch('resources').get(slug='my_resource')
url = transifex_api.ResourceTranslationsAsyncDownload.download(
    resource=resource, language=lang,content_encoding = 'text',
    file_type = 'default', pseudo = False
)

class TransifexHandler(BaseHookHandler):

    def __init__(self, transifex_dict, secret, master,):
        self.secret = secret
        self.master = master
        self.transifex_dict = transifex_dict

    def verifyTransifexSignature(
        self, request, content, rendered_secret, signature, header_signature
        ):
        http_verb = 'POST'
        http_url_path = request.headers('X-TX-Url')
        http_gmt_date = request.headers('Date')
        content_md5 = hashlib.md5(content).hexdigest()
        msg = b'\n'.join([
            http_verb, http_url_path, http_gmt_date, content_md5
        ])
        tx_signature = base64.b64encode(
            hmac.new(
                key=rendered_secret,
                msg=msg,
                digestmod=hashlib.sha256
            ).digest()
        )
        if tx_signature() != header_signature:
            raise ValueError('Invalid secret')
    
        try:
            if signature != os.environ.get('HTTP_X_TX_SIGNATURE'):
                return False
            return True
        except:
            return False

    def updateLocalTranslationDestination(self, ts, lang, branch):
        call(["git", "pull", "--all", "-q"])
        call(["git", "stash", "-q"])
        call(["git", "checkout", "-q", branch])
        call(["git", "checkout", "-q", "-b", "translate_" + lang + "_" + str(ts)])
        call(["git", "branch", "-q", "--set-upstream-to=origin/v0.6"])

    def downloadTranslatedLanguage(ts, lang):
        headers = {"Authorization": "Basic " + b64encode(transifexUsername + ":" + transifexPassword)}
        fname = "bitmessage_" + lang.lower() + ".po"
        with open("src/translations/" + fname, "wt") as handle:
            response = requests.get("https://www.transifex.com/api/2/project/pybitmessage/resource/pybitmessage/translation/" + lang + "/",
                headers=headers)
            if response.ok:
                content = json.loads(response.content)["content"]
                handle.write(content.encode("utf-8"))
        return response

    def process_translation_completed(self, transifex_dict, event_type, codebase):
        # refname = payload["ref"]
        payload = transifex_dict
        changes = []
        # We only care about regular heads or tags
        # match = re.match(r"^refs/(heads|tags)/(.+)$", refname)
        if "Transifex" in os.environ.get("HTTP_USER_AGENT"):
            if not self.verifyTransifexSignature(
                request, content, rendered_secret, signature, header_signature):
                print('Invalid transifex signature!')
            else:
                payload = self.transifex_dict
                if 'pybitmessage-test' in payload['transifex_data']['project'] and 'messagespot' in payload['transifex_data']['translation_completed']['resource']:
                    if 'translation_completed' in payload['transifex_data'] and '100' in payload['transifex_data']['translated']:
                        ts = int(time.time())
                        lang = payload['transifex_data']['translation_completed']['language']
                        branch = payload['transifex_data']['branch']
                        self.updateLocalTranslationDestination(ts, lang.lower(), branch)
                        self.downloadTranslatedLanguage(ts, lang.lower())


        repository = payload['repository']
        repo_url = repository['ssh_url']
        project = repository['full_name']

        commits = payload['commits']
        # if isinstance(self.options, dict) and self.options.get('onlyIncludePushCommit', False):
        #     commits = commits[:1]

        # for commit in commits:
        #     files = []
        #     for kind in ('added', 'modified', 'removed'):
        #         files.extend(commit.get(kind, []) or [])
        #     timestamp = dateparse(commit['timestamp'])
        #     change = {
        #         'author': '{} <{}>'.format(commit['author']['name'],
        #                                    commit['author']['email']),
        #         'files': files,
        #         'comments': commit['message'],
        #         'revision': commit['id'],
        #         'when_timestamp': timestamp,
        #         'branch': branch,
        #         'revlink': commit['url'],
        #         'repository': repo_url,
        #         'project': project,
        #         'category': event_type,
        #         'properties': {
        #             'event': event_type,
        #             'repository_name': repository['name'],
        #             'owner': repository["owner"]["username"]
        #         },
        #     }
        #     if codebase is not None:
        #         change['codebase'] = codebase
        #     changes.insert(0, change)
        # return changes

    def process_review_completed(self, payload, transifex_data):
        pass


    def _transform_variables(self, payload=transifex_dict):
        retval = {
            'project': payload[''].get('project'),
            'repository': [payload.get('resource')],
            'branch': payload.get('language')
        }
        return retval

    @defer.inlineCallbacks
    def getChanges(self, request):
        self.secret = None
        if isinstance(self.options, dict):
            self.secret = self.options.get('secret')
        try:
            content = request.content.read()
            content_text = bytes2unicode(content)
            payload = json.loads(content_text)
        except Exception as exception:
            raise ValueError('Error loading JSON: ' + str(exception))

        if self.secret is not None:
            p = Properties()
            p.master = self.master
            rendered_secret = yield p.render(self.secret)
            signature = hmac.new(
                unicode2bytes(rendered_secret),
                unicode2bytes(content_text.strip()),
                digestmod=hashlib.sha256)
            header_signature = bytes2unicode(
                request.getHeader(_HEADER_SIGNATURE))
        event_type = bytes2unicode(payload.get(_EVENT_KEY), "None")
        log.msg("Received event '{}' from transifex".format(event_type))

        codebase = ""
        changes = []

        handler_function = getattr(self, 'process_{}'.format(event_type), None)
        if not handler_function:
            log.msg("Ignoring transifex event '{}'".format(event_type))
        else:
            changes = handler_function(payload, event_type, codebase)

        return (changes, 'transifex')


transifex = TransifexHandler(transifex_dict)