Compare commits

...

3 Commits

Author SHA1 Message Date
d54419fd35 Add MAX_AGE config parameter 2024-01-13 12:44:28 +01:00
acaa8367d6 Fix more pylint warnings 2023-12-15 13:38:32 +01:00
11ddbf0169 Take advice from pylint 2023-12-15 11:36:34 +01:00
18 changed files with 314 additions and 281 deletions

5
.pylintrc Normal file
View File

@ -0,0 +1,5 @@
[BASIC]
good-names=db, v, v1, v2, f, id, i, j, k
[MESSAGES CONTROL]
disable=pointless-string-statement

View File

@ -16,3 +16,6 @@ DISABLE_REGISTER=False
# Front URL of the application (for QRCodes generation)
BASE_URL="http://localhost:5000"
# Session expiration, in days
MAX_AGE=31

View File

@ -3,6 +3,7 @@
Main file
"""
import os
import datetime
import subprocess
from flask import Flask, g, redirect, render_template, request, send_file, flash, session, abort
@ -32,9 +33,10 @@ app.register_blueprint(partition.bp)
try:
result = subprocess.run(["git", "describe", "--tags"], stdout=subprocess.PIPE)
result = subprocess.run(["git", "describe", "--tags"], stdout=subprocess.PIPE, check=True)
__version__ = result.stdout.decode('utf8')
except FileNotFoundError: # In case git not found, which would be strange
except (FileNotFoundError, subprocess.CalledProcessError):
# In case git not found or any platform specific weird error
__version__ = "unknown"
@ -56,26 +58,10 @@ def add_user():
username = request.form["username"]
password = request.form["password"]
album_uuid = request.form["album_uuid"]
db = get_db()
error = None
if not username:
error = "Un nom d'utilisateur est requis."
elif not password:
error = "Un mot de passe est requis."
error = auth.create_user(username, password)
if error is None:
try:
db.execute(
"INSERT INTO user (username, password) VALUES (?, ?)",
(username, generate_password_hash(password)),
)
db.commit()
except db.IntegrityError:
# The username was already taken, which caused the
# commit to fail. Show a validation error.
error = f"Le nom d'utilisateur {username} est déjà pris."
else:
# Success, go to the login page.
user = User(name=username)
try:
@ -94,8 +80,11 @@ def add_user():
@app.route("/static/search-thumbnails/<uuid>.jpg")
@login_required
def search_thumbnail(uuid):
"""
Renvoie l'apercu d'un résultat de recherche
"""
db = get_db()
partition = db.execute(
part = db.execute(
"""
SELECT uuid, url FROM search_results
WHERE uuid = ?
@ -103,7 +92,7 @@ def search_thumbnail(uuid):
(uuid,)
).fetchone()
if partition is None:
if part is None:
abort(404)
if not os.path.exists(os.path.join(app.static_folder, "search-thumbnails", f"{uuid}.jpg")):
os.system(
@ -117,11 +106,18 @@ def search_thumbnail(uuid):
return send_file(os.path.join(app.static_folder, "search-thumbnails", f"{uuid}.jpg"))
@app.before_request
def before_request():
"""Set cookie max age to 31 days"""
session.permanent = True
app.permanent_session_lifetime = datetime.timedelta(days=int(app.config["MAX_AGE"]))
@app.context_processor
def inject_default_variables():
"""Inject the version number in the template variables"""
if __version__ == "unknown":
return dict(version="")
return dict(version=__version__)
return {"version": ''}
return {"version": __version__}
@app.after_request

View File

@ -2,8 +2,7 @@
"""
Admin Panel
"""
import os
from flask import Blueprint, abort, send_file, render_template, session
from flask import Blueprint, render_template, session
from .db import get_db
from .auth import admin_required
@ -15,6 +14,9 @@ bp = Blueprint("admin", __name__, url_prefix="/admin")
@bp.route("/")
@admin_required
def index():
"""
Admin panel home page
"""
current_user = User(user_id=session.get("user_id"))
current_user.get_albums() # We need to do that before we close the db
db = get_db()
@ -23,10 +25,10 @@ def index():
SELECT id FROM user
"""
)
users = [User(user_id=u["id"]) for u in users_id]
for u in users:
u.albums = u.get_albums()
u.partitions = u.get_partitions()
users = [User(user_id=user["id"]) for user in users_id]
for user in users:
user.get_albums()
user.get_partitions()
return render_template(
"admin/index.html",

View File

@ -6,8 +6,8 @@ import os
import shutil
from uuid import uuid4
from flask import (Blueprint, abort, flash, redirect, render_template, request,
send_file, session, current_app)
from flask import (Blueprint, abort, flash, redirect, render_template,
request, session, current_app)
from .auth import login_required
from .db import get_db
@ -20,13 +20,10 @@ bp = Blueprint("albums", __name__, url_prefix="/albums")
@bp.route("/")
@login_required
def index():
"""
Albums home page
"""
user = User(user_id=session.get("user_id"))
albums = user.get_albums()
if user.access_level == 1:
max_queries = 10
else:
max_queries = current_app.config["MAX_ONLINE_QUERIES"]
return render_template("albums/index.html", user=user)
@ -34,6 +31,9 @@ def index():
@bp.route("/search", methods=["POST"])
@login_required
def search_page():
"""
Résultats de recherche
"""
if "query" not in request.form or request.form["query"] == "":
flash("Aucun terme de recherche spécifié.")
return redirect("/albums")
@ -65,7 +65,7 @@ def search_page():
)
@bp.route("/<uuid>")
def album(uuid):
def get_album(uuid):
"""
Album page
"""
@ -98,14 +98,18 @@ def album(uuid):
@bp.route("/<uuid>/qr")
def qr_code(uuid):
"""
Renvoie le QR Code d'un album
"""
return get_qrcode(f"/albums/{uuid}")
@bp.route("/create-album", methods=["POST"])
@login_required
def create_album():
current_user = User(user_id=session.get("user_id"))
"""
Création d'un album
"""
name = request.form["name"]
db = get_db()
error = None
@ -145,7 +149,6 @@ def create_album():
"status": "ok",
"uuid": uuid
}
else:
return redirect(f"/albums/{uuid}")
flash(error)
@ -155,6 +158,9 @@ def create_album():
@bp.route("/<uuid>/join")
@login_required
def join_album(uuid):
"""
Rejoindre un album
"""
user = User(user_id=session.get("user_id"))
try:
user.join_album(uuid)
@ -169,6 +175,9 @@ def join_album(uuid):
@bp.route("/<uuid>/quit")
@login_required
def quit_album(uuid):
"""
Quitter un album
"""
user = User(user_id=session.get("user_id"))
album = Album(uuid=uuid)
users = album.get_users()
@ -182,13 +191,15 @@ def quit_album(uuid):
user.quit_album(uuid)
flash("Album quitté.")
return redirect(f"/albums")
return redirect("/albums")
@bp.route("/<uuid>/delete", methods=["GET", "POST"])
@login_required
def delete_album(uuid):
db = get_db()
"""
Supprimer un album
"""
album = Album(uuid=uuid)
user = User(user_id=session.get("user_id"))
@ -218,6 +229,9 @@ def delete_album(uuid):
@bp.route("/<album_uuid>/add-partition", methods=["POST"])
@login_required
def add_partition(album_uuid):
"""
Ajouter une partition à un album (par upload)
"""
db = get_db()
user = User(user_id=session.get("user_id"))
album = Album(uuid=album_uuid)
@ -281,7 +295,10 @@ def add_partition(album_uuid):
file = request.files["file"]
file.save(f"partitioncloud/partitions/{partition_uuid}.pdf")
else:
shutil.copyfile(f"partitioncloud/search-partitions/{search_uuid}.pdf", f"partitioncloud/partitions/{partition_uuid}.pdf")
shutil.copyfile(
f"partitioncloud/search-partitions/{search_uuid}.pdf",
f"partitioncloud/partitions/{partition_uuid}.pdf"
)
os.system(
f'/usr/bin/convert -thumbnail\
@ -290,14 +307,6 @@ def add_partition(album_uuid):
partitioncloud/partitions/{partition_uuid}.pdf[0] \
partitioncloud/static/thumbnails/{partition_uuid}.jpg'
)
album_id = db.execute(
"""
SELECT id FROM album
WHERE uuid = ?
""",
(album.uuid,)
).fetchone()["id"]
db.commit()
album.add_partition(partition_uuid)
@ -311,7 +320,6 @@ def add_partition(album_uuid):
"status": "ok",
"uuid": partition_uuid
}
else:
flash(f"Partition {request.form['name']} ajoutée")
return redirect(f"/albums/{album.uuid}")
@ -319,6 +327,9 @@ def add_partition(album_uuid):
@bp.route("/add-partition", methods=["POST"])
@login_required
def add_partition_from_search():
"""
Ajout d'une partition (depuis la recherche)
"""
user = User(user_id=session.get("user_id"))
error = None
@ -355,7 +366,7 @@ def add_partition_from_search():
return redirect(f"/albums/{album.uuid}")
elif request.form["partition-type"] == "online_search":
if request.form["partition-type"] == "online_search":
return render_template(
"albums/add-partition.html",
album=album,
@ -363,6 +374,5 @@ def add_partition_from_search():
user=user
)
else:
flash("Type de partition inconnu.")
return redirect("/albums")

View File

@ -3,19 +3,11 @@
Authentification module
"""
import functools
from typing import Optional
from flask import (Blueprint, flash, g, redirect, render_template,
request, session, url_for, current_app)
from flask import (
Blueprint,
flash,
g,
redirect,
render_template,
request,
session,
url_for,
flash,
current_app
)
from werkzeug.security import check_password_hash, generate_password_hash
from .db import get_db
@ -84,6 +76,30 @@ def load_logged_in_user():
)
def create_user(username: str, password: str) -> Optional[str]:
"""Adds a new user to the database"""
if not username:
error = "Un nom d'utilisateur est requis."
elif not password:
error = "Un mot de passe est requis."
try:
db = get_db()
db.execute(
"INSERT INTO user (username, password) VALUES (?, ?)",
(username, generate_password_hash(password)),
)
db.commit()
except db.IntegrityError:
# The username was already taken, which caused the
# commit to fail. Show a validation error.
error = f"Le nom d'utilisateur {username} est déjà pris."
if error is not None:
return error
@bp.route("/register", methods=("GET", "POST"))
@anon_required
def register():
@ -98,31 +114,13 @@ def register():
if request.method == "POST":
username = request.form["username"]
password = request.form["password"]
db = get_db()
error = None
if not username:
error = "Un nom d'utilisateur est requis."
elif not password:
error = "Un mot de passe est requis."
if error is None:
try:
db.execute(
"INSERT INTO user (username, password) VALUES (?, ?)",
(username, generate_password_hash(password)),
)
db.commit()
flash(f"Utilisateur {username} créé avec succès. Vous pouvez vous connecter.")
except db.IntegrityError:
# The username was already taken, which caused the
# commit to fail. Show a validation error.
error = f"Le nom d'utilisateur {username} est déjà pris. Vous souhaitez peut-être vous connecter"
else:
# Success, go to the login page.
return redirect(url_for("auth.login"))
error = create_user(username, password)
if error is not None:
flash(error)
else:
flash("Utilisateur créé avec succès. Vous pouvez vous connecter.")
return render_template("auth/register.html")

View File

@ -1,3 +1,6 @@
"""
Classe Album
"""
import os
from ..db import get_db

View File

@ -1,7 +1,5 @@
import os
from flask import current_app
from ..db import get_db

View File

@ -1,5 +1,4 @@
import os
from flask import current_app
from ..db import get_db
from .user import User
@ -61,7 +60,7 @@ class Partition():
def update(self, name=None, author="", body=""):
if name is None:
return Exception("name cannot be None")
raise ValueError("name cannot be None")
db = get_db()
db.execute(

View File

@ -226,5 +226,4 @@ class User():
if len(colors) == 0:
integer = hash(self.username) % 16777215
return "#" + str(hex(integer))[2:]
else:
return f"var({colors[hash(self.username) %len(colors)]})"

View File

@ -16,13 +16,3 @@ def get_db():
g.db.row_factory = sqlite3.Row
return g.db
def close_db(e=None):
"""If this request connected to the database, close the
connection.
"""
db = g.pop("db", None)
if db is not None:
db.close()

View File

@ -2,15 +2,12 @@
"""
Groupe module
"""
import os
from flask import (Blueprint, abort, flash, redirect, render_template, request,
send_file, session, current_app)
from flask import (Blueprint, abort, flash, redirect, render_template,
request, session)
from .auth import login_required
from .db import get_db
from .utils import User, Album, get_all_partitions, Groupe, new_uuid, get_qrcode, format_uuid
from . import search
from .utils import User, Album, Groupe, new_uuid, get_qrcode, format_uuid
bp = Blueprint("groupe", __name__, url_prefix="/groupe")
@ -21,7 +18,7 @@ def index():
@bp.route("/<uuid>")
def groupe(uuid):
def get_groupe(uuid):
"""
Groupe page
"""
@ -61,8 +58,6 @@ def album_qr_code(uuid):
@bp.route("/create-groupe", methods=["POST"])
@login_required
def create_groupe():
current_user = User(user_id=session.get("user_id"))
name = request.form["name"]
db = get_db()
error = None
@ -102,7 +97,6 @@ def create_groupe():
"status": "ok",
"uuid": uuid
}
else:
return redirect(f"/groupe/{uuid}")
flash(error)
@ -139,13 +133,12 @@ def quit_groupe(uuid):
user.quit_groupe(groupe.uuid)
flash("Groupe quitté.")
return redirect(f"/albums")
return redirect("/albums")
@bp.route("/<uuid>/delete", methods=["POST"])
@login_required
def delete_groupe(uuid):
db = get_db()
groupe = Groupe(uuid=uuid)
user = User(user_id=session.get("user_id"))
@ -220,7 +213,6 @@ def create_album(groupe_uuid):
"status": "ok",
"uuid": uuid
}
else:
return redirect(f"/groupe/{groupe.uuid}/{uuid}")
flash(error)
@ -229,7 +221,7 @@ def create_album(groupe_uuid):
@bp.route("/<groupe_uuid>/<album_uuid>")
def album(groupe_uuid, album_uuid):
def get_album(groupe_uuid, album_uuid):
"""
Album page
"""
@ -254,7 +246,7 @@ def album(groupe_uuid, album_uuid):
user = User(user_id=session.get("user_id"))
# List of users without duplicate
users_id = list(set([i["id"] for i in album.get_users()+groupe.get_users()]))
users_id = list({i["id"] for i in album.get_users()+groupe.get_users()})
album.users = [User(user_id=id) for id in users_id]
partitions = album.get_partitions()

View File

@ -14,8 +14,7 @@ from .utils import get_all_partitions, User, Partition, Attachment
bp = Blueprint("partition", __name__, url_prefix="/partition")
@bp.route("/<uuid>")
def partition(uuid):
db = get_db()
def get_partition(uuid):
try:
partition = Partition(uuid=uuid)
except LookupError:
@ -29,7 +28,6 @@ def partition(uuid):
@bp.route("/<uuid>/attachments")
def attachments(uuid):
db = get_db()
try:
partition = Partition(uuid=uuid)
except LookupError:
@ -105,13 +103,11 @@ def add_attachment(uuid):
"status": "ok",
"uuid": attachment_uuid
}
else:
return redirect(f"/partition/{partition.uuid}/attachments")
@bp.route("/attachment/<uuid>.<filetype>")
def attachment(uuid, filetype):
db = get_db()
def get_attachment(uuid, filetype):
try:
attachment = Attachment(uuid=uuid)
except LookupError:
@ -129,7 +125,6 @@ def attachment(uuid, filetype):
@bp.route("/<uuid>/edit", methods=["GET", "POST"])
@login_required
def edit(uuid):
db = get_db()
try:
partition = Partition(uuid=uuid)
except LookupError:
@ -169,7 +164,6 @@ def edit(uuid):
@bp.route("/<uuid>/details", methods=["GET", "POST"])
@admin_required
def details(uuid):
db = get_db()
try:
partition = Partition(uuid=uuid)
except LookupError:

View File

@ -56,8 +56,8 @@ def download_search_result(element):
try:
urllib.request.urlretrieve(url, f"partitioncloud/search-partitions/{uuid}.pdf")
except (urllib.error.HTTPError, urllib.error.URLError) as e:
with open(f"partitioncloud/search-partitions/{uuid}.pdf",'a') as f:
except (urllib.error.HTTPError, urllib.error.URLError):
with open(f"partitioncloud/search-partitions/{uuid}.pdf", 'a', encoding="utf8") as _:
pass # Create empty file
@ -112,7 +112,6 @@ def online_search(query, num_queries):
thread.join()
for element in partitions:
pass
uuid = element["uuid"]
url = element["url"]
if os.stat(f"partitioncloud/search-partitions/{uuid}.pdf").st_size == 0:

View File

@ -1,12 +1,11 @@
#!/usr/bin/python3
import os
import io
import random
import string
import qrcode
from .db import get_db
from flask import current_app, send_file
from .db import get_db
def new_uuid():

View File

@ -2,6 +2,7 @@ import random
import string
import sqlite3
def run_sqlite_command(*args):
"""Run a command against the database"""
con = sqlite3.connect("instance/partitioncloud.sqlite")
@ -10,17 +11,22 @@ def run_sqlite_command(*args):
con.commit()
con.close()
def get_sqlite_data(*args):
"""Get data from the db"""
con = sqlite3.connect("instance/partitioncloud.sqlite")
cur = con.cursor()
data = cur.execute(*args)
new_data = [i for i in data]
new_data = list(data)
con.close()
return new_data
def new_uuid():
return ''.join([random.choice(string.ascii_uppercase + string.digits) for _ in range(6)])
return "".join(
[random.choice(string.ascii_uppercase + string.digits) for _ in range(6)]
)
def format_uuid(uuid):
"""Format old uuid4 format"""

View File

@ -6,10 +6,11 @@ from colorama import Fore, Style
"""
v1.3.*
"""
def add_source():
utils.run_sqlite_command(
"ALTER TABLE partition ADD source TEXT DEFAULT 'unknown'"
)
utils.run_sqlite_command("ALTER TABLE partition ADD source TEXT DEFAULT 'unknown'")
def add_groupes():
utils.run_sqlite_command(
@ -35,6 +36,7 @@ def add_groupes():
);"""
)
def add_attachments():
os.makedirs("partitioncloud/attachments", exist_ok=True)
utils.run_sqlite_command(
@ -47,6 +49,7 @@ def add_attachments():
);"""
)
def install_colorama():
os.system("pip install colorama -qq")
@ -54,17 +57,15 @@ def install_colorama():
"""
v1.4.*
"""
def mass_rename():
"""Rename all albums & groupes to use a shorter uuid"""
albums = utils.get_sqlite_data("SELECT * FROM album")
groupes = utils.get_sqlite_data("SELECT * FROM groupe")
utils.run_sqlite_command(
"ALTER TABLE groupe RENAME TO _groupe_old"
)
utils.run_sqlite_command(
"ALTER TABLE album RENAME TO _album_old"
)
utils.run_sqlite_command("ALTER TABLE groupe RENAME TO _groupe_old")
utils.run_sqlite_command("ALTER TABLE album RENAME TO _album_old")
utils.run_sqlite_command( # Add UNIQUE constraint & change uuid length
"""CREATE TABLE groupe (
@ -89,17 +90,20 @@ def mass_rename():
INSERT INTO album (id, name, uuid)
VALUES (?, ?, ?)
""",
(album[0], album[1], utils.format_uuid(album[2]))
(album[0], album[1], utils.format_uuid(album[2])),
)
except sqlite3.IntegrityError:
uuid = new_uuid()
print(f"{Fore.RED}Collision on {album[1]}{Style.RESET_ALL} ({album[2][:10]} renaming to {uuid})")
uuid = utils.new_uuid()
print(
f"{Fore.RED}Collision on {album[1]}{Style.RESET_ALL} \
({album[2][:10]} renaming to {uuid})"
)
utils.run_sqlite_command(
"""
INSERT INTO album (id, name, uuid)
VALUES (?, ?, ?)
""",
(album[0], album[1], uuid)
(album[0], album[1], uuid),
)
for groupe in groupes:
@ -109,28 +113,32 @@ def mass_rename():
INSERT INTO groupe (id, name, uuid)
VALUES (?, ?, ?)
""",
(groupe[0], groupe[1], utils.format_uuid(groupe[2]))
(groupe[0], groupe[1], utils.format_uuid(groupe[2])),
)
except sqlite3.IntegrityError:
uuid = new_uuid()
print(f"{Fore.RED}Collision on {groupe[1]}{Style.RESET_ALL} ({groupe[2][:10]} renaming to {uuid})")
uuid = utils.new_uuid()
print(
f"{Fore.RED}Collision on {groupe[1]}{Style.RESET_ALL} \
({groupe[2][:10]} renaming to {uuid})"
)
utils.run_sqlite_command(
"""
INSERT INTO groupe (id, name, uuid)
VALUES (?, ?, ?)
""",
(groupe[0], groupe[1], uuid)
(groupe[0], groupe[1], uuid),
)
utils.run_sqlite_command(
"DROP TABLE _groupe_old"
)
utils.run_sqlite_command(
"DROP TABLE _album_old"
)
utils.run_sqlite_command("DROP TABLE _groupe_old")
utils.run_sqlite_command("DROP TABLE _album_old")
def base_url_parameter_added():
print(f"{Style.BRIGHT}{Fore.YELLOW}The parameter BASE_URL has been added, reference your front url in it{Style.RESET_ALL}")
print(
f"{Style.BRIGHT}{Fore.YELLOW}The parameter BASE_URL has been added, \
reference your front url in it{Style.RESET_ALL}"
)
def install_qrcode():
os.system("pip install qrcode -qq")

View File

@ -1,18 +1,18 @@
#!/usr/bin/python3
import os
import sys
import shutil
import argparse
from functools import cmp_to_key
from distutils.dir_util import copy_tree
from colorama import Fore, Style
from hooks import v1
from hooks import v1 as v1_hooks
def get_version(v: str) -> (int, int, int):
"""Returns a tuple (major, minor, patch from the string v{major}.{minor}.{patch})"""
assert (v[0] == 'v') # Check if the version is correctly formatted
return tuple(map(int, v[1:].split('.')))
assert v[0] == "v" # Check if the version is correctly formatted
return tuple(map(int, v[1:].split(".")))
def is_newer(v1: str, v2: str) -> bool:
@ -21,41 +21,36 @@ def is_newer(v1: str, v2: str) -> bool:
hooks = [
("v1.3.0", [
("add SOURCE column", v1.add_source)
]),
("v1.2.0", [
("create groupe structure", v1.add_groupes)
]),
("v1.3.0", [
("create attachment table", v1.add_attachments)
]),
("v1.3.3", [
("Install colorama", v1.install_colorama)
]),
("v1.4.0", [
("Change all albums & groupes uuids", v1.mass_rename),
("Warn new parameter", v1.base_url_parameter_added)
]),
("v1.4.1", [
("Install qrcode", v1.install_qrcode)
]),
("v1.3.0", [("add SOURCE column", v1_hooks.add_source)]),
("v1.2.0", [("create groupe structure", v1_hooks.add_groupes)]),
("v1.3.0", [("create attachment table", v1_hooks.add_attachments)]),
("v1.3.3", [("Install colorama", v1_hooks.install_colorama)]),
(
"v1.4.0",
[
("Change all albums & groupes uuids", v1_hooks.mass_rename),
("Warn new parameter", v1_hooks.base_url_parameter_added),
],
),
("v1.4.1", [("Install qrcode", v1_hooks.install_qrcode)]),
]
def get_hooks(current, target):
"""Returns a list of hooks needed to migrate"""
def compare(v1: str, v2: str):
if is_newer(v2[0], v1[0]):
return -1
elif is_newer(v1[0], v2[0]):
if is_newer(v1[0], v2[0]):
return 1
else:
return 0
applied_hooks = []
for hook in hooks:
if is_newer(hook[0], current) and (target == hook[0] or is_newer(target, hook[0])):
if is_newer(hook[0], current) and (
target == hook[0] or is_newer(target, hook[0])
):
applied_hooks.append(hook)
return sorted(applied_hooks, key=cmp_to_key(compare))
@ -63,67 +58,82 @@ def get_hooks(current, target):
def backup_instance(version, verbose=True):
"""Backs up current instance in backups/{version}"""
def print_verbose(*args):
def print_verbose(*f_args):
if verbose:
print(*args)
print(*f_args)
print_verbose("\nBacking up current instance")
dest = os.path.join("backups", version)
if os.path.exists(dest):
print(f"{Fore.RED}Backup directory already exists{Style.RESET_ALL}")
exit(1)
sys.exit(1)
os.makedirs(dest)
paths = [
("instance", os.path.join(dest, "instance")),
(os.path.join("partitioncloud", "partitions"), os.path.join(dest, "partitions")),
(os.path.join("partitioncloud", "attachments"), os.path.join(dest, "attachments")),
(os.path.join("partitioncloud", "search-partitions"), os.path.join(dest, "search-partitions"))
(
os.path.join("partitioncloud", "partitions"),
os.path.join(dest, "partitions"),
),
(
os.path.join("partitioncloud", "attachments"),
os.path.join(dest, "attachments"),
),
(
os.path.join("partitioncloud", "search-partitions"),
os.path.join(dest, "search-partitions"),
),
]
for src, dst in paths:
if os.path.exists(src):
print_verbose(f"\tBacking up {src}")
copy_tree(src, dst)
shutil.copy_tree(src, dst)
def print_hooks(hooks):
for hook in hooks:
def print_hooks(hooks_list):
for hook in hooks_list:
print(f"- {Fore.BLUE}{hook[0]}{Style.RESET_ALL}:")
for subhook in hook[1]:
print("\t", subhook[0])
for sub_hook in hook[1]:
print("\t", sub_hook[0])
def apply_hooks(hooks):
for hook in hooks:
def apply_hooks(hooks_list):
for hook in hooks_list:
print(f"Migrating to {Fore.BLUE}{hook[0]}{Style.RESET_ALL}:")
for subhook in hook[1]:
print(f"\tApplying '{subhook[0]}'")
subhook[1]()
for sub_hook in hook[1]:
print(f"\tApplying '{sub_hook[0]}'")
sub_hook[1]()
def migrate(current, target, skip_backup=False, prog_name="scripts/migration.py"):
""""""
print(f"Trying to migrate from {args.current} to {args.target}")
print(f"Trying to migrate from {current} to {target}")
assert is_newer(args.target, args.current)
assert is_newer(target, current)
applied_hooks = get_hooks(args.current, args.target)
if (len(applied_hooks) == 0):
applied_hooks = get_hooks(current, target)
if len(applied_hooks) == 0:
print(f"{Fore.GREEN}No hook to apply{Style.RESET_ALL}")
exit(0)
sys.exit(0)
print("The following hooks will be applied:")
print_hooks(applied_hooks)
if input("Apply these hooks ? [y/N] ") != "y":
print(f"{Fore.RED}Aborting !{Style.RESET_ALL}")
exit(1)
sys.exit(1)
if not skip_backup:
backup_instance(current)
print(f"Instance backed up in {Style.BRIGHT}backups/{current}{Style.RESET_ALL}\n")
print(f"If something goes wrong, recover with {Style.BRIGHT}{Fore.BLUE}{prog_name} --restore {current}{Style.RESET_ALL}")
print(
f"Instance backed up in {Style.BRIGHT}backups/{current}{Style.RESET_ALL}\n"
)
print(
f"If something goes wrong, recover with {Style.BRIGHT}{Fore.BLUE}{prog_name}\
--restore {current}{Style.RESET_ALL}"
)
else:
print("Skipping automatic backup")
@ -132,17 +142,31 @@ def migrate(current, target, skip_backup=False, prog_name="scripts/migration.py"
def restore(version):
if input("Do you really want to restore from backup ? Your current data will be deleted [y/N] ") != "y":
if (
input(
"Do you really want to restore from backup ? Your current data will be deleted [y/N] "
)
!= "y"
):
print(f"{Fore.RED}Aborting !{Style.RESET_ALL}")
exit(1)
sys.exit(1)
dest = os.path.join("backups", version)
print(f"Restoring from {dest}")
paths = [
("instance", os.path.join(dest, "instance")),
(os.path.join("partitioncloud", "partitions"), os.path.join(dest, "partitions")),
(os.path.join("partitioncloud", "attachments"), os.path.join(dest, "attachments")),
(os.path.join("partitioncloud", "search-partitions"), os.path.join(dest, "search-partitions"))
(
os.path.join("partitioncloud", "partitions"),
os.path.join(dest, "partitions"),
),
(
os.path.join("partitioncloud", "attachments"),
os.path.join(dest, "attachments"),
),
(
os.path.join("partitioncloud", "search-partitions"),
os.path.join(dest, "search-partitions"),
),
]
for src, dst in paths:
if os.path.exists(src):
@ -150,20 +174,28 @@ def restore(version):
if os.path.exists(dst):
print(f"\tRestoring {src}")
copy_tree(dst, src)
shutil.copy_tree(dst, src)
else:
print(f"\t{Fore.RED}No available backup for {src}, deleting current content to avoid any conflict{Style.RESET_ALL}")
print(
f"\t{Fore.RED}No available backup for {src}, \
deleting current content to avoid any conflict{Style.RESET_ALL}"
)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
prog='PartitionCloud Migration tool',
description='Helps you migrate from one version to another')
prog="PartitionCloud Migration tool",
description="Helps you migrate from one version to another",
)
parser.add_argument('-c', '--current', help="current version (vx.y.z)")
parser.add_argument('-t', '--target', help="target version (vx.y.z)")
parser.add_argument('-s', '--skip-backup', action='store_true')
parser.add_argument('-r', '--restore', help='restore from specific version backup, will not apply any hook (vx.y.z)')
parser.add_argument("-c", "--current", help="current version (vx.y.z)")
parser.add_argument("-t", "--target", help="target version (vx.y.z)")
parser.add_argument("-s", "--skip-backup", action="store_true")
parser.add_argument(
"-r",
"--restore",
help="restore from specific version backup, will not apply any hook (vx.y.z)",
)
args = parser.parse_args()