code.delx.au - offlineimap/blob - offlineimap/repository/Maildir.py

   1 # Maildir repository support
   2 # Copyright (C) 2002 John Goerzen
   3 # <jgoerzen@complete.org>
   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., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  18
  19 from Base import BaseRepository
  20 from offlineimap import folder, imaputil
  21 from offlineimap.ui import UIBase
  22 from mailbox import Maildir
  23 import os
  24 from stat import *
  25
  26 class MaildirRepository(BaseRepository):
  27     def __init__(self, reposname, account):
  28         """Initialize a MaildirRepository object.  Takes a path name
  29         to the directory holding all the Maildir directories."""
  30         BaseRepository.__init__(self, reposname, account)
  31
  32         self.root = self.getlocalroot()
  33         self.folders = None
  34         self.ui = UIBase.getglobalui()
  35         self.debug("MaildirRepository initialized, sep is " + repr(self.getsep()))
  36         self.folder_atimes = []
  37
  38     def _append_folder_atimes(self, foldername):
  39         p = os.path.join(self.root, foldername)
  40         new = os.path.join(p, 'new')
  41         cur = os.path.join(p, 'cur')
  42         f = p, os.stat(new)[ST_ATIME], os.stat(cur)[ST_ATIME]
  43         self.folder_atimes.append(f)
  44
  45     def restore_folder_atimes(self):
  46         if not self.folder_atimes:
  47             return
  48
  49         for f in self.folder_atimes:
  50             t = f[1], os.stat(os.path.join(f[0], 'new'))[ST_MTIME]
  51             os.utime(os.path.join(f[0], 'new'), t)
  52             t = f[2], os.stat(os.path.join(f[0], 'cur'))[ST_MTIME]
  53             os.utime(os.path.join(f[0], 'cur'), t)
  54
  55     def getlocalroot(self):
  56         return os.path.expanduser(self.getconf('localfolders'))
  57
  58     def debug(self, msg):
  59         self.ui.debug('maildir', msg)
  60
  61     def getsep(self):
  62         return self.getconf('sep', '.').strip()
  63
  64     def makefolder(self, foldername):
  65         self.debug("makefolder called with arg " + repr(foldername))
  66         # Do the chdir thing so the call to makedirs does not make the
  67         # self.root directory (we'd prefer to raise an error in that case),
  68         # but will make the (relative) paths underneath it.  Need to use
  69         # makedirs to support a / separator.
  70         if self.getsep() == '/':
  71             for invalid in ['new', 'cur', 'tmp', 'offlineimap.uidvalidity']:
  72                 for component in foldername.split('/'):
  73                     assert component != invalid, "When using nested folders (/ as a separator in the account config), your folder names may not contain 'new', 'cur', 'tmp', or 'offlineimap.uidvalidity'."
  74
  75         assert foldername.find('./') == -1, "Folder names may not contain ../"
  76         assert not foldername.startswith('/'), "Folder names may not begin with /"
  77
  78         oldcwd = os.getcwd()
  79         os.chdir(self.root)
  80
  81         # If we're using hierarchical folders, it's possible that sub-folders
  82         # may be created before higher-up ones.  If this is the case,
  83         # makedirs will fail because the higher-up dir already exists.
  84         # So, check to see if this is indeed the case.
  85
  86         if (self.getsep() == '/' or self.getconfboolean('existsok', 0)) \
  87             and os.path.isdir(foldername):
  88             self.debug("makefolder: %s already is a directory" % foldername)
  89             # Already exists.  Sanity-check that it's not a Maildir.
  90             for subdir in ['cur', 'new', 'tmp']:
  91                 assert not os.path.isdir(os.path.join(foldername, subdir)), \
  92                        "Tried to create folder %s but it already had dir %s" %\
  93                        (foldername, subdir)
  94         else:
  95             self.debug("makefolder: calling makedirs %s" % foldername)
  96             os.makedirs(foldername, 0700)
  97         self.debug("makefolder: creating cur, new, tmp")
  98         for subdir in ['cur', 'new', 'tmp']:
  99             os.mkdir(os.path.join(foldername, subdir), 0700)
 100         # Invalidate the cache
 101         self.folders = None
 102         os.chdir(oldcwd)
 103
 104     def deletefolder(self, foldername):
 105         self.ui.warn("NOT YET IMPLEMENTED: DELETE FOLDER %s" % foldername)
 106
 107     def getfolder(self, foldername):
 108         if self.config.has_option('Repository ' + self.name, 'restoreatime') and self.config.getboolean('Repository ' + self.name, 'restoreatime'):
 109             self._append_folder_atimes(foldername)
 110         return folder.Maildir.MaildirFolder(self.root, foldername,
 111                                             self.getsep(), self, self.accountname)
 112
 113     def _getfolders_scandir(self, root, extension = None):
 114         self.debug("_GETFOLDERS_SCANDIR STARTING. root = %s, extension = %s" \
 115                    % (root, extension))
 116         # extension willl only be non-None when called recursively when
 117         # getsep() returns '/'.
 118         retval = []
 119
 120         # Configure the full path to this repository -- "toppath"
 121
 122         if extension == None:
 123             toppath = root
 124         else:
 125             toppath = os.path.join(root, extension)
 126
 127         self.debug("  toppath = %s" % toppath)
 128
 129         # Iterate over directories in top.
 130         for dirname in os.listdir(toppath) + ['.']:
 131             self.debug("  *** top of loop")
 132             self.debug("  dirname = %s" % dirname)
 133             if dirname in ['cur', 'new', 'tmp', 'offlineimap.uidvalidity']:
 134                 self.debug("  skipping this dir (Maildir special)")
 135                 # Bypass special files.
 136                 continue
 137             fullname = os.path.join(toppath, dirname)
 138             self.debug("  fullname = %s" % fullname)
 139             if not os.path.isdir(fullname):
 140                 self.debug("  skipping this entry (not a directory)")
 141                 # Not a directory -- not a folder.
 142                 continue
 143             foldername = dirname
 144             if extension != None:
 145                 foldername = os.path.join(extension, dirname)
 146             if (os.path.isdir(os.path.join(fullname, 'cur')) and
 147                 os.path.isdir(os.path.join(fullname, 'new')) and
 148                 os.path.isdir(os.path.join(fullname, 'tmp'))):
 149                 # This directory has maildir stuff -- process
 150                 self.debug("  This is a maildir folder.")
 151
 152                 self.debug("  foldername = %s" % foldername)
 153
 154                 if self.config.has_option('Repository ' + self.name, 'restoreatime') and self.config.getboolean('Repository ' + self.name, 'restoreatime'):
 155                     self._append_folder_atimes(foldername)
 156                 retval.append(folder.Maildir.MaildirFolder(self.root, foldername,
 157                                                            self.getsep(), self, self.accountname))
 158             if self.getsep() == '/' and dirname != '.':
 159                 # Check sub-directories for folders.
 160                 retval.extend(self._getfolders_scandir(root, foldername))
 161         self.debug("_GETFOLDERS_SCANDIR RETURNING %s" % \
 162                    repr([x.getname() for x in retval]))
 163         return retval
 164
 165     def getfolders(self):
 166         if self.folders == None:
 167             self.folders = self._getfolders_scandir(self.root)
 168         return self.folders
 169