zanata_userinfo.py: get user name and email
The script generates a csv file by reading a yaml file which contains the list of translators (e.g., translation_team.yaml) with user name and e-mail addresses by interacting with Zanata API. The script utilizes ZanataUtils in openstack-infra/project-config to read zanata.ini configuration file which contains Zanata API key. Note that the Zanata API key is required to use Zanata accounts API. Change-Id: I078742c5956b2c72eb1dabb10b9d3b4667b8f7e6
This commit is contained in:
Ian Y. Choi
parent
8ae974d80e
commit
e7806cd03a
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.')
|
||||
|
|
@ -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()
|
||||
Loading…
Reference in New Issue