|
#!/usr/bin/env python
# coding: utf-8
"""
This is the 'keepcool' module.
This module is a git repository inspection package
that aims to avoid developper's burnout.
"""
from datetime import datetime, time
from optparse import OptionParser
from git import Git
class Date(object):
"""
Class that handle date from a unix timestamp.
This class is used to determine if a date
belongs or not to the working hours
Please note that datetime is an immuatable object :
it is complicated to inherit from it...
"""
def __init__(self, unix_timestamp):
"""
It is complicated to inherit from datetime
wich is an immuatable object.
"""
self.date = datetime.fromtimestamp(unix_timestamp)
def __repr__(self):
return self.date.strftime("Date(%s)")
def __str__(self):
ret = self.date.strftime("%A %Y-%m-%d %H:%M:%S")
extra = []
if not self.belongs_to_workin_hours():
if self.belongs_to_week_end():
extra.append("week-end")
if self.belongs_to_early_morning():
extra.append("early morning")
if self.belongs_to_late_evening():
extra.append("late in the evening")
ret = ret + " -> " + str(extra)
return ret
def belongs_to_week_end(self):
"""
Return True if date on saterday or sunday
"""
return self.date.weekday() >= 5
def belongs_to_early_morning(self):
"""
Return True if date is before 8:00 am
"""
return self.date.time() < time(hour=8)
def belongs_to_late_evening(self):
"""
Return True if date is after 20:00
"""
return self.date.time() > time(hour=20)
def belongs_to_workin_hours(self):
"""
Return TRUE if date belongs to workings hours.
Working hours include 8 am to 8 pm from monday to friday.
"""
ret = True
if (self.belongs_to_week_end()
or self.belongs_to_early_morning()
or self.belongs_to_late_evening()):
ret = False
return ret
class Commit(Date):
"""
This class is used to store GIT each commit object
"""
def __init__(self, uuid, user, unix_timestamp):
Date.__init__(self, unix_timestamp)
self.uuid = uuid
self.user = user
def __repr__(self):
return "Commit('%s', %s, %s)" % (
self.uuid, repr(self.user), self.date.strftime("%s"))
def __str__(self):
return Date.__str__(self)
class User(object):
"""
This class is used to store GIT committers
"""
def __init__(self, name):
self.name = name
self.nb_in = 0
self.nb_out = 0
self.commits = {}
def __repr__(self):
return "User('%s')" % self.name
def __str__(self):
# still not very well understand
return (self.name).encode('utf-8')
def add_commit(self, commit):
"""
Add a commit object if not already there.
Throw KeyError if already there.
"""
if commit.uuid in self.commits:
raise IndexError(
"'%s' commit's key is already recorded for '%s'" % (
commit.uuid, self.name))
self.commits[commit.uuid] = commit
def compute_status(self):
"""
Compute commits done by the user
"""
self.nb_in = 0
self.nb_out = 0
for key in self.commits:
if self.commits[key].belongs_to_workin_hours():
self.nb_in = self.nb_in + 1
else:
self.nb_out = self.nb_out + 1
def print_status(self, trace=False):
"""
Display commits done by the user
"""
nb_sum = self.nb_in + self.nb_out
ratio = 100 * self.nb_out / nb_sum if nb_sum else 0
ret = ""
if trace is True:
def get_date(item):
"""sort commits by date"""
return item[1].date
ret = ret + ".......................................\n"
items = sorted(self.commits.items(), key=get_date)
for asso in items:
ret = ret + str(asso[1]) + "\n"
ret = ret + "%25s: %4i / %4i%%\n" % (str(self), nb_sum, ratio)
return ret
class KeepCool(object):
""" is the module's class.
Unique namespace to import :
>>> from keepcoll import KeepCool
"""
def __init__(self, repo, **kwargs):
self.repo = repo
self.options = kwargs
self.users = {}
self.commits = {}
self.nb_in = 0
self.nb_out = 0
def __repr__(self):
return "KeepCool('%s')" % self.repo
def parse_logs(self, **kwargs):
"""
Load logs into objects.
"""
git = Git(self.repo)
log_args = ["--encoding=UTF-8", "--pretty=%H; %cN; %ct"]
only_user = None
after = None
before = None
# get context
if 'user' in kwargs:
only_user = kwargs['user']
elif 'user' in self.options:
only_user = self.options['user']
if 'after' in kwargs:
after = kwargs['after']
elif 'after' in self.options:
after = self.options['after']
if 'before' in kwargs:
before = kwargs['before']
elif 'before' in self.options:
before = self.options['before']
# add command line options
if only_user is not None:
log_args.append("--committer=" + only_user)
if after is not None:
log_args.append("--after=" + after)
if before is not None:
log_args.append("--before=" + before)
# query, parse and record git logs
string = git.log(*log_args)
for log in string.splitlines():
(uuid, name, unix_timestamp_string) = log.split(';')
self.add_commit(uuid, name, int(unix_timestamp_string))
def get_user(self, name):
"""
Find and return an already recorded user.
Return None if not already there
"""
ret = None
if name in self.users:
ret = self.users[name]
return ret
def add_user(self, name):
"""
Add a new user object if not already there.
Return the user object.
"""
ret = self.get_user(name)
if ret is None:
ret = self.users[name] = User(name)
return ret
def add_commit(self, uuid, name, unix_timestamp):
"""
Add a new commit object if not already there.
Return the commit objet,
or throw KeyError if already there.
"""
if uuid in self.commits:
raise IndexError("'%s' commit's key is already recorded" % uuid)
user = self.add_user(name)
ret = self.commits[uuid] = Commit(uuid, user, unix_timestamp)
user.add_commit(ret)
return ret
def compute_status(self):
"""
Compute commits done by users
"""
self.nb_in = 0
self.nb_out = 0
for key in self.users:
self.users[key].compute_status()
self.nb_in = self.nb_in + self.users[key].nb_in
self.nb_out = self.nb_out + self.users[key].nb_out
def print_status(self, **kwargs):
"""
Display commits done by all user
"""
def get_name(item):
"""sort users by name"""
return item[1].name
def get_sum(item):
"""sort users by sum"""
return item[1].nb_in + item[1].nb_out
def get_ratio(item):
"""sort users by ratio"""
nb_sum = item[1].nb_in + item[1].nb_out
return 100 * item[1].nb_out / nb_sum if nb_sum else 0
trace = False
sort_field = 'name'
only_user = None
# get context
if 'trace' in kwargs:
trace = kwargs['trace']
elif 'trace' in self.options:
trace = self.options['trace']
if 'sort_field' in kwargs:
sort_field = kwargs['sort_field']
elif 'sort_field' in self.options:
sort_field = self.options['sort_field']
if 'user' in kwargs:
only_user = kwargs['user']
elif 'user' in self.options:
only_user = self.options['user']
nb_sum = self.nb_in + self.nb_out
ratio = 100 * self.nb_out / nb_sum if nb_sum else 0
ret = ""
ret = ret + "---------------------------------------\n"
ret = ret + "%25s: %4s / %4s\n" % (
self.repo + "'s commiter", "sum", "ratio")
ret = ret + "---------------------------------------\n"
# sort users
if sort_field == 'sum':
key_func = get_sum
elif sort_field == 'ratio':
key_func = get_ratio
else:
key_func = get_name
# loop on user
items = sorted(self.users.items(), key=key_func)
for asso in items:
ret = ret + asso[1].print_status(trace)
ret = ret + "---------------------------------------\n"
if only_user is None:
ret = ret + "%25s: %4i / %4i%%\n" % ("all", nb_sum, ratio)
ret = ret + "---------------------------------------\n"
return ret
def main():
"""
Entry point if called as an executable
Note: optparse is deprecated, but fails to use argparse
within Debian Stretch using official packages.
"""
parser = OptionParser(usage="usage: %prog [options] [git-path]",
version="%prog 0.1")
# get options and parameters
parser.add_option(
"-v", "--verbose", action="store_true", dest="trace",
default="False", help="print commits too")
parser.add_option(
"-u", "--user", dest="user", default=None,
help="limit search to a USER", metavar="USER")
parser.add_option(
"-a", "--after", dest="after", default=None,
help="limit search from DATE", metavar="DATE")
parser.add_option(
"-b", "--before", dest="before", default=None,
help="limit search to DATE", metavar="DATE")
parser.add_option(
"-s", "--sort", dest="sort_field", default='name',
help="sort on FIELD: 'name' (default), 'sum', or 'ratio'",
metavar="FIELD")
(options, args) = parser.parse_args()
repo = args[0] if len(args) > 0 else "repo"
# call main API
keepcool = KeepCool(repo, user=options.user)
keepcool.parse_logs(after=options.after, before=options.before)
keepcool.compute_status()
print keepcool.print_status(sort_field=options.sort_field,
trace=options.trace)
if __name__ == "__main__":
main()
|
