--- /dev/null
+try:
+# Using Psyco makes it about 25% faster, but there's a bug in psyco in
+# handling of eval causing it to use unlimited memory with the magic
+# file enabled.
+# import psyco
+# psyco.full()
+# from psyco.classes import *
+ pass
+except:
+ pass
+import re
+import base64
+import cStringIO
+import specialuu
+import array
+import email.Utils
+import zlib
+import magic
+
+# Comment out if you don't want magic detection
+magicf = magic.MagicFile()
+
+# Open our output file
+outfile = open("gnats2bz_data.sql", "w")
+
+# List of GNATS fields
+fieldnames = ("Number", "Category", "Synopsis", "Confidential", "Severity",
+ "Priority", "Responsible", "State", "Quarter", "Keywords",
+ "Date-Required", "Class", "Submitter-Id", "Arrival-Date",
+ "Closed-Date", "Last-Modified", "Originator", "Release",
+ "Organization", "Environment", "Description", "How-To-Repeat",
+ "Fix", "Release-Note", "Audit-Trail", "Unformatted")
+
+# Dictionary telling us which GNATS fields are multiline
+multilinefields = {"Organization":1, "Environment":1, "Description":1,
+ "How-To-Repeat":1, "Fix":1, "Release-Note":1,
+ "Audit-Trail":1, "Unformatted":1}
+
+# Mapping of GCC release to version. Our version string is updated every
+# so we need to funnel all release's with 3.4 in the string to be version
+# 3.4 for bug tracking purposes
+# The key is a regex to match, the value is the version it corresponds
+# with
+releasetovermap = {r"3\.4":"3.4", r"3\.3":"3.3", r"3\.2\.2":"3.2.2",
+ r"3\.2\.1":"3.2.1", r"3\.2":"3.2", r"3\.1\.2":"3.1.2",
+ r"3\.1\.1":"3.1.1", r"3\.1":"3.1", r"3\.0\.4":"3.0.4",
+ r"3\.0\.3":"3.0.3", r"3\.0\.2":"3.0.2", r"3\.0\.1":"3.0.1",
+ r"3\.0":"3.0", r"2\.95\.4":"2.95.4", r"2\.95\.3":"2.95.3",
+ r"2\.95\.2":"2.95.2", r"2\.95\.1":"2.95.1",
+ r"2\.95":"2.95", r"2\.97":"2.97",
+ r"2\.96.*[rR][eE][dD].*[hH][aA][tT]":"2.96 (redhat)",
+ r"2\.96":"2.96"}
+
+# These map the field name to the field id bugzilla assigns. We need
+# the id when doing bug activity.
+fieldids = {"State":8, "Responsible":15}
+
+# These are the keywords we use in gcc bug tracking. They are transformed
+# into bugzilla keywords. The format here is <keyword>-><bugzilla keyword id>
+keywordids = {"wrong-code":1, "ice-on-legal-code":2, "ice-on-illegal-code":3,
+ "rejects-legal":4, "accepts-illegal":5, "pessimizes-code":6}
+
+# Map from GNATS states to Bugzilla states. Duplicates and reopened bugs
+# are handled when parsing the audit trail, so no need for them here.
+state_lookup = {"":"NEW", "open":"ASSIGNED", "analyzed":"ASSIGNED",
+ "feedback":"WAITING", "closed":"CLOSED",
+ "suspended":"SUSPENDED"}
+
+# Table of versions that exist in the bugs, built up as we go along
+versions_table = {}
+
+# Delimiter gnatsweb uses for attachments
+attachment_delimiter = "----gnatsweb-attachment----\n"
+
+# Here starts the various regular expressions we use
+# Matches an entire GNATS single line field
+gnatfieldre = re.compile(r"""^([>\w\-]+)\s*:\s*(.*)\s*$""")
+
+# Matches the name of a GNATS field
+fieldnamere = re.compile(r"""^>(.*)$""")
+
+# Matches the useless part of an envelope
+uselessre = re.compile(r"""^(\S*?):\s*""", re.MULTILINE)
+
+# Matches the filename in a content disposition
+dispositionre = re.compile("(\\S+);\\s*filename=\"([^\"]+)\"")
+
+# Matches the last changed date in the entire text of a bug
+# If you have other editable fields that get audit trail entries, modify this
+# The field names are explicitly listed in order to speed up matching
+lastdatere = re.compile(r"""^(?:(?:State|Responsible|Priority|Severity)-Changed-When: )(.+?)$""", re.MULTILINE)
+
+# Matches the From line of an email or the first line of an audit trail entry
+# We use this re to find the begin lines of all the audit trail entries
+# The field names are explicitly listed in order to speed up matching
+fromtore=re.compile(r"""^(?:(?:State|Responsible|Priority|Severity)-Changed-From-To: |From: )""", re.MULTILINE)
+
+# These re's match the various parts of an audit trail entry
+changedfromtore=re.compile(r"""^(\w+?)-Changed-From-To: (.+?)$""", re.MULTILINE)
+changedbyre=re.compile(r"""^\w+?-Changed-By: (.+?)$""", re.MULTILINE)
+changedwhenre=re.compile(r"""^\w+?-Changed-When: (.+?)$""", re.MULTILINE)
+changedwhyre=re.compile(r"""^\w+?-Changed-Why:\s*(.*?)$""", re.MULTILINE)
+
+# This re matches audit trail text saying that the current bug is a duplicate of another
+duplicatere=re.compile(r"""(?:")?Dup(?:licate)?(?:d)?(?:")? of .*?(\d+)""", re.IGNORECASE | re.MULTILINE)
+
+# Get the text of a From: line
+fromre=re.compile(r"""^From: (.*?)$""", re.MULTILINE)
+
+# Get the text of a Date: Line
+datere=re.compile(r"""^Date: (.*?)$""", re.MULTILINE)
+
+# Map of the responsible file to email addresses
+responsible_map = {}
+# List of records in the responsible file
+responsible_list = []
+# List of records in the categories file
+categories_list = []
+# List of pr's in the index
+pr_list = []
+# Map usernames to user ids
+usermapping = {}
+# Start with this user id
+userid_base = 2
+
+# Name of gnats user
+gnats_username = "gnats@gcc.gnu.org"
+# Name of unassigned user
+unassigned_username = "unassigned@gcc.gnu.org"
+
+gnats_db_dir = "."
+product = "gcc"
+productdesc = "GNU Compiler Connection"
+milestoneurl = "http://gcc/gnu.org"
+defaultmilestone = "3.4"
+
+def write_non_bug_tables():
+ """ Write out the non-bug related tables, such as products, profiles, etc."""
+ # Set all non-unconfirmed bugs's everconfirmed flag
+ print >>outfile, "update bugs set everconfirmed=1 where bug_status != 'UNCONFIRMED';"
+
+ # Set all bugs assigned to the unassigned user to NEW
+ print >>outfile, "update bugs set bug_status='NEW',assigned_to='NULL' where bug_status='ASSIGNED' AND assigned_to=3;"
+
+ # Insert the products
+ print >>outfile, "\ninsert into products ("
+ print >>outfile, " product, description, milestoneurl, disallownew,"
+ print >>outfile, " defaultmilestone, votestoconfirm) values ("
+ print >>outfile, " '%s', '%s', '%s', 0, '%s', 1);" % (product,
+ productdesc,
+ milestoneurl,
+ defaultmilestone)
+
+ # Insert the components
+ for category in categories_list:
+ component = SqlQuote(category[0])
+ productstr = SqlQuote(product)
+ description = SqlQuote(category[1])
+ initialowner = SqlQuote("3")
+ print >>outfile, "\ninsert into components (";
+ print >>outfile, " value, program, initialowner, initialqacontact,"
+ print >>outfile, " description) values ("
+ print >>outfile, " %s, %s, %s, '', %s);" % (component, productstr,
+ initialowner, description)
+
+ # Insert the versions
+ for productstr, version_list in versions_table.items():
+ productstr = SqlQuote(productstr)
+ for version in version_list:
+ version = SqlQuote(version)
+ print >>outfile, "\ninsert into versions (value, program) "
+ print >>outfile, " values (%s, %s);" % (version, productstr)
+
+ # Insert the users
+ for username, userid in usermapping.items():
+ realname = map_username_to_realname(username)
+ username = SqlQuote(username)
+ realname = SqlQuote(realname)
+ print >>outfile, "\ninsert into profiles ("
+ print >>outfile, " userid, login_name, password, cryptpassword, realname, groupset"
+ print >>outfile, ") values ("
+ print >>outfile, "%s,%s,'password',encrypt('password'), %s, 0);" % (userid, username, realname)
+ print >>outfile, "update profiles set groupset=1 << 32 where login_name like '%\@gcc.gnu.org';"
+
+def unixdate2datetime(unixdate):
+ """ Convert a unix date to a datetime value """
+ year, month, day, hour, min, sec, x, x, x, x = email.Utils.parsedate_tz(unixdate)
+ return "%d-%02d-%02d %02d:%02d:%02d" % (year,month,day,hour,min,sec)
+
+def unixdate2timestamp(unixdate):
+ """ Convert a unix date to a timestamp value """
+ year, month, day, hour, min, sec, x, x, x, x = email.Utils.parsedate_tz(unixdate)
+ return "%d%02d%02d%02d%02d%02d" % (year,month,day,hour,min,sec)
+
+def SqlQuote(str):
+ """ Perform SQL quoting on a string """
+ return "'%s'" % str.replace("'", """''""").replace("\\", "\\\\").replace("\0","\\0")
+
+def convert_gccver_to_ver(gccver):
+ """ Given a gcc version, convert it to a Bugzilla version. """
+ for k in releasetovermap.keys():
+ if re.search(".*%s.*" % k, gccver) is not None:
+ return releasetovermap[k]
+ result = re.search(r""".*(\d\.\d) \d+ \(experimental\).*""", gccver)
+ if result is not None:
+ return result.group(1)
+ return "unknown"
+
+def load_index(fname):
+ """ Load in the GNATS index file """
+ global pr_list
+ ifp = open(fname)
+ for record in ifp.xreadlines():
+ fields = record.split("|")
+ pr_list.append(fields[0])
+ ifp.close()
+
+def load_categories(fname):
+ """ Load in the GNATS categories file """
+ global categories_list
+ cfp = open(fname)
+ for record in cfp.xreadlines():
+ if re.search("^#", record) is not None:
+ continue
+ categories_list.append(record.split(":"))
+ cfp.close()
+
+def map_username_to_realname(username):
+ """ Given a username, find the real name """
+ name = username
+ name = re.sub("@.*", "", name)
+ for responsible_record in responsible_list:
+ if responsible_record[0] == name:
+ return responsible_record[1]
+ if len(responsible_record) > 2:
+ if responsible_record[2] == username:
+ return responsible_record[1]
+ return ""
+
+
+def get_userid(responsible):
+ """ Given an email address, get the user id """
+ global responsible_map
+ global usermapping
+ global userid_base
+ if responsible is None:
+ return -1
+ responsible = responsible.lower()
+ responsible = re.sub("sources.redhat.com", "gcc.gnu.org", responsible)
+ if responsible_map.has_key(responsible):
+ responsible = responsible_map[responsible]
+ if usermapping.has_key(responsible):
+ return usermapping[responsible]
+ else:
+ usermapping[responsible] = userid_base
+ userid_base += 1
+ return usermapping[responsible]
+
+def load_responsible(fname):
+ """ Load in the GNATS responsible file """
+ global responsible_map
+ global responsible_list
+ rfp = open(fname)
+ for record in rfp.xreadlines():
+ if re.search("^#", record) is not None:
+ continue
+ split_record = record.split(":")
+ responsible_map[split_record[0]] = split_record[2].rstrip()
+ responsible_list.append(record.split(":"))
+ rfp.close()
+
+def split_csl(list):
+ """ Split a comma seperated list """
+ newlist = re.split(r"""\s*,\s*""", list)
+ return newlist
+
+def fix_email_addrs(addrs):
+ """ Perform various fixups and cleaning on an e-mail address """
+ addrs = split_csl(addrs)
+ trimmed_addrs = []
+ for addr in addrs:
+ addr = re.sub(r"""\(.*\)""","",addr)
+ addr = re.sub(r""".*<(.*)>.*""","\\1",addr)
+ addr = addr.rstrip()
+ addr = addr.lstrip()
+ trimmed_addrs.append(addr)
+ addrs = ", ".join(trimmed_addrs)
+ return addrs
+
+class Bugzillabug(object):
+ """ Class representing a bugzilla bug """
+ def __init__(self, gbug):
+ """ Initialize a bugzilla bug from a GNATS bug. """
+ self.bug_id = gbug.bug_id
+ self.long_descs = []
+ self.bug_ccs = [get_userid("gcc-bugs@gcc.gnu.org")]
+ self.bug_activity = []
+ self.attachments = gbug.attachments
+ self.gnatsfields = gbug.fields
+ self.need_unformatted = gbug.has_unformatted_attach == 0
+ self.need_unformatted &= gbug.fields.has_key("Unformatted")
+ self.translate_pr()
+ self.update_versions()
+ if self.fields.has_key("Audit-Trail"):
+ self.parse_audit_trail()
+ self.write_bug()
+
+ def parse_fromto(type, string):
+ """ Parses the from and to parts of a changed-from-to line """
+ fromstr = ""
+ tostr = ""
+
+ # Some slightly messed up changed lines have unassigned-new,
+ # instead of unassigned->new. So we make the > optional.
+ result = re.search(r"""(.*)-(?:>?)(.*)""", string)
+
+ # Only know how to handle parsing of State and Responsible
+ # changed-from-to right now
+ if type == "State":
+ fromstr = state_lookup[result.group(1)]
+ tostr = state_lookup[result.group(2)]
+ elif type == "Responsible":
+ if result.group(1) != "":
+ fromstr = result.group(1)
+ if result.group(2) != "":
+ tostr = result.group(2)
+ if responsible_map.has_key(fromstr):
+ fromstr = responsible_map[fromstr]
+ if responsible_map.has_key(tostr):
+ tostr = responsible_map[tostr]
+ return (fromstr, tostr)
+ parse_fromto = staticmethod(parse_fromto)
+
+ def parse_audit_trail(self):
+ """ Parse a GNATS audit trail """
+ trail = self.fields["Audit-Trail"]
+ # Begin to split the audit trail into pieces
+ result = fromtore.finditer(trail)
+ starts = []
+ ends = []
+ pieces = []
+ # Make a list of the pieces
+ for x in result:
+ pieces.append (x)
+ # Find the start and end of each piece
+ if len(pieces) > 0:
+ for x in xrange(len(pieces)-1):
+ starts.append(pieces[x].start())
+ ends.append(pieces[x+1].start())
+ starts.append(pieces[-1].start())
+ ends.append(len(trail))
+ pieces = []
+ # Now make the list of actual text of the pieces
+ for x in xrange(len(starts)):
+ pieces.append(trail[starts[x]:ends[x]])
+ # And parse the actual pieces
+ for piece in pieces:
+ result = changedfromtore.search(piece)
+ # See what things we actually have inside this entry, and
+ # handle them approriately
+ if result is not None:
+ type = result.group(1)
+ changedfromto = result.group(2)
+ # If the bug was reopened, mark it as such
+ if changedfromto.find("closed->analyzed") != -1:
+ if self.fields["bug_status"] == "'NEW'":
+ self.fields["bug_status"] = "'REOPENED'"
+ if type == "State" or type == "Responsible":
+ oldstate, newstate = self.parse_fromto (type, changedfromto)
+ result = changedbyre.search(piece)
+ if result is not None:
+ changedby = result.group(1)
+ result = changedwhenre.search(piece)
+ if result is not None:
+ changedwhen = result.group(1)
+ changedwhen = unixdate2datetime(changedwhen)
+ changedwhen = SqlQuote(changedwhen)
+ result = changedwhyre.search(piece)
+ changedwhy = piece[result.start(1):]
+ #changedwhy = changedwhy.lstrip()
+ changedwhy = changedwhy.rstrip()
+ changedby = get_userid(changedby)
+ # Put us on the cc list if we aren't there already
+ if changedby != self.fields["userid"] \
+ and changedby not in self.bug_ccs:
+ self.bug_ccs.append(changedby)
+ # If it's a duplicate, mark it as such
+ result = duplicatere.search(changedwhy)
+ if result is not None:
+ newtext = "*** This bug has been marked as a duplicate of %s ***" % result.group(1)
+ newtext = SqlQuote(newtext)
+ self.long_descs.append((self.bug_id, changedby,
+ changedwhen, newtext))
+ self.fields["bug_status"] = "'RESOLVED'"
+ self.fields["resolution"] = "'DUPLICATE'"
+ self.fields["userid"] = changedby
+ else:
+ newtext = "%s-Changed-From-To: %s\n%s-Changed-Why: %s\n" % (type, changedfromto, type, changedwhy)
+ newtext = SqlQuote(newtext)
+ self.long_descs.append((self.bug_id, changedby,
+ changedwhen, newtext))
+ if type == "State" or type == "Responsible":
+ newstate = SqlQuote("%s" % newstate)
+ oldstate = SqlQuote("%s" % oldstate)
+ fieldid = fieldids[type]
+ self.bug_activity.append((newstate, oldstate, fieldid, changedby, changedwhen))
+
+ else:
+ # It's an email
+ result = fromre.search(piece)
+ if result is None:
+ continue
+ fromstr = result.group(1)
+ fromstr = fix_email_addrs(fromstr)
+ fromstr = get_userid(fromstr)
+ result = datere.search(piece)
+ if result is None:
+ continue
+ datestr = result.group(1)
+ datestr = SqlQuote(unixdate2timestamp(datestr))
+ if fromstr != self.fields["userid"] \
+ and fromstr not in self.bug_ccs:
+ self.bug_ccs.append(fromstr)
+ self.long_descs.append((self.bug_id, fromstr, datestr,
+ SqlQuote(piece)))
+
+
+
+ def write_bug(self):
+ """ Output a bug to the data file """
+ fields = self.fields
+ print >>outfile, "\ninsert into bugs("
+ print >>outfile, " bug_id, assigned_to, bug_severity, priority, bug_status, creation_ts, delta_ts,"
+ print >>outfile, " short_desc,"
+ print >>outfile, " reporter, version,"
+ print >>outfile, " product, component, resolution, target_milestone, qa_contact,"
+ print >>outfile, " gccbuild, gcctarget, gcchost, keywords"
+ print >>outfile, " ) values ("
+ print >>outfile, "%s, %s, %s, %s, %s, %s, %s," % (self.bug_id, fields["userid"], fields["bug_severity"], fields["priority"], fields["bug_status"], fields["creation_ts"], fields["delta_ts"])
+ print >>outfile, "%s," % (fields["short_desc"])
+ print >>outfile, "%s, %s," % (fields["reporter"], fields["version"])
+ print >>outfile, "%s, %s, %s, %s, 0," %(fields["product"], fields["component"], fields["resolution"], fields["target_milestone"])
+ print >>outfile, "%s, %s, %s, %s" % (fields["gccbuild"], fields["gcctarget"], fields["gcchost"], fields["keywords"])
+ print >>outfile, ");"
+ if self.fields["keywords"] != 0:
+ print >>outfile, "\ninsert into keywords (bug_id, keywordid) values ("
+ print >>outfile, " %s, %s);" % (self.bug_id, fields["keywordid"])
+ for id, who, when, text in self.long_descs:
+ print >>outfile, "\ninsert into longdescs ("
+ print >>outfile, " bug_id, who, bug_when, thetext) values("
+ print >>outfile, " %s, %s, %s, %s);" % (id, who, when, text)
+ for name, data, who in self.attachments:
+ print >>outfile, "\ninsert into attachments ("
+ print >>outfile, " bug_id, filename, description, mimetype, ispatch, submitter_id, thedata) values ("
+ ftype = None
+ # It's *magic*!
+ if name.endswith(".ii") == 1:
+ ftype = "text/x-c++"
+ elif name.endswith(".i") == 1:
+ ftype = "text/x-c"
+ else:
+ ftype = magicf.detect(cStringIO.StringIO(data))
+ if ftype is None:
+ ftype = "application/octet-stream"
+
+ print >>outfile, "%s,%s,%s, %s,0, %s,%s);" %(self.bug_id, SqlQuote(name), SqlQuote(name), SqlQuote (ftype), who, SqlQuote(zlib.compress(data)))
+ for newstate, oldstate, fieldid, changedby, changedwhen in self.bug_activity:
+ print >>outfile, "\ninsert into bugs_activity ("
+ print >>outfile, " bug_id, who, bug_when, fieldid, added, removed) values ("
+ print >>outfile, " %s, %s, %s, %s, %s, %s);" % (self.bug_id,
+ changedby,
+ changedwhen,
+ fieldid,
+ newstate,
+ oldstate)
+ for cc in self.bug_ccs:
+ print >>outfile, "\ninsert into cc(bug_id, who) values (%s, %s);" %(self.bug_id, cc)
+ def update_versions(self):
+ """ Update the versions table to account for the version on this bug """
+ global versions_table
+ if self.fields.has_key("Release") == 0 \
+ or self.fields.has_key("Category") == 0:
+ return
+ curr_product = "gcc"
+ curr_version = self.fields["Release"]
+ if curr_version == "":
+ return
+ curr_version = convert_gccver_to_ver (curr_version)
+ if versions_table.has_key(curr_product) == 0:
+ versions_table[curr_product] = []
+ for version in versions_table[curr_product]:
+ if version == curr_version:
+ return
+ versions_table[curr_product].append(curr_version)
+ def translate_pr(self):
+ """ Transform a GNATS PR into a Bugzilla bug """
+ self.fields = self.gnatsfields
+ if (self.fields.has_key("Organization") == 0) \
+ or self.fields["Organization"].find("GCC"):
+ self.fields["Originator"] = ""
+ self.fields["Organization"] = ""
+ self.fields["Organization"].lstrip()
+ if (self.fields.has_key("Release") == 0) \
+ or self.fields["Release"] == "" \
+ or self.fields["Release"].find("unknown-1.0") != -1:
+ self.fields["Release"]="unknown"
+ if self.fields.has_key("Responsible"):
+ result = re.search(r"""\w+""", self.fields["Responsible"])
+ self.fields["Responsible"] = "%s%s" % (result.group(0), "@gcc.gnu.org")
+ self.fields["gcchost"] = ""
+ self.fields["gcctarget"] = ""
+ self.fields["gccbuild"] = ""
+ if self.fields.has_key("Environment"):
+ result = re.search("^host: (.+?)$", self.fields["Environment"],
+ re.MULTILINE)
+ if result is not None:
+ self.fields["gcchost"] = result.group(1)
+ result = re.search("^target: (.+?)$", self.fields["Environment"],
+ re.MULTILINE)
+ if result is not None:
+ self.fields["gcctarget"] = result.group(1)
+ result = re.search("^build: (.+?)$", self.fields["Environment"],
+ re.MULTILINE)
+ if result is not None:
+ self.fields["gccbuild"] = result.group(1)
+ self.fields["userid"] = get_userid(self.fields["Responsible"])
+ self.fields["bug_severity"] = "normal"
+ if self.fields["Class"] == "change-request":
+ self.fields["bug_severity"] = "enhancement"
+ elif self.fields.has_key("Severity"):
+ if self.fields["Severity"] == "critical":
+ self.fields["bug_severity"] = "critical"
+ elif self.fields["Severity"] == "serious":
+ self.fields["bug_severity"] = "major"
+ elif self.fields.has_key("Synopsis"):
+ if re.search("crash|assert", self.fields["Synopsis"]):
+ self.fields["bug_severity"] = "critical"
+ elif re.search("wrong|error", self.fields["Synopsis"]):
+ self.fields["bug_severity"] = "major"
+ self.fields["bug_severity"] = SqlQuote(self.fields["bug_severity"])
+ self.fields["keywords"] = 0
+ if keywordids.has_key(self.fields["Class"]):
+ self.fields["keywords"] = self.fields["Class"]
+ self.fields["keywordid"] = keywordids[self.fields["Class"]]
+ self.fields["keywords"] = SqlQuote(self.fields["keywords"])
+ self.fields["priority"] = "P1"
+ if self.fields.has_key("Severity") and self.fields.has_key("Priority"):
+ severity = self.fields["Severity"]
+ priority = self.fields["Priority"]
+ if severity == "critical":
+ if priority == "high":
+ self.fields["priority"] = "P1"
+ else:
+ self.fields["priority"] = "P2"
+ elif severity == "serious":
+ if priority == "low":
+ self.fields["priority"] = "P4"
+ else:
+ self.fields["priority"] = "P3"
+ else:
+ if priority == "high":
+ self.fields["priority"] = "P4"
+ else:
+ self.fields["priority"] = "P5"
+ self.fields["priority"] = SqlQuote(self.fields["priority"])
+ state = self.fields["State"]
+ if (state == "open" or state == "analyzed") and self.fields["userid"] != 3:
+ self.fields["bug_status"] = "ASSIGNED"
+ self.fields["resolution"] = ""
+ elif state == "feedback":
+ self.fields["bug_status"] = "WAITING"
+ self.fields["resolution"] = ""
+ elif state == "closed":
+ self.fields["bug_status"] = "CLOSED"
+ if self.fields.has_key("Class"):
+ theclass = self.fields["Class"]
+ if theclass.find("duplicate") != -1:
+ self.fields["resolution"]="DUPLICATE"
+ elif theclass.find("mistaken") != -1:
+ self.fields["resolution"]="INVALID"
+ else:
+ self.fields["resolution"]="FIXED"
+ else:
+ self.fields["resolution"]="FIXED"
+ elif state == "suspended":
+ self.fields["bug_status"] = "SUSPENDED"
+ self.fields["resolution"] = ""
+ elif state == "analyzed" and self.fields["userid"] == 3:
+ self.fields["bug_status"] = "NEW"
+ self.fields["resolution"] = ""
+ else:
+ self.fields["bug_status"] = "UNCONFIRMED"
+ self.fields["resolution"] = ""
+ self.fields["bug_status"] = SqlQuote(self.fields["bug_status"])
+ self.fields["resolution"] = SqlQuote(self.fields["resolution"])
+ self.fields["creation_ts"] = ""
+ if self.fields.has_key("Arrival-Date") and self.fields["Arrival-Date"] != "":
+ self.fields["creation_ts"] = unixdate2datetime(self.fields["Arrival-Date"])
+ self.fields["creation_ts"] = SqlQuote(self.fields["creation_ts"])
+ self.fields["delta_ts"] = ""
+ if self.fields.has_key("Audit-Trail"):
+ result = lastdatere.findall(self.fields["Audit-Trail"])
+ result.reverse()
+ if len(result) > 0:
+ self.fields["delta_ts"] = unixdate2timestamp(result[0])
+ if self.fields["delta_ts"] == "":
+ if self.fields.has_key("Arrival-Date") and self.fields["Arrival-Date"] != "":
+ self.fields["delta_ts"] = unixdate2timestamp(self.fields["Arrival-Date"])
+ self.fields["delta_ts"] = SqlQuote(self.fields["delta_ts"])
+ self.fields["short_desc"] = SqlQuote(self.fields["Synopsis"])
+ if self.fields.has_key("Reply-To") and self.fields["Reply-To"] != "":
+ self.fields["reporter"] = get_userid(self.fields["Reply-To"])
+ elif self.fields.has_key("Mail-Header"):
+ result = re.search(r"""From .*?([\w.]+@[\w.]+)""", self.fields["Mail-Header"])
+ if result:
+ self.fields["reporter"] = get_userid(result.group(1))
+ else:
+ self.fields["reporter"] = get_userid(gnats_username)
+ else:
+ self.fields["reporter"] = get_userid(gnats_username)
+ long_desc = self.fields["Description"]
+ long_desc2 = ""
+ for field in ["Release", "Environment", "How-To-Repeat"]:
+ if self.fields.has_key(field) and self.fields[field] != "":
+ long_desc += ("\n\n%s:\n" % field) + self.fields[field]
+ if self.fields.has_key("Fix") and self.fields["Fix"] != "":
+ long_desc2 = "Fix:\n" + self.fields["Fix"]
+ if self.need_unformatted == 1 and self.fields["Unformatted"] != "":
+ long_desc += "\n\nUnformatted:\n" + self.fields["Unformatted"]
+ if long_desc != "":
+ self.long_descs.append((self.bug_id, self.fields["reporter"],
+ self.fields["creation_ts"],
+ SqlQuote(long_desc)))
+ if long_desc2 != "":
+ self.long_descs.append((self.bug_id, self.fields["reporter"],
+ self.fields["creation_ts"],
+ SqlQuote(long_desc2)))
+ for field in ["gcchost", "gccbuild", "gcctarget"]:
+ self.fields[field] = SqlQuote(self.fields[field])
+ self.fields["version"] = ""
+ if self.fields["Release"] != "":
+ self.fields["version"] = convert_gccver_to_ver (self.fields["Release"])
+ self.fields["version"] = SqlQuote(self.fields["version"])
+ self.fields["product"] = SqlQuote("gcc")
+ self.fields["component"] = "invalid"
+ if self.fields.has_key("Category"):
+ self.fields["component"] = self.fields["Category"]
+ self.fields["component"] = SqlQuote(self.fields["component"])
+ self.fields["target_milestone"] = "---"
+ if self.fields["version"].find("3.4") != -1:
+ self.fields["target_milestone"] = "3.4"
+ self.fields["target_milestone"] = SqlQuote(self.fields["target_milestone"])
+ if self.fields["userid"] == 2:
+ self.fields["userid"] = "\'NULL\'"
+
+class GNATSbug(object):
+ """ Represents a single GNATS PR """
+ def __init__(self, filename):
+ self.attachments = []
+ self.has_unformatted_attach = 0
+ fp = open (filename)
+ self.fields = self.parse_pr(fp.xreadlines())
+ self.bug_id = int(self.fields["Number"])
+ if self.fields.has_key("Unformatted"):
+ self.find_gnatsweb_attachments()
+ if self.fields.has_key("How-To-Repeat"):
+ self.find_regular_attachments("How-To-Repeat")
+ if self.fields.has_key("Fix"):
+ self.find_regular_attachments("Fix")
+
+ def get_attacher(fields):
+ if fields.has_key("Reply-To") and fields["Reply-To"] != "":
+ return get_userid(fields["Reply-To"])
+ else:
+ result = None
+ if fields.has_key("Mail-Header"):
+ result = re.search(r"""From .*?([\w.]+\@[\w.]+)""",
+ fields["Mail-Header"])
+ if result is not None:
+ reporter = get_userid(result.group(1))
+ else:
+ reporter = get_userid(gnats_username)
+ get_attacher = staticmethod(get_attacher)
+ def find_regular_attachments(self, which):
+ fields = self.fields
+ while re.search("^begin [0-7]{3}", fields[which],
+ re.DOTALL | re.MULTILINE):
+ outfp = cStringIO.StringIO()
+ infp = cStringIO.StringIO(fields[which])
+ filename, start, end = specialuu.decode(infp, outfp, quiet=0)
+ fields[which]=fields[which].replace(fields[which][start:end],
+ "See attachments for %s\n" % filename)
+ self.attachments.append((filename, outfp.getvalue(),
+ self.get_attacher(fields)))
+
+ def decode_gnatsweb_attachment(self, attachment):
+ result = re.split(r"""\n\n""", attachment, 1)
+ if len(result) == 1:
+ return -1
+ envelope, body = result
+ envelope = uselessre.split(envelope)
+ envelope.pop(0)
+ # Turn the list of key, value into a dict of key => value
+ attachinfo = dict([(envelope[i], envelope[i+1]) for i in xrange(0,len(envelope),2)])
+ for x in attachinfo.keys():
+ attachinfo[x] = attachinfo[x].rstrip()
+ if (attachinfo.has_key("Content-Type") == 0) or \
+ (attachinfo.has_key("Content-Disposition") == 0):
+ raise ValueError, "Unable to parse file attachment"
+ result = dispositionre.search(attachinfo["Content-Disposition"])
+ filename = result.group(2)
+ filename = re.sub(".*/","", filename)
+ filename = re.sub(".*\\\\","", filename)
+ attachinfo["filename"]=filename
+ result = re.search("""(\S+);.*""", attachinfo["Content-Type"])
+ if result is not None:
+ attachinfo["Content-Type"] = result.group(1)
+ if attachinfo.has_key("Content-Transfer-Encoding"):
+ if attachinfo["Content-Transfer-Encoding"] == "base64":
+ attachinfo["data"] = base64.decodestring(body)
+ else:
+ attachinfo["data"]=body
+
+ return (attachinfo["filename"], attachinfo["data"],
+ self.get_attacher(self.fields))
+
+ def find_gnatsweb_attachments(self):
+ fields = self.fields
+ attachments = re.split(attachment_delimiter, fields["Unformatted"])
+ fields["Unformatted"] = attachments.pop(0)
+ for attachment in attachments:
+ result = self.decode_gnatsweb_attachment (attachment)
+ if result != -1:
+ self.attachments.append(result)
+ self.has_unformatted_attach = 1
+ def parse_pr(lines):
+ #fields = {"envelope":[]}
+ fields = {"envelope":array.array("c")}
+ hdrmulti = "envelope"
+ for line in lines:
+ line = line.rstrip('\n')
+ line += '\n'
+ result = gnatfieldre.search(line)
+ if result is None:
+ if hdrmulti != "":
+ if fields.has_key(hdrmulti):
+ #fields[hdrmulti].append(line)
+ fields[hdrmulti].fromstring(line)
+ else:
+ #fields[hdrmulti] = [line]
+ fields[hdrmulti] = array.array("c", line)
+ continue
+ hdr, arg = result.groups()
+ ghdr = "*not valid*"
+ result = fieldnamere.search(hdr)
+ if result != None:
+ ghdr = result.groups()[0]
+ if ghdr in fieldnames:
+ if multilinefields.has_key(ghdr):
+ hdrmulti = ghdr
+ #fields[ghdr] = [""]
+ fields[ghdr] = array.array("c")
+ else:
+ hdrmulti = ""
+ #fields[ghdr] = [arg]
+ fields[ghdr] = array.array("c", arg)
+ elif hdrmulti != "":
+ #fields[hdrmulti].append(line)
+ fields[hdrmulti].fromstring(line)
+ if hdrmulti == "envelope" and \
+ (hdr == "Reply-To" or hdr == "From" \
+ or hdr == "X-GNATS-Notify"):
+ arg = fix_email_addrs(arg)
+ #fields[hdr] = [arg]
+ fields[hdr] = array.array("c", arg)
+ if fields.has_key("Reply-To") and len(fields["Reply-To"]) > 0:
+ fields["Reply-To"] = fields["Reply-To"]
+ else:
+ fields["Reply-To"] = fields["From"]
+ if fields.has_key("From"):
+ del fields["From"]
+ if fields.has_key("X-GNATS-Notify") == 0:
+ fields["X-GNATS-Notify"] = array.array("c")
+ #fields["X-GNATS-Notify"] = ""
+ for x in fields.keys():
+ fields[x] = fields[x].tostring()
+ #fields[x] = "".join(fields[x])
+ for x in fields.keys():
+ if multilinefields.has_key(x):
+ fields[x] = fields[x].rstrip()
+
+ return fields
+ parse_pr = staticmethod(parse_pr)
+load_index("%s/gnats-adm/index" % gnats_db_dir)
+load_categories("%s/gnats-adm/categories" % gnats_db_dir)
+load_responsible("%s/gnats-adm/responsible" % gnats_db_dir)
+get_userid(gnats_username)
+get_userid(unassigned_username)
+for x in pr_list:
+ print "Processing %s..." % x
+ a = GNATSbug ("%s/%s" % (gnats_db_dir, x))
+ b = Bugzillabug(a)
+write_non_bug_tables()
+outfile.close()
--- /dev/null
+# Found on a russian zope mailing list, and modified to fix bugs in parsing
+# the magic file and string making
+# -- Daniel Berlin <dberlin@dberlin.org>
+import sys, struct, time, re, exceptions, pprint, stat, os, pwd, grp
+
+_mew = 0
+
+# _magic='/tmp/magic'
+# _magic='/usr/share/magic.mime'
+_magic='/usr/share/magic.mime'
+mime = 1
+
+_ldate_adjust = lambda x: time.mktime( time.gmtime(x) )
+
+BUFFER_SIZE = 1024 * 128 # 128K should be enough...
+
+class MagicError(exceptions.Exception): pass
+
+def _handle(fmt='@x',adj=None): return fmt, struct.calcsize(fmt), adj
+
+KnownTypes = {
+ # 'byte':_handle('@b'),
+ 'byte':_handle('@B'),
+ 'ubyte':_handle('@B'),
+
+ 'string':('s',0,None),
+ 'pstring':_handle('p'),
+
+# 'short':_handle('@h'),
+# 'beshort':_handle('>h'),
+# 'leshort':_handle('<h'),
+ 'short':_handle('@H'),
+ 'beshort':_handle('>H'),
+ 'leshort':_handle('<H'),
+ 'ushort':_handle('@H'),
+ 'ubeshort':_handle('>H'),
+ 'uleshort':_handle('<H'),
+
+ 'long':_handle('@l'),
+ 'belong':_handle('>l'),
+ 'lelong':_handle('<l'),
+ 'ulong':_handle('@L'),
+ 'ubelong':_handle('>L'),
+ 'ulelong':_handle('<L'),
+
+ 'date':_handle('=l'),
+ 'bedate':_handle('>l'),
+ 'ledate':_handle('<l'),
+ 'ldate':_handle('=l',_ldate_adjust),
+ 'beldate':_handle('>l',_ldate_adjust),
+ 'leldate':_handle('<l',_ldate_adjust),
+}
+
+_mew_cnt = 0
+def mew(x):
+ global _mew_cnt
+ if _mew :
+ if x=='.' :
+ _mew_cnt += 1
+ if _mew_cnt % 64 == 0 : sys.stderr.write( '\n' )
+ sys.stderr.write( '.' )
+ else:
+ sys.stderr.write( '\b'+x )
+
+def has_format(s):
+ n = 0
+ l = None
+ for c in s :
+ if c == '%' :
+ if l == '%' : n -= 1
+ else : n += 1
+ l = c
+ return n
+
+def read_asciiz(file,size=None,pos=None):
+ s = []
+ if pos :
+ mew('s')
+ file.seek( pos, 0 )
+ mew('z')
+ if size is not None :
+ s = [file.read( size ).split('\0')[0]]
+ else:
+ while 1 :
+ c = file.read(1)
+ if (not c) or (ord(c)==0) or (c=='\n') : break
+ s.append (c)
+ mew('Z')
+ return ''.join(s)
+
+def a2i(v,base=0):
+ if v[-1:] in 'lL' : v = v[:-1]
+ return int( v, base )
+
+_cmap = {
+ '\\' : '\\',
+ '0' : '\0',
+}
+for c in range(ord('a'),ord('z')+1) :
+ try : e = eval('"\\%c"' % chr(c))
+ except ValueError : pass
+ else : _cmap[chr(c)] = e
+else:
+ del c
+ del e
+
+def make_string(s):
+ return eval( '"'+s.replace('"','\\"')+'"')
+
+class MagicTestError(MagicError): pass
+
+class MagicTest:
+ def __init__(self,offset,mtype,test,message,line=None,level=None):
+ self.line, self.level = line, level
+ self.mtype = mtype
+ self.mtest = test
+ self.subtests = []
+ self.mask = None
+ self.smod = None
+ self.nmod = None
+ self.offset, self.type, self.test, self.message = \
+ offset,mtype,test,message
+ if self.mtype == 'true' : return # XXX hack to enable level skips
+ if test[-1:]=='\\' and test[-2:]!='\\\\' :
+ self.test += 'n' # looks like someone wanted EOL to match?
+ if mtype[:6]=='string' :
+ if '/' in mtype : # for strings
+ self.type, self.smod = \
+ mtype[:mtype.find('/')], mtype[mtype.find('/')+1:]
+ else:
+ for nm in '&+-' :
+ if nm in mtype : # for integer-based
+ self.nmod, self.type, self.mask = (
+ nm,
+ mtype[:mtype.find(nm)],
+ # convert mask to int, autodetect base
+ int( mtype[mtype.find(nm)+1:], 0 )
+ )
+ break
+ self.struct, self.size, self.cast = KnownTypes[ self.type ]
+ def __str__(self):
+ return '%s %s %s %s' % (
+ self.offset, self.mtype, self.mtest, self.message
+ )
+ def __repr__(self):
+ return 'MagicTest(%s,%s,%s,%s,line=%s,level=%s,subtests=\n%s%s)' % (
+ `self.offset`, `self.mtype`, `self.mtest`, `self.message`,
+ `self.line`, `self.level`,
+ '\t'*self.level, pprint.pformat(self.subtests)
+ )
+ def run(self,file):
+ result = ''
+ do_close = 0
+ try:
+ if type(file) == type('x') :
+ file = open( file, 'r', BUFFER_SIZE )
+ do_close = 1
+# else:
+# saved_pos = file.tell()
+ if self.mtype != 'true' :
+ data = self.read(file)
+ last = file.tell()
+ else:
+ data = last = None
+ if self.check( data ) :
+ result = self.message+' '
+ if has_format( result ) : result %= data
+ for test in self.subtests :
+ m = test.run(file)
+ if m is not None : result += m
+ return make_string( result )
+ finally:
+ if do_close :
+ file.close()
+# else:
+# file.seek( saved_pos, 0 )
+ def get_mod_and_value(self):
+ if self.type[-6:] == 'string' :
+ # "something like\tthis\n"
+ if self.test[0] in '=<>' :
+ mod, value = self.test[0], make_string( self.test[1:] )
+ else:
+ mod, value = '=', make_string( self.test )
+ else:
+ if self.test[0] in '=<>&^' :
+ mod, value = self.test[0], a2i(self.test[1:])
+ elif self.test[0] == 'x':
+ mod = self.test[0]
+ value = 0
+ else:
+ mod, value = '=', a2i(self.test)
+ return mod, value
+ def read(self,file):
+ mew( 's' )
+ file.seek( self.offset(file), 0 ) # SEEK_SET
+ mew( 'r' )
+ try:
+ data = rdata = None
+ # XXX self.size might be 0 here...
+ if self.size == 0 :
+ # this is an ASCIIZ string...
+ size = None
+ if self.test != '>\\0' : # magic's hack for string read...
+ value = self.get_mod_and_value()[1]
+ size = (value=='\0') and None or len(value)
+ rdata = data = read_asciiz( file, size=size )
+ else:
+ rdata = file.read( self.size )
+ if not rdata or (len(rdata)!=self.size) : return None
+ data = struct.unpack( self.struct, rdata )[0] # XXX hack??
+ except:
+ print >>sys.stderr, self
+ print >>sys.stderr, '@%s struct=%s size=%d rdata=%s' % (
+ self.offset, `self.struct`, self.size,`rdata`)
+ raise
+ mew( 'R' )
+ if self.cast : data = self.cast( data )
+ if self.mask :
+ try:
+ if self.nmod == '&' : data &= self.mask
+ elif self.nmod == '+' : data += self.mask
+ elif self.nmod == '-' : data -= self.mask
+ else: raise MagicTestError(self.nmod)
+ except:
+ print >>sys.stderr,'data=%s nmod=%s mask=%s' % (
+ `data`, `self.nmod`, `self.mask`
+ )
+ raise
+ return data
+ def check(self,data):
+ mew('.')
+ if self.mtype == 'true' :
+ return '' # not None !
+ mod, value = self.get_mod_and_value()
+ if self.type[-6:] == 'string' :
+ # "something like\tthis\n"
+ if self.smod :
+ xdata = data
+ if 'b' in self.smod : # all blanks are optional
+ xdata = ''.join( data.split() )
+ value = ''.join( value.split() )
+ if 'c' in self.smod : # all blanks are optional
+ xdata = xdata.upper()
+ value = value.upper()
+ # if 'B' in self.smod : # compact blanks
+ ### XXX sorry, i don't understand this :-(
+ # data = ' '.join( data.split() )
+ # if ' ' not in data : return None
+ else:
+ xdata = data
+ try:
+ if mod == '=' : result = data == value
+ elif mod == '<' : result = data < value
+ elif mod == '>' : result = data > value
+ elif mod == '&' : result = data & value
+ elif mod == '^' : result = (data & (~value)) == 0
+ elif mod == 'x' : result = 1
+ else : raise MagicTestError(self.test)
+ if result :
+ zdata, zval = `data`, `value`
+ if self.mtype[-6:]!='string' :
+ try: zdata, zval = hex(data), hex(value)
+ except: zdata, zval = `data`, `value`
+ if 0 : print >>sys.stderr, '%s @%s %s:%s %s %s => %s (%s)' % (
+ '>'*self.level, self.offset,
+ zdata, self.mtype, `mod`, zval, `result`,
+ self.message
+ )
+ return result
+ except:
+ print >>sys.stderr,'mtype=%s data=%s mod=%s value=%s' % (
+ `self.mtype`, `data`, `mod`, `value`
+ )
+ raise
+ def add(self,mt):
+ if not isinstance(mt,MagicTest) :
+ raise MagicTestError((mt,'incorrect subtest type %s'%(type(mt),)))
+ if mt.level == self.level+1 :
+ self.subtests.append( mt )
+ elif self.subtests :
+ self.subtests[-1].add( mt )
+ elif mt.level > self.level+1 :
+ # it's possible to get level 3 just after level 1 !!! :-(
+ level = self.level + 1
+ while level < mt.level :
+ xmt = MagicTest(None,'true','x','',line=self.line,level=level)
+ self.add( xmt )
+ level += 1
+ else:
+ self.add( mt ) # retry...
+ else:
+ raise MagicTestError((mt,'incorrect subtest level %s'%(`mt.level`,)))
+ def last_test(self):
+ return self.subtests[-1]
+#end class MagicTest
+
+class OffsetError(MagicError): pass
+
+class Offset:
+ pos_format = {'b':'<B','B':'>B','s':'<H','S':'>H','l':'<I','L':'>I',}
+ pattern0 = re.compile(r''' # mere offset
+ ^
+ &? # possible ampersand
+ ( 0 # just zero
+ | [1-9]{1,1}[0-9]* # decimal
+ | 0[0-7]+ # octal
+ | 0x[0-9a-f]+ # hex
+ )
+ $
+ ''', re.X|re.I
+ )
+ pattern1 = re.compile(r''' # indirect offset
+ ^\(
+ (?P<base>&?0 # just zero
+ |&?[1-9]{1,1}[0-9]* # decimal
+ |&?0[0-7]* # octal
+ |&?0x[0-9A-F]+ # hex
+ )
+ (?P<type>
+ \. # this dot might be alone
+ [BSL]? # one of this chars in either case
+ )?
+ (?P<sign>
+ [-+]{0,1}
+ )?
+ (?P<off>0 # just zero
+ |[1-9]{1,1}[0-9]* # decimal
+ |0[0-7]* # octal
+ |0x[0-9a-f]+ # hex
+ )?
+ \)$''', re.X|re.I
+ )
+ def __init__(self,s):
+ self.source = s
+ self.value = None
+ self.relative = 0
+ self.base = self.type = self.sign = self.offs = None
+ m = Offset.pattern0.match( s )
+ if m : # just a number
+ if s[0] == '&' :
+ self.relative, self.value = 1, int( s[1:], 0 )
+ else:
+ self.value = int( s, 0 )
+ return
+ m = Offset.pattern1.match( s )
+ if m : # real indirect offset
+ try:
+ self.base = m.group('base')
+ if self.base[0] == '&' :
+ self.relative, self.base = 1, int( self.base[1:], 0 )
+ else:
+ self.base = int( self.base, 0 )
+ if m.group('type') : self.type = m.group('type')[1:]
+ self.sign = m.group('sign')
+ if m.group('off') : self.offs = int( m.group('off'), 0 )
+ if self.sign == '-' : self.offs = 0 - self.offs
+ except:
+ print >>sys.stderr, '$$', m.groupdict()
+ raise
+ return
+ raise OffsetError(`s`)
+ def __call__(self,file=None):
+ if self.value is not None : return self.value
+ pos = file.tell()
+ try:
+ if not self.relative : file.seek( self.offset, 0 )
+ frmt = Offset.pos_format.get( self.type, 'I' )
+ size = struct.calcsize( frmt )
+ data = struct.unpack( frmt, file.read( size ) )
+ if self.offs : data += self.offs
+ return data
+ finally:
+ file.seek( pos, 0 )
+ def __str__(self): return self.source
+ def __repr__(self): return 'Offset(%s)' % `self.source`
+#end class Offset
+
+class MagicFileError(MagicError): pass
+
+class MagicFile:
+ def __init__(self,filename=_magic):
+ self.file = None
+ self.tests = []
+ self.total_tests = 0
+ self.load( filename )
+ self.ack_tests = None
+ self.nak_tests = None
+ def __del__(self):
+ self.close()
+ def load(self,filename=None):
+ self.open( filename )
+ self.parse()
+ self.close()
+ def open(self,filename=None):
+ self.close()
+ if filename is not None :
+ self.filename = filename
+ self.file = open( self.filename, 'r', BUFFER_SIZE )
+ def close(self):
+ if self.file :
+ self.file.close()
+ self.file = None
+ def parse(self):
+ line_no = 0
+ for line in self.file.xreadlines() :
+ line_no += 1
+ if not line or line[0]=='#' : continue
+ line = line.lstrip().rstrip('\r\n')
+ if not line or line[0]=='#' : continue
+ try:
+ x = self.parse_line( line )
+ if x is None :
+ print >>sys.stderr, '#[%04d]#'%line_no, line
+ continue
+ except:
+ print >>sys.stderr, '###[%04d]###'%line_no, line
+ raise
+ self.total_tests += 1
+ level, offset, mtype, test, message = x
+ new_test = MagicTest(offset,mtype,test,message,
+ line=line_no,level=level)
+ try:
+ if level == 0 :
+ self.tests.append( new_test )
+ else:
+ self.tests[-1].add( new_test )
+ except:
+ if 1 :
+ print >>sys.stderr, 'total tests=%s' % (
+ `self.total_tests`,
+ )
+ print >>sys.stderr, 'level=%s' % (
+ `level`,
+ )
+ print >>sys.stderr, 'tests=%s' % (
+ pprint.pformat(self.tests),
+ )
+ raise
+ else:
+ while self.tests[-1].level > 0 :
+ self.tests.pop()
+ def parse_line(self,line):
+ # print >>sys.stderr, 'line=[%s]' % line
+ if (not line) or line[0]=='#' : return None
+ level = 0
+ offset = mtype = test = message = ''
+ mask = None
+ # get optional level (count leading '>')
+ while line and line[0]=='>' :
+ line, level = line[1:], level+1
+ # get offset
+ while line and not line[0].isspace() :
+ offset, line = offset+line[0], line[1:]
+ try:
+ offset = Offset(offset)
+ except:
+ print >>sys.stderr, 'line=[%s]' % line
+ raise
+ # skip spaces
+ line = line.lstrip()
+ # get type
+ c = None
+ while line :
+ last_c, c, line = c, line[0], line[1:]
+ if last_c!='\\' and c.isspace() :
+ break # unescaped space - end of field
+ else:
+ mtype += c
+ if last_c == '\\' :
+ c = None # don't fuck my brain with sequential backslashes
+ # skip spaces
+ line = line.lstrip()
+ # get test
+ c = None
+ while line :
+ last_c, c, line = c, line[0], line[1:]
+ if last_c!='\\' and c.isspace() :
+ break # unescaped space - end of field
+ else:
+ test += c
+ if last_c == '\\' :
+ c = None # don't fuck my brain with sequential backslashes
+ # skip spaces
+ line = line.lstrip()
+ # get message
+ message = line
+ if mime and line.find("\t") != -1:
+ message=line[0:line.find("\t")]
+ #
+ # print '>>', level, offset, mtype, test, message
+ return level, offset, mtype, test, message
+ def detect(self,file):
+ self.ack_tests = 0
+ self.nak_tests = 0
+ answers = []
+ for test in self.tests :
+ message = test.run( file )
+ if message :
+ self.ack_tests += 1
+ answers.append( message )
+ else:
+ self.nak_tests += 1
+ if answers :
+ return '; '.join( answers )
+#end class MagicFile
+
+def username(uid):
+ try:
+ return pwd.getpwuid( uid )[0]
+ except:
+ return '#%s'%uid
+
+def groupname(gid):
+ try:
+ return grp.getgrgid( gid )[0]
+ except:
+ return '#%s'%gid
+
+def get_file_type(fname,follow):
+ t = None
+ if not follow :
+ try:
+ st = os.lstat( fname ) # stat that entry, don't follow links!
+ except os.error, why :
+ pass
+ else:
+ if stat.S_ISLNK(st[stat.ST_MODE]) :
+ t = 'symbolic link'
+ try:
+ lnk = os.readlink( fname )
+ except:
+ t += ' (unreadable)'
+ else:
+ t += ' to '+lnk
+ if t is None :
+ try:
+ st = os.stat( fname )
+ except os.error, why :
+ return "can't stat `%s' (%s)." % (why.filename,why.strerror)
+
+ dmaj, dmin = (st.st_rdev>>8)&0x0FF, st.st_rdev&0x0FF
+
+ if 0 : pass
+ elif stat.S_ISSOCK(st.st_mode) : t = 'socket'
+ elif stat.S_ISLNK (st.st_mode) : t = follow and 'symbolic link' or t
+ elif stat.S_ISREG (st.st_mode) : t = 'file'
+ elif stat.S_ISBLK (st.st_mode) : t = 'block special (%d/%d)'%(dmaj,dmin)
+ elif stat.S_ISDIR (st.st_mode) : t = 'directory'
+ elif stat.S_ISCHR (st.st_mode) : t = 'character special (%d/%d)'%(dmaj,dmin)
+ elif stat.S_ISFIFO(st.st_mode) : t = 'pipe'
+ else: t = '<unknown>'
+
+ if st.st_mode & stat.S_ISUID :
+ t = 'setuid(%d=%s) %s'%(st.st_uid,username(st.st_uid),t)
+ if st.st_mode & stat.S_ISGID :
+ t = 'setgid(%d=%s) %s'%(st.st_gid,groupname(st.st_gid),t)
+ if st.st_mode & stat.S_ISVTX :
+ t = 'sticky '+t
+
+ return t
+
+HELP = '''%s [options] [files...]
+
+Options:
+
+ -?, --help -- this help
+ -m, --magic=<file> -- use this magic <file> instead of %s
+ -f, --files=<namefile> -- read filenames for <namefile>
+* -C, --compile -- write "compiled" magic file
+ -b, --brief -- don't prepend filenames to output lines
++ -c, --check -- check the magic file
+ -i, --mime -- output MIME types
+* -k, --keep-going -- don't stop st the first match
+ -n, --flush -- flush stdout after each line
+ -v, --verson -- print version and exit
+* -z, --compressed -- try to look inside compressed files
+ -L, --follow -- follow symlinks
+ -s, --special -- don't skip special files
+
+* -- not implemented so far ;-)
++ -- implemented, but in another way...
+'''
+
+def main():
+ import getopt
+ global _magic
+ try:
+ brief = 0
+ flush = 0
+ follow= 0
+ mime = 0
+ check = 0
+ special=0
+ try:
+ opts, args = getopt.getopt(
+ sys.argv[1:],
+ '?m:f:CbciknvzLs',
+ ( 'help',
+ 'magic=',
+ 'names=',
+ 'compile',
+ 'brief',
+ 'check',
+ 'mime',
+ 'keep-going',
+ 'flush',
+ 'version',
+ 'compressed',
+ 'follow',
+ 'special',
+ )
+ )
+ except getopt.error, why:
+ print >>sys.stderr, sys.argv[0], why
+ return 1
+ else:
+ files = None
+ for o,v in opts :
+ if o in ('-?','--help'):
+ print HELP % (
+ sys.argv[0],
+ _magic,
+ )
+ return 0
+ elif o in ('-f','--files='):
+ files = v
+ elif o in ('-m','--magic='):
+ _magic = v[:]
+ elif o in ('-C','--compile'):
+ pass
+ elif o in ('-b','--brief'):
+ brief = 1
+ elif o in ('-c','--check'):
+ check = 1
+ elif o in ('-i','--mime'):
+ mime = 1
+ if os.path.exists( _magic+'.mime' ) :
+ _magic += '.mime'
+ print >>sys.stderr,sys.argv[0]+':',\
+ "Using regular magic file `%s'" % _magic
+ elif o in ('-k','--keep-going'):
+ pass
+ elif o in ('-n','--flush'):
+ flush = 1
+ elif o in ('-v','--version'):
+ print 'VERSION'
+ return 0
+ elif o in ('-z','--compressed'):
+ pass
+ elif o in ('-L','--follow'):
+ follow = 1
+ elif o in ('-s','--special'):
+ special = 1
+ else:
+ if files :
+ files = map(lambda x: x.strip(), v.split(','))
+ if '-' in files and '-' in args :
+ error( 1, 'cannot use STDIN simultaneously for file list and data' )
+ for file in files :
+ for name in (
+ (file=='-')
+ and sys.stdin
+ or open(file,'r',BUFFER_SIZE)
+ ).xreadlines():
+ name = name.strip()
+ if name not in args :
+ args.append( name )
+ try:
+ if check : print >>sys.stderr, 'Loading magic database...'
+ t0 = time.time()
+ m = MagicFile(_magic)
+ t1 = time.time()
+ if check :
+ print >>sys.stderr, \
+ m.total_tests, 'tests loaded', \
+ 'for', '%.2f' % (t1-t0), 'seconds'
+ print >>sys.stderr, len(m.tests), 'tests at top level'
+ return 0 # XXX "shortened" form ;-)
+
+ mlen = max( map(len, args) )+1
+ for arg in args :
+ if not brief : print (arg + ':').ljust(mlen),
+ ftype = get_file_type( arg, follow )
+ if (special and ftype.find('special')>=0) \
+ or ftype[-4:] == 'file' :
+ t0 = time.time()
+ try:
+ t = m.detect( arg )
+ except (IOError,os.error), why:
+ t = "can't read `%s' (%s)" % (why.filename,why.strerror)
+ if ftype[-4:] == 'file' : t = ftype[:-4] + t
+ t1 = time.time()
+ print t and t or 'data'
+ if 0 : print \
+ '#\t%d tests ok, %d tests failed for %.2f seconds'%\
+ (m.ack_tests, m.nak_tests, t1-t0)
+ else:
+ print mime and 'application/x-not-regular-file' or ftype
+ if flush : sys.stdout.flush()
+ # print >>sys.stderr, 'DONE'
+ except:
+ if check : return 1
+ raise
+ else:
+ return 0
+ finally:
+ pass
+
+if __name__ == '__main__' :
+ sys.exit( main() )
+# vim:ai
+# EOF #
projects / thirdparty / bugzilla.git / commitdiff
