"""Project file system commands.This modules implements file system operations used by rope. Differentversion control systems can be supported by implementing the interfaceprovided by `FileSystemCommands` class. See `SubversionCommands` and`MercurialCommands` for example."""importosimportshutilimportsubprocessdefcreate_fscommands(root):dirlist=os.listdir(root)commands={'.hg':MercurialCommands,'.svn':SubversionCommands,'.git':GITCommands,'_svn':SubversionCommands,'_darcs':DarcsCommands}forkeyincommands:ifkeyindirlist:try:returncommands[key](root)except(ImportError,OSError):passreturnFileSystemCommands()classFileSystemCommands(object):defcreate_file(self,path):open(path,'w').close()defcreate_folder(self,path):os.mkdir(path)defmove(self,path,new_location):shutil.move(path,new_location)defremove(self,path):ifos.path.isfile(path):os.remove(path)else:shutil.rmtree(path)defwrite(self,path,data):file_=open(path,'wb')try:file_.write(data)finally:file_.close()classSubversionCommands(object):def__init__(self,*args):self.normal_actions=FileSystemCommands()importpysvnself.client=pysvn.Client()defcreate_file(self,path):self.normal_actions.create_file(path)self.client.add(path,force=True)defcreate_folder(self,path):self.normal_actions.create_folder(path)self.client.add(path,force=True)defmove(self,path,new_location):self.client.move(path,new_location,force=True)defremove(self,path):self.client.remove(path,force=True)defwrite(self,path,data):self.normal_actions.write(path,data)classMercurialCommands(object):def__init__(self,root):self.hg=self._import_mercurial()self.normal_actions=FileSystemCommands()try:self.ui=self.hg.ui.ui(verbose=False,debug=False,quiet=True,interactive=False,traceback=False,report_untrusted=False)except:self.ui=self.hg.ui.ui()self.ui.setconfig('ui','interactive','no')self.ui.setconfig('ui','debug','no')self.ui.setconfig('ui','traceback','no')self.ui.setconfig('ui','verbose','no')self.ui.setconfig('ui','report_untrusted','no')self.ui.setconfig('ui','quiet','yes')self.repo=self.hg.hg.repository(self.ui,root)def_import_mercurial(self):importmercurial.commandsimportmercurial.hgimportmercurial.uireturnmercurialdefcreate_file(self,path):self.normal_actions.create_file(path)self.hg.commands.add(self.ui,self.repo,path)defcreate_folder(self,path):self.normal_actions.create_folder(path)defmove(self,path,new_location):self.hg.commands.rename(self.ui,self.repo,path,new_location,after=False)defremove(self,path):self.hg.commands.remove(self.ui,self.repo,path)defwrite(self,path,data):self.normal_actions.write(path,data)classGITCommands(object):def__init__(self,root):self.root=rootself._do(['version'])self.normal_actions=FileSystemCommands()defcreate_file(self,path):self.normal_actions.create_file(path)self._do(['add',self._in_dir(path)])defcreate_folder(self,path):self.normal_actions.create_folder(path)defmove(self,path,new_location):self._do(['mv',self._in_dir(path),self._in_dir(new_location)])defremove(self,path):self._do(['rm',self._in_dir(path)])defwrite(self,path,data):# XXX: should we use ``git add``?self.normal_actions.write(path,data)def_do(self,args):_execute(['git']+args,cwd=self.root)def_in_dir(self,path):ifpath.startswith(self.root):returnpath[len(self.root)+1:]returnself.rootclassDarcsCommands(object):def__init__(self,root):self.root=rootself.normal_actions=FileSystemCommands()defcreate_file(self,path):self.normal_actions.create_file(path)self._do(['add',path])defcreate_folder(self,path):self.normal_actions.create_folder(path)self._do(['add',path])defmove(self,path,new_location):self._do(['mv',path,new_location])defremove(self,path):self.normal_actions.remove(path)defwrite(self,path,data):self.normal_actions.write(path,data)def_do(self,args):_execute(['darcs']+args,cwd=self.root)def_execute(args,cwd=None):process=subprocess.Popen(args,cwd=cwd,stdout=subprocess.PIPE)process.wait()returnprocess.returncodedefunicode_to_file_data(contents,encoding=None):ifnotisinstance(contents,unicode):returncontentsifencodingisNone:encoding=read_str_coding(contents)ifencodingisnotNone:returncontents.encode(encoding)try:returncontents.encode()exceptUnicodeEncodeError:returncontents.encode('utf-8')deffile_data_to_unicode(data,encoding=None):result=_decode_data(data,encoding)if'\r'inresult:result=result.replace('\r\n','\n').replace('\r','\n')returnresultdef_decode_data(data,encoding):ifisinstance(data,unicode):returndataifencodingisNone:encoding=read_str_coding(data)ifencodingisNone:# there is no encoding tip, we need to guess.# PEP263 says that "encoding not explicitly defined" means it is ascii,# but we will use utf8 instead since utf8 fully covers ascii and btw is# the only non-latin sane encoding.encoding='utf-8'try:returndata.decode(encoding)except(UnicodeError,LookupError):# fallback to latin1: it should never failreturndata.decode('latin1')defread_file_coding(path):file=open(path,'b')count=0result=[]buffsize=10whileTrue:current=file.read(10)ifnotcurrent:breakcount+=current.count('\n')result.append(current)file.close()return_find_coding(''.join(result))defread_str_coding(source):try:first=source.index('\n')+1second=source.index('\n',first)+1exceptValueError:second=len(source)return_find_coding(source[:second])def_find_coding(text):coding='coding'try:start=text.index(coding)+len(coding)iftext[start]notin'=:':returnstart+=1whilestart<len(text)andtext[start].isspace():start+=1end=startwhileend<len(text):c=text[end]ifnotc.isalnum()andcnotin'-_':breakend+=1returntext[start:end]exceptValueError:pass