-
Notifications
You must be signed in to change notification settings - Fork 0
/
dbs3.py
executable file
·124 lines (91 loc) · 2.98 KB
/
dbs3.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
import sqlite3
import json
class Db:
schema_file = "schema.sql"
def __init__(self, databaseName="files.db"):
self.databaseName = databaseName
self.conn = None
self.cur = None
def connect(self):
with sqlite3.connect(self.databaseName) as self.conn:
self.cur = self.conn.cursor()
def checkTableExists(self, tableName):
if not self.conn:
self.connect()
self.cur.execute("""
SELECT count(*) as cnt
FROM sqlite_master
WHERE type='table' AND name='files'
""")
return self.cur.fetchone()[0]
def read_schema(self):
with open(self.schema_file) as sf:
lines = sf.readlines()
copy_started = False
par = 0
tables = []
collect = []
for line in lines:
row = line.strip()
if row.startswith("CREATE TABLE"):
if par != 0:
collect = []
copy_started = True
collect.append(row)
par = row.count("(") - row.count(")")
continue
if copy_started:
if par != 0:
collect.append(row)
par += row.count("(") - row.count(")")
else:
copy_started = False
tables.append("".join(collect))
collect = []
# flush collect if has lines and parantesis = 0
if copy_started and par == 0:
copy_started = False
tables.append("".join(collect))
return tables
def createFilesTable(self):
tables = self.read_schema()
for table in tables:
try:
self.cur.execute(table)
except sqlite3.OperationalError as err:
print("Table cannot be created:\n{}".format(table))
print("Error: {}".format(err))
continue
def checkConnection(self):
if not self.checkTableExists("files"):
self.createFilesTable()
def saveFiles(self, files):
self.checkConnection()
for fileName, md5 in files.items():
# Insert a row of data
self.cur.execute(
"""
INSERT OR REPLACE
INTO files (fileName, md5)
VALUES (?, ?)
""", (fileName, md5)
)
print("{} = {}".format(fileName, md5))
# Save (commit) the changes
self.conn.commit()
def test_file(self, fileName, testValue) -> int:
self.checkConnection()
self.cur.execute(
"""
SELECT filename, md5
FROM files
WHERE fileName = ?
""", (fileName,)
)
record = self.cur.fetchone()
if record is None:
return -1
return 1 if record[1] == testValue else 0
if __name__ == "__main__":
db = Db()
db.checkConnection()