shekhar-cis/buildbot-transifex
1
0
Fork 0
forked from Bitmessage/buildbot-transifex
Code Issues Pull Requests Packages Projects Releases Wiki Activity

WIP[downloadTranslatedLanguage() and updateLocalTranslationDestination()]
All checks were successful
buildbot/multibuild_parent Build done.
Details
buildbot/travis_bionic Build done.
Details

Browse Source
This commit is contained in:
shekhar-cis 2023-01-04 20:51:06 +05:30
parent f2587e071d
commit 1b29a5639c
Signed by: shekhar-cis
GPG Key ID: F4F00AB04E83F9A7
1 changed files with 147 additions and 194 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

341
buildbot_transifex/webhook.py
View File

@ -1,8 +1,19 @@
import sys
import os
import base64 import base64
import time
import json import json
import re import re
import hmac import hmac
import pprint
import hashlib import hashlib
import requests
from subprocess import call
from base64 import b64encode
from buildbot.process.properties import Properties from buildbot.process.properties import Properties
from buildbot.util import bytes2unicode, unicode2bytes from buildbot.util import bytes2unicode, unicode2bytes
from buildbot.www.hooks.base import BaseHookHandler from buildbot.www.hooks.base import BaseHookHandler
@ -15,201 +26,157 @@ from dateutil.parser import parse as dateparse
_HEADER_USER_AGENT = 'User-Agent' _HEADER_USER_AGENT = 'User-Agent'
_HEADER_SIGNATURE = 'X-TX-Signature' _HEADER_SIGNATURE = 'X-TX-Signature'
_EVENT_KEY = 'event' _EVENT_KEY = 'event'
transifexSecret = ""
transifexUsername = ""
transifexPassword = ""
transifex_dict = {}
secret = ""
master = ""
gitHubToken = os.environ('gitHubToken')
class TransifexHandler(BaseHookHandler): class TransifexHandler(BaseHookHandler):
# def verifyGitHubSignature (environ, payload_body): def __init__(self, master, secret, transifex_dict):
# signature = 'sha1=' + hmac.new(gitHubSecret, payload_body, sha1).hexdigest() self.secret = secret
# try: self.master = master
# if signature != environ.get('X-TX-Signature'): self.transifex_dict = transifex_dict
# return False
# return True
# except:
# return False
# def verifyTransifexSignature (environ, payload_body):
# signature = b64encode(hmac.new(transifexSecret, payload_body, sha1).digest())
# try:
# debug(signature)
# if signature != environ.get('HTTP_X_TX_SIGNATURE'):
# return False
# return True
# except:
# return False
# def returnMessage(status = False, message = "Unimplemented"): def returnMessage(self, status = False, message = "Unimplemented"):
# output = json.dumps({"status": "OK" if status else "FAIL", "message": message}) output = json.dumps({"status": "OK" if status else "FAIL", "message": message})
# return [output, [('Content-type', 'text/plain'), return [output, [('Content-type', 'text/plain'),
# ('Content-Length', str(len(output))) ('Content-Length', str(len(output)))
# ]] ]]
# def application(environ, start_response): def verifyTransifexSignature(
# status = '200 OK' self, request, content, rendered_secret, signature, header_signature
# output = '' ):
# lockWait() http_verb = 'POST'
# length = int(environ.get('CONTENT_LENGTH', '0')) http_url_path = request.headers('X-TX-Url')
# body = environ['wsgi.input'].read(length) 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')
# if "Transifex" in environ.get("HTTP_USER_AGENT"): try:
# # debug(environ) if signature != os.environ.get('HTTP_X_TX_SIGNATURE'):
# # debug(body) return False
# if not verifyTransifexSignature(environ, body): return True
# debug ("Verify Transifex Signature fail, but fuck them") except:
# else: return False
# debug ("Verify Transifex Signature ok")
# # output, responseHeaders = returnMessage(False, "Checksum bad")
# # start_response(status, responseHeaders)
# # unlock()
# # return [output]
# try:
# # debug(body)
# payload = parse_qs(body)
# # debug(payload)
# if 'pybitmessage' in payload['project'] and 'pybitmessage' in payload['resource']:
# if 'translated' in payload and '100' in payload['translated']:
# ts = int(time.time())
# updateLocalTranslationDestination(ts, payload['language'][0].lower())
# downloadTranslatedLanguage(ts, payload['language'][0])
# response = commitTranslatedLanguage(ts, payload['language'][0].lower())
# if response.ok:
# output, responseHeaders = returnMessage(True, "Processed.")
# else:
# output, responseHeaders = returnMessage(False, "Error: %i." % (response.status_code))
# else:
# output, responseHeaders = returnMessage(False, "Nothing to do")
# else:
# output, responseHeaders = returnMessage(False, "Nothing to do")
# except:
# output, responseHeaders = returnMessage(True, "Not processing")
# else:
# debug("Unknown command %s" % (environ.get("HTTP_X_GITHUB_EVENT")))
# output, responseHeaders = returnMessage(True, "Unknown command, ignoring")
def __init__(self, transifex_dict=None):
super(TransifexHandler, self).__init__(*args, **kwargs)
def process_translation_completed(self, payload, event_type, codebase): def downloadTranslatedLanguage(self, ts, lang):
refname = payload["ref"] 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 commitTranslatedLanguage(self, ts, lang):
call(["kivy", "src/translations/messages.pro"])
call(["git", "add", "src/translations/bitmessage_" + lang + ".ts", "src/translations/bitmessage_" + lang + ".qm"])
call(["git", "commit", "-q", "-S", "-m", "Auto-updated language %s from transifex" % (lang)])
newbranch = "translate_" + lang + "_" + str(ts)
call(["git", "push", "-q", "translations", newbranch + ":" + newbranch])
branch = transifex_dict['branch']
request = {
"title": "Translation update " + lang,
"body": "Auto-updated from transifex",
"head": "PyBitmessageTranslations:" + newbranch,
"base": branch
}
headers = {"Authorization": "token " + gitHubToken}
response = requests.post("https://api.github.com/repos/Bitmessage/PyBitmessage/pulls",
headers=headers, data=json.dumps(request))
return response
def process_translation_completed(self, payload, transifex_dict, event_type, codebase):
changes = [] changes = []
transifex_response = self._transform_variables(payload, transifex_dict)
if 'pybitmessage-test' in transifex_response['project'] and 'messagespot' in transifex_response['resource']:
if 'translation_completed' in transifex_response['event'] and 100 in transifex_response['translated']:
ts = int(time.time())
lang = transifex_response['language']
branch = transifex_dict['branch']
self.downloadTranslatedLanguage(ts, lang.lower())
response = self.commitTranslatedLanguage(ts, lang.lower())
if response.ok:
output, responseHeaders = self.returnMessage(True, "Processed.")
else:
output, responseHeaders = self.returnMessage(False, "Error: %i." % (response.status_code))
else:
output, responseHeaders = self.returnMessage(False, "Nothing to do")
else:
output, responseHeaders = self.returnMessage(False, "Nothing to do")
# We only care about regular heads or tags # if isinstance(self.options, dict) and self.options.get('onlyIncludePushCommit', False):
match = re.match(r"^refs/(heads|tags)/(.+)$", refname) # commits = commits[:1]
if event_type == 'translation_completed':
pass
if not match: # for commit in commits:
log.msg("Ignoring refname '{}': Not a branch or tag".format(refname)) # files = []
return changes # for kind in ('added', 'modified', 'removed'):
# files.extend(commit.get(kind, []) or [])
branch = match.group(2) # timestamp = dateparse(commit['timestamp'])
# change = {
repository = payload['repository'] # 'author': '{} <{}>'.format(commit['author']['name'],
repo_url = repository['ssh_url'] # commit['author']['email']),
project = repository['full_name'] # 'files': files,
# 'comments': commit['message'],
commits = payload['commits'] # 'revision': commit['id'],
if isinstance(self.options, dict) and self.options.get('onlyIncludePushCommit', False): # 'when_timestamp': timestamp,
commits = commits[:1] # 'branch': branch,
# 'revlink': commit['url'],
for commit in commits: # 'repository': repo_url,
files = [] # 'project': project,
for kind in ('added', 'modified', 'removed'): # 'category': event_type,
files.extend(commit.get(kind, []) or []) # 'properties': {
timestamp = dateparse(commit['timestamp']) # 'event': event_type,
change = { # 'repository_name': repository['name'],
'author': '{} <{}>'.format(commit['author']['name'], # 'owner': repository["owner"]["username"]
commit['author']['email']), # },
'files': files, # }
'comments': commit['message'], # if codebase is not None:
'revision': commit['id'], # change['codebase'] = codebase
'when_timestamp': timestamp, # changes.insert(0, change)
'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 return changes
def process_review_completed(self, payload, event_type, codebase): def process_review_completed(self, payload, transifex_data):
action = payload['action'] pass
if event_type == 'review_completed':
pass
# Only handle potential new stuff, ignore close/.
# Merge itself is handled by the regular branch push message
if action not in ['opened', 'synchronized', 'edited', 'reopened']:
log.msg("Transifex Pull Request event '{}' ignored".format(action))
return []
# pull_request = payload['pull_request']
# if not pull_request['mergeable']:
# log.msg("Transifex Pull Request ignored because it is not mergeable.")
# return []
# if pull_request['merged']:
# log.msg("Transifex Pull Request ignored because it is already merged.")
# return []
# timestamp = dateparse(pull_request['updated_at'])
# base = pull_request['base']
# head = pull_request['head']
repository = payload['repository']
change = {
'author': '{} <{}>'.format(pull_request['user']['full_name'],
pull_request['user']['email']),
# 'comments': 'PR#{}: {}\n\n{}'.format(
# pull_request['number'],
# pull_request['title'],
# pull_request['body']),
'revision': base['sha'],
'when_timestamp': timestamp,
'branch': base['ref'],
# 'revlink': pull_request['html_url'],
'repository': base['repo']['ssh_url'],
'project': repository['full_name'],
'category': event_type,
'properties': {
'event': event_type,
'base_branch': base['ref'],
'base_sha': base['sha'],
'base_repo_id': base['repo_id'],
'base_repository': base['repo']['clone_url'],
'base_git_ssh_url': base['repo']['ssh_url'],
'head_branch': head['ref'],
'head_sha': head['sha'],
'head_repo_id': head['repo_id'],
'head_repository': head['repo']['clone_url'],
'head_git_ssh_url': head['repo']['ssh_url'],
'head_owner': head['repo']['owner']['username'],
'head_reponame': head['repo']['name'],
# 'pr_id': pull_request['id'],
# 'pr_number': pull_request['number'],
'repository_name': repository['name'],
'owner': repository["owner"]["username"],
},
}
if codebase is not None:
change['codebase'] = codebase
return [change]
def _transform_variables(payload):
retval = { def _transform_variables(self, payload, transifex_dict):
project: payload.get('project'), transifex_variables = {
repository = [payload.get('resource')], 'project': payload['project'],
branch = payload.get('language') "translated": payload['translated'],
"resource": payload['resource'],
"event": payload['event'],
"language": payload['language']
} }
return retval
return transifex_variables
@defer.inlineCallbacks @defer.inlineCallbacks
def getChanges(self, request): def getChanges(self, request):
secret = None self.secret = None
if isinstance(self.options, dict): if isinstance(self.options, dict):
secret = self.options.get('secret') self.secret = self.options.get('secret')
try: try:
content = request.content.read() content = request.content.read()
content_text = bytes2unicode(content) content_text = bytes2unicode(content)
@ -217,34 +184,20 @@ class TransifexHandler(BaseHookHandler):
except Exception as exception: except Exception as exception:
raise ValueError('Error loading JSON: ' + str(exception)) raise ValueError('Error loading JSON: ' + str(exception))
if self.secret is not None:
if secret is not None:
p = Properties() p = Properties()
p.master = self.master p.master = self.master
rendered_secret = yield p.render(secret) rendered_secret = yield p.render(self.secret)
signature = hmac.new( signature = hmac.new(
unicode2bytes(rendered_secret), unicode2bytes(rendered_secret),
unicode2bytes(content_text.strip()), unicode2bytes(content_text.strip()),
digestmod=hashlib.sha256) digestmod=hashlib.sha256)
header_signature = bytes2unicode( header_signature = bytes2unicode(
request.getHeader(_HEADER_SIGNATURE)) request.getHeader(_HEADER_SIGNATURE))
self.verifyTransifexSignature(
http_verb = 'POST' request, content, rendered_secret,
http_url_path = request.headers('X-TX-Url') signature, header_signature
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')
event_type = bytes2unicode(payload.get(_EVENT_KEY), "None") event_type = bytes2unicode(payload.get(_EVENT_KEY), "None")
log.msg("Received event '{}' from transifex".format(event_type)) log.msg("Received event '{}' from transifex".format(event_type))