mirror of
https://github.com/arsenetar/dupeguru.git
synced 2024-11-16 12:19:03 +00:00
309 lines
11 KiB
Python
309 lines
11 KiB
Python
# Created By: Virgil Dupras
|
|
# Created On: 2006/11/11
|
|
# $Id$
|
|
# Copyright 2009 Hardcoded Software (http://www.hardcoded.net)
|
|
#
|
|
# This software is licensed under the "HS" License as described in the "LICENSE" file,
|
|
# which should be included with this package. The terms are also available at
|
|
# http://www.hardcoded.net/licenses/hs_license
|
|
|
|
from Foundation import *
|
|
from AppKit import *
|
|
import logging
|
|
import os.path as op
|
|
|
|
import hsfs as fs
|
|
from hsutil import io, cocoa, job
|
|
from hsutil.cocoa import install_exception_hook
|
|
from hsutil.misc import stripnone
|
|
from hsutil.reg import RegistrationRequired
|
|
|
|
import app, data
|
|
|
|
JOBID2TITLE = {
|
|
app.JOB_SCAN: "Scanning for duplicates",
|
|
app.JOB_LOAD: "Loading",
|
|
app.JOB_MOVE: "Moving",
|
|
app.JOB_COPY: "Copying",
|
|
app.JOB_DELETE: "Sending to Trash",
|
|
}
|
|
|
|
def demo_method(method):
|
|
def wrapper(self, *args, **kwargs):
|
|
try:
|
|
return method(self, *args, **kwargs)
|
|
except RegistrationRequired:
|
|
NSNotificationCenter.defaultCenter().postNotificationName_object_('RegistrationRequired', self)
|
|
|
|
return wrapper
|
|
|
|
class DupeGuru(app.DupeGuru):
|
|
def __init__(self, data_module, appdata_subdir, appid):
|
|
LOGGING_LEVEL = logging.DEBUG if NSUserDefaults.standardUserDefaults().boolForKey_('debug') else logging.WARNING
|
|
logging.basicConfig(level=LOGGING_LEVEL, format='%(levelname)s %(message)s')
|
|
logging.debug('started in debug mode')
|
|
install_exception_hook()
|
|
if data_module is None:
|
|
data_module = data
|
|
appsupport = NSSearchPathForDirectoriesInDomains(NSApplicationSupportDirectory, NSUserDomainMask, True)[0]
|
|
appdata = op.join(appsupport, appdata_subdir)
|
|
app.DupeGuru.__init__(self, data_module, appdata, appid)
|
|
self.progress = cocoa.ThreadedJobPerformer()
|
|
self.display_delta_values = False
|
|
self.selected_dupes = []
|
|
self.RefreshDetailsTable(None,None)
|
|
|
|
#--- Override
|
|
@staticmethod
|
|
def _recycle_dupe(dupe):
|
|
if not io.exists(dupe.path):
|
|
dupe.parent = None
|
|
return True
|
|
directory = unicode(dupe.parent.path)
|
|
filename = dupe.name
|
|
result, tag = NSWorkspace.sharedWorkspace().performFileOperation_source_destination_files_tag_(
|
|
NSWorkspaceRecycleOperation, directory, '', [filename])
|
|
if not io.exists(dupe.path):
|
|
dupe.parent = None
|
|
return True
|
|
logging.warning('Could not send %s to trash. tag: %d' % (unicode(dupe.path), tag))
|
|
return False
|
|
|
|
def _start_job(self, jobid, func):
|
|
try:
|
|
j = self.progress.create_job()
|
|
self.progress.run_threaded(func, args=(j, ))
|
|
except job.JobInProgressError:
|
|
NSNotificationCenter.defaultCenter().postNotificationName_object_('JobInProgress', self)
|
|
else:
|
|
ud = {'desc': JOBID2TITLE[jobid], 'jobid':jobid}
|
|
NSNotificationCenter.defaultCenter().postNotificationName_object_userInfo_('JobStarted', self, ud)
|
|
|
|
#---Helpers
|
|
def GetObjects(self,node_path):
|
|
#returns a tuple g,d
|
|
try:
|
|
g = self.results.groups[node_path[0]]
|
|
if len(node_path) == 2:
|
|
return (g,self.results.groups[node_path[0]].dupes[node_path[1]])
|
|
else:
|
|
return (g,None)
|
|
except IndexError:
|
|
return (None,None)
|
|
|
|
def GetDirectory(self,node_path,curr_dir=None):
|
|
if not node_path:
|
|
return curr_dir
|
|
if curr_dir is not None:
|
|
l = curr_dir.dirs
|
|
else:
|
|
l = self.directories
|
|
d = l[node_path[0]]
|
|
return self.GetDirectory(node_path[1:],d)
|
|
|
|
def RefreshDetailsTable(self,dupe,group):
|
|
l1 = self._get_display_info(dupe, group, False)
|
|
# we don't want the two sides of the table to display the stats for the same file
|
|
ref = group.ref if group is not None and group.ref is not dupe else None
|
|
l2 = self._get_display_info(ref, group, False)
|
|
names = [c['display'] for c in self.data.COLUMNS]
|
|
self.details_table = zip(names,l1,l2)
|
|
|
|
#---Public
|
|
def AddSelectedToIgnoreList(self):
|
|
for dupe in self.selected_dupes:
|
|
self.add_to_ignore_list(dupe)
|
|
|
|
copy_or_move_marked = demo_method(app.DupeGuru.copy_or_move_marked)
|
|
delete_marked = demo_method(app.DupeGuru.delete_marked)
|
|
|
|
def MakeSelectedReference(self):
|
|
self.make_reference(self.selected_dupes)
|
|
|
|
def OpenSelected(self):
|
|
if self.selected_dupes:
|
|
path = unicode(self.selected_dupes[0].path)
|
|
NSWorkspace.sharedWorkspace().openFile_(path)
|
|
|
|
def PurgeIgnoreList(self):
|
|
self.scanner.ignore_list.Filter(lambda f,s:op.exists(f) and op.exists(s))
|
|
|
|
def RefreshDetailsWithSelected(self):
|
|
if self.selected_dupes:
|
|
self.RefreshDetailsTable(
|
|
self.selected_dupes[0],
|
|
self.results.get_group_of_duplicate(self.selected_dupes[0])
|
|
)
|
|
else:
|
|
self.RefreshDetailsTable(None,None)
|
|
|
|
def RemoveDirectory(self,index):
|
|
try:
|
|
del self.directories[index]
|
|
except IndexError:
|
|
pass
|
|
|
|
def RemoveSelected(self):
|
|
self.results.remove_duplicates(self.selected_dupes)
|
|
|
|
def RenameSelected(self,newname):
|
|
try:
|
|
d = self.selected_dupes[0]
|
|
d = d.move(d.parent,newname)
|
|
return True
|
|
except (IndexError,fs.FSError),e:
|
|
logging.warning("dupeGuru Warning: %s" % str(e))
|
|
return False
|
|
|
|
def RevealSelected(self):
|
|
if self.selected_dupes:
|
|
path = unicode(self.selected_dupes[0].path)
|
|
NSWorkspace.sharedWorkspace().selectFile_inFileViewerRootedAtPath_(path,'')
|
|
|
|
def start_scanning(self):
|
|
self.RefreshDetailsTable(None, None)
|
|
try:
|
|
app.DupeGuru.start_scanning(self)
|
|
return 0
|
|
except app.NoScannableFileError:
|
|
return 3
|
|
except app.AllFilesAreRefError:
|
|
return 1
|
|
|
|
def selected_result_node_paths(self):
|
|
def get_path(dupe):
|
|
try:
|
|
group = self.results.get_group_of_duplicate(dupe)
|
|
groupindex = self.results.groups.index(group)
|
|
if dupe is group.ref:
|
|
return [groupindex]
|
|
dupeindex = group.dupes.index(dupe)
|
|
return [groupindex, dupeindex]
|
|
except ValueError: # dupe not in there
|
|
return None
|
|
|
|
dupes = self.selected_dupes
|
|
return stripnone(get_path(dupe) for dupe in dupes)
|
|
|
|
def selected_powermarker_node_paths(self):
|
|
def get_path(dupe):
|
|
try:
|
|
dupeindex = self.results.dupes.index(dupe)
|
|
return [dupeindex]
|
|
except ValueError: # dupe not in there
|
|
return None
|
|
|
|
dupes = self.selected_dupes
|
|
return stripnone(get_path(dupe) for dupe in dupes)
|
|
|
|
def SelectResultNodePaths(self,node_paths):
|
|
def extract_dupe(t):
|
|
g,d = t
|
|
if d is not None:
|
|
return d
|
|
else:
|
|
if g is not None:
|
|
return g.ref
|
|
|
|
selected = [extract_dupe(self.GetObjects(p)) for p in node_paths]
|
|
self.selected_dupes = [dupe for dupe in selected if dupe is not None]
|
|
|
|
def SelectPowerMarkerNodePaths(self,node_paths):
|
|
rows = [p[0] for p in node_paths]
|
|
self.selected_dupes = [
|
|
self.results.dupes[row] for row in rows if row in xrange(len(self.results.dupes))
|
|
]
|
|
|
|
def SetDirectoryState(self,node_path,state):
|
|
d = self.GetDirectory(node_path)
|
|
self.directories.set_state(d.path,state)
|
|
|
|
def sort_dupes(self,key,asc):
|
|
self.results.sort_dupes(key,asc,self.display_delta_values)
|
|
|
|
def sort_groups(self,key,asc):
|
|
self.results.sort_groups(key,asc)
|
|
|
|
def ToggleSelectedMarkState(self):
|
|
for dupe in self.selected_dupes:
|
|
self.results.mark_toggle(dupe)
|
|
|
|
#---Data
|
|
def GetOutlineViewMaxLevel(self, tag):
|
|
if tag == 0:
|
|
return 2
|
|
elif tag == 1:
|
|
return 0
|
|
elif tag == 2:
|
|
return 1
|
|
|
|
def GetOutlineViewChildCounts(self, tag, node_path):
|
|
if self.progress._job_running:
|
|
return []
|
|
if tag == 0: #Normal results
|
|
assert not node_path # no other value is possible
|
|
return [len(g.dupes) for g in self.results.groups]
|
|
elif tag == 1: #Directories
|
|
try:
|
|
dirs = self.GetDirectory(node_path).dirs if node_path else self.directories
|
|
return [d.dircount for d in dirs]
|
|
except IndexError: # node_path out of range
|
|
return []
|
|
else: #Power Marker
|
|
assert not node_path # no other value is possible
|
|
return [0 for d in self.results.dupes]
|
|
|
|
def GetOutlineViewValues(self, tag, node_path):
|
|
if self.progress._job_running:
|
|
return
|
|
if not node_path:
|
|
return
|
|
if tag in (0,2): #Normal results / Power Marker
|
|
if tag == 0:
|
|
g, d = self.GetObjects(node_path)
|
|
if d is None:
|
|
d = g.ref
|
|
else:
|
|
d = self.results.dupes[node_path[0]]
|
|
g = self.results.get_group_of_duplicate(d)
|
|
result = self._get_display_info(d, g, self.display_delta_values)
|
|
return result
|
|
elif tag == 1: #Directories
|
|
try:
|
|
d = self.GetDirectory(node_path)
|
|
return [d.name, self.directories.get_state(d.path)]
|
|
except IndexError: # node_path out of range
|
|
return []
|
|
|
|
def GetOutlineViewMarked(self, tag, node_path):
|
|
# 0=unmarked 1=marked 2=unmarkable
|
|
if self.progress._job_running:
|
|
return
|
|
if not node_path:
|
|
return 2
|
|
if tag == 1: #Directories
|
|
return 2
|
|
if tag == 0: #Normal results
|
|
g, d = self.GetObjects(node_path)
|
|
else: #Power Marker
|
|
d = self.results.dupes[node_path[0]]
|
|
if (d is None) or (not self.results.is_markable(d)):
|
|
return 2
|
|
elif self.results.is_marked(d):
|
|
return 1
|
|
else:
|
|
return 0
|
|
|
|
def GetTableViewCount(self, tag):
|
|
if self.progress._job_running:
|
|
return 0
|
|
return len(self.details_table)
|
|
|
|
def GetTableViewMarkedIndexes(self,tag):
|
|
return []
|
|
|
|
def GetTableViewValues(self,tag,row):
|
|
return self.details_table[row]
|
|
|
|
|