75 lines
3.6 KiB
Python
75 lines
3.6 KiB
Python
from dotenv import load_dotenv
|
||
load_dotenv()
|
||
import os
|
||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
||
from sqlalchemy.orm import declarative_base
|
||
from sqlalchemy.ext.asyncio import async_sessionmaker
|
||
|
||
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite+aiosqlite:///bot.db")
|
||
|
||
if DATABASE_URL.startswith("sqlite+aiosqlite:///"):
|
||
db_path = DATABASE_URL.replace("sqlite+aiosqlite:///", "")
|
||
abs_db_path = os.path.abspath(db_path)
|
||
db_dir = os.path.dirname(abs_db_path)
|
||
# Создаём директорию только если она не равна текущей ('.') и не пустая
|
||
if db_dir and db_dir != os.path.abspath("") and db_dir != '.' and not os.path.exists(db_dir):
|
||
os.makedirs(db_dir, exist_ok=True)
|
||
# Если по этому пути уже есть папка, удаляем её
|
||
if os.path.exists(abs_db_path) and os.path.isdir(abs_db_path):
|
||
import shutil
|
||
shutil.rmtree(abs_db_path)
|
||
# Если файла нет, создаём пустой файл
|
||
if not os.path.exists(abs_db_path):
|
||
open(abs_db_path, 'a').close()
|
||
|
||
engine = create_async_engine(DATABASE_URL, future=True, echo=False)
|
||
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
||
Base = declarative_base()
|
||
|
||
async def init_db():
|
||
print(f'База данных: {DATABASE_URL}')
|
||
need_create = False
|
||
if DATABASE_URL.startswith("sqlite+aiosqlite:///"):
|
||
db_path = DATABASE_URL.replace("sqlite+aiosqlite:///", "")
|
||
abs_db_path = os.path.abspath(db_path)
|
||
print(f"Абсолютный путь к базе данных: {abs_db_path}")
|
||
# Если файл отсутствует или пустой, или это папка — создаём таблицы
|
||
if not os.path.exists(abs_db_path):
|
||
print("Файл базы данных отсутствует, будет создан.")
|
||
need_create = True
|
||
elif os.path.isdir(abs_db_path):
|
||
print("Вместо файла обнаружена папка, будет удалена и создан файл.")
|
||
import shutil
|
||
shutil.rmtree(abs_db_path)
|
||
open(abs_db_path, 'a').close()
|
||
need_create = True
|
||
elif os.path.getsize(abs_db_path) == 0:
|
||
print("Файл базы данных пустой, будут созданы таблицы.")
|
||
need_create = True
|
||
else:
|
||
# Проверяем наличие таблиц
|
||
import sqlite3
|
||
try:
|
||
conn = sqlite3.connect(abs_db_path)
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
|
||
tables = cursor.fetchall()
|
||
conn.close()
|
||
if not tables:
|
||
print("В базе нет таблиц, будут созданы.")
|
||
need_create = True
|
||
except Exception as e:
|
||
print(f"Ошибка проверки таблиц: {e}")
|
||
need_create = True
|
||
else:
|
||
print(f"База данных: {DATABASE_URL}")
|
||
# Для других СУБД всегда пытаемся создать таблицы
|
||
need_create = True
|
||
|
||
if need_create:
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(Base.metadata.create_all)
|
||
tables = Base.metadata.tables.keys()
|
||
print(f"Созданы таблицы: {', '.join(tables)}")
|
||
else:
|
||
print("База данных уже существует и содержит таблицы, создание пропущено.") |