openstack/i18n
Code Issues Proposed changes

Merge "zanata_userinfo.py: get user name and email"

Browse Source
This commit is contained in:
Jenkins 2017-03-19 08:11:35 +00:00 committed by Gerrit Code Review
commit 371d47ad38
3 changed files with 407 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
requirements.txt
View File

@ -2,6 +2,7 @@
# of appearance. Changing the order has an impact on the overall integration
# process, which may cause wedges in the gate later.
beautifulsoup4 # MIT
lxml>=2.3,!=3.7.0 # BSD
oslo.log>=3.11.0 # Apache-2.0
requests!=2.12.2,!=2.13.0,>=2.10.0 # Apache-2.0
six>=1.9.0 # MIT

203
tools/zanata/ZanataUtils.py Executable file
View File

@ -0,0 +1,203 @@
# Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from io import BytesIO
import json
from lxml import etree
import os
import re
import requests
try:
import configparser
except ImportError:
import ConfigParser as configparser
try:
from urllib.parse import urljoin
except ImportError:
from urlparse import urljoin
class IniConfig(object):
"""Object that stores zanata.ini configuration
Read zanata.ini and make its values available.
Attributes:
inifile: The path to the ini file to load values from.
"""
def __init__(self, inifile):
self.inifile = inifile
self._load_config()
def _load_config(self):
"""Load configuration from the zanata.ini file
Parses the ini file and stores its data.
"""
if not os.path.isfile(self.inifile):
raise ValueError('zanata.ini file not found.')
config = configparser.ConfigParser()
try:
config.read(self.inifile)
except configparser.Error:
raise ValueError('zanata.ini could not be parsed, please check '
'format.')
for item in config.items('servers'):
item_type = item[0].split('.')[1]
if item_type in ('username', 'key', 'url'):
setattr(self, item_type, item[1])
class ZanataRestService(object):
def __init__(self, zconfig, accept='application/xml',
content_type='application/xml', verify=True):
self.url = zconfig.url
if "charset" not in content_type:
content_type = "%s;charset=utf8" % content_type
self.headers = {'Accept': accept,
'Content-Type': content_type,
'X-Auth-User': zconfig.username,
'X-Auth-Token': zconfig.key}
self.verify = verify
def _construct_url(self, url_fragment):
return urljoin(self.url, url_fragment)
def query(self, url_fragment, raise_errors=True):
request_url = self._construct_url(url_fragment)
try:
r = requests.get(request_url, verify=self.verify,
headers=self.headers)
except requests.exceptions.ConnectionError:
raise ValueError('Connection error')
if raise_errors and r.status_code != 200:
raise ValueError('Got status code %s for %s' %
(r.status_code, request_url))
if raise_errors and not r.content:
raise ValueError('Did not receive any data from %s' % request_url)
return r
def push(self, url_fragment, data):
request_url = self._construct_url(url_fragment)
try:
return requests.put(request_url, verify=self.verify,
headers=self.headers, data=json.dumps(data))
except requests.exceptions.ConnectionError:
raise ValueError('Connection error')
class ProjectConfig(object):
"""Object that stores zanata.xml per-project configuration.
Write out a zanata.xml file for the project given the supplied values.
Attributes:
zconfig (IniConfig): zanata.ini values
xmlfile (str): path to zanata.xml to read or write
rules (list): list of two-ples with pattern and rules
"""
def __init__(self, zconfig, xmlfile, rules, verify, **kwargs):
self.rest_service = ZanataRestService(zconfig, verify=verify)
self.xmlfile = xmlfile
self.rules = self._parse_rules(rules)
for key, value in kwargs.items():
setattr(self, key, value)
self._create_config()
def _get_tag_prefix(self, root):
"""XML utility method
Get the namespace of the XML file so we can
use it to search for tags.
"""
return '{%s}' % etree.QName(root).namespace
def _parse_rules(self, rules):
"""Parse a two-ple of pattern, rule.
Returns a list of dictionaries with 'pattern' and 'rule' keys.
"""
return [{'pattern': rule[0], 'rule': rule[1]} for rule in rules]
def _create_config(self):
"""Create zanata.xml
Use the supplied parameters to create zanata.xml by downloading
a base version of the file and adding customizations.
"""
xml = self._fetch_zanata_xml()
self._add_configuration(xml)
self._write_xml(xml)
def _fetch_zanata_xml(self):
"""Get base zanata.xml
Download a basic version of the configuration for the project
using Zanata's REST API.
"""
r = self.rest_service.query(
'/rest/projects/p/%s/iterations/i/%s/config'
% (self.project, self.version))
project_config = r.content
p = etree.XMLParser(remove_blank_text=True)
try:
xml = etree.parse(BytesIO(project_config), p)
except etree.ParseError:
raise ValueError('Error parsing xml output')
return xml
def _add_configuration(self, xml):
"""Insert additional configuration
Add locale mapping rules to the base zanata.xml retrieved from
the server.
Args:
xml (etree): zanata.xml file contents
"""
root = xml.getroot()
s = etree.SubElement(root, 'src-dir')
s.text = self.srcdir
t = etree.SubElement(root, 'trans-dir')
t.text = self.txdir
rules = etree.SubElement(root, 'rules')
for rule in self.rules:
new_rule = etree.SubElement(rules, 'rule')
new_rule.attrib['pattern'] = rule['pattern']
new_rule.text = rule['rule']
if self.excludes:
excludes = etree.SubElement(root, 'excludes')
excludes.text = self.excludes
tag_prefix = self._get_tag_prefix(root)
# Work around https://bugzilla.redhat.com/show_bug.cgi?id=1219624
# by removing port number in URL if it's there
url = root.find('%surl' % tag_prefix)
url.text = re.sub(':443', '', url.text)
def _write_xml(self, xml):
"""Write xml
Write out xml to zanata.xml.
"""
try:
xml.write(self.xmlfile, pretty_print=True)
except IOError:
raise ValueError('Error writing zanata.xml.')

203
tools/zanata/zanata_userinfo.py Executable file
View File

@ -0,0 +1,203 @@
#!/usr/bin/env python
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import argparse
import csv
import io
import operator
import os
import sys
from oslo_log import log as logging
import yaml
from ZanataUtils import IniConfig
from ZanataUtils import ZanataRestService
LOG = logging.getLogger(__name__)
class ZanataAccounts(object):
"""Object that retrieves Zanata account information.
Retrieve name and e-mail address using Zanata ID.
Attributes:
zconfig (IniConfig): zanata.ini values
verify (Bool): True if communicating with non-SSL server
"""
def __init__(self, zconfig, verify, **kwargs):
accept = 'application/vnd.zanata.account+json'
content_type = 'application/json'
self.rest_service = ZanataRestService(zconfig, accept=accept,
content_type=content_type,
verify=verify)
for key, value in kwargs.items():
setattr(self, key, value)
def get_account_data(self, zanata_id):
"""Get detail account information
Retrieve name and e-mail address information by Zanata ID
using Zanata's REST API.
"""
r = self.rest_service.query(
'/rest/accounts/u/%s'
% (zanata_id))
account_data = r.json()
return account_data
def _make_language_team(name, team_info):
return {
'tag': 'language_team',
'language_code': name,
'language': team_info['language'],
# Zanata ID which only consists of numbers is a valid ID
# and such entry is interpreted as integer unless it is
# quoted in the YAML file. Ensure to stringify them.
'translators': [str(i) for i in team_info['translators']],
'reviewers': [str(i) for i in team_info.get('reviewers', [])],
'coordinators': [str(i) for i in team_info.get('coordinators', [])],
}
def _make_user(user_id, language_code, language):
return {
'user_id': user_id,
'lang_code': language_code,
'lang': language,
'name': '',
'email': ''
}
def read_language_team_yaml(translation_team_uri, lang_list):
LOG.debug('Process list of language team from uri: %s',
translation_team_uri)
content = yaml.safe_load(io.open(translation_team_uri, 'r'))
language_teams = {}
if lang_list:
lang_notfound = [lang_code for lang_code in lang_list
if lang_code not in content]
if lang_notfound:
print('Language %s not tound in %s.' %
(', '.join(lang_notfound),
translation_team_uri))
sys.exit(1)
for lang_code, team_info in content.items():
if lang_list and lang_code not in lang_list:
continue
language_teams[lang_code] = _make_language_team(lang_code, team_info)
return language_teams
def get_zanata_userdata(zc, verify, role, language_teams):
print('Getting user data in Zanata...')
accounts = ZanataAccounts(zc, verify)
users = {}
if not role:
role = 'translators'
for language_code in language_teams:
language_team = language_teams[language_code]
language_name = language_team['language']
for user in language_team[role]:
users[user] = _make_user(user, language_code, language_name)
for user_id in users:
user = users.get(user_id)
print('Getting user detail data for user %(user_id)s'
% {'user_id': user_id})
user_data = accounts.get_account_data(user_id)
if user_data:
user['name'] = user_data['name'].encode('utf-8')
user['email'] = user_data['email']
return users
def write_userdata_to_file(users, output_file):
userdata = [user for user in
sorted(users.values(),
key=operator.itemgetter('lang', 'user_id'))]
_write_userdata_to_csvfile(userdata, output_file)
print('Userdata has been written to %s' % output_file)
def _write_userdata_to_csvfile(userdata, output_file):
with open(output_file, 'wb') as csvfile:
writer = csv.writer(csvfile)
writer.writerow(['user_id', 'lang_code', 'lang',
'name', 'email'])
for data in userdata:
writer.writerow([data['user_id'], data['lang_code'],
data['lang'], data['name'], data['email']])
def _comma_separated_list(s):
return s.split(',')
def main():
# Loads zanata.ini configuration file
try:
zc = IniConfig(os.path.expanduser('~/.config/zanata.ini'))
except ValueError as e:
sys.exit(e)
# Parses command option(s)
parser = argparse.ArgumentParser(description='Generate a csv file which '
'contains the list of translators for '
'a specified target role with name and '
'e-mail address. Require a privilege '
'to access Zanata accounts API.')
parser.add_argument("-o", "--output-file",
help=("Specify the output file. "
"Default: zanata_userinfo_output.csv."))
parser.add_argument("-r", "--role",
help=("Specify the target role. "
"Roles: coordinators, translators, reviewers."
"Default: translators."))
parser.add_argument("-l", "--lang",
type=_comma_separated_list,
help=("Specify language(s). Comma-separated list. "
"Language code like zh-CN, ja needs to be used. "
"Otherwise all languages are processed."))
parser.add_argument('--no-verify', action='store_false', dest='verify',
help='Do not perform HTTPS certificate verification')
parser.add_argument("user_yaml",
help="YAML file of the user list")
options = parser.parse_args()
# Reads langauge team information
language_teams = read_language_team_yaml(options.user_yaml, options.lang)
users = get_zanata_userdata(zc, options.verify, options.role,
language_teams)
output_file = (options.output_file or 'zanata_userinfo_output.csv')
write_userdata_to_file(users, output_file)
if __name__ == '__main__':
main()