New command: dvc fsck. Close #381.

dvc/cache.py

@@ -30,7 +30,7 @@ class Cache(object):
         file_set = set(files)
         cached = {}
         for cache_file in self.all():
-            cached_files = list(filter(lambda f: System.samefile(cache_file, f, True), file_set))
+            cached_files = list(filter(lambda f: System.samefile(cache_file, f), file_set))
             cached.update(dict((f, os.path.basename(cache_file)) for f in cached_files))
             file_set = file_set - set(cached_files)
         return cached

dvc/cli.py

@@ -6,6 +6,7 @@ import argparse
 from multiprocessing import cpu_count
 
 from dvc.command.find import CmdFind
+from dvc.command.fsck import CmdFsck
 from dvc.command.init import CmdInit
 from dvc.command.remove import CmdRemove
 from dvc.command.run import CmdRun
@@ -414,6 +415,28 @@ def parse_args(argv=None):
                         default=False,
                         help='Show metrics value.')
 
+    fsck_parser = subparsers.add_parser(
+                        'fsck',
+                        parents=[parent_parser],
+                        help='Data file consistency check')
+    fsck_parser.add_argument(
+                        'targets',
+                        nargs='*',
+                        help='Data files to check')
+    fsck_parser.add_argument(
+                        '-a',
+                        '--all',
+                        action='store_true',
+                        default=False,
+                        help='Check all data files')
+    fsck_parser.add_argument(
+                        '-p',
+                        '--physical',
+                        action='store_true',
+                        default=False,
+                        help='Compute actual md5')
+    fsck_parser.set_defaults(func=CmdFsck)
+
     if isinstance(argv, str):
         argv = argv.split()
 

dvc/command/fsck.py

@@ -0,0 +1,75 @@
+import os
+
+from dvc.command.common.base import CmdBase
+from dvc.data_cloud import file_md5
+
+
+class CmdFsck(CmdBase):
+    def run(self):
+        if not self.args.targets and not self.args.all:
+            self.project.logger.error(u'Data files are not specified')
+            return 1
+
+        directions = self.all_directions()
+        files_and_stages = self.directions_by_datafile(directions)
+
+        if self.args.all:
+            dvc_files = files_and_stages.keys()
+        else:
+            dvc_files = [os.path.relpath(os.path.abspath(f), self.project.root_dir)
+                         for f in self.args.targets]
+
+        caches = self.project.cache.find_cache(dvc_files)
+
+        self.print_fsck(caches, dvc_files, files_and_stages)
+        return 0
+
+    def all_directions(self):
+        result = []
+        for stage in self.project.stages():
+            for dep in stage.deps + stage.outs:
+                result.append((stage, dep))
+        return result
+
+    def print_fsck(self, caches, dvc_files, files_and_stages):
+        for file in dvc_files:
+            print('File {}:'.format(file))
+
+            full_path = os.path.join(self.project.root_dir, file)
+            if self.args.physical:
+                md5 = file_md5(full_path)[0]
+                print(u'    Actual checksum:        {}'.format(md5))
+            else:
+                md5 = None
+
+            hardlink_md5 = caches.get(file)
+            hardlink_msg = hardlink_md5 if caches else 'Error - cache file not found!'
+            hardlink_error = '!!!' if hardlink_md5 and md5 and md5 != hardlink_md5 else ''
+            print(u'    Hardlink to cache file: {} {}'.format(hardlink_msg, hardlink_error))
+
+            state = self.project.state.get(file)
+            local_state_error = '!!!' if state.md5 and hardlink_md5 and state.md5 != hardlink_md5 else ''
+            print(u'    Local state checksum:   {} {}'.format(state.md5 if state else '', local_state_error))
+            print(u'    Local state mtime:      {}'.format(state.mtime if state else ''))
+            mtime = os.path.getmtime(full_path) if os.path.exists(full_path) else 'None'
+            print(u'    Actual mtime:           {}'.format(mtime))
+
+            for stage, dep in files_and_stages.get(file, []):
+                direction = type(dep).__name__
+                stage_error = '!!!' if dep.md5 and hardlink_md5 and dep.md5 != hardlink_md5 else ''
+                print(u'    Stage file: {}'.format(stage.dvc_path))
+                print(u'        Type:               {}'.format(direction))
+                print(u'        Checksum:           {} {}'.format(dep.md5, stage_error))
+                print(u'        Use cache:          {}'.format(str(dep.use_cache).lower()))
+        pass
+
+    @staticmethod
+    def directions_by_datafile(directions):
+        result = {}
+
+        for stage, dep in directions:
+            if dep.dvc_path not in result:
+                result[dep.dvc_path] = []
+            result[dep.dvc_path].append((stage, dep))
+
+        return result

dvc/stage.py

@@ -54,7 +54,14 @@ class Output(object):
         self.md5 = md5
         self.use_cache = use_cache
 
+    @property
+    def dvc_path(self):
+        return os.path.relpath(self.path, self.project.root_dir)
+
     def _changed_md5(self):
+        if not os.path.exists(self.path):
+            return True
+
         state = self.project.state.get(self.path)
         if state and state.mtime == self.mtime():
             md5 = state.md5
@@ -213,7 +220,11 @@ class Stage(object):
 
     @property
     def relpath(self):
-        return os.path.relpath(self.path, '.')
+        return os.path.relpath(self.path)
+
+    @property
+    def dvc_path(self):
+        return os.path.relpath(self.path, self.project.root_dir)
 
     @staticmethod
     def is_stage_file(path):

dvc/system.py

@@ -27,6 +27,9 @@ class System(object):
 
     @staticmethod
     def samefile(path1, path2):
+        if not os.path.exists(path1) or not os.path.exists(path2):
+            return False
+
         if System.is_unix():
             return os.path.samefile(path1, path2)