Created
August 19, 2014 08:45
-
-
Save jcollado/89d5334211f09a0422b8 to your computer and use it in GitHub Desktop.
Look for sqlite database files and check their integrity
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/python | |
"""Look for sqlite database files and check their integrity.""" | |
import argparse | |
import logging | |
import os | |
import sqlite3 | |
from contextlib import closing | |
from fnmatch import fnmatch | |
def main(args): | |
"""Walk from base directory and check files that match pattern.""" | |
check_passed = 0 | |
check_failed = 0 | |
passed_paths = [] | |
for path in look_for_sqlite_files(args.directory, args.patterns): | |
if run_integrity_check(path): | |
check_passed += 1 | |
passed_paths.append(path) | |
else: | |
check_failed += 1 | |
logging.info( | |
'%d files checked, %d passed, %d failed', | |
check_passed + check_failed, check_passed, check_failed) | |
logging.info('Passed paths:\n%s', '\n'.join(passed_paths)) | |
def look_for_sqlite_files(directory, patterns): | |
"""Walk from base directory and yield files that match pattern. | |
:param directory: Base directory to walk from | |
:type directory: str | |
:param patterns: Patterns that files should match against | |
:type patterns: list(str) | |
""" | |
for (dirpath, _dirnames, filenames) in os.walk(directory): | |
logging.debug('Exploring %s...', dirpath) | |
for filename in filenames: | |
if any(fnmatch(filename, pattern) | |
for pattern in patterns): | |
yield os.path.join(dirpath, filename) | |
def run_integrity_check(path): | |
"""Check potential sqlite database. | |
:param path: Path to file | |
:type path: str | |
""" | |
logging.debug('Checking %s...', path) | |
with closing(sqlite3.connect(path)) as connection: | |
with closing(connection.cursor()) as cursor: | |
try: | |
cursor.execute('PRAGMA integrity_check;') | |
return cursor.fetchone()[0] == 'ok' | |
except sqlite3.DatabaseError: | |
return False | |
def parse_arguments(): | |
"""Parse command line arguments. | |
:return: Parsed arguments | |
:rtype: argparse.Namespace | |
""" | |
parser = argparse.ArgumentParser(description=__doc__) | |
parser.add_argument( | |
'directory', | |
help='Base directory to look for sqlite database') | |
parser.add_argument( | |
'patterns', | |
metavar='pattern', | |
nargs='+', | |
help='Pattern that file name should match to check it') | |
return parser.parse_args() | |
if __name__ == '__main__': | |
logging.basicConfig( | |
format='%(levelname)s: %(message)s', | |
level=logging.DEBUG, | |
) | |
args = parse_arguments() | |
main(args) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment