trunk/mia/status.py

# The module which forms the heart of the MIA scripts
# Copyright (C) 2001, 2002  Martin Michlmayr <tbm@cyrius.com>
# $Id$

# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.

# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA


# This is a collection of software to track down developers who are
# Missing In Action (MIA).


import ldap, os, re, string, sys, time
import apt_pkg


config = "/org/qa.debian.org/mia/mia.conf"

apt_pkg.init()
Cnf = apt_pkg.newConfiguration()
apt_pkg.ReadConfigFileISC(Cnf, config)

info = { }
matches = { }

def find_status(path=Cnf["Dir::Database"]):
    files = []

    try:
        for file in os.listdir(path):
            root, ext = os.path.splitext(file)
            if ext == ".summary":
                files.append(root)
    except OSError:
        print "Cannot find directory of database: %s" % path
        sys.exit(1)

    return files


def read_status(files):
    for file in files:
        try:
            summary = open(Cnf["Dir::Database"] + "/" + file + ".summary", "r")
        except IOError:
            print "Cannot open summary file %s" % file
            sys.exit(1)

        info[file] = { }

        for line in summary.readlines():
            result = re.search("(\d+):\s*(.*)", line)

            if result is not None:
                info[file][int(result.group(1))] = result.group(2)
            else:
                print "Error parsing file %s: %s" % (file, line)
                sys.exit(1)

        summary.close()


def write_status(file, id, summary):
    try:
        status = open(Cnf["Dir::Database"] + "/" + file + ".summary", "a")
    except IOError:
        print "Cannot open summary file to write!"
        sys.exit(1)

    string = str(id) + ": " + summary + "\n"
    status.write(string)
    status.close()


def get_all_files():
    files = info.keys()
    files.sort()
    return files


def get_selected_files(pattern, age):
    files = get_all_files()
    selection = [ ]
    cpattern = re.compile(pattern, re.I)
    now = time.time()
    for file in files:
        if grep_items(file, cpattern) and (not age or (int(age) * 3600 * 24 > now - latest(file))):
            selection.append(file)
    return selection


def grep_items(file, cpattern):
    for item in info[file].values():
        if cpattern.search(item):
            return 1


def get_history(file):
    history = [ ]
    all_contacts = info[file].keys()
    all_contacts.sort()
    for when in all_contacts:
        history.append("    %s: %s" % (get_time(when, "%Y-%m-%d"), info[file][when]))
    return history


def latest(file):
    return max(info[file].keys())


def should_check(file, ttl):
    if (time.time() - latest(file)) > (int(ttl) * 3600 * 24):
        return 1


def compile_matches():
    matches['ignore'] = [ 'willfix' ]
    matches['in'] = [ 'S-V', 'FHS', 'Old', 'RC', 'ITH', 'NMU', 'Hint', 'email', 'Contact' ]
    matches['out'] = [ 'Fixed', 'Orphaned', 'Orphaning', 'Removed',
                        'Removing', 'Hijack', 'Hijacked' ]
    matches['latest'] = [ 'Maintainer has no packages anymore', 'Retired',
                          'Retiring', 'Active' ]

    for key in matches.keys():
        matches[key] = map(string.lower, matches[key])


def last_is_final(file):
    if string.lower(info[file][latest(file)]) in matches["latest"]:
        return 1


def has_stuff_todo(file):
    if last_is_final(file):
        return

    found = { }
    for key in matches.keys():
        found[key] = { }

    for entry in info[file].values():
        fields = re.split(":\s+", entry, 2)
        if len(fields) > 1:
            what = re.split("\s+|&", fields[0])[0]
            packages = re.split(",\s*", string.lower(fields[1]))
            for key in matches.keys():
                if string.lower(what) in matches[key]:
                    for package in packages:
                        found[key][package] = 1
                    break
            else:
                print "Error: do not know what to do with '%s'" % what
                sys.exit(1)

    # show cases without any INs.
    if not found["in"]:
        return [ "no-todo-list" ]

    for item in found["out"].keys():
        try:
            del found["in"][item]
        except KeyError:
            print "Warning: '%s' was not on the TODO list" % item

    if found["in"]:
        # something has not been dealt with yet
        outstanding = found["in"].keys()
        outstanding.sort()
        return outstanding


def get_name(uid):
    try:
        query = ldap.open("db.debian.org")
    except SystemError:
        print "Cannot connect to LDAP server.  Use --no-ldap (not implemented yet)"
        sys.exit(1)
    base = "ou=users,dc=debian,dc=org"
    query.bind(base, "", ldap.AUTH_SIMPLE)
    result = query.search_s(base, ldap.SCOPE_SUBTREE, "(uid=%s)" % uid, ("cn", "sn"))
    for _, dict in result:
        cn = dict["cn"][0]
        sn = dict["sn"][0]

    return (cn, sn)


def get_time(when, format):
    return time.strftime(format, time.gmtime(when))


# main()

compile_matches()


# vim:set ts=4:
# vim:set expandtab:
# vim:set shiftwidth=4:
1	# The module which forms the heart of the MIA scripts
2	# Copyright (C) 2001, 2002 Martin Michlmayr <tbm@cyrius.com>
3	# $Id$
4
5	# This program is free software; you can redistribute it and/or modify
6	# it under the terms of the GNU General Public License as published by
7	# the Free Software Foundation; either version 2 of the License, or
8	# (at your option) any later version.
9
10	# This program is distributed in the hope that it will be useful,
11	# but WITHOUT ANY WARRANTY; without even the implied warranty of
12	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13	# GNU General Public License for more details.
14
15	# You should have received a copy of the GNU General Public License
16	# along with this program; if not, write to the Free Software
17	# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18
19
20	# This is a collection of software to track down developers who are
21	# Missing In Action (MIA).
22
23
24	import ldap, os, re, string, sys, time
25	import apt_pkg
26
27
28	config = "/org/qa.debian.org/mia/mia.conf"
29
30	apt_pkg.init()
31	Cnf = apt_pkg.newConfiguration()
32	apt_pkg.ReadConfigFileISC(Cnf, config)
33
34	info = { }
35	matches = { }
36
37	def find_status(path=Cnf["Dir::Database"]):
38	files = []
39
40	try:
41	for file in os.listdir(path):
42	root, ext = os.path.splitext(file)
43	if ext == ".summary":
44	files.append(root)
45	except OSError:
46	print "Cannot find directory of database: %s" % path
47	sys.exit(1)
48
49	return files
50
51
52	def read_status(files):
53	for file in files:
54	try:
55	summary = open(Cnf["Dir::Database"] + "/" + file + ".summary", "r")
56	except IOError:
57	print "Cannot open summary file %s" % file
58	sys.exit(1)
59
60	info[file] = { }
61
62	for line in summary.readlines():
63	result = re.search("(\d+):\s(.)", line)
64
65	if result is not None:
66	info[file][int(result.group(1))] = result.group(2)
67	else:
68	print "Error parsing file %s: %s" % (file, line)
69	sys.exit(1)
70
71	summary.close()
72
73
74	def write_status(file, id, summary):
75	try:
76	status = open(Cnf["Dir::Database"] + "/" + file + ".summary", "a")
77	except IOError:
78	print "Cannot open summary file to write!"
79	sys.exit(1)
80
81	string = str(id) + ": " + summary + "\n"
82	status.write(string)
83	status.close()
84
85
86	def get_all_files():
87	files = info.keys()
88	files.sort()
89	return files
90
91
92	def get_selected_files(pattern, age):
93	files = get_all_files()
94	selection = [ ]
95	cpattern = re.compile(pattern, re.I)
96	now = time.time()
97	for file in files:
98	if grep_items(file, cpattern) and (not age or (int(age) * 3600 * 24 > now - latest(file))):
99	selection.append(file)
100	return selection
101
102
103	def grep_items(file, cpattern):
104	for item in info[file].values():
105	if cpattern.search(item):
106	return 1
107
108
109	def get_history(file):
110	history = [ ]
111	all_contacts = info[file].keys()
112	all_contacts.sort()
113	for when in all_contacts:
114	history.append(" %s: %s" % (get_time(when, "%Y-%m-%d"), info[file][when]))
115	return history
116
117
118	def latest(file):
119	return max(info[file].keys())
120
121
122	def should_check(file, ttl):
123	if (time.time() - latest(file)) > (int(ttl) * 3600 * 24):
124	return 1
125
126
127	def compile_matches():
128	matches['ignore'] = [ 'willfix' ]
129	matches['in'] = [ 'S-V', 'FHS', 'Old', 'RC', 'ITH', 'NMU', 'Hint', 'email', 'Contact' ]
130	matches['out'] = [ 'Fixed', 'Orphaned', 'Orphaning', 'Removed',
131	'Removing', 'Hijack', 'Hijacked' ]
132	matches['latest'] = [ 'Maintainer has no packages anymore', 'Retired',
133	'Retiring', 'Active' ]
134
135	for key in matches.keys():
136	matches[key] = map(string.lower, matches[key])
137
138
139	def last_is_final(file):
140	if string.lower(info[file][latest(file)]) in matches["latest"]:
141	return 1
142
143
144	def has_stuff_todo(file):
145	if last_is_final(file):
146	return
147
148	found = { }
149	for key in matches.keys():
150	found[key] = { }
151
152	for entry in info[file].values():
153	fields = re.split(":\s+", entry, 2)
154	if len(fields) > 1:
155	what = re.split("\s+\|&", fields[0])[0]
156	packages = re.split(",\s*", string.lower(fields[1]))
157	for key in matches.keys():
158	if string.lower(what) in matches[key]:
159	for package in packages:
160	found[key][package] = 1
161	break
162	else:
163	print "Error: do not know what to do with '%s'" % what
164	sys.exit(1)
165
166	# show cases without any INs.
167	if not found["in"]:
168	return [ "no-todo-list" ]
169
170	for item in found["out"].keys():
171	try:
172	del found["in"][item]
173	except KeyError:
174	print "Warning: '%s' was not on the TODO list" % item
175
176	if found["in"]:
177	# something has not been dealt with yet
178	outstanding = found["in"].keys()
179	outstanding.sort()
180	return outstanding
181
182
183	def get_name(uid):
184	try:
185	query = ldap.open("db.debian.org")
186	except SystemError:
187	print "Cannot connect to LDAP server. Use --no-ldap (not implemented yet)"
188	sys.exit(1)
189	base = "ou=users,dc=debian,dc=org"
190	query.bind(base, "", ldap.AUTH_SIMPLE)
191	result = query.search_s(base, ldap.SCOPE_SUBTREE, "(uid=%s)" % uid, ("cn", "sn"))
192	for _, dict in result:
193	cn = dict["cn"][0]
194	sn = dict["sn"][0]
195
196	return (cn, sn)
197
198
199	def get_time(when, format):
200	return time.strftime(format, time.gmtime(when))
201
202
203	# main()
204
205	compile_matches()
206
207
208	# vim:set ts=4:
209	# vim:set expandtab:
210	# vim:set shiftwidth=4:
Name	Value
svn:eol-style	native
svn:executable	*
svn:keywords	Author Date Id Revision