Update Webhook response variabel
All checks were successful
buildbot/multibuild_parent Build done.
buildbot/travis_bionic Build done.

This commit is contained in:
shekhar-cis 2023-01-06 20:38:43 +05:30
parent 1b29a5639c
commit 89305ec602
Signed by: shekhar-cis
GPG Key ID: F4F00AB04E83F9A7

View File

@ -1,18 +1,14 @@
import sys
"""Transifex webhook handler """
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
import requests
from buildbot.process.properties import Properties
from buildbot.util import bytes2unicode, unicode2bytes
@ -25,152 +21,94 @@ from dateutil.parser import parse as dateparse
_HEADER_USER_AGENT = 'User-Agent'
_HEADER_SIGNATURE = 'X-TX-Signature'
_HEADER_URL_PATH = 'X-TX-Url'
HTTP_DATE = 'date'
_EVENT_KEY = 'event'
transifexSecret = ""
transifexUsername = ""
transifexPassword = ""
transifex_dict = {}
secret = ""
master = ""
transifex_request_data = {}
gitHubToken = os.environ('gitHubToken')
class TransifexHandler(BaseHookHandler):
def __init__(self, master, secret, transifex_dict):
def __init__(self, master, secret, options, transifex_dict):
if not options:
options = {}
self.secret = secret
self.master = master
self.options = options
self.transifex_dict = transifex_dict
def returnMessage(self, status = False, message = "Unimplemented"):
output = json.dumps({"status": "OK" if status else "FAIL", "message": message})
return [output, [('Content-type', 'text/plain'),
('Content-Length', str(len(output)))
]]
return [output, [('Content-type', 'application/json')]]
def verifyTransifexSignature(
self, request, content, rendered_secret, signature, header_signature
self, request, content, signature, header_signature
):
http_verb = 'POST'
http_url_path = request.headers('X-TX-Url')
http_gmt_date = request.headers('Date')
http_url_path = request.getHeader(_HEADER_URL_PATH)
http_gmt_date = request.getHeader(HTTP_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,
key=self.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:
if tx_signature != header_signature:
return False
def downloadTranslatedLanguage(self, 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 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
if signature != request.getHeader(_HEADER_SIGNATURE):
return False
def process_translation_completed(self, payload, transifex_dict, event_type, codebase):
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']:
if 'translation_completed' in transifex_response['event']:
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")
return
# if isinstance(self.options, dict) and self.options.get('onlyIncludePushCommit', False):
# if isinstance(self.options, dict):
# 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)
# # for commit in commits:
# # files = []
# # for kind in ('added', 'modified', 'removed'):
# # files.extend(commit.get(kind, []) or [])
change = {
'author': 'buildbot-transifex,
'resource': transifex_response['resource'],
'branch': transifex_dict['branch'],
'project': transifex_response['project'],
'event': event_type,
'properties': {
'branch': branch,
'revision': revision,
'language': lang,
'resource': resource,
'project': project
}
}
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):
transifex_variables = {
'project': payload['project'],
"translated": payload['translated'],
"resource": payload['resource'],
"event": payload['event'],
"language": payload['language']
project = payload.get('project', 'None')
transform_values = {
project: {
"repository": "https://github.com/Bitmessage/PyBitmessage",
"branch": "v0.6"
}
return transifex_variables
}
return transform_values
@defer.inlineCallbacks
def getChanges(self, request):
@ -187,19 +125,39 @@ class TransifexHandler(BaseHookHandler):
if self.secret is not None:
p = Properties()
p.master = self.master
rendered_secret = yield p.render(self.secret)
option = self.options
self.rendered_secret = yield p.render(self.secret)
signature = hmac.new(
unicode2bytes(rendered_secret),
unicode2bytes(self.rendered_secret),
unicode2bytes(content_text.strip()),
digestmod=hashlib.sha256)
header_signature = bytes2unicode(
request.getHeader(_HEADER_SIGNATURE))
self.verifyTransifexSignature(
request, content, rendered_secret,
request, content, self.rendered_secret,
signature, header_signature
)
event_type = bytes2unicode(payload.get(_EVENT_KEY), "None")
event_type = payload.get("event", "None")
language = payload.get("language", 'None')
project = payload.get("project", 'None')
resource = payload.get("resource", 'None')
transifex_request_data['branch'] = "v0.6"
transifex_request_data['revision'] = ""
transifex_request_data["properties"] = "langugage"
transifex_request_data["properties"] = "resource"
transifex_request_data["properties"] = "project"
transifex_request_data["properties"] = {
"branch": branch,
"revision": revision
}
transifex_request_data["changes"] = {
"author": "buildbot-transifex",
"repository": project,
}
log.msg("Received event '{}' from transifex".format(event_type))
codebase = ""