"""HyperArch: Pipermail archiving for Mailman
- The Dragon De Monsyne <dragondm@integral.org>
TODO:
- Should be able to force all HTML to be regenerated next time the
archive is run, in case a template is changed.
- Run a command to generate tarball of html archives for downloading
(probably in the 'update_dirty_archives' method).
"""
from __future__ import nested_scopes
import sys
import re
import errno
import urllib
import time
import os
import types
import HyperDatabase
import pipermail
import weakref
import binascii
from email.Header import decode_header, make_header
from Mailman import mm_cfg
from Mailman import Utils
from Mailman import LockFile
from Mailman import MailList
from Mailman import i18n
from Mailman.SafeDict import SafeDict
from Mailman.Logging.Syslog import syslog
from Mailman.Mailbox import ArchiverMailbox
_ = i18n._
gzip = None
if mm_cfg.GZIP_ARCHIVE_TXT_FILES:
try:
import gzip
except ImportError:
pass
EMPTYSTRING = ''
NL = '\n'
if sys.platform == 'darwin':
try:
import resource
except ImportError:
pass
else:
soft, hard = resource.getrlimit(resource.RLIMIT_STACK)
newsoft = min(hard, max(soft, 1024*2048))
resource.setrlimit(resource.RLIMIT_STACK, (newsoft, hard))
def html_quote(s, lang=None):
repls = ( ('&', '&'),
("<", '<'),
(">", '>'),
('"', '"'))
for thing, repl in repls:
s = s.replace(thing, repl)
return Utils.uncanonstr(s, lang)
def url_quote(s):
return urllib.quote(s)
def null_to_space(s):
return s.replace('\000', ' ')
def sizeof(filename, lang):
try:
size = os.path.getsize(filename)
except OSError, e:
if e.errno <> errno.ENOENT: raise
return _('size not available')
if size < 1000:
otrans = i18n.get_translation()
try:
i18n.set_language(lang)
out = _(' %(size)i bytes ')
finally:
i18n.set_translation(otrans)
return out
elif size < 1000000:
return ' %d KB ' % (size / 1000)
return ' %d MB ' % (size / 1000000)
html_charset = '<META http-equiv="Content-Type" ' \
'content="text/html; charset=%s">'
def CGIescape(arg, lang=None):
if isinstance(arg, types.UnicodeType):
s = Utils.websafe(arg)
else:
s = Utils.websafe(str(arg))
return Utils.uncanonstr(s.replace('"', '"'), lang)
paren_name_pat = re.compile(r'([(].*[)])')
REpat = re.compile( r"\s*RE\s*(\[\d+\]\s*)?:\s*", re.IGNORECASE)
emailpat = re.compile(r'([-+,.\w]+@[-+.\w]+)')
urlpat = re.compile(r'(\w+://[^>)\s]+)')
blankpat = re.compile(r'^\s*$')
htmlpat = re.compile(r'^\s*<HTML>\s*$', re.IGNORECASE)
nohtmlpat = re.compile(r'^\s*</HTML>\s*$', re.IGNORECASE)
quotedpat = re.compile(r'^([>|:]|>)+')
_templatecache = {}
def quick_maketext(templatefile, dict=None, lang=None, mlist=None):
if lang is None:
if mlist is None:
lang = mm_cfg.DEFAULT_SERVER_LANGUAGE
else:
lang = mlist.preferred_language
template = _templatecache.get((templatefile, lang))
if template is None:
template = Utils.maketext(templatefile, lang=lang, raw=1)
_templatecache[(templatefile, lang)] = template
text = template
if dict is not None:
try:
sdict = SafeDict(dict)
try:
text = sdict.interpolate(template)
except UnicodeError:
utemplate = unicode(template,
Utils.GetCharSet(lang),
'replace')
text = sdict.interpolate(utemplate)
except (TypeError, ValueError):
pass
return Utils.uncanonstr(text, lang)
class Article(pipermail.Article):
__super_init = pipermail.Article.__init__
__super_set_date = pipermail.Article._set_date
_last_article_time = time.time()
def __init__(self, message=None, sequence=0, keepHeaders=[],
lang=mm_cfg.DEFAULT_SERVER_LANGUAGE, mlist=None):
self.__super_init(message, sequence, keepHeaders)
self.prev = None
self.next = None
i = 0
while i != -1:
result = REpat.match(self.subject)
if result:
i = result.end(0)
self.subject = self.subject[i:]
else:
i = -1
self._lang = lang
self._mlist = mlist
if mm_cfg.ARCHIVER_OBSCURES_EMAILADDRS:
otrans = i18n.get_translation()
try:
i18n.set_language(lang)
self.email = re.sub('@', _(' at '), self.email)
finally:
i18n.set_translation(otrans)
ctype = message.get('Content-Type', 'text/plain')
cenc = message.get('Content-Transfer-Encoding', '')
self.ctype = ctype.lower()
self.cenc = cenc.lower()
self.decoded = {}
charset = message.get_param('charset')
if isinstance(charset, types.TupleType):
charset = unicode(charset[2], charset[0])
if charset:
charset = charset.lower().strip()
if charset[0]=='"' and charset[-1]=='"':
charset = charset[1:-1]
if charset[0]=="'" and charset[-1]=="'":
charset = charset[1:-1]
try:
body = message.get_payload(decode=1)
except binascii.Error:
body = None
if body and charset != Utils.GetCharSet(self._lang):
try:
body = unicode(body, charset)
except (UnicodeError, LookupError):
body = None
if body:
self.body = [l + "\n" for l in body.splitlines()]
self.decode_headers()
_listcache = weakref.WeakValueDictionary()
def _open_list(self, listname):
mlist = self._listcache.get(listname)
if not mlist:
try:
mlist = MailList.MailList(listname, lock=0)
except Errors.MMListError, e:
syslog('error', 'error opening list: %s\n%s', listname, e)
return None
else:
self._listcache[listname] = mlist
return mlist
def __getstate__(self):
d = self.__dict__.copy()
if d.has_key('_mlist'):
mlist = d['_mlist']
del d['_mlist']
else:
mlist = None
if mlist:
d['__listname'] = self._mlist.internal_name()
else:
d['__listname'] = None
for attr in ('prev', 'next', 'body'):
if d.has_key(attr):
del d[attr]
d['body'] = []
return d
def __setstate__(self, d):
self.__dict__ = d
listname = d.get('__listname')
if listname:
del d['__listname']
d['_mlist'] = self._open_list(listname)
if not d.has_key('_lang'):
if hasattr(self, '_mlist'):
self._lang = self._mlist.preferred_language
else:
self._lang = mm_cfg.DEFAULT_SERVER_LANGUAGE
if not d.has_key('cenc'):
self.cenc = None
if not d.has_key('decoded'):
self.decoded = {}
def setListIfUnset(self, mlist):
if getattr(self, '_mlist', None) is None:
self._mlist = mlist
def quote(self, buf):
return html_quote(buf, self._lang)
def decode_headers(self):
"""MIME-decode headers.
If the email, subject, or author attributes contain non-ASCII
characters using the encoded-word syntax of RFC 2047, decoded versions
of those attributes are placed in the self.decoded (a dictionary).
If the list's charset differs from the header charset, an attempt is
made to decode the headers as Unicode. If that fails, they are left
undecoded.
"""
author = self.decode_charset(self.author)
subject = self.decode_charset(self.subject)
if author:
self.decoded['author'] = author
email = self.decode_charset(self.email)
if email:
self.decoded['email'] = email
if subject:
self.decoded['subject'] = subject
def decode_charset(self, field):
if field.find("=?") == -1:
return None
pairs = decode_header(field)
try:
h = make_header(pairs, 99999)
return h.__unicode__()
except (UnicodeError, LookupError):
return None
return EMPTYSTRING.join([s for s, c in pairs])
def as_html(self):
d = self.__dict__.copy()
otrans = i18n.get_translation()
i18n.set_language(self._lang)
try:
d["prev"], d["prev_wsubj"] = self._get_prev()
d["next"], d["next_wsubj"] = self._get_next()
d["email_html"] = self.quote(self.email)
d["title"] = self.quote(self.subject)
d["subject_html"] = self.quote(self.subject)
d["subject_url"] = url_quote(self.subject)
d["in_reply_to_url"] = url_quote(self.in_reply_to)
if mm_cfg.ARCHIVER_OBSCURES_EMAILADDRS:
author = re.sub('@', _(' at '), self.author)
emailurl = self._mlist.GetListEmail()
else:
author = self.author
emailurl = self.email
d["author_html"] = self.quote(author)
d["email_url"] = url_quote(emailurl)
d["datestr_html"] = self.quote(i18n.ctime(int(self.date)))
d["body"] = self._get_body()
d['listurl'] = self._mlist.GetScriptURL('listinfo', absolute=1)
d['listname'] = self._mlist.real_name
d['encoding'] = ''
finally:
i18n.set_translation(otrans)
charset = Utils.GetCharSet(self._lang)
d["encoding"] = html_charset % charset
self._add_decoded(d)
return quick_maketext(
'article.html', d,
lang=self._lang, mlist=self._mlist)
def _get_prev(self):
"""Return the href and subject for the previous message"""
if self.prev:
subject = self._get_subject_enc(self.prev)
prev = ('<LINK REL="Previous" HREF="%s">'
% (url_quote(self.prev.filename)))
prev_wsubj = ('<LI>' + _('Previous message:') +
' <A HREF="%s">%s\n</A></li>'
% (url_quote(self.prev.filename),
self.quote(subject)))
else:
prev = prev_wsubj = ""
return prev, prev_wsubj
def _get_subject_enc(self, art):
"""Return the subject of art, decoded if possible.
If the charset of the current message and art match and the
article's subject is encoded, decode it.
"""
return art.decoded.get('subject', art.subject)
def _get_next(self):
"""Return the href and subject for the previous message"""
if self.next:
subject = self._get_subject_enc(self.next)
next = ('<LINK REL="Next" HREF="%s">'
% (url_quote(self.next.filename)))
next_wsubj = ('<LI>' + _('Next message:') +
' <A HREF="%s">%s\n</A></li>'
% (url_quote(self.next.filename),
self.quote(subject)))
else:
next = next_wsubj = ""
return next, next_wsubj
_rx_quote = re.compile('=([A-F0-9][A-F0-9])')
_rx_softline = re.compile('=[ \t]*$')
def _get_body(self):
"""Return the message body ready for HTML, decoded if necessary"""
try:
body = self.html_body
except AttributeError:
body = self.body
return null_to_space(EMPTYSTRING.join(body))
def _add_decoded(self, d):
"""Add encoded-word keys to HTML output"""
for src, dst in (('author', 'author_html'),
('email', 'email_html'),
('subject', 'subject_html'),
('subject', 'title')):
if self.decoded.has_key(src):
d[dst] = self.quote(self.decoded[src])
def as_text(self):
d = self.__dict__.copy()
if not d.get('fromdate', '').strip():
d['fromdate'] = time.ctime(time.time())
if not d.get('email', '').strip():
d['email'] = 'bogus@does.not.exist.com'
if not d.get('datestr', '').strip():
d['datestr'] = time.ctime(time.time())
headers = ['From %(email)s %(fromdate)s',
'From: %(email)s (%(author)s)',
'Date: %(datestr)s',
'Subject: %(subject)s']
if d['_in_reply_to']:
headers.append('In-Reply-To: %(_in_reply_to)s')
if d['_references']:
headers.append('References: %(_references)s')
if d['_message_id']:
headers.append('Message-ID: %(_message_id)s')
body = EMPTYSTRING.join(self.body)
if isinstance(body, types.UnicodeType):
body = body.encode(Utils.GetCharSet(self._lang), 'replace')
return NL.join(headers) % d + '\n\n' + body
def _set_date(self, message):
self.__super_set_date(message)
self.fromdate = time.ctime(int(self.date))
def loadbody_fromHTML(self,fileobj):
self.body = []
begin = 0
while 1:
line = fileobj.readline()
if not line:
break
if not begin:
if line.strip() == '<!--beginarticle-->':
begin = 1
continue
if line.strip() == '<!--endarticle-->':
break
self.body.append(line)
class HyperArchive(pipermail.T):
__super_init = pipermail.T.__init__
__super_update_archive = pipermail.T.update_archive
__super_update_dirty_archives = pipermail.T.update_dirty_archives
__super_add_article = pipermail.T.add_article
DIRMODE = 02775
FILEMODE = 0660
VERBOSE = 0
DEFAULTINDEX = 'thread'
ARCHIVE_PERIOD = 'month'
THREADLAZY = 0
THREADLEVELS = 3
ALLOWHTML = 1 SHOWHTML = 0 IQUOTES = 1 SHOWBR = 0
def __init__(self, maillist):
dir = maillist.archive_dir()
db = HyperDatabase.HyperDatabase(dir, maillist)
self.__super_init(dir, reload=1, database=db)
self.maillist = maillist
self._lock_file = None
self.lang = maillist.preferred_language
self.charset = Utils.GetCharSet(maillist.preferred_language)
if hasattr(self.maillist,'archive_volume_frequency'):
if self.maillist.archive_volume_frequency == 0:
self.ARCHIVE_PERIOD='year'
elif self.maillist.archive_volume_frequency == 2:
self.ARCHIVE_PERIOD='quarter'
elif self.maillist.archive_volume_frequency == 3:
self.ARCHIVE_PERIOD='week'
elif self.maillist.archive_volume_frequency == 4:
self.ARCHIVE_PERIOD='day'
else:
self.ARCHIVE_PERIOD='month'
yre = r'(?P<year>[0-9]{4,4})'
mre = r'(?P<month>[01][0-9])'
dre = r'(?P<day>[0123][0-9])'
self._volre = {
'year': '^' + yre + '$',
'quarter': '^' + yre + r'q(?P<quarter>[1234])$',
'month': '^' + yre + r'-(?P<month>[a-zA-Z]+)$',
'week': r'^Week-of-Mon-' + yre + mre + dre,
'day': '^' + yre + mre + dre + '$'
}
def _makeArticle(self, msg, sequence):
return Article(msg, sequence,
lang=self.maillist.preferred_language,
mlist=self.maillist)
def html_foot(self):
mlist = self.maillist
otrans = i18n.get_translation()
i18n.set_language(mlist.preferred_language)
def quotetime(s):
return html_quote(i18n.ctime(s), self.lang)
try:
d = {"lastdate": quotetime(self.lastdate),
"archivedate": quotetime(self.archivedate),
"listinfo": mlist.GetScriptURL('listinfo', absolute=1),
"version": self.version,
}
i = {"thread": _("thread"),
"subject": _("subject"),
"author": _("author"),
"date": _("date")
}
finally:
i18n.set_translation(otrans)
for t in i.keys():
cap = t[0].upper() + t[1:]
if self.type == cap:
d["%s_ref" % (t)] = ""
else:
d["%s_ref" % (t)] = ('<a href="%s.html#start">[ %s ]</a>'
% (t, i[t]))
return quick_maketext(
'archidxfoot.html', d,
mlist=mlist)
def html_head(self):
mlist = self.maillist
otrans = i18n.get_translation()
i18n.set_language(mlist.preferred_language)
def quotetime(s):
return html_quote(i18n.ctime(s), self.lang)
try:
d = {"listname": html_quote(mlist.real_name, self.lang),
"archtype": self.type,
"archive": self.volNameToDesc(self.archive),
"listinfo": mlist.GetScriptURL('listinfo', absolute=1),
"firstdate": quotetime(self.firstdate),
"lastdate": quotetime(self.lastdate),
"size": self.size,
}
i = {"thread": _("thread"),
"subject": _("subject"),
"author": _("author"),
"date": _("date"),
}
finally:
i18n.set_translation(otrans)
for t in i.keys():
cap = t[0].upper() + t[1:]
if self.type == cap:
d["%s_ref" % (t)] = ""
d["archtype"] = i[t]
else:
d["%s_ref" % (t)] = ('<a href="%s.html#start">[ %s ]</a>'
% (t, i[t]))
if self.charset:
d["encoding"] = html_charset % self.charset
else:
d["encoding"] = ""
return quick_maketext(
'archidxhead.html', d,
mlist=mlist)
def html_TOC(self):
mlist = self.maillist
listname = mlist.internal_name()
mbox = os.path.join(mlist.archive_dir()+'.mbox', listname+'.mbox')
d = {"listname": mlist.real_name,
"listinfo": mlist.GetScriptURL('listinfo', absolute=1),
"fullarch": '../%s.mbox/%s.mbox' % (listname, listname),
"size": sizeof(mbox, mlist.preferred_language),
'meta': '',
}
otrans = i18n.get_translation()
i18n.set_language(mlist.preferred_language)
try:
if not self.archives:
d["noarchive_msg"] = _(
'<P>Currently, there are no archives. </P>')
d["archive_listing_start"] = ""
d["archive_listing_end"] = ""
d["archive_listing"] = ""
else:
d["noarchive_msg"] = ""
d["archive_listing_start"] = quick_maketext(
'archliststart.html',
lang=mlist.preferred_language,
mlist=mlist)
d["archive_listing_end"] = quick_maketext(
'archlistend.html',
mlist=mlist)
accum = []
for a in self.archives:
accum.append(self.html_TOC_entry(a))
d["archive_listing"] = EMPTYSTRING.join(accum)
finally:
i18n.set_translation(otrans)
d['meta'] += html_charset % Utils.GetCharSet(mlist.preferred_language)
return quick_maketext(
'archtoc.html', d,
mlist=mlist)
def html_TOC_entry(self, arch):
txtfile = os.path.join(self.maillist.archive_dir(), arch + '.txt')
gzfile = txtfile + '.gz'
if os.path.exists(gzfile):
file = gzfile
url = arch + '.txt.gz'
templ = '<td><A href="%(url)s">[ ' + _('Gzip\'d Text%(sz)s') \
+ ']</a></td>'
elif os.path.exists(txtfile):
file = txtfile
url = arch + '.txt'
templ = '<td><A href="%(url)s">[ ' + _('Text%(sz)s') + ']</a></td>'
else:
file = None
if file:
textlink = templ % {
'url': url,
'sz' : sizeof(file, self.maillist.preferred_language)
}
else:
textlink = ''
return quick_maketext(
'archtocentry.html',
{'archive': arch,
'archivelabel': self.volNameToDesc(arch),
'textlink': textlink
},
mlist=self.maillist)
def GetArchLock(self):
if self._lock_file:
return 1
self._lock_file = LockFile.LockFile(
os.path.join(mm_cfg.LOCK_DIR,
self.maillist.internal_name() + '-arch.lock'))
try:
self._lock_file.lock(timeout=0.5)
except LockFile.TimeOutError:
return 0
return 1
def DropArchLock(self):
if self._lock_file:
self._lock_file.unlock(unconditionally=1)
self._lock_file = None
def processListArch(self):
name = self.maillist.ArchiveFileName()
wname= name+'.working'
ename= name+'.err_unarchived'
try:
os.stat(name)
except (IOError,os.error):
return
if not self.GetArchLock():
return
try:
wf = open(wname)
syslog('error',
'Archive working file %s present. '
'Check %s for possibly unarchived msgs',
wname, ename)
omask = os.umask(007)
try:
ef = open(ename, 'a+')
finally:
os.umask(omask)
ef.seek(1,2)
if ef.read(1) <> '\n':
ef.write('\n')
ef.write(wf.read())
ef.close()
wf.close()
os.unlink(wname)
except IOError:
pass
os.rename(name,wname)
archfile = open(wname)
self.processUnixMailbox(archfile)
archfile.close()
os.unlink(wname)
self.DropArchLock()
def get_filename(self, article):
return '%06i.html' % (article.sequence,)
def get_archives(self, article):
"""Return a list of indexes where the article should be filed.
A string can be returned if the list only contains one entry,
and the empty list is legal."""
res = self.dateToVolName(float(article.date))
self.message(_("figuring article archives\n"))
self.message(res + "\n")
return res
def volNameToDesc(self, volname):
volname = volname.strip()
monthdict = [
'',
_('January'), _('February'), _('March'), _('April'),
_('May'), _('June'), _('July'), _('August'),
_('September'), _('October'), _('November'), _('December')
]
for each in self._volre.keys():
match = re.match(self._volre[each], volname)
if match:
year = int(match.group('year'))
if each == 'quarter':
d =["", _("First"), _("Second"), _("Third"), _("Fourth") ]
ord = d[int(match.group('quarter'))]
return _("%(ord)s quarter %(year)i")
elif each == 'month':
monthstr = match.group('month').lower()
for i in range(1, 13):
monthname = time.strftime("%B", (1999,i,1,0,0,0,0,1,0))
if monthstr.lower() == monthname.lower():
month = monthdict[i]
return _("%(month)s %(year)i")
raise ValueError, "%s is not a month!" % monthstr
elif each == 'week':
month = monthdict[int(match.group("month"))]
day = int(match.group("day"))
return _("The Week Of Monday %(day)i %(month)s %(year)i")
elif each == 'day':
month = monthdict[int(match.group("month"))]
day = int(match.group("day"))
return _("%(day)i %(month)s %(year)i")
else:
return match.group('year')
raise ValueError, "%s is not a valid volname" % volname
def dateToVolName(self,date):
datetuple=time.localtime(date)
if self.ARCHIVE_PERIOD=='year':
return time.strftime("%Y",datetuple)
elif self.ARCHIVE_PERIOD=='quarter':
if datetuple[1] in [1,2,3]:
return time.strftime("%Yq1",datetuple)
elif datetuple[1] in [4,5,6]:
return time.strftime("%Yq2",datetuple)
elif datetuple[1] in [7,8,9]:
return time.strftime("%Yq3",datetuple)
else:
return time.strftime("%Yq4",datetuple)
elif self.ARCHIVE_PERIOD == 'day':
return time.strftime("%Y%m%d", datetuple)
elif self.ARCHIVE_PERIOD == 'week':
monday = time.mktime(datetuple) - datetuple[6] * 24 * 60 * 60
datetuple = time.localtime(monday)
return time.strftime("Week-of-Mon-%Y%m%d", datetuple)
else:
return time.strftime("%Y-%B",datetuple)
def volNameToDate(self, volname):
volname = volname.strip()
for each in self._volre.keys():
match = re.match(self._volre[each],volname)
if match:
year = int(match.group('year'))
month = 1
day = 1
if each == 'quarter':
q = int(match.group('quarter'))
month = (q * 3) - 2
elif each == 'month':
monthstr = match.group('month').lower()
m = []
for i in range(1,13):
m.append(
time.strftime("%B",(1999,i,1,0,0,0,0,1,0)).lower())
try:
month = m.index(monthstr) + 1
except ValueError:
pass
elif each == 'week' or each == 'day':
month = int(match.group("month"))
day = int(match.group("day"))
try:
return time.mktime((year,month,1,0,0,0,0,1,-1))
except OverflowError:
return 0.0
return 0.0
def sortarchives(self):
def sf(a, b):
al = self.volNameToDate(a)
bl = self.volNameToDate(b)
if al > bl:
return 1
elif al < bl:
return -1
else:
return 0
if self.ARCHIVE_PERIOD in ('month','year','quarter'):
self.archives.sort(sf)
else:
self.archives.sort()
self.archives.reverse()
def message(self, msg):
if self.VERBOSE:
f = sys.stderr
f.write(msg)
if msg[-1:] != '\n':
f.write('\n')
f.flush()
def open_new_archive(self, archive, archivedir):
index_html = os.path.join(archivedir, 'index.html')
try:
os.unlink(index_html)
except:
pass
os.symlink(self.DEFAULTINDEX+'.html',index_html)
def write_index_header(self):
self.depth=0
print self.html_head()
if not self.THREADLAZY and self.type=='Thread':
self.message(_("Computing threaded index\n"))
self.updateThreadedIndex()
def write_index_footer(self):
for i in range(self.depth):
print '</UL>'
print self.html_foot()
def write_index_entry(self, article):
subject = self.get_header("subject", article)
author = self.get_header("author", article)
if mm_cfg.ARCHIVER_OBSCURES_EMAILADDRS:
author = re.sub('@', _(' at '), author)
subject = CGIescape(subject, self.lang)
author = CGIescape(author, self.lang)
d = {
'filename': urllib.quote(article.filename),
'subject': subject,
'sequence': article.sequence,
'author': author
}
print quick_maketext(
'archidxentry.html', d,
mlist=self.maillist)
def get_header(self, field, article):
result = article.decoded.get(field)
if result is None:
return getattr(article, field)
return result
def write_threadindex_entry(self, article, depth):
if depth < 0:
self.message('depth<0')
depth = 0
if depth > self.THREADLEVELS:
depth = self.THREADLEVELS
if depth < self.depth:
for i in range(self.depth-depth):
print '</UL>'
elif depth > self.depth:
for i in range(depth-self.depth):
print '<UL>'
print '<!--%i %s -->' % (depth, article.threadKey)
self.depth = depth
self.write_index_entry(article)
def write_TOC(self):
self.sortarchives()
omask = os.umask(002)
try:
toc = open(os.path.join(self.basedir, 'index.html'), 'w')
finally:
os.umask(omask)
toc.write(self.html_TOC())
toc.close()
def write_article(self, index, article, path):
omask = os.umask(002)
try:
f = open(path, 'w')
finally:
os.umask(omask)
f.write(article.as_html())
f.close()
path = os.path.join(self.basedir, "%s.txt" % index)
omask = os.umask(002)
try:
f = open(path, 'a+')
finally:
os.umask(omask)
f.write(article.as_text())
f.close()
def update_archive(self, archive):
self.__super_update_archive(archive)
if gzip:
archz = None
archt = None
txtfile = os.path.join(self.basedir, '%s.txt' % archive)
gzipfile = os.path.join(self.basedir, '%s.txt.gz' % archive)
oldgzip = os.path.join(self.basedir, '%s.old.txt.gz' % archive)
try:
archt = open(txtfile)
except IOError:
return
try:
os.rename(gzipfile, oldgzip)
archz = gzip.open(oldgzip)
except (IOError, RuntimeError, os.error):
pass
try:
ou = os.umask(002)
newz = gzip.open(gzipfile, 'w')
finally:
os.umask(ou)
if archz:
newz.write(archz.read())
archz.close()
os.unlink(oldgzip)
try:
newz.write(archt.read())
newz.close()
archt.close()
except IOError:
pass
os.unlink(txtfile)
_skip_attrs = ('maillist', '_lock_file', 'charset')
def getstate(self):
d={}
for each in self.__dict__.keys():
if not (each in self._skip_attrs
or each.upper() == each):
d[each] = self.__dict__[each]
return d
def __processbody_URLquote(self, lines):
source = lines[:]
dest = lines
last_line_was_quoted = 0
for i in xrange(0, len(source)):
Lorig = L = source[i]
prefix = suffix = ""
if L is None:
continue
if self.IQUOTES:
quoted = quotedpat.match(L)
if quoted is None:
last_line_was_quoted = 0
else:
quoted = quoted.end(0)
prefix = CGIescape(L[:quoted], self.lang) + '<i>'
suffix = '</I>'
if self.SHOWHTML:
suffix += '<BR>'
if not last_line_was_quoted:
prefix = '<BR>' + prefix
L = L[quoted:]
last_line_was_quoted = 1
L2 = ""
jr = emailpat.search(L)
kr = urlpat.search(L)
while jr is not None or kr is not None:
if jr == None:
j = -1
else:
j = jr.start(0)
if kr is None:
k = -1
else:
k = kr.start(0)
if j != -1 and (j < k or k == -1):
text = jr.group(1)
length = len(text)
if mm_cfg.ARCHIVER_OBSCURES_EMAILADDRS:
text = re.sub('@', _(' at '), text)
URL = self.maillist.GetScriptURL(
'listinfo', absolute=1)
else:
URL = 'mailto:' + text
pos = j
elif k != -1 and (j > k or j == -1):
text = URL = kr.group(1)
length = len(text)
pos = k
else: raise ValueError, "j==k: This can't happen!"
L2 += '%s<A HREF="%s">%s</A>' % (
CGIescape(L[:pos], self.lang),
html_quote(URL), CGIescape(text, self.lang))
L = L[pos+length:]
jr = emailpat.search(L)
kr = urlpat.search(L)
if jr is None and kr is None:
L = CGIescape(L, self.lang)
L = prefix + L2 + L + suffix
source[i] = None
dest[i] = L
def __processbody_HTML(self, lines):
source = lines[:]
dest = lines
l = len(source)
i = 0
while i < l:
while i < l and htmlpat.match(source[i]) is None:
i = i + 1
if i < l:
source[i] = None
i = i + 1
while i < l and nohtmlpat.match(source[i]) is None:
dest[i], source[i] = source[i], None
i = i + 1
if i < l:
source[i] = None
i = i + 1
def format_article(self, article):
lines = filter(None, article.body)
if self.ALLOWHTML:
self.__processbody_HTML(lines)
self.__processbody_URLquote(lines)
if not self.SHOWHTML and lines:
lines.insert(0, '<PRE>')
lines.append('</PRE>')
else:
if self.SHOWBR:
lines = map(lambda x:x + "<BR>", lines)
else:
for i in range(0, len(lines)):
s = lines[i]
if s[0:1] in ' \t\n':
lines[i] = '<P>' + s
article.html_body = lines
return article
def update_article(self, arcdir, article, prev, next):
seq = article.sequence
filename = os.path.join(arcdir, article.filename)
self.message(_('Updating HTML for article %(seq)s'))
try:
f = open(filename)
article.loadbody_fromHTML(f)
f.close()
except IOError, e:
if e.errno <> errno.ENOENT: raise
self.message(_('article file %(filename)s is missing!'))
article.prev = prev
article.next = next
omask = os.umask(002)
try:
f = open(filename, 'w')
finally:
os.umask(omask)
f.write(article.as_html())
f.close()