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