mastofeeds/app/libraries/database.py

229 líneas
5,8 KiB
Python

2023-05-06 15:05:41 +02:00
import os
import sys
import psycopg2
from psycopg2 import sql
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
import uuid
from datetime import datetime
import pytz
import pdb
tz = pytz.timezone('Europe/Madrid')
class Database():
name = 'mastofeeds database library'
def __init__(self, config_file=None, mastofeeds_db=None, mastofeeds_db_user=None, mastofeeds_db_user_password=None, feeds_url=None):
self.config_file = "config/db_config.txt"
self.mastofeeds_db = self.__get_parameter("mastofeeds_db", self.config_file)
self.mastofeeds_db_user = self.__get_parameter("mastofeeds_db_user", self.config_file)
self.mastofeeds_db_user_password = self.__get_parameter("mastofeeds_db_user_password", self.config_file)
self.feeds_url = self.__get_parameter("feeds_url", self.config_file)
db_setup = self.__check_dbsetup(self)
if not db_setup:
self.mastofeeds_db = input("\nmastofeeds database name: ")
self.mastofeeds_db_user = input("\nmastofeeds database user: ")
self.mastofeeds_db_user_password = input("\nmastofeeds database user password: ")
self.feeds_url = input("\nenter feed URL: ")
self.__createdb(self)
self.__create_config(self)
self.__write_config(self)
def published(self, link):
# check database if feed is already published
publish = False
try:
conn = None
conn = psycopg2.connect(database = self.mastofeeds_db, user = self.mastofeeds_db_user, password = "", host = "/var/run/postgresql", port = "5432")
cur = conn.cursor()
cur.execute('select link from mastofeeds where link=(%s)', (link,))
row = cur.fetchone()
if row == None:
publish = True
cur.close()
return publish
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
def write_feed(self, link):
insert_line = 'INSERT INTO mastofeeds(link) VALUES (%s)'
conn = None
try:
conn = psycopg2.connect(database = self.mastofeeds_db, user = self.mastofeeds_db_user, password = "", host = "/var/run/postgresql", port = "5432")
cur = conn.cursor()
cur.execute(insert_line, (link,))
conn.commit()
cur.close()
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
@staticmethod
def __check_dbsetup(self):
db_setup = False
try:
conn = None
conn = psycopg2.connect(database = self.mastofeeds_db, user = self.mastofeeds_db_user, password = self.mastofeeds_db_user_password, host = "/var/run/postgresql", port = "5432")
db_setup = True
except (Exception, psycopg2.DatabaseError) as error:
print(error)
return db_setup
@staticmethod
def __createdb(self):
conn = None
try:
conn = psycopg2.connect(dbname='postgres',
user=self.mastofeeds_db_user, host='',
password=self.mastofeeds_db_user_password)
conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
cur = conn.cursor()
print(f"Creating database {self.mastofeeds_db}. Please wait...")
cur.execute(sql.SQL("CREATE DATABASE {}").format(
sql.Identifier(self.mastofeeds_db))
)
print(f"Database {self.mastofeeds_db} created!\n")
self.__dbtables_schemes(self)
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
@staticmethod
def __dbtables_schemes(self):
db = self.mastofeeds_db
table = "mastofeeds"
sql = "create table "+table+" (link varchar(300) PRIMARY KEY)"
self.__create_table(self, table, sql)
@staticmethod
def __create_table(self, table, sql):
conn = None
try:
conn = psycopg2.connect(database = self.mastofeeds_db, user = self.mastofeeds_db_user, password = self.mastofeeds_db_user_password, host = "/var/run/postgresql", port = "5432")
cur = conn.cursor()
print(f"Creating table {table}")
cur.execute(sql)
conn.commit()
print(f"Table {table} created!\n")
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
def __get_parameter(self, parameter, config_file):
if not os.path.isfile(config_file):
print(f"File {config_file} not found..")
return
with open( config_file ) as f:
for line in f:
if line.startswith( parameter ):
return line.replace(parameter + ":", "").strip()
print(f"{config_file} Missing parameter {parameter}")
sys.exit(0)
@staticmethod
def __create_config(self):
if not os.path.exists('config'):
os.makedirs('config')
if not os.path.exists(self.config_file):
print(self.config_file + " created!")
with open(self.config_file, 'w'): pass
@staticmethod
def __write_config(self):
with open(self.config_file, 'a') as the_file:
the_file.write(f'mastofeeds_db: {self.mastofeeds_db}\nmastofeeds_db_user: {self.mastofeeds_db_user}\nmastofeeds_db_user_password: {self.mastofeeds_db_user_password}\nfeeds_url: {self.feeds_url}')
print(f"adding parameters to {self.config_file}\n")