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 "";