291 lines
10 KiB
Python
291 lines
10 KiB
Python
import json
|
|
import os
|
|
from werkzeug.utils import secure_filename
|
|
from app.forms import *
|
|
from PIL import Image
|
|
from pathlib import Path, PurePath, PurePosixPath
|
|
from shutil import rmtree, copytree
|
|
import subprocess
|
|
from flask import session
|
|
|
|
|
|
class DataSite:
|
|
user_folder = None
|
|
username = str()
|
|
datas = dict()
|
|
static_path = None
|
|
trash = list()
|
|
filej= None
|
|
|
|
def __init__(self, username :str , users_folder :str ):
|
|
self.username = username
|
|
self.user_folder = Path(PurePosixPath(users_folder).joinpath(username))
|
|
self.filej = Path(PurePosixPath(self.user_folder).joinpath('ident.json'))
|
|
self.static_path = Path(PurePosixPath(self.user_folder.joinpath('public')))
|
|
|
|
def create(self, theme:str):
|
|
|
|
themesImgFolder = Path(PurePosixPath('app').joinpath('static','themes', theme, 'img'))
|
|
imgFolder = Path(PurePosixPath(self.user_folder).joinpath('public', 'img'))
|
|
#imgFolder.mkdir(exist_ok=True)
|
|
|
|
site_default = Path('./app/templates/themes/'+theme+'/ident-default.json')
|
|
|
|
if self.user_folder.exists():
|
|
raise FileNotFoundError(f"Ce site existe déjà")
|
|
return False
|
|
else:
|
|
self.user_folder.mkdir(parents=True, exist_ok=False)
|
|
self.filej.write_text(site_default.read_text())
|
|
self.static_path.mkdir(parents=True, exist_ok=True)
|
|
copytree(themesImgFolder, imgFolder)
|
|
|
|
|
|
if not self.filej.exists():
|
|
raise FileNotFoundError(f"Le site n'a pas pu etre créé")
|
|
return False
|
|
else:
|
|
self.load_file()
|
|
return True
|
|
|
|
def load_file(self):
|
|
if not self.filej.is_file():
|
|
raise FileNotFoundError(f"Le fichier ident.json n'existe pas : {self.filej}")
|
|
|
|
with self.filej.open(newline='', mode='r', encoding='utf-8') as f:
|
|
self.datas = json.load(f)
|
|
|
|
def update_json(self):
|
|
self.filej.write_text(json.dumps(self.datas, indent=5))
|
|
|
|
def set_passwd(self, passwd:str):
|
|
self.datas['config']['passwd']
|
|
self.update_json()
|
|
|
|
def get_passwd(self):
|
|
return self.datas['config']['passwd']
|
|
|
|
def get_blocs(self, page: str):
|
|
|
|
if page != '':
|
|
page = Path(self.user_folder.joinpath(page, 'page.json'))
|
|
datas = ""
|
|
if not page.exists():
|
|
raise FileNotFoundError(f"Le fichier page.json n'existe pas : {page}")
|
|
|
|
datas = json.loads(page.read_text())['blocs']
|
|
else:
|
|
datas = self.datas['blocs'].copy()
|
|
if 'footer' in datas.keys():
|
|
del datas['footer']
|
|
|
|
|
|
return datas
|
|
|
|
def get_footer(self):
|
|
datas = self.datas['blocs']
|
|
if 'footer' in datas.keys():
|
|
return datas['footer']
|
|
else:
|
|
return None
|
|
|
|
|
|
def update_bloc(self, bloc, form, files=None):
|
|
|
|
if files != None:
|
|
f = files['image']
|
|
filename = secure_filename(f.filename)
|
|
extension = filename.rsplit('.', 1)[1].lower()
|
|
bg_custom = 'bg_custom_'+bloc['name']+'.'+extension
|
|
f.save(PurePath.joinpath(
|
|
self.static_path, 'img', bg_custom))
|
|
|
|
bloc['bg_img'] = bg_custom
|
|
|
|
return bloc
|
|
|
|
def update_img(self, image, form):
|
|
img = dict()
|
|
img['title'] = form.title.data
|
|
img['subtitle'] = form.subtitle.data
|
|
img['description'] = form.description.data
|
|
img['category'] = form.category.data
|
|
img['file'] = image
|
|
return img
|
|
|
|
def add_img(self, image_uploaded):
|
|
img = dict()
|
|
|
|
img['title'] = ""
|
|
img['subtitle']=""
|
|
img['text'] = ""
|
|
img['category'] = ""
|
|
img['file'] = '/img/' + image_uploaded
|
|
|
|
return img
|
|
|
|
|
|
def del_img(self, bloc, img):
|
|
|
|
if bloc['type'] == 'gallery':
|
|
img_file = Path(PurePosixPath(self.user_folder).joinpath('public', 'img', bloc[img]['file'] ))
|
|
img_thumb_file = Path(PurePosixPath(self.user_folder).joinpath('public','img', 'thumbnails', bloc[img]['file']))
|
|
if img_file.exists():
|
|
img_file.unlink()
|
|
|
|
if img_thumb_file.exists():
|
|
img_thumb_file.unlink()
|
|
|
|
bloc.pop(img)
|
|
else:
|
|
img_file = Path(PurePosixPath(self.user_folder).joinpath('public', 'img', img ))
|
|
|
|
return bloc
|
|
|
|
|
|
|
|
def save_menu(self, menu:dict):
|
|
self.datas['menu'] = menu
|
|
self.update_json()
|
|
|
|
|
|
def save_blocs(self, page='', blocs=dict()):
|
|
if page != '':
|
|
datas = dict()
|
|
datas['blocs'] = blocs
|
|
|
|
page_info = Path(PurePosixPath(self.user_folder).joinpath(page, 'page.json'))
|
|
|
|
if not page_info.exists():
|
|
raise FileNotFoundError(f"Le fichier page.json n'existe pas : {page}")
|
|
|
|
page_info.write_text(json.dumps(datas, indent=3))
|
|
else:
|
|
self.datas['blocs']= blocs
|
|
self.update_json()
|
|
|
|
def save_server_info(self, serverInfos=dict()):
|
|
self.datas['publish'] = serverInfos
|
|
self.update_json()
|
|
|
|
|
|
def get_theme(self):
|
|
return self.datas['config']['theme']
|
|
|
|
def get_pages(self):
|
|
mypages = list()
|
|
folders = Path(self.user_folder)
|
|
for folder in folders.iterdir():
|
|
if folder.name != 'public' and folder.is_dir():
|
|
mypages.append(folder.name)
|
|
|
|
return mypages
|
|
|
|
def new_page(self, pageName:str, pageType:str):
|
|
page_folder = Path(PurePosixPath(self.user_folder).joinpath(pageName))
|
|
page = Path(PurePosixPath(self.user_folder).joinpath(pageName, 'page.json'))
|
|
page_type = Path('./app/templates/themes/'+self.datas['config']['theme']+'/pages/'+pageType+'.json')
|
|
|
|
if not page_type.exists():
|
|
raise FileNotFoundError(f"Le type de page n'existe pas : {page_type}")
|
|
else:
|
|
page_folder.mkdir()
|
|
page.write_text(page_type.read_text())
|
|
|
|
if not page.exists():
|
|
raise FileNotFoundError(f"La page n'a pas pu etre créé : {page}")
|
|
return False
|
|
return True
|
|
|
|
def rm_page(self, pageName:str):
|
|
page_folder = Path(PurePosixPath(self.user_folder).joinpath(pageName))
|
|
page = Path(PurePosixPath(self.user_folder).joinpath(pageName, 'page.json'))
|
|
|
|
#First delete the json page
|
|
try:
|
|
page.unlink()
|
|
print(f"Successfully deleted: {page}")
|
|
except FileNotFoundError:
|
|
print(f"File not found: {page}. Nothing to delete.")
|
|
# Delete the empty directory
|
|
if page_folder.is_dir():
|
|
try:
|
|
page_folder.rmdir()
|
|
print(f"Successfully removed empty directory: {page_folder}")
|
|
except OSError as e:
|
|
print(f"Error removing directory (might not be empty): {e}")
|
|
|
|
|
|
def writeHTML(self, theme:str, pageName :str, htmlExport :str):
|
|
publicFolder = Path(PurePosixPath(self.user_folder).joinpath('public'))
|
|
themesFolder = Path(PurePosixPath('app').joinpath('templates','themes', self.datas['config']['theme'],'static'))
|
|
|
|
if pageName == '/':
|
|
pageIndex = publicFolder / 'index.html'
|
|
pageIndex.write_text(htmlExport)
|
|
else:
|
|
pageFolder = Path(PurePosixPath(self.user_folder).joinpath('public', pageName))
|
|
pageFolder.mkdir(exist_ok=True)
|
|
pageIndex = pageFolder / 'index.html'
|
|
pageIndex.write_text(htmlExport)
|
|
|
|
def copy_static_files(self):
|
|
themesStaticFolder = Path(PurePosixPath('app').joinpath('static','themes', self.datas['config']['theme']))
|
|
publicStaticFolder = Path(PurePosixPath(self.user_folder).joinpath('public', 'static', 'themes', self.datas['config']['theme']))
|
|
|
|
rmtree (publicStaticFolder, ignore_errors=True)
|
|
|
|
copytree(themesStaticFolder, publicStaticFolder)
|
|
|
|
|
|
def test_server_login(self, serverInfos: dict(), password:str()):
|
|
|
|
test_connect = "lftp -u \"{}\",\"{}\" -p {} {} -e 'ls; bye;'".format(serverInfos['username'],
|
|
password,
|
|
serverInfos['Port'],
|
|
serverInfos['Address'])
|
|
test_login = subprocess.check_output(test_connect, shell=True, stderr=subprocess.STDOUT)
|
|
if "Login incorrect" in str(test_login) :
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
|
|
|
|
def publish(self, serverInfos=dict(), password=str()):
|
|
userDir = Path(PurePosixPath(self.user_folder))
|
|
upload = "lftp -u \"{}\",\"{}\" -p {} {}:{} -e 'mirror -R ./public ./; bye;'".format(serverInfos['username'],
|
|
password,
|
|
serverInfos['Port'],
|
|
serverInfos['Address'],
|
|
serverInfos['folder'])
|
|
|
|
set_permissions = "lftp -u \"{}\",\"{}\" -p {} {}:{} -e 'chmod -R o+rx ./; bye;'".format(serverInfos['username'],
|
|
password,
|
|
serverInfos['Port'],
|
|
serverInfos['Address'],
|
|
serverInfos['folder'])
|
|
try:
|
|
subprocess.call(upload, shell=True, cwd=str(userDir))
|
|
self.update_status_publish("Uploaded")
|
|
subprocess.call(set_permissions, shell=True, cwd=str(userDir))
|
|
|
|
except ChildProcessError:
|
|
return False
|
|
|
|
return True
|
|
|
|
def update_status_publish (self, status: str):
|
|
self.datas['publication_status'] = status
|
|
self.filej.write_text(json.dumps(self.datas, indent=5))
|
|
|
|
def get_status_publish(self):
|
|
if 'publication_status' in self.datas.keys():
|
|
return self.datas['publication_status']
|
|
else:
|
|
return "";
|
|
|
|
|
|
|
|
|