import sqlite3 import UtilPack as util import MgrSQLiteDB as MyDB class MgrPupilColDB(MyDB.MgrSQLiteDB): def __init__(self, path: str): super().__init__(path) self.init() def init(self): # 데이터베이스 연결 (파일이 없으면 새로 생성됨) self.conn = sqlite3.connect(f'file:{self.path}?charset=UTF-8', uri=True) self.cursor = self.conn.cursor() # ID -> HitomiID, ArchiveID, Title, Authors, Group, Series, Type. Language, Tags, Description, def GetBookByID(self, nID: int) -> dict: pass def GetArchiveBrBookID(self, nID: int ) -> list[str]: pass def GetAuthorByID(self, nID: int) -> list[str]: pass def GetTagByID(self, nID: int) -> list[str]: pass def GetSeriesByID(self, nID: int) -> list[str]: pass def GetGroupByID(self, nID: int) -> list[str]: pass # def GetIDByName(self, strTableName : str, strName: str) -> int: if True == util.IsEmptyStr(strName) or True == util.IsEmptyStr(strTableName): return -1 strSQL = f"SELECT id FROM \'{strTableName}\' WHERE name = \'{strName}\';" self.SQLExecute(self.cursor, strSQL, True) result = self.cursor.fetchone() if result is None: return -1 return result[0] # def GetAuthorIDByName(self, strName: str) -> int: return self.GetIDByName("authors", strName) # def GetGroupIDByName(self, strName: str) -> int: return self.GetIDByName("groups", strName) # def GetLanguageIDByName(self, strName: str) -> int: return self.GetIDByName("languages", strName) # def GetTagIDByName(self, strTag: str) -> int: return self.GetIDByName("tags", strTag) # ["id"] ["title"] ["language"] ["type"] ["date"] ["artists"] ["artist"] ["groups"] ["group"] ["url"] ["parodys"] ["parody"] ["url"] ["tags"] ["tag"] ["url"] ["female"] ["male"] ["related"] ["languages"] ["galleryid"] ["url"] ["language_localname"] ["name"] ["characters"] ["character"] ["url"] ["files"] def AddBook(self, nHitomiID: int, strArchiveID: str, strTitle: str, listAuthors: list[str], listGroups: list[str], listSeries: list[str], strType: str, strLanguage: str, listTags: list[str], strDescription: str) -> int: pass # 작가 정보 # Language 는 작가가 작품을 주로 작성하는 언어다. 한국인이라도 일본어로 작품을 내면 일본어로 설정 def AddAuthor(self, strName: str, strGroup: str, strLanguage: str, strDesc: str) -> int: if True == util.IsEmptyStr(strName): return -1 # 이미 들어있나? nAuthorID = self.GetAuthorIDByName(strName) if nAuthorID > 0: util.DbgOut(f"이미 존재하는 작가: {strName}", True) return nAuthorID # 그룹이 없으면 추가 nGroupID = self.GetGroupIDByName(strGroup) if nGroupID <= 0: util.DbgOut(f"그룹이 존재하지 않음: {strGroup}", True) nGroupID = self.AddGroup(strGroup) # 그룹 추가 확인, 여기서 잘못되면 더 위에서 잘못된 것. if nGroupID <= 0: util.DbgOut(f"그룹 추가 실패: {strGroup}", True) return -1 # 언어는 오타가 나거나 모르면 unknown : 0 으로 설정 nLanguageID = self.GetLanguageIDByName(strLanguage) if nLanguageID <= 0: util.DbgOut(f"언어가 존재하지 않음: {strLanguage}", True) strLanguage = "unknown" strTableName = "authors" try: strSQL = f"INSERT OR IGNORE INTO \'{strTableName}\' (name, group, language, desc) VALUES (?, ?, ?, ?)" self.cursor.execute(strSQL, (strName, strGroup, strLanguage, strDesc)) nID = self.cursor.lastrowid self.conn.commit() return nID if nID is not None else -1 except sqlite3.Error as e: util.DbgOut(f"SQLite Error occurred: {e}", True) return -1 def AddTag(self, strTag: str) -> int: if True == util.IsEmptyStr(strTag): return -1 nTagID = self.GetTagIDByName(strTag) if nTagID > 0: util.DbgOut(f"이미 존재하는 태그: {strTag}", True) return nTagID strSep = "" strValue = "" if strTag.find(":") >= 0: strSep = strTag.split(":")[0] strValue = strTag.split(":")[1] return self.AddTagRaw(strTag, strSep, strValue) def AddTagRaw(self, strName: str, strSep: str, strValue: str) -> int: if True == util.IsEmptyStr(strName): return -1 strTableName = "tags" try: strSQL = f"INSERT INTO \'{strTableName}\' (name, sep_title, value) VALUES (?)" self.cursor.execute(strSQL, (strName, strSep, strValue,)) nID = self.cursor.lastrowid self.conn.commit() return nID if nID is not None else -1 except sqlite3.Error as e: util.DbgOut(f"SQLite Error occurred: {e}", True) return -1 def AddsSeries(self, strName: str) -> int: pass def AddGroup(self, strName: str) -> int: if True == util.IsEmptyStr(strName): return -1 strTableName = "groups" try: strSQL = f"INSERT INTO \'{strTableName}\' (name) VALUES (?)" self.cursor.execute(strSQL, (strName,)) nID = self.cursor.lastrowid self.conn.commit() return nID if nID is not None else -1 except sqlite3.Error as e: util.DbgOut(f"SQLite Error occurred: {e}", True) return -1 def AddArchive(self, strName: str, setArcPath: str) -> int: pass # ["width"] ["hash"] ["name"] ["height"] ["hasavif"] # filename, hash, ext, hasavif, width, height def AddFileInfo(self, strName: str, strHash: str, bHasAvif: bool, nWidth: int, nHeight: int) -> int: strTableName = "files" try: strSQL = f"INSERT INTO \'{strTableName}\' (filename, hash, hasavif, width, height) VALUES (?)" self.cursor.execute(strSQL, (strName, strHash, bHasAvif, nWidth, nHeight)) nID = self.cursor.lastrowid self.conn.commit() return nID if nID is not None else -1 except sqlite3.Error as e: util.DbgOut(f"SQLite Error occurred: {e}", True) return -1