90 lines
3.2 KiB
Python
90 lines
3.2 KiB
Python
import sys
|
|
from sqlalchemy import create_engine, text, inspect
|
|
from sqlalchemy.orm import sessionmaker
|
|
from backend.database import SessionLocal as NewSession, engine as new_engine
|
|
from backend import models
|
|
from pathlib import Path
|
|
|
|
|
|
OLD_DB_URL = "sqlite:///./instance/project.-10-11-25.db"
|
|
old_engine = create_engine(OLD_DB_URL, connect_args={"check_same_thread": False})
|
|
OldSession = sessionmaker(bind=old_engine)
|
|
old_db = OldSession()
|
|
|
|
new_db = NewSession()
|
|
|
|
def log(msg: str):
|
|
print(f"[INFO] {msg}", file=sys.stderr)
|
|
|
|
|
|
def ensure_schema():
|
|
"""Ensure new DB has required tables/columns (owners table and oboruds.owner_id)."""
|
|
# Create any missing tables defined in models (e.g., owners)
|
|
models.Base.metadata.create_all(bind=new_engine)
|
|
|
|
# If oboruds.owner_id is missing (older DB), add the column (SQLite allows simple ALTER)
|
|
try:
|
|
inspector = inspect(new_engine)
|
|
cols = [c["name"] for c in inspector.get_columns("oboruds")]
|
|
if "owner_id" not in cols:
|
|
with new_engine.begin() as conn:
|
|
conn.execute(text("ALTER TABLE oboruds ADD COLUMN owner_id INTEGER"))
|
|
log("Добавлен столбец oboruds.owner_id")
|
|
except Exception as e:
|
|
log(f"Предупреждение: проверка/добавление owner_id не выполнена: {e}")
|
|
|
|
def migrate():
|
|
ensure_schema()
|
|
log("Запуск переноса данных из old_app.db → app.db")
|
|
|
|
# Тип оборудования по умолчанию
|
|
log("Добавление типа оборудования по умолчанию: 'Неизвестно'")
|
|
default_type = models.EquipmentType(name="Неизвестно")
|
|
new_db.add(default_type)
|
|
new_db.commit()
|
|
|
|
# Перенос аудиторий
|
|
log("Перенос аудиторий...")
|
|
auditory_map = {}
|
|
aud_rows = old_db.execute(text("SELECT id, audnazvanie FROM auditory")).fetchall()
|
|
for aud in aud_rows:
|
|
new_aud = models.Auditory(audnazvanie=aud.audnazvanie)
|
|
new_db.add(new_aud)
|
|
new_db.flush()
|
|
auditory_map[aud.id] = new_aud.id
|
|
log(f" → перенесено: {len(aud_rows)}")
|
|
|
|
# Перенос оборудования
|
|
log("Перенос оборудования...")
|
|
ob_rows = old_db.execute(text("SELECT * FROM oboruds")).fetchall()
|
|
for ob in ob_rows:
|
|
new_ob = models.Oboruds(
|
|
invNumber=ob.invNumber,
|
|
nazvanie=ob.nazvanie,
|
|
raspologenie=ob.raspologenie,
|
|
numberppasu=ob.numberppasu,
|
|
kolichestvo=ob.kolichestvo,
|
|
aud_id=auditory_map.get(ob.aud_id),
|
|
type_id=default_type.id
|
|
)
|
|
new_db.add(new_ob)
|
|
log(f" → перенесено: {len(ob_rows)}")
|
|
|
|
# Перенос заметок
|
|
log("Перенос заметок...")
|
|
z_rows = old_db.execute(text("SELECT * FROM zametki")).fetchall()
|
|
for z in z_rows:
|
|
new_z = models.Zametki(
|
|
txtzam=z.txtzam,
|
|
created_date=z.created_date,
|
|
rmdt=z.rmdt
|
|
)
|
|
new_db.add(new_z)
|
|
log(f" → перенесено: {len(z_rows)}")
|
|
|
|
new_db.commit()
|
|
log("✅ Перенос завершён успешно.")
|
|
|
|
if __name__ == "__main__":
|
|
migrate()
|