First working scanner.
This commit is contained in:
Birth Markus
parent
4fc03a7a77
commit
805fed2294
102
MosDb.py
Normal file
102
MosDb.py
Normal file
@ -0,0 +1,102 @@
|
||||
from datetime import datetime
|
||||
import sqlite3
|
||||
|
||||
class MosDb():
|
||||
def __init__(self, dbfile = "mosobjects.db3"):
|
||||
self.idcache = {}
|
||||
self.ridcache = {}
|
||||
self.dbfile = dbfile
|
||||
self.conn = sqlite3.connect(self.dbfile)
|
||||
self.c = self.conn.cursor()
|
||||
self.c.execute("PRAGMA foreign_keys=on;")
|
||||
|
||||
def commit(self):
|
||||
self.conn.commit()
|
||||
|
||||
def finish(self):
|
||||
self.c.execute("VACUUM")
|
||||
self.conn.commit()
|
||||
self.conn.close()
|
||||
|
||||
def build_id_cache(self, table):
|
||||
self.idcache[table] = {}
|
||||
self.ridcache[table] = {}
|
||||
self.c.execute("SELECT * FROM " + table)
|
||||
for row in self.c.fetchall():
|
||||
self.idcache[table][row[1]] = row[0]
|
||||
self.ridcache[table][row[0]] = row[1]
|
||||
|
||||
def add_new_id(self, table, col, value):
|
||||
self.c.execute("INSERT INTO " + table + " (" + col + ") VALUES (?)", (value,))
|
||||
self.conn.commit()
|
||||
rid = self.c.lastrowid
|
||||
self.idcache[table][value] = rid
|
||||
self.ridcache[table][rid] = value
|
||||
|
||||
def get_id(self, table, col, value):
|
||||
if not table in self.idcache:
|
||||
self.build_id_cache(table)
|
||||
if not value in self.idcache[table]:
|
||||
# New value we didn't know yet, add it
|
||||
self.add_new_id(table, col, value)
|
||||
return self.idcache[table][value]
|
||||
|
||||
def get_mos_id(self, mos_server):
|
||||
return self.get_id("mosservers", "mosServer", mos_server)
|
||||
|
||||
def get_status_id(self, status):
|
||||
return self.get_id("mosstatus", "status", status)
|
||||
|
||||
def get_objair_id(self, obj_air):
|
||||
return self.get_id("mosair", "objAir", obj_air)
|
||||
|
||||
def get_objgroup_id(self, obj_group):
|
||||
return self.get_id("mosgroups", "objGroup", obj_group)
|
||||
|
||||
def get_user_id(self, username):
|
||||
return self.get_id("mosusers", "name", username)
|
||||
|
||||
def date_to_stamp(self, datestring):
|
||||
dt = datetime.strptime(datestring, "%Y-%m-%dT%H:%M:%S")
|
||||
return dt.timestamp()
|
||||
|
||||
def add_link(self, obj_id, planning_id):
|
||||
self.db_insert_or_replace("mosobjlinks", {"objId": obj_id, "planningId": planning_id})
|
||||
|
||||
def db_insert_or_replace(self, table, data):
|
||||
fieldsList = []
|
||||
placeHolders = []
|
||||
valuesTuple = ()
|
||||
for k in data.keys():
|
||||
fieldsList.append(k)
|
||||
placeHolders.append("?")
|
||||
valuesTuple += (data[k],)
|
||||
|
||||
fieldsList = ", ".join(fieldsList)
|
||||
placeHolders = ", ".join(placeHolders)
|
||||
self.c.execute(
|
||||
"INSERT OR REPLACE INTO " + table + " (" + fieldsList + ") VALUES (" + placeHolders + ")",
|
||||
valuesTuple
|
||||
)
|
||||
|
||||
def add_or_replace(self, objData):
|
||||
insertObj = {
|
||||
"objId": objData["objId"],
|
||||
"mosId": self.get_mos_id(objData["mosId"]),
|
||||
"statusId": self.get_status_id(objData["status"]),
|
||||
"objSlug": objData["objSlug"],
|
||||
"objPath": objData["objPath"],
|
||||
"description": objData["description"],
|
||||
"duration": objData["duration"],
|
||||
"objAir": self.get_objair_id(objData["objAir"]),
|
||||
"created": self.date_to_stamp(objData["created"]),
|
||||
"createdBy": self.get_user_id(objData["createdBy"]),
|
||||
"changed": self.date_to_stamp(objData["changed"]),
|
||||
"changedBy": self.get_user_id(objData["changedBy"]),
|
||||
"groupId": self.get_objgroup_id(objData["objGroup"])
|
||||
}
|
||||
|
||||
self.db_insert_or_replace("mosobjects", insertObj)
|
||||
|
||||
if objData["planningId"] != "":
|
||||
self.add_link(objData["objId"], objData["planningId"])
|
||||
39
MosXml.py
Normal file
39
MosXml.py
Normal file
@ -0,0 +1,39 @@
|
||||
import xml.etree.ElementTree
|
||||
|
||||
def get_text_or_default(parent_node: xml.etree.ElementTree.Element, node_names, default_value):
|
||||
if not isinstance(node_names, list):
|
||||
node_names = [node_names]
|
||||
node = parent_node
|
||||
for n in node_names:
|
||||
node = node.find(n)
|
||||
if node is None:
|
||||
return default_value
|
||||
return node.text
|
||||
|
||||
def parse_mos_xml(mosElem: xml.etree.ElementTree.Element) -> bool:
|
||||
mosId = mosElem.find("mosID").text
|
||||
mosObj = mosElem.find("mosObj")
|
||||
objId = mosObj.find("objID").text
|
||||
status = mosObj.find("status").text
|
||||
|
||||
if status == "DELETED":
|
||||
return False
|
||||
|
||||
objData = {
|
||||
"mosId": mosId,
|
||||
"objId": objId,
|
||||
"status": status,
|
||||
"objSlug": get_text_or_default(mosObj, "objSlug", ""),
|
||||
"objPath": get_text_or_default(mosObj, ["objPaths", "objPath"], ""),
|
||||
"objGroup": get_text_or_default(mosObj, "objGroup", ""),
|
||||
"objAir": get_text_or_default(mosObj, "objAir", ""),
|
||||
"created": get_text_or_default(mosObj, "created", ""),
|
||||
"createdBy": get_text_or_default(mosObj, "createdBy", ""),
|
||||
"changed": get_text_or_default(mosObj, "changed", ""),
|
||||
"changedBy": get_text_or_default(mosObj, "changedBy", ""),
|
||||
"description": get_text_or_default(mosObj, "description", ""),
|
||||
"duration": get_text_or_default(mosObj, "objDur", "0"),
|
||||
"planningId": get_text_or_default(mosObj, ["mosExternalMetadata", "mosPayload", "planningID"], "")
|
||||
}
|
||||
#print(repr(objData))
|
||||
return objData
|
||||
17
dbschema.sql
17
dbschema.sql
@ -40,6 +40,11 @@ CREATE TABLE "mosgroups" (
|
||||
"objGroup" TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE "mosusers" (
|
||||
"userId" INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
"name" TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE "mosobjects" (
|
||||
"objId" TEXT PRIMARY KEY,
|
||||
"mosId" INTEGER REFERENCES "mosservers" ("mosId"),
|
||||
@ -48,9 +53,11 @@ CREATE TABLE "mosobjects" (
|
||||
"objPath" TEXT,
|
||||
"objAir" INTEGER REFERENCES "mosair" ("airId"),
|
||||
"created" INTEGER,
|
||||
"createdBy" TEXT,
|
||||
"createdBy" INTEGER REFERENCES "mosusers" ("userId"),
|
||||
"changed" INTEGER,
|
||||
"changedBy" TEXT,
|
||||
"changedBy" INTEGER REFERENCES "mosusers" ("userId"),
|
||||
"description" TEXT,
|
||||
"duration" INTEGER,
|
||||
"groupId" INTEGER REFERENCES "mosgroups" ("groupId")
|
||||
);
|
||||
|
||||
@ -59,6 +66,12 @@ CREATE TABLE "mosobjlinks" (
|
||||
"planningId" TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE "moslufs" (
|
||||
"objId" TEXT PRIMARY KEY,
|
||||
"lufs" REAL,
|
||||
"ludiff" REAL
|
||||
);
|
||||
|
||||
CREATE TABLE "todo" (
|
||||
"objId" TEXT PRIMARY KEY
|
||||
);
|
||||
|
||||
@ -5,10 +5,10 @@ import configparser
|
||||
import datetime
|
||||
import fnmatch
|
||||
import os
|
||||
import re
|
||||
import sqlite3
|
||||
import sys
|
||||
import xml.etree.ElementTree as ET
|
||||
import MosDb
|
||||
import MosXml
|
||||
|
||||
c = configparser.ConfigParser()
|
||||
c.read("config.ini")
|
||||
@ -33,41 +33,7 @@ print("Found {} files to be parsed.".format(num_files))
|
||||
|
||||
#sys.exit(0)
|
||||
|
||||
def get_text_or_default(parent_node, node_names, default_value):
|
||||
if not isinstance(node_names, list):
|
||||
node_names = [node_names]
|
||||
node = parent_node
|
||||
for n in node_names:
|
||||
node = node.find(n)
|
||||
if node is None:
|
||||
return default_value
|
||||
return node.text
|
||||
|
||||
def parse_mos_xml(mosElem):
|
||||
mosId = mosElem.find("mosID").text
|
||||
mosObj = mosElem.find("mosObj")
|
||||
objId = mosObj.find("objID").text
|
||||
status = mosObj.find("status").text
|
||||
|
||||
if status == "DELETED":
|
||||
return False
|
||||
|
||||
objData = {
|
||||
"mosId": mosId,
|
||||
"objId": objId,
|
||||
"status": status,
|
||||
"objSlug": get_text_or_default(mosObj, "objSlug", ""),
|
||||
"objPath": get_text_or_default(mosObj, ["objPaths", "objPath"], ""),
|
||||
"objGroup": get_text_or_default(mosObj, "objGroup", ""),
|
||||
"objAir": get_text_or_default(mosObj, "objAir", ""),
|
||||
"created": get_text_or_default(mosObj, "created", ""),
|
||||
"createdBy": get_text_or_default(mosObj, "createdBy", ""),
|
||||
"changed": get_text_or_default(mosObj, "changed", ""),
|
||||
"changedBy": get_text_or_default(mosObj, "changedBy", ""),
|
||||
"planningId": get_text_or_default(mosObj, ["mosExternalMetadata", "mosPayload", "planningID"], "")
|
||||
}
|
||||
#print(repr(objData))
|
||||
return objData
|
||||
db = MosDb.MosDb()
|
||||
|
||||
file_count = 0
|
||||
last_file = ""
|
||||
@ -85,12 +51,13 @@ for file in files_to_parse:
|
||||
continue
|
||||
|
||||
mos = elem.find("mos")
|
||||
objData = parse_mos_xml(mos)
|
||||
objData = MosXml.parse_mos_xml(mos)
|
||||
if objData:
|
||||
# TODO: Write objData to database
|
||||
db.add_or_replace(objData)
|
||||
print(".", end="")
|
||||
else:
|
||||
print("x", end="")
|
||||
|
||||
db.finish()
|
||||
print("")
|
||||
print("Last file: {}".format(last_file))
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user