vnshed/backend/db.py

88 lines
2.7 KiB
Python

import sqlite3
from utils import asset
from vntypes import Mark
class VNDB:
def __init__(self, db_name=asset('vn_database.db')):
self.__db_name = db_name
with sqlite3.connect(self.__db_name) as connection:
cursor = connection.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS marks (
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
type TEXT NOT NULL,
value TEXT NOT NULL
);
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS authors (
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
description TEXT NOT NULL,
thumbnail TEXT NOT NULL
);
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS novels (
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
author INTEGER NOT NULL
);
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS posts_log (
novel INTEGER NOT NULL,
channel INTEGER NOT NULL,
marks JSON NOT NULL,
title TEXT NOT NULL,
description TEXT NOT NULL,
author INTEGER NOT NULL,
files_on_disk JSON NOT NULL,
post_info JSON,
post_at INTEGER NOT NULL,
created_at INTEGER NOT NULL
);
''')
connection.commit()
def __execute(self, sql: str, params: tuple):
with sqlite3.connect(self.__db_name) as connection:
connection.cursor().execute(sql, params)
connection.commit()
def __fetch(self, sql: str, params: tuple):
with sqlite3.connect(self.__db_name) as connection:
cursor = connection.cursor()
cursor.execute(sql, params)
return cursor.fetchall()
def insert_mark(self, type: str, value: str):
self.__execute("INSERT INTO marks (type, value) "
"VALUES (?, ?)", (type, value))
def search_mark(self, query: Mark):
return self.__fetch("SELECT type, value FROM marks "
"WHERE value LIKE ? "
"AND type = ?", ('%'+query.value+'%',query.type))
def insert_novel(self, title: str, author_id: int):
self.__execute("INSERT INTO novels (title, author) "
"VALUES (?, ?)", (title, author_id))
def insert_author(self, title: str, desc: str, thumb: str):
self.__execute("INSERT INTO authors (title, description, thumbnail) "
"VALUES (?, ?, ?)", (title, desc, thumb))