mirror of https://github.com/tLDP/LDP
894 lines
23 KiB
Python
Executable File
894 lines
23 KiB
Python
Executable File
#!/usr/bin/python
|
|
|
|
"""
|
|
Lampadas Object Hierarchy Module
|
|
|
|
This module defines Data Objects (Users, Documents, Notes, Topics, etc.)
|
|
for the Lampadas system. All access to the underlying database should be
|
|
performed through this layer.
|
|
"""
|
|
|
|
# Modules
|
|
|
|
import Config
|
|
import Database
|
|
import Log
|
|
from string import strip
|
|
from types import StringType
|
|
|
|
|
|
# Globals
|
|
|
|
Config = Config.Config()
|
|
DB = Database.Database()
|
|
DB.Connect(Config.DBType, Config.DBName)
|
|
Log = Log.Log()
|
|
Log.Truncate()
|
|
|
|
|
|
# Base Classes
|
|
|
|
class LampadasList:
|
|
"""
|
|
Base class for Lampadas list objects, which are cached in RAM
|
|
for high performance.
|
|
|
|
Classes based on this one emulate lists, with additional methods.
|
|
"""
|
|
|
|
list = []
|
|
|
|
def __len__(self):
|
|
return len(self.list)
|
|
|
|
def __getitem__(self, key):
|
|
return self.list[key]
|
|
|
|
def __setitem__(self, key, value):
|
|
self.list[key] = value
|
|
|
|
def __delitem__(self, key):
|
|
del self.list[key]
|
|
|
|
def items(self):
|
|
return self.list.items()
|
|
|
|
def append(self, item):
|
|
self.list.append(item)
|
|
|
|
def Count(self):
|
|
return len(self.list)
|
|
|
|
|
|
class LampadasCollection:
|
|
"""
|
|
Base class for Lampadas collection objects, which are cached in RAM
|
|
for high performance.
|
|
|
|
Classes based on this one become pseudo-dictionaries, providing
|
|
iteration and similar methods. This is done by providing a wrapper to
|
|
the built-in dictionary type. In Python 2.2, dictionaries will be
|
|
subclassable, so this can be rewritten to take advantage of that.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.data = {}
|
|
|
|
def __getitem__(self, key):
|
|
try:
|
|
item = self.data[key]
|
|
except KeyError:
|
|
item = None
|
|
return item
|
|
|
|
def __setitem__(self, key, item):
|
|
self.data[key] = item
|
|
|
|
def __delitem__(self, key):
|
|
del self.data[key]
|
|
|
|
def keys(self):
|
|
return self.data.keys()
|
|
|
|
def Count(self):
|
|
return len(self.data)
|
|
|
|
|
|
# Lampadas
|
|
|
|
class Lampadas:
|
|
"""
|
|
This is the top level container class for all Lampadas objects.
|
|
While you can also create User, Doc, and other classes independently,
|
|
this class can be instantiated and all those objects accessed as part
|
|
of a single object hierarchy.
|
|
|
|
Using this method gives you complete data caching capabilities and a
|
|
single, global access route to all Lampadas data.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.Classes = Classes()
|
|
self.Classes.Load()
|
|
self.Config = Config()
|
|
self.Config.Load()
|
|
self.Docs = Docs()
|
|
self.Docs.Load()
|
|
self.DTDs = DTDs()
|
|
self.Formats = Formats()
|
|
self.Languages = Languages()
|
|
self.PubStatuses = PubStatuses()
|
|
self.Strings = Strings()
|
|
self.Topics = Topics()
|
|
self.Users = Users()
|
|
|
|
def User(self, UserID):
|
|
return User(UserID)
|
|
|
|
def Doc(self, DocID):
|
|
return Doc(DocID)
|
|
|
|
|
|
# Class
|
|
|
|
class Classes(LampadasCollection):
|
|
"""
|
|
A collection object of all document classes (HOWTO, FAQ, etc).
|
|
"""
|
|
|
|
def Load(self):
|
|
self.sql = "SELECT class_id FROM class"
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
row = self.cursor.fetchone()
|
|
if row == None: break
|
|
newClass = Class()
|
|
newClass.Load(row)
|
|
self.data[newClass.ID] = newClass
|
|
|
|
class Class:
|
|
|
|
def __init__(self, ClassID=None):
|
|
self.I18n = {}
|
|
if ClassID==None: return
|
|
self.ID = ClassID
|
|
|
|
def Load(self, row):
|
|
self.ID = row[0]
|
|
self.sql = "SELECT lang, class_name, class_description FROM class_i18n WHERE class_id=" + str(self.ID)
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
self.row = self.cursor.fetchone()
|
|
if self.row == None: break
|
|
newClassI18n = ClassI18n()
|
|
newClassI18n.Load(self.row)
|
|
self.I18n[newClassI18n.Lang] = newClassI18n
|
|
|
|
class ClassI18n:
|
|
|
|
def Load(self, row):
|
|
self.Lang = row[0]
|
|
self.Name = trim(row[1])
|
|
self.Description = trim(row[2])
|
|
|
|
|
|
# Config
|
|
|
|
class Config(LampadasCollection):
|
|
|
|
def __call__(self, key):
|
|
return self[key]
|
|
|
|
def Load(self):
|
|
self.sql = "SELECT name, value FROM config"
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
row = self.cursor.fetchone()
|
|
if row == None: break
|
|
self[trim(row[0])] = trim(row[1])
|
|
|
|
|
|
# Documents
|
|
|
|
class Docs(LampadasCollection):
|
|
"""
|
|
A collection object providing access to all documents.
|
|
"""
|
|
|
|
def Load(self):
|
|
self.sql = "SELECT doc_id, title, class_id, format, dtd, dtd_version, version, last_update, url, isbn, pub_status, review_status, tickle_date, pub_date, ref_url, tech_review_status, maintained, license, abstract, rating, lang, sk_seriesid FROM document"
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
self.row = self.cursor.fetchone()
|
|
if self.row == None: break
|
|
newDoc = Doc()
|
|
newDoc.Load(self.row)
|
|
self[newDoc.ID] = newDoc
|
|
|
|
def Add(self, Title, ClassID, Format, DTD, DTDVersion, Version, LastUpdate, URL, ISBN, PubStatus, ReviewStatus, TickleDate, PubDate, HomeURL, TechReviewStatus, License, Abstract, LanguageCode, SeriesID):
|
|
self.id = DB.Value('SELECT max(doc_id) from document') + 1
|
|
self.sql = "INSERT INTO document(doc_id, title, class_id, format, dtd, dtd_version, version, last_update, url, isbn, pub_status, review_status, tickle_date, pub_date, ref_url, tech_review_status, license, abstract, lang, sk_seriesid) VALUES (" + str(self.id) + ", " + wsq(Title) + ", " + str(ClassID) + ", " + wsq(Format) + ", " + wsq(DTD) + ", " + wsq(DTDVersion) + ", " + wsq(Version) + ", " + wsq(LastUpdate) + ", " + wsq(URL) + ", " + wsq(ISBN) + ", " + wsq(PubStatus) + ", " + wsq(ReviewStatus) + ", " + wsq(TickleDate) + ", " + wsq(PubDate) + ", " + wsq(HomeURL) + ", " + wsq(TechReviewStatus) + ", " + wsq(License) + ", " + wsq(Abstract) + ", " + wsq(LanguageCode) + ", " + wsq(SeriesID) + ")"
|
|
assert DB.Exec(self.sql) == 1
|
|
DB.Commit()
|
|
self.NewID = DB.Value('SELECT MAX(doc_id) from document')
|
|
newDoc = Doc(self.NewID)
|
|
self[self.NewID] = newDoc
|
|
return self.NewID
|
|
|
|
def Del(self, id):
|
|
self.sql = ('DELETE from document WHERE doc_id=' + str(id))
|
|
assert DB.Exec(self.sql) == 1
|
|
DB.Commit()
|
|
del self[id]
|
|
|
|
class Doc:
|
|
"""
|
|
A document in any format, whether local or remote.
|
|
"""
|
|
|
|
def __init__(self, id=None):
|
|
if id == None: return
|
|
self.sql = "SELECT doc_id, title, class_id, format, dtd, dtd_version, version, last_update, url, isbn, pub_status, review_status, tickle_date, pub_date, ref_url, tech_review_status, maintained, license, abstract, rating, lang, sk_seriesid FROM document WHERE doc_id=" + str(id)
|
|
self.cursor = DB.Select(self.sql)
|
|
self.row = self.cursor.fetchone()
|
|
self.Load(self.row)
|
|
|
|
def Load(self, row):
|
|
self.ID = row[0]
|
|
self.Title = trim(row[1])
|
|
self.ClassID = row[2]
|
|
self.Format = trim(row[3])
|
|
self.DTD = trim(row[4])
|
|
self.DTDVersion = trim(row[5])
|
|
self.Version = trim(row[6])
|
|
self.LastUpdate = trim(row[7])
|
|
self.URL = trim(row[8])
|
|
self.ISBN = trim(row[9])
|
|
self.PubStatus = trim(row[10])
|
|
self.ReviewStatus = trim(row[11])
|
|
self.TickleDate = trim(row[12])
|
|
self.PubDate = trim(row[13])
|
|
self.HomeURL = trim(row[14])
|
|
self.TechReviewStatus = trim(row[15])
|
|
self.Maintained = tf2bool(row[16])
|
|
self.License = trim(row[17])
|
|
self.Abstract = trim(row[18])
|
|
self.Rating = safeint(row[19])
|
|
self.LanguageCode = trim(row[20])
|
|
self.SeriesID = trim(row[21])
|
|
|
|
self.Errors = DocErrors(self.ID)
|
|
self.Files = DocFiles(self.ID)
|
|
self.Ratings = DocRatings(self.ID)
|
|
self.Ratings.Parent = self
|
|
self.Versions = DocVersions(self.ID)
|
|
|
|
def Save(self):
|
|
self.sql = "UPDATE document SET title=" + wsq(self.Title) + ", class_id=" + str(self.ClassID) + ", format=" + wsq(self.Format) + ", dtd=" + wsq(self.DTD) + ", dtd_version=" + wsq(self.DTDVersion) + ", version=" + wsq(self.Version) + ", last_update=" + wsq(self.LastUpdate) + ", url=" + wsq(self.URL) + ", isbn=" + wsq(self.ISBN) + ", pub_status=" + wsq(self.PubStatus) + ", review_status=" + wsq(self.ReviewStatus) + ", tickle_date=" + wsq(self.TickleDate) + ", pub_date=" + wsq(self.PubDate) + ", ref_url=" + wsq(self.HomeURL) + ", tech_review_status=" + wsq(self.TechReviewStatus) + ", maintained=" + wsq(bool2tf(self.Maintained)) + ", license=" + wsq(self.License) + ", abstract=" + wsq(self.Abstract) + ", rating=" + dbint(self.Rating) + ", lang=" + wsq(self.LanguageCode) + ", sk_seriesid=" + wsq(self.SeriesID) + " WHERE doc_id=" + str(self.ID)
|
|
DB.Exec(self.sql)
|
|
DB.Commit()
|
|
|
|
|
|
# DocErrors
|
|
|
|
class DocErrors(LampadasList):
|
|
"""
|
|
A collection object providing access to all document errors, as identified by the
|
|
Lintadas subsystem.
|
|
"""
|
|
|
|
def __init__(self, DocID):
|
|
assert not DocID == None
|
|
self.DocID = DocID
|
|
self.sql = "SELECT error FROM document_error WHERE doc_id=" + str(DocID)
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
row = self.cursor.fetchone()
|
|
if row == None: break
|
|
newDocError = DocError()
|
|
newDocError.Load(DocID, row)
|
|
self.list = self.list + [newDocError]
|
|
|
|
def Clear(self):
|
|
self.sql = "DELETE FROM document_error WHERE doc_id=" + str(self.DocID)
|
|
DB.Exec(self.sql)
|
|
self.list = []
|
|
|
|
def Add(self, Error):
|
|
self.sql = "INSERT INTO document_error(doc_id, error) VALUES (" + str(self.DocID) + ", " + wsq(Error)
|
|
assert DB.Exec(self.sql) == 1
|
|
newDocError = DocError()
|
|
newDocError.DocID = self.DocID
|
|
newDocError.Error = Error
|
|
self.list = self.list + [newDocError]
|
|
DB.Commit()
|
|
|
|
class DocError:
|
|
"""
|
|
An error filed against a document by the Lintadas subsystem.
|
|
"""
|
|
|
|
def Load(self, DocID, row):
|
|
assert not DocID == None
|
|
assert not row == None
|
|
self.DocID = DocID
|
|
self.Error = trim(row[0])
|
|
|
|
|
|
# DocFiles
|
|
|
|
class DocFiles(LampadasCollection):
|
|
"""
|
|
A collection object providing access to all document source files.
|
|
"""
|
|
|
|
def __init__(self, DocID):
|
|
self.data = {}
|
|
assert not DocID == None
|
|
self.DocID = DocID
|
|
self.sql = "SELECT filename, format FROM document_file WHERE doc_id=" + str(DocID)
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
row = self.cursor.fetchone()
|
|
if row == None: break
|
|
newDocFile = DocFile()
|
|
newDocFile.Load(DocID, row)
|
|
self.data[newDocFile.Filename] = newDocFile
|
|
|
|
class DocFile:
|
|
"""
|
|
An association between a document and a file.
|
|
"""
|
|
|
|
def Load(self, DocID, row):
|
|
assert not DocID == None
|
|
assert not row == None
|
|
self.DocID = DocID
|
|
self.Filename = trim(row[0])
|
|
self.Format = trim(row[1])
|
|
|
|
def Save(self):
|
|
self.sql = "UPDATE document_file SET format=" + wsq(self.Format) + " WHERE doc_id=" + str(self.DocID) + " AND filename=" + wsq(self.Filename)
|
|
assert DB.Exec(self.sql) == 1
|
|
DB.Commit()
|
|
|
|
|
|
# DocRatings
|
|
|
|
class DocRatings(LampadasCollection):
|
|
"""
|
|
A collection object providing access to all ratings placed on documents by users.
|
|
"""
|
|
|
|
def __init__(self, DocID):
|
|
self.data = {}
|
|
self.Parent = None
|
|
assert not DocID == None
|
|
self.DocID = DocID
|
|
self.sql = "SELECT user_id, date_entered, vote FROM doc_vote WHERE doc_id=" + str(DocID)
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
self.row = self.cursor.fetchone()
|
|
if self.row == None: break
|
|
newDocRating = DocRating()
|
|
newDocRating.Load(DocID, self.row)
|
|
self.data[newDocRating.UserID] = newDocRating
|
|
self.CalcAverage()
|
|
|
|
def Add(self, UserID, Rating):
|
|
newDocRating = DocRating()
|
|
newDocRating.DocID = self.DocID
|
|
newDocRating.UserID = UserID
|
|
newDocRating.Rating = Rating
|
|
newDocRating.Save()
|
|
self.data[newDocRating.UserID] = newDocRating
|
|
self.CalcAverage()
|
|
|
|
def Del(self, UserID):
|
|
if self.data[UserID] == None: return
|
|
del self.data[UserID]
|
|
self.sql = 'DELETE FROM doc_vote WHERE doc_id=' + str(self.DocID) + ' AND user_id=' + str(UserID)
|
|
DB.Exec(self.sql)
|
|
self.CalcAverage()
|
|
|
|
def Clear(self):
|
|
self.sql = "DELETE FROM doc_vote WHERE doc_id=" + str(self.DocID)
|
|
DB.Exec(self.sql)
|
|
self.data = {}
|
|
self.CalcAverage()
|
|
|
|
def CalcAverage(self):
|
|
self.Average = 0
|
|
if self.Count() > 0:
|
|
keys = self.data.keys()
|
|
for key in keys:
|
|
self.Average = self.Average + self.data[key].Rating
|
|
self.Average = self.Average / self.Count()
|
|
self.sql = "UPDATE document SET rating=" + str(self.Average) + " WHERE doc_id=" + str(self.DocID)
|
|
DB.Exec(self.sql)
|
|
if not self.Parent == None:
|
|
self.Parent.Rating = self.Average
|
|
|
|
class DocRating:
|
|
"""
|
|
A rating of a document, assigned by a registered user.
|
|
"""
|
|
|
|
def Load(self, DocID, row):
|
|
assert not DocID == None
|
|
assert not row == None
|
|
self.DocID = DocID
|
|
self.UserID = row[0]
|
|
self.DateEntered = trim(row[1])
|
|
self.Rating = row[2]
|
|
|
|
def Save(self):
|
|
self.sql = "DELETE from doc_vote WHERE doc_id=" + str(self.DocID) + " AND user_id=" + str(self.UserID)
|
|
DB.Exec(self.sql)
|
|
self.sql = "INSERT INTO doc_vote (doc_id, user_id, vote) VALUES (" + str(self.DocID) + ", " + str(self.UserID) + ", " + str(self.Rating) + ")"
|
|
DB.Exec(self.sql)
|
|
DB.Commit()
|
|
|
|
|
|
# DocVersions
|
|
|
|
class DocVersions(LampadasCollection):
|
|
"""
|
|
A collection object providing access to document revisions.
|
|
"""
|
|
|
|
def __init__(self, DocID):
|
|
self.data = {}
|
|
assert not DocID == None
|
|
self.DocID = DocID
|
|
self.sql = "SELECT rev_id, version, pub_date, initials, notes FROM document_rev WHERE doc_id=" + str(DocID)
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
row = self.cursor.fetchone()
|
|
if row == None: break
|
|
newDocVersion = DocVersion()
|
|
newDocVersion.Load(DocID, row)
|
|
self.data[newDocVersion.ID] = newDocVersion
|
|
|
|
class DocVersion:
|
|
"""
|
|
A release of the document.
|
|
"""
|
|
|
|
def Load(self, DocID, row):
|
|
assert not DocID == None
|
|
assert not row == None
|
|
self.DocID = DocID
|
|
self.ID = row[0]
|
|
self.Version = trim(row[1])
|
|
self.PubDate = trim(row[2])
|
|
self.Initials = trim(row[3])
|
|
self.Notes = trim(row[4])
|
|
|
|
def Save(self):
|
|
self.sql = "UPDATE document_rev SET version=" + wsq(self.Version) + ", pub_date=" + wsq(self.PubDate) + ", initials=" + wsq(self.Initials) + ", notes=" + wsq(self.Notes) + "WHERE doc_id=" + str(self.DocID) + " AND rev_id" + wsq(self.ID)
|
|
assert DB.Exec(self.sql) == 1
|
|
DB.Commit()
|
|
|
|
|
|
# DTDs
|
|
|
|
class DTDs(LampadasCollection):
|
|
"""
|
|
A collection object of all DTDs.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.data = {}
|
|
self.sql = "SELECT dtd from dtd"
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
row = self.cursor.fetchone()
|
|
if row == None: break
|
|
newDTD = DTD()
|
|
newDTD.Load(row)
|
|
self.data[newDTD.DTD] = newDTD
|
|
|
|
class DTD:
|
|
|
|
def __init__(self, DTD=None):
|
|
if DTD==None: return
|
|
self.DTD = DTD
|
|
|
|
def Load(self, row):
|
|
self.DTD = trim(row[0])
|
|
|
|
|
|
# Formats
|
|
|
|
class Formats(LampadasCollection):
|
|
"""
|
|
A collection object of all formats.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.data = {}
|
|
self.sql = "SELECT format FROM format"
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
row = self.cursor.fetchone()
|
|
if row == None: break
|
|
newFormat = Format()
|
|
newFormat.Load(row)
|
|
self.data[newFormat.Format] = newFormat
|
|
|
|
class Format:
|
|
|
|
def __init__(self, Format=None):
|
|
self.I18n = {}
|
|
if Format==None: return
|
|
self.Code = Format
|
|
|
|
def Load(self, row):
|
|
self.Format = trim(row[0])
|
|
self.sql = "SELECT lang, format_name FROM format_i18n WHERE format=" + wsq(self.Format)
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
self.row = self.cursor.fetchone()
|
|
if self.row == None: break
|
|
newFormatI18n = FormatI18n()
|
|
newFormatI18n.Load(self.row)
|
|
self.I18n[newFormatI18n.Lang] = newFormatI18n
|
|
|
|
# FormatI18n
|
|
|
|
class FormatI18n:
|
|
|
|
def Load(self, row):
|
|
self.Lang = row[0]
|
|
self.Name = trim(row[1])
|
|
|
|
|
|
# Languages
|
|
|
|
class Languages(LampadasCollection):
|
|
"""
|
|
A collection object of all languages.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.data = {}
|
|
self.sql = "SELECT isocode, language_name FROM language"
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
row = self.cursor.fetchone()
|
|
if row == None: break
|
|
newLanguage = Language()
|
|
newLanguage.Load(row)
|
|
self.data[newLanguage.Code] = newLanguage
|
|
|
|
class Language:
|
|
|
|
def __init__(self, LanguageCode=None):
|
|
self.I18n = {}
|
|
if LanguageCode == None: return
|
|
self.Code = LanguageCode
|
|
self.sql = "SELECT isocode FROM language WHERE isocode= " + wsq(LanguageCode)
|
|
self.cursor = DB.Select(self.sql)
|
|
self.Load(self.sql)
|
|
|
|
def Load(self, row):
|
|
self.Code = trim(row[0])
|
|
self.sql = "SELECT lang, language_name FROM language_i18n WHERE isocode=" + wsq(self.Code)
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
self.row = self.cursor.fetchone()
|
|
if self.row == None: break
|
|
newLanguageI18n = LanguageI18n()
|
|
newLanguageI18n.Load(self.row)
|
|
self.I18n[newLanguageI18n.Lang] = newLanguageI18n
|
|
|
|
# LanguageI18n
|
|
|
|
class LanguageI18n:
|
|
|
|
def Load(self, row):
|
|
self.Lang = row[0]
|
|
self.Name = trim(row[1])
|
|
|
|
|
|
# PubStatuses
|
|
|
|
class PubStatuses(LampadasCollection):
|
|
"""
|
|
A collection object of all publication statuses.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.data = {}
|
|
self.sql = "SELECT pub_status FROM pub_status"
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
row = self.cursor.fetchone()
|
|
if row == None: break
|
|
newPubStatus = PubStatus()
|
|
newPubStatus.Load(row)
|
|
self.data[newPubStatus.PubStatus] = newPubStatus
|
|
|
|
class PubStatus:
|
|
|
|
def __init__(self, PubStatus=None):
|
|
self.I18n = {}
|
|
if PubStatus==None: return
|
|
self.PubStatus = PubStatus
|
|
|
|
def Load(self, row):
|
|
self.PubStatus = trim(row[0])
|
|
self.sql = "SELECT lang, pub_status_name, pub_status_desc FROM pub_status_i18n WHERE pub_status=" + wsq(self.PubStatus)
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
self.row = self.cursor.fetchone()
|
|
if self.row == None: break
|
|
newPubStatusI18n = PubStatusI18n()
|
|
newPubStatusI18n.Load(self.row)
|
|
self.I18n[newPubStatusI18n.Lang] = newPubStatusI18n
|
|
|
|
# PubStatusI18n
|
|
|
|
class PubStatusI18n:
|
|
|
|
def Load(self, row):
|
|
self.Lang = row[0]
|
|
self.Name = trim(row[1])
|
|
self.Description = trim(row[2])
|
|
|
|
|
|
# Strings
|
|
|
|
class Strings(LampadasCollection):
|
|
"""
|
|
A collection object of all localized strings.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.data = {}
|
|
self.sql = "SELECT string_code FROM string"
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
row = self.cursor.fetchone()
|
|
if row == None: break
|
|
newString = String()
|
|
newString.Load(row)
|
|
self.data[newString.Code] = newString
|
|
|
|
class String:
|
|
|
|
def __init__(self, StringCode=None):
|
|
self.I18n = {}
|
|
if StringCode==None: return
|
|
self.Code = StringCode
|
|
|
|
def Load(self, row):
|
|
self.Code = trim(row[0])
|
|
self.sql = "SELECT lang, string FROM string_i18n WHERE string_code=" + wsq(self.Code)
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
self.row = self.cursor.fetchone()
|
|
if self.row == None: break
|
|
newStringI18n = StringI18n()
|
|
newStringI18n.Load(self.row)
|
|
self.I18n[newStringI18n.Lang] = newStringI18n
|
|
|
|
# StringI18n
|
|
|
|
class StringI18n:
|
|
|
|
def Load(self, row):
|
|
self.Lang = row[0]
|
|
self.Text = trim(row[1])
|
|
|
|
|
|
# Topics
|
|
|
|
class Topics(LampadasCollection):
|
|
"""
|
|
A collection object of all topics.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.data = {}
|
|
self.sql = "SELECT topic_num FROM topic"
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
row = self.cursor.fetchone()
|
|
if row == None: break
|
|
newTopic = Topic()
|
|
newTopic.Load(row)
|
|
self.data[newTopic.Num] = newTopic
|
|
|
|
class Topic:
|
|
|
|
def __init__(self, TopicNum=None):
|
|
self.I18n = {}
|
|
if TopicNum==None: return
|
|
self.Num = TopicNum
|
|
|
|
def Load(self, row):
|
|
self.Num = trim(row[0])
|
|
self.sql = "SELECT lang, topic_name, topic_description FROM topic_i18n string_i18n WHERE topic_num=" + wsq(self.Num)
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
self.row = self.cursor.fetchone()
|
|
if self.row == None: break
|
|
newTopicI18n = TopicI18n()
|
|
newTopicI18n.Load(self.row)
|
|
self.I18n[newTopicI18n.Lang] = newTopicI18n
|
|
|
|
class TopicI18n:
|
|
|
|
def Load(self, row):
|
|
self.Lang = row[0]
|
|
self.Name = trim(row[1])
|
|
self.Description = trim(row[2])
|
|
|
|
|
|
# Users
|
|
|
|
class Users:
|
|
"""
|
|
A collection object providing access to registered users.
|
|
"""
|
|
|
|
def Count(self):
|
|
return DB.Value('SELECT count(*) from username')
|
|
|
|
def Add(self, Username, FirstName, MiddleName, Surname, Email, IsAdmin, IsSysadmin, Password, Notes, Stylesheet):
|
|
self.id = DB.Value('SELECT max(user_id) from username') + 1
|
|
self.sql = "INSERT INTO username (user_id, username, first_name, middle_name, surname, email, admin, sysadmin, password, notes, stylesheet) VALUES (" + str(self.id) + ", " + wsq(Username) + ", " + wsq(FirstName) + ", " + wsq(MiddleName) + ", " + wsq(Surname) + ", " + wsq(Email) + ", " + wsq(bool2tf(IsAdmin)) + ", " + wsq(bool2tf(IsSysadmin)) + ", " + wsq(Password) + ", " + wsq(Notes) + ", " + wsq(Stylesheet) + ")"
|
|
assert DB.Exec(self.sql) == 1
|
|
DB.Commit()
|
|
return DB.Value('SELECT max(user_id) from username')
|
|
|
|
def Del(self, id):
|
|
self.sql = ('DELETE from username WHERE user_id=' + str(id))
|
|
assert DB.Exec(self.sql) == 1
|
|
DB.Commit()
|
|
|
|
class User:
|
|
"""
|
|
A user is known by the system and can login to manipulate documents
|
|
and act on the database according to his rights.
|
|
"""
|
|
|
|
def __init__(self, id) :
|
|
self.sql = 'SELECT user_id, username, session_id, first_name, middle_name, surname, email, admin, sysadmin, password, notes, stylesheet FROM username WHERE user_id=' + str(id)
|
|
self.cursor = DB.Select(self.sql)
|
|
row = self.cursor.fetchone()
|
|
self.ID = row[0]
|
|
self.Username = trim(row[1])
|
|
self.SessionID = trim(row[2])
|
|
self.FirstName = trim(row[3])
|
|
self.MiddleName = trim(row[4])
|
|
self.Surname = trim(row[5])
|
|
self.Email = trim(row[6])
|
|
self.IsAdmin = tf2bool(row[7])
|
|
self.IsSyadmin = tf2bool(row[8])
|
|
self.Password = trim(row[9])
|
|
self.Notes = trim(row[10])
|
|
self.Stylesheet = trim(row[11])
|
|
self.Name = trim(trim(self.FirstName + ' ' + self.MiddleName) + ' ' + self.Surname)
|
|
|
|
self.Docs = UserDocs(self.ID)
|
|
|
|
|
|
# UserDocs
|
|
|
|
class UserDocs(LampadasList):
|
|
"""
|
|
A collection object providing access to all user document associations.
|
|
"""
|
|
|
|
def __init__(self, UserID):
|
|
assert not UserID == None
|
|
self.UserID = UserID
|
|
self.sql = "SELECT doc_id, user_id, role, email, active FROM document_user WHERE user_id=" + str(self.UserID)
|
|
self.cursor = DB.Select(self.sql)
|
|
while (1):
|
|
row = self.cursor.fetchone()
|
|
if row == None: break
|
|
newUserDoc = UserDoc(UserID, row[0])
|
|
newUserDoc.Load(row)
|
|
self.list = self.list + [newUserDoc]
|
|
|
|
def Add(self, DocID, Role, Email, Active):
|
|
self.sql = "INSERT INTO document_user(doc_id, user_id, role, email, active) VALUES (" + str(DocID) + ", " + str(self.UserID) + ", " + wsq(Role) + ", " + wsq(Email) + ", " + wsq(bool2tf(Active)) + " )"
|
|
assert DB.Exec(self.sql) == 1
|
|
DB.Commit()
|
|
|
|
def Del(self, DocID):
|
|
self.sql = ('DELETE from document_user WHERE doc_id=' + str(DocID) + ' AND user_id=' + str(self.UserID))
|
|
assert DB.Exec(self.sql) == 1
|
|
DB.Commit()
|
|
del self.col[DocID]
|
|
|
|
class UserDoc:
|
|
"""
|
|
An association between a user and a document.
|
|
"""
|
|
|
|
def __init__(self, UserID=None, DocID=None):
|
|
self.UserID = UserID
|
|
self.DocID = DocID
|
|
if DocID == None: return
|
|
if UserID == None: return
|
|
self.sql = "SELECT doc_id, user_id, role, email, active FROM document_user WHERE doc_id=" + str(DocID) + " AND user_id=" + str(UserID)
|
|
self.cursor = DB.Select(self.sql)
|
|
row = self.cursor.fetchone()
|
|
self.Load(row)
|
|
|
|
def Load(self, row):
|
|
assert not row == None
|
|
self.DocID = row[0]
|
|
self.UserID = row[1]
|
|
self.Role = trim(row[2])
|
|
self.Email = trim(row[3])
|
|
self.Active = tf2bool(row[4])
|
|
|
|
def Save(self):
|
|
self.sql = "UPDATE document_user SET role=" + wsq(self.Role) + ", email=" + wsq(self.Email) + ", active=" + wsq(bool2tf(self.Active)) + " WHERE doc_id=" + str(self.DocID) + " AND user_id=" + str(self.UserID)
|
|
DB.Exec(self.sql)
|
|
DB.Commit()
|
|
|
|
|
|
# Utility routines
|
|
|
|
def wsq(astring):
|
|
if astring == None:
|
|
return 'NULL'
|
|
elif astring == '':
|
|
return 'NULL'
|
|
else:
|
|
return "'" + astring.replace("'", "''") + "'"
|
|
|
|
def dbint(anint):
|
|
if anint == None:
|
|
temp = 'NULL'
|
|
else:
|
|
temp = str(anint)
|
|
return temp
|
|
|
|
def safeint(anint):
|
|
if anint == None:
|
|
return 0
|
|
elif anint == '':
|
|
return 0
|
|
else:
|
|
return int(anint)
|
|
|
|
def bool2tf(bool):
|
|
if bool == 1:
|
|
return 't'
|
|
else:
|
|
return 'f'
|
|
|
|
def tf2bool(tf):
|
|
if tf == 't':
|
|
return 1
|
|
else:
|
|
return 0
|
|
|
|
def trim(astring):
|
|
if astring == None:
|
|
temp = ''
|
|
else:
|
|
temp = str(astring)
|
|
return strip(temp)
|
|
|
|
|
|
# main
|
|
if __name__ == '__main__' :
|
|
print "Running unit tests..."
|
|
string = "foo"
|
|
assert wsq(string) == "'foo'"
|
|
string = "it's"
|
|
assert wsq(string) == "'it''s'"
|
|
string = "it's that's"
|
|
assert wsq(string) == "'it''s that''s'"
|
|
print "End unit test run."
|
|
|