ada/sun_project
1
0
Fork 0
mirror of https://github.com/ada-dmitry/sun_project.git synced 2026-09-24 00:30:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

init

Browse Source
This commit is contained in:
ada-dmitry
2024-09-13 00:10:39 +03:00
commit 5ea57e2f4a
1626 changed files with 289568 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
final/__pycache__/add_func.cpython-311.pyc
BIN
View File
Binary file not shown.
final/__pycache__/dbconnection.cpython-311.pyc
BIN
View File
Binary file not shown.
final/__pycache__/dbtable.cpython-311.pyc
BIN
View File
Binary file not shown.
final/__pycache__/project_config.cpython-311.pyc
BIN
View File
Binary file not shown.
final/__pycache__/texts.cpython-311.pyc
BIN
View File
Binary file not shown.
final/add_func.py
+49
View File
@@ -0,0 +1,49 @@
import re
def validate_input(prompt, min_value, max_value):
"""Проверка ввода для выбора номера строки
Args:
prompt (string): "Промт" или запрос пользователю
min_value (int): Нижняя граница, по умолчанию 0
max_value (int): Верхняя граница
Returns:
int: Либо число, введенное пользователем, либо -1 для выхода
"""
while True:
user_input = input(prompt)
try:
value = int(user_input)
if min_value <= value <= max_value and value != 0:
return value
elif value == 0:
return -1
else:
print(f"Введенное значение должно быть в диапазоне от {min_value} до {max_value}")
except ValueError:
print("Введенное значение должно быть числом.")
def add_zero_before(old_index: str, max_len: str) -> str:
"""Функция для "красивого" вывода индексов
Args:
old_index (str): текущий индекс для вывода
max_len (int): максимальная длина индексов
Returns:
new_index (str): окончательная версия индекса
"""
ln = len(old_index)
new_index = old_index[:]
for i in range(max_len-ln):
new_index = '0' + new_index
return new_index
# test1 = 'erhwogngonwro'
# test2 = '1 minute'
# test3 = '2 hours 3 minutes'
# print(validate_time_format(test1), validate_time_format(test2), validate_time_format(test3))
final/config.yaml
+6
View File
@@ -0,0 +1,6 @@
# Базовые параметры настройки конфигурации проекта
dbname: Test # В классе менять не надо
user: postgres # Замените на свой логин от БД в классе
password: Qwerty # Замените на свой пароль от БД в классе
host: localhost # Замените на 192.168.0.48 в классе
dbtableprefix: "public." # Замените на свой логин от БД
final/dbconnection.py
+34
View File
@@ -0,0 +1,34 @@
# Установка соединения с базой данных
# (параметры передаются через класс конфиг).
import psycopg2
class DbConnection:
def __init__(self, config):
self.dbname = config.dbname
self.user = config.user
self.password = config.password
self.host = config.host
self.prefix = config.dbtableprefix
self.conn = psycopg2.connect(dbname = self.dbname,
user = self.user,
password = self.password,
host = self.host)
def __del__(self):
if self.conn:
self.conn.close()
def test(self):
cur = self.conn.cursor()
cur.execute("DROP TABLE IF EXISTS test CASCADE")
cur.execute("CREATE TABLE test(test integer)")
cur.execute("INSERT INTO test(test) VALUES(1)")
self.conn.commit()
cur.execute("SELECT * FROM test")
result = cur.fetchall()
cur.execute("DROP TABLE test")
self.conn.commit()
return (result[0][0] == 1)
final/dbtable.py
+124
View File
@@ -0,0 +1,124 @@
# Базовые действия с таблицами
from dbconnection import *
'''
FIXME:
'''
class DbTable:
dbconn = None
def __init__(self):
return
def table_name(self):
return self.dbconn.prefix + "table"
def columns(self):
return {"test": ["integer", "PRIMARY KEY"]}
def column_names(self):
return sorted(self.columns().keys(), key = lambda x: x)
def primary_key(self):
return ['id']
def column_names_without_id(self):
res = sorted(self.columns().keys(), key = lambda x: x)
if 'id' in res:
res.remove('id')
return res
def table_constraints(self):
return []
def create(self):
sql = "CREATE TABLE " + self.table_name() + "("
arr = [k + " " + " ".join(v) for k, v in sorted(self.columns().items(), key = lambda x: x[0])]
sql += ", ".join(arr + self.table_constraints())
sql += ")"
cur = self.dbconn.conn.cursor()
cur.execute(sql)
self.dbconn.conn.commit()
return
def drop(self):
sql = "DROP TABLE IF EXISTS " + self.table_name()
cur = self.dbconn.conn.cursor()
cur.execute(sql)
self.dbconn.conn.commit()
return
def insert_one(self, vals):
for i in range(len(vals)):
if type(vals[i]) == str:
vals[i] = "'" + vals[i] + "'"
else:
vals[i] = str(vals[i])
values = ", ".join(vals)
query = f"""INSERT INTO {self.table_name()}({", ".join(self.column_names_without_id())}) VALUES({values})"""
cur = self.dbconn.conn.cursor()
# cur.execute(sql)
try:
cur.execute(query)
self.dbconn.conn.commit()
except psycopg2.errors.UniqueViolation:
self.dbconn.conn.rollback()
print('Такая строка уже существует. Повторите попытку')
return
def update(self, column, values, wh):
sql = f"UPDATE {self.table_name()} SET {column} = {values} WHERE id = {wh};"
cur = self.dbconn.conn.cursor()
cur.execute(sql)
self.dbconn.conn.commit()
return
def first(self):
sql = "SELECT * FROM " + self.table_name()
sql += " ORDER BY "
sql += ", ".join(self.primary_key())
cur = self.dbconn.conn.cursor()
cur.execute(sql)
return cur.fetchone()
def last(self):
sql = "SELECT * FROM " + self.table_name()
sql += " ORDER BY "
sql += ", ".join([x + " DESC" for x in self.primary_key()])
cur = self.dbconn.conn.cursor()
cur.execute(sql)
return cur.fetchone()
def all(self):
sql = "SELECT * FROM " + self.table_name()
sql += " ORDER BY "
sql += ", ".join(self.primary_key())
cur = self.dbconn.conn.cursor()
cur.execute(sql)
return cur.fetchall()
def select_one(self, **kwargs):
conditions = []
values = []
sorted_kwargs = sorted(kwargs.items(), key=lambda x: x[0])
for key, value in sorted_kwargs():
conditions.append(f"{key}=%s")
values.append(value)
sql = f"SELECT * FROM {self.table_name()} WHERE " + " AND ".join(conditions)
cur = self.dbconn.conn.cursor()
cur.execute(sql, tuple(values))
result = cur.fetchone()
cur.close()
if result:
return True
else:
return False
final/main.py
+265
View File
@@ -0,0 +1,265 @@
import sys
sys.path.append('tables')
from project_config import *
from dbconnection import *
from tables.routes_table import *
from tables.station_table import *
class Main:
config = ProjectConfig()
connection = DbConnection(config)
def __init__(self):
DbTable.dbconn = self.connection
return
def db_init(self):
stat = StationTable()
rout = RoutesTable()
stat.create()
rout.create()
return
def db_insert_somethings(self):
stat = StationTable()
rout = RoutesTable()
stat.example_insert()
rout.example_insert()
def db_drop(self):
stat = StationTable()
rout = RoutesTable()
rout.drop()
stat.drop()
return
def show_main_menu(self):
menu = """Привутствуем в меню, выберите действие:
1 - просмотр станций;
2 - сброс и инициализация БД;
9 - выход;"""
print(menu)
return
def read_next_step(self):
return input("=> ").strip()
def after_main_menu(self, next_step):
if next_step == "2":
self.db_drop()
self.db_init()
self.db_insert_somethings()
print("Таблицы созданы заново!")
return "0"
elif next_step != "1" and next_step != "9":
print("Выбрано неверное число! Повторите ввод!")
return "0"
else:
return next_step
def show_station(self):
self.station_id = -1
self.station_arr = []
self.max_len_name = -9000
lst = StationTable().all()
self.max_station_index = (len(lst))
for i in lst:
self.station_arr.append([i[0], str(i[2]), str(i[3]), str(i[1])])
# 0 - id, 1 - name, 2 - tarrif_id, 3 - index
self.max_len_name = max(self.max_len_name, len(str(i[2])))
menu = f"""Просмотр списка станций.
№\tНазвание {" "*(self.max_len_name-4)}Тарифная зона Индекс\n{"-"*100}"""
print(menu)
for i in range(len(self.station_arr)):
output = f"""{str(i+1)}\t{self.station_arr[i][1]} {" "*(4+self.max_len_name - len(self.station_arr[i][1]))}\
{str(self.station_arr[i][2])} {" "*(13-len(str(self.station_arr[i][2])))}{str(self.station_arr[i][3])}"""
print(output)
menu = f"""{"-"*100}\nДальнейшие операции:
0 - возврат в главное меню;
3 - добавление новой станции;
4 - удаление станции;
5 - просмотр маршрутов по станции;
6 - добавление маршрута;
8 - обновить станцию;
9 - выход."""
print(menu)
return
def after_show_station(self, next_step):
"""Выбор действий после вывода категорий
"""
while True:
if next_step == "4": # Удаление станции
x = add_func.validate_input('Введите номер удаляемой станции (0 - для отмены): ', 0, self.max_station_index)
if(x!=-1):
StationTable().delete(self.station_arr[int(x)-1])
return "1"
elif next_step == "6": #Добавление маршрута
RoutesTable().insert_route_one(self.max_station_index)
next_step = "5"
elif next_step == "7":#Удаление маршрута
x = add_func.validate_input('Введите номер удаляемого маршрута (0 - для отмены): ', 0, self.max_routes_index)
if(x!=-1):
RoutesTable().delete(self.routes_arr[int(x)-1][2])
else:
return "1"
next_step = "5"
elif next_step == "5":
next_step = self.show_routes_with_station()
elif next_step == "8": # Обновление названия станции
x = add_func.validate_input('Введите номер обновляемой станции (0 - для отмены): ', 0, self.max_station_index)
if(x!=-1):
StationTable().update(self.station_arr[int(x)-1])
return "1"
elif next_step != "0" and next_step != "9" and next_step != "3":
print("Выбрано неверное число! Повторите ввод!")
return "1"
else:
return next_step
def add_station(self):
while True:
st_name = input('Введите название добавляемой станции (0 - отмена): ').strip()
if len(st_name) > 20:
print('Недопустмая длина названия станции')
elif st_name == '0':
return "1"
else:
break
tarrif_zone_id = add_func.validate_input('Введите номер тарифной зоны(0-отмена): ', 0, 7)
if(tarrif_zone_id==-1) or tarrif_zone_id == '1':
return "1"
st_index = add_func.validate_input('Введите индекс добавляемой станции(0-отмена): ', 0, 100)
if(st_index==-1) or st_index == '0':
return "1"
insert = [st_index, st_name, tarrif_zone_id]
StationTable().insert_one(insert)
def show_routes_with_station(self):
"""Вывод всех маршрутов по введенной станции
"""
self.routes_arr = []
if self.station_id == -1:
while True:
t = add_func.validate_input('Вывести маршрут по:\
\n1 - начальной станции;\
\n2 - конечной станции;\n(0 - отмена);\n', 0, 2)
if t == 1:
x = add_func.validate_input('Выберите номер интересуемой строки с начальной станцией (0 - отмена): ', 0, self.max_station_index)
if(x==-1):
return
self.station_id = self.station_arr[x-1][0]
self.station_obj = self.station_arr[x-1]
space_lenth = len(self.station_obj[1])
lst = RoutesTable().all_by_station_id(self.station_id, 1)
self.max_routes_index = len(lst)
if lst == []:
print('Нет маршрута с такой начальной станцией')
else:
print("Выбрана station: " + self.station_obj[1])
print("Маршруты:")
print(f"""№\tНомер маршрута\tВыбранная станция {" "*8} Конечная станция\n{"-"*100}-""")
for i in lst:
self.routes_arr.append([str(i[0]), str(i[2]), i[1]])
for i in range(len(self.routes_arr)):
output = f"""{str(i+1)}\t{str(self.routes_arr[i][2])}\
{" " * 15}{self.station_obj[1]} {"-" * (24 - space_lenth)}> {str(StationTable().name_by_id(self.routes_arr[i][1])[0])}"""
print(output)
menu = """Дальнейшие операции:
0 - возврат в главное меню;
1 - возврат в просмотр станций;
7 - удаление маршрута;
9 - выход."""
print(menu)
return self.read_next_step()
elif t == 2:
x = add_func.validate_input('Выберите номер интересуемой строки с конечной станцией (0 - отмена): ', 0, self.max_station_index)
if(x==0):
return
self.station_id = self.station_arr[x-1][0]
self.station_obj = self.station_arr[x-1]
space_lenth = len(self.station_obj[1])
lst = RoutesTable().all_by_station_id(self.station_id, 2)
self.max_routes_index = len(lst)
if lst == []:
print('Нет маршрута с такой конечной станцией')
else:
print("Выбрана station: " + self.station_obj[1])
print("Маршруты:")
print(f"""№\tНомер маршрута\tВыбранная станция {" "*8} Начальная станция\n{"-"*100}""")
for i in lst:
self.routes_arr.append([str(i[0]), str(i[2]), i[1]])
for i in range(len(self.routes_arr)):
output = f"""{str(i+1)}\t{str(self.routes_arr[i][2])}\
{" " * 15}{self.station_obj[1]} <{"-" * (24 - space_lenth)} {str(StationTable().name_by_id(self.routes_arr[i][2])[0])}"""
print(output)
menu = f"""{"-"*100}\nДальнейшие операции:
0 - возврат в главное меню;
1 - возврат в просмотр станций;
7 - удаление маршрута;
9 - выход."""
print(menu)
return self.read_next_step()
else:
return "5"
def main_cycle(self):
"""Основной цикл программы, регулирующий порядок действий
"""
current_menu = "0"
next_step = None
while(current_menu != "9"):
if current_menu == "0":
self.show_main_menu()
next_step = self.read_next_step()
current_menu = self.after_main_menu(next_step)
elif current_menu == "1":
self.show_station()
next_step = self.read_next_step()
current_menu = self.after_show_station(next_step)
elif current_menu == "2":
self.show_main_menu()
elif current_menu == "3":
self.add_station()
current_menu = "1"
print("До свидания!")
return
def test(self):
DbTable.dbconn.test()
m = Main()
# m.test()
m.main_cycle()
final/project_config.py
+21
View File
@@ -0,0 +1,21 @@
# Загрузка настроек проекта (в данном случае только настроек соединения с БД)
# из файла config.yaml.
import yaml
class ProjectConfig:
"""Класс считывает базовые настройки из файла config.yaml"""
def __init__(self):
with open('config.yaml') as f:
config = yaml.safe_load(f)
self.dbname = config['dbname']
self.user = config['user']
self.password = config['password']
self.host = config['host']
self.dbtableprefix = config['dbtableprefix']
# Этот метод запускается только, если запускать
# данный файл, а не подключать его.
if __name__ == "__main__":
x = ProjectConfig()
print(x.dbfilepath)
final/tables/__pycache__/cath_table.cpython-311.pyc
BIN
View File
Binary file not shown.
final/tables/__pycache__/dish_table.cpython-311.pyc
BIN
View File
Binary file not shown.
final/tables/__pycache__/routes_table.cpython-311.pyc
BIN
View File
Binary file not shown.
final/tables/__pycache__/station_table.cpython-311.pyc
BIN
View File
Binary file not shown.
final/tables/routes_table.py
+91
View File
@@ -0,0 +1,91 @@
from tables.station_table import *
from dbtable import *
import add_func
class RoutesTable(DbTable):
def table_name(self):
return self.dbconn.prefix + "routes"
def columns(self):
return {"id": ["serial", "PRIMARY KEY"],
"first_st_id": ["integer", "NOT NULL", "REFERENCES stations(id) ON DELETE CASCADE"],
"last_st_id": ["integer", "NOT NULL", "REFERENCES stations(id) ON DELETE CASCADE"]}
def primary_key(self):
return ['id']
'''def table_constraints(self):
return ["PRIMARY KEY(route_id)"]'''
def all_by_station_id(self, pid, t):
if(t == 1):
sql = f"""SELECT * FROM {self.table_name()} WHERE first_st_id = {str(pid)} ORDER BY {", ".join(self.primary_key())};"""
else:
sql = f"""SELECT * FROM {self.table_name()} WHERE last_st_id = {str(pid)} ORDER BY {", ".join(self.primary_key())};"""
cur = self.dbconn.conn.cursor()
cur.execute(sql)
return cur.fetchall()
def delete(self, id):
par_sql = f"DELETE FROM routes WHERE id = {str(id)}"
cur = self.dbconn.conn.cursor()
cur.execute(par_sql)
self.dbconn.conn.commit()
def insert_route_one(self, max_index):
while True:
first_st_id = add_func.validate_input('Введите номер добавляемой начальной станции (0 - для отмены): ', 0, max_index)
if(first_st_id == -1): return "1"
last_st_id = add_func.validate_input('Введите номер добавляемой конечной станции (0 - для отмены): ', 0, max_index)
if(last_st_id == -1): return "1"
t1 = StationTable().index_by_id(first_st_id)
t2 = StationTable().index_by_id(last_st_id)
if t1>t2:
print('Нельзя добавить такой маршрут')
else:
insert = [first_st_id, last_st_id]
self.insert_one(insert)
return
def update(self, max_index):
while True:
t = add_func.validate_input('Изменить маршрут по:\
\n1 - начальной станции;\
\n2- конечной станции;\n(0 - отмена);\n', 0, 2)
if(t==1):
n1 = add_func.validate_input('Введите номер начальной станции, который вы хотите заменить (0 - для отмены): ', 0, max_index)
if(n1 == -1): return "1"
n2 = add_func.validate_input('Введите номер, на который вы хотите поменять (0 - для отмены): ', 0, max_index)
if(n2 == -1): return "1"
if n1 != n2:
sql = f"""UPDATE {self.table_name()} SET first_st_id = '{n2}' WHERE first_st_id = '{n1}';"""
break
else:
print('Невозможно поменять на такой номер')
elif(t==2):
n1 = add_func.validate_input('Введите номер конечной станции, который вы хотите заменить (0 - для отмены): ', 0, max_index)
if(n1 == -1): return "1"
n2 = add_func.validate_input('Введите номер, на который вы хотите поменять (0 - для отмены): ', 0, max_index)
if(n2 == -1): return "1"
if n1 != n2:
sql = f"""UPDATE {self.table_name()} SET last_st_id = '{n2}' WHERE last_st_id = '{n1}';"""
# sql2 = f"""UPDATE stations SET id = '{n2}' WHERE id = '{n1}';"""
break
else:
print('Невозможно поменять на такой номер')
else:
return "1"
cur = self.dbconn.conn.cursor()
cur.execute(sql)
self.dbconn.conn.commit()
return
def example_insert(self):
self.insert_one([1,2])
self.insert_one([1,6])
self.insert_one([1,7])
self.insert_one([1,8])
self.insert_one([1,9])
self.insert_one([1,6])
return
final/tables/station_table.py
+120
View File
@@ -0,0 +1,120 @@
from dbtable import *
import add_func
class StationTable(DbTable):
'''
Это класс для таблиц
'''
def table_name(self):
'''
Это метод для названия таблицы
'''
return self.dbconn.prefix + "stations"
def columns(self):
return {"id": ["serial", "PRIMARY KEY"],
"st_name": ["varchar(20)", "NOT NULL"],
"tarrif_zone_id": ["integer", "NOT NULL"],
"st_index": ["integer", "NOT NULL"]}
def table_constraint(self):
return ['CONTRAINT "Name" UNIQUE (st_name)']
def find_by_position(self, num):
sql = "SELECT * FROM " + self.table_name()
sql += " ORDER BY "
sql += ", ".join(self.primary_key())
sql += " LIMIT 1 OFFSET %(offset)s"
cur = self.dbconn.conn.cursor()
cur.execute(sql, {"offset": num - 1})
return cur.fetchone()
def delete(self, val):
par_sql = f"DELETE FROM stations WHERE st_name = '{val[1]}';"
cur = self.dbconn.conn.cursor()
cur.execute(par_sql)
self.dbconn.conn.commit()
def insert(self):
while True:
st_name = input('Введите название добавляемой станции (enter - отмена): ').strip()
if len(st_name) > 20:
print('Недопустмая длина названия станции')
elif st_name == '':
return "1"
else:
break
tarrif_zone_id = add_func.validate_input('Введите номер тарифной зоны: ', 0, 7)
if(tarrif_zone_id==-1):
return "1"
st_index = add_func.validate_input('Введите индекс добавляемой станции: ', 0, 100)
if(st_index==-1):
return "1"
insert = [st_index, st_name, tarrif_zone_id]
self.insert_one(insert)
def update(self, old):
"""Функция для обновления станции
"""
print(f'''Выбрана станция для изменения: {old[1]}''')
cur = self.dbconn.conn.cursor()
while True:
data = input("Введите название (enter - отмена): ").strip()
if(len(data.strip()) > 20):
data = input("Название слишком длинное! Введите название заново (enter - отмена):").strip()
elif(self.check_by_name(data.strip())):
data = input("Такое название уже существует. Введите новое (enter - отмена):").strip()
elif(data==''): return
else:
param_sql = f"UPDATE {self.table_name()} SET st_name = '{data}' WHERE st_name = '{old[1]}';"
cur.execute(param_sql)
self.dbconn.conn.commit()
return
def check_by_name(self, value):
sql = f"SELECT * FROM {self.table_name()} WHERE st_name='{value}'"
cur = self.dbconn.conn.cursor()
cur.execute(sql)
result = cur.fetchone()
cur.close()
if result:
return True
else:
return False
def find_by_name(self, name):
cur = self.dbconn.conn.cursor()
param_query = f"SELECT id FROM Stations WHERE st_name = '{name}';"
# sql_sel = "SELECT id FROM " + self.table_name()
# sql_sel += " WHERE cath_name = " + "'" + name + "'" + ";"
cur.execute(param_query)
ret = cur.fetchone()
return ret
def name_by_id(self, id):
cur = self.dbconn.conn.cursor()
query = "SELECT st_name FROM Stations WHERE id = %s;"
cur.execute(query, str(id))
return cur.fetchone()
def index_by_id(self,id):
cur = self.dbconn.conn.cursor()
query = f"SELECT st_index FROM Stations WHERE id = {id};"
cur.execute(query)
return cur.fetchone()
def example_insert(self):
self.insert_one([1, "Нахабино", 1])
self.insert_one([2, "Стрешнево", 2])
self.insert_one([2, "Царицыно", 2])
self.insert_one([3, "Яуза", 3])
self.insert_one([2, "Щукинская", 3])
self.insert_one([4, "Москва-Курская", 4])
self.insert_one([4, "Дмитровская", 4])
self.insert_one([4, "Остафьево", 4])
self.insert_one([4, "Подольск", 4])
return
final/test.py
+1
View File
@@ -0,0 +1 @@
print(len("-------------------------------------------------------------------------------------------"))