Utilities, assets and configurations used by I18n team
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

407 lines
16 KiB

#!/usr/bin/env python3
# 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 collections
import csv
import datetime
import io
import json
import logging
import random
import re
import sys
import requests
import yaml
ZANATA_URI = 'https://translate.openstack.org/rest/%s'
LOG = logging.getLogger('zanata_stats')
ZANATA_VERSION_EXPR = r'^(master[-,a-z]*|stable-[a-z]+|openstack-user-survey)$'
ZANATA_VERSION_PATTERN = re.compile(ZANATA_VERSION_EXPR)
class ZanataUtility(object):
"""Utilities to invoke Zanata REST API."""
user_agents = [
'Mozilla/5.0 (X11; Ubuntu; Linux x86_64) Gecko/20100101 Firefox/32.0',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_6) AppleWebKit/537.78.2',
'Mozilla/5.0 (Windows NT 6.3; WOW64) Gecko/20100101 Firefox/32.0',
'Mozilla/5.0 (Macintosh; Intel Mac OS X) Chrome/37.0.2062.120',
'Mozilla/5.0 (Windows NT 6.1; WOW64; Trident/7.0; rv:11.0) like Gecko'
]
def read_uri(self, uri, headers):
try:
headers['User-Agent'] = random.choice(ZanataUtility.user_agents)
req = requests.get(uri, headers=headers)
return req.text
except Exception as e:
LOG.error('Error "%(error)s" while reading uri %(uri)s',
{'error': e, 'uri': uri})
raise
def read_json_from_uri(self, uri):
data = self.read_uri(uri, {'Accept': 'application/json'})
try:
return json.loads(data)
except Exception as e:
LOG.error('Error "%(error)s" parsing json from uri %(uri)s',
{'error': e, 'uri': uri})
raise
def get_projects(self):
uri = ZANATA_URI % ('projects')
LOG.debug("Reading projects from %s" % uri)
projects_data = self.read_json_from_uri(uri)
return [project['id'] for project in projects_data]
@staticmethod
def _is_valid_version(version):
return bool(ZANATA_VERSION_PATTERN.match(version))
def get_project_versions(self, project_id):
uri = ZANATA_URI % ('projects/p/%s' % project_id)
LOG.debug("Reading iterations for project %s" % project_id)
project_data = self.read_json_from_uri(uri)
if 'iterations' in project_data:
return [interation_data['id']
for interation_data in project_data['iterations']
if self._is_valid_version(interation_data['id'])]
else:
return []
def get_user_stats(self, project_id, iteration_id, zanata_user_id,
start_date, end_date):
uri = ZANATA_URI % ('stats/project/%s/version/%s/contributor/%s/%s..%s'
% (project_id, iteration_id, zanata_user_id,
start_date, end_date))
return self.read_json_from_uri(uri)
class LanguageTeam(object):
def __init__(self, language_code, team_info):
self.language_code = language_code
self.language = team_info['language']
# Zanata ID which only consists of numbers is a valid ID in Zanata.
# Such entry is interpreted as integer unless it is quoted
# in the YAML file. Ensure to stringify them.
self.translators = [str(i) for i in team_info['translators']]
self.reviewers = [str(i) for i in team_info.get('reviewers', [])]
self.coordinators = [str(i) for i in team_info.get('coordinators', [])]
@classmethod
def load_from_language_team_yaml(cls, 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'))
if lang_list:
lang_notfound = [lang_code for lang_code in lang_list
if lang_code not in content]
if lang_notfound:
LOG.error('Language %s not tound in %s.',
', '.join(lang_notfound),
translation_team_uri)
sys.exit(1)
return [cls(lang_code, team_info)
for lang_code, team_info in content.items()
if not lang_list or lang_code in lang_list]
class User(object):
trans_fields = ['total', 'translated', 'needReview',
'approved', 'rejected']
review_fields = ['total', 'approved', 'rejected']
def __init__(self, user_id, language_code):
self.user_id = user_id
self.lang = language_code
self.stats = collections.defaultdict(dict)
def __str__(self):
return ("<%s: user_id=%s, lang=%s, stats=%s" %
(self.__class__.__name__,
self.user_id, self.lang, self.stats,))
def __repr__(self):
return repr(self.convert_to_serializable_data())
def __lt__(self, other):
if self.lang != other.lang:
return self.lang < other.lang
else:
return self.user_id < other.user_id
def read_from_zanata_stats(self, zanata_stats, project_id, version):
# data format (Zanata 3.9.6)
# {
# "username": "amotoki",
# "contributions": [
# {
# "locale": "ja",
# "translation-stats": {
# "translated": 7360,
# "needReview": 0,
# "approved": 152,
# "rejected": 0
# },
# "review-stats": {
# "approved": 220,
# "rejected": 0
# }
# }
# ]
# }
stats = [d for d in zanata_stats['contributions']
if d['locale'] == self.lang]
if not stats:
return
stats = stats[0]
trans_stats = stats.get('translation-stats', {})
if trans_stats:
trans_stats['total'] = sum(trans_stats.values())
review_stats = stats.get('review-stats', {})
if review_stats:
review_stats['total'] = sum(review_stats.values())
self.stats[project_id][version] = {'translation-stats': trans_stats,
'review-stats': review_stats}
def populate_total_stats(self):
total_trans = dict([(k, 0) for k in self.trans_fields])
total_review = dict([(k, 0) for k in self.review_fields])
for project_id, versions in self.stats.items():
for version, stats in versions.items():
trans_stats = stats.get('translation-stats', {})
for k in self.trans_fields:
total_trans[k] += trans_stats.get(k, 0)
review_stats = stats.get('review-stats', {})
for k in self.review_fields:
total_review[k] += review_stats.get(k, 0)
self.stats['__total__']['translation-stats'] = total_trans
self.stats['__total__']['review-stats'] = total_review
def needs_output(self, include_no_activities):
if include_no_activities:
return True
return bool(self.stats) and all(self.stats.values())
@staticmethod
def get_flattened_data_title():
return [
'user_id',
'lang',
'project',
'version',
'translation-total',
'translated',
'needReview',
'approved',
'rejected',
'review-total',
'review-approved',
'review-rejected'
]
def convert_to_flattened_data(self, detail=False):
self.populate_total_stats()
data = []
for project_id, versions in self.stats.items():
if project_id == '__total__':
continue
for version, stats in versions.items():
trans_stats = stats.get('translation-stats', {})
review_stats = stats.get('review-stats', {})
if detail:
data.append(
[self.user_id, self.lang, project_id, version] +
[trans_stats.get(k, 0) for k in self.trans_fields] +
[review_stats.get(k, 0) for k in self.review_fields])
data.append([self.user_id, self.lang, '-', '-'] +
[self.stats['__total__']['translation-stats'][k]
for k in self.trans_fields] +
[self.stats['__total__']['review-stats'][k]
for k in self.review_fields])
return data
def convert_to_serializable_data(self, detail):
self.populate_total_stats()
return {'user_id': self.user_id,
'lang': self.lang,
'stats': (self.stats if detail
else self.stats['__total__'])}
def get_zanata_stats(start_date, end_date, language_teams, project_list,
version_list, user_list):
LOG.info('Getting Zanata contributors statistics (from %s to %s) ...',
start_date, end_date)
zanataUtil = ZanataUtility()
users = []
for team in language_teams:
users += [User(user_id, team.language_code)
for user_id in team.translators]
if not project_list:
project_list = zanataUtil.get_projects()
for project_id in project_list:
for version in zanataUtil.get_project_versions(project_id):
if version_list and version not in version_list:
continue
for user in users:
if user_list and user.user_id not in user_list:
continue
LOG.info('Getting %(project_id)s %(version)s '
'for user %(user_id)s %(user_lang)s',
{'project_id': project_id,
'version': version,
'user_id': user.user_id,
'user_lang': user.lang})
data = zanataUtil.get_user_stats(
project_id, version, user.user_id, start_date, end_date)
LOG.debug('Got: %s', data)
user.read_from_zanata_stats(data, project_id, version)
LOG.debug('=> %s', user)
return users
def write_stats_to_file(users, output_file, file_format,
include_no_activities, detail):
users = sorted([user for user in users
if user.needs_output(include_no_activities)])
if file_format == 'csv':
_write_stats_to_csvfile(users, output_file, detail)
else:
_write_stats_to_jsonfile(users, output_file, detail)
LOG.info('Stats has been written to %s', output_file)
def _write_stats_to_csvfile(users, output_file, detail):
with open(output_file, 'w') as csvfile:
writer = csv.writer(csvfile)
writer.writerow(User.get_flattened_data_title())
for user in users:
writer.writerows(user.convert_to_flattened_data(detail))
def _write_stats_to_jsonfile(users, output_file, detail):
users = [user.convert_to_serializable_data(detail)
for user in users]
with open(output_file, 'w') as f:
f.write(json.dumps(users, indent=4, sort_keys=True))
def _comma_separated_list(s):
return s.split(',')
def main():
default_end_date = datetime.datetime.now()
default_start_date = default_end_date - datetime.timedelta(days=180)
default_start_date = default_start_date.strftime('%Y-%m-%d')
default_end_date = default_end_date.strftime('%Y-%m-%d')
parser = argparse.ArgumentParser()
parser.add_argument("-s", "--start-date",
default=default_start_date,
help=("Specify the start date. "
"Default:%s" % default_start_date))
parser.add_argument("-e", "--end-date",
default=default_end_date,
help=("Specify the end date. "
"Default:%s" % default_end_date))
parser.add_argument("-o", "--output-file",
help=("Specify the output file. "
"Default: zanata_stats_output.{csv,json}."))
parser.add_argument("-p", "--project",
type=_comma_separated_list,
help=("Specify project(s). Comma-separated list. "
"Otherwise all Zanata projects are processed."))
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("-t", "--target-version",
type=_comma_separated_list,
help=("Specify version(s). Comma-separated list. "
"Otherwise all available versions are "
"processed."))
parser.add_argument("-u", "--user",
type=_comma_separated_list,
help=("Specify user(s). Comma-separated list. "
"Otherwise all users are processed."))
parser.add_argument('--detail',
action='store_true',
help=("If specified, statistics per project "
"and version are output in addition to "
"total statistics."))
parser.add_argument("--include-no-activities",
action='store_true',
help=("If specified, stats for users with no "
"activities are output as well."
"By default, stats only for users with "
"any activities are output."))
parser.add_argument("-f", "--format",
default='csv', choices=['csv', 'json'],
help="Output file format.")
parser.add_argument("--debug",
action='store_true',
help="Enable debug message.")
parser.add_argument("user_yaml",
help="YAML file of the user list")
options = parser.parse_args()
logging_level = logging.DEBUG if options.debug else logging.INFO
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
handler = logging.StreamHandler()
handler.setLevel(logging_level)
handler.setFormatter(formatter)
LOG.setLevel(logging_level)
LOG.addHandler(handler)
language_teams = LanguageTeam.load_from_language_team_yaml(
options.user_yaml, options.lang)
versions = [v.replace('/', '-') for v in options.target_version or []]
users = get_zanata_stats(options.start_date, options.end_date,
language_teams, options.project,
versions, options.user)
output_file = (options.output_file or
'zanata_stats_output.%s' % options.format)
write_stats_to_file(users, output_file, options.format,
options.include_no_activities,
options.detail)
if __name__ == '__main__':
main()