319 lines
13 KiB
Python
319 lines
13 KiB
Python
from pyramid.view import view_config
|
|
import os
|
|
import shutil
|
|
import magic
|
|
import datetime
|
|
import time
|
|
import charset_normalizer
|
|
import re
|
|
|
|
@view_config(route_name='home', renderer='html_cleanup:templates/home.jinja2')
|
|
def home(request):
|
|
|
|
message = ''
|
|
output_name = ''
|
|
file_url = ''
|
|
input_encoding = ''
|
|
|
|
if 'form.submitted' in request.params:
|
|
if request.POST['uploadfile'] != b'':
|
|
input_file = request.POST['uploadfile'].file
|
|
input_name = request.POST['uploadfile'].filename
|
|
# Sanitize filename to prevent path traversal
|
|
input_name = os.path.basename(input_name)
|
|
# récupère le fichier download, faire les controles et traiter
|
|
message, input_encoding, output_name = process_file(request, input_file, input_name)
|
|
# process without error ?
|
|
if message == '':
|
|
file_url = request.static_url('html_cleanup:static/temp/') + output_name
|
|
if input_encoding == 'utf-8':
|
|
message = "Votre fichier est nettoyé : " + output_name
|
|
else:
|
|
message = "Votre fichier est converti et nettoyé : " + output_name
|
|
|
|
return {
|
|
'page_title': "HTML cleanup 26.01",
|
|
'message': message,
|
|
'file_url': file_url,
|
|
'file_name': output_name,
|
|
}
|
|
|
|
def process_file(request, input_file, input_name):
|
|
# check input_file validity
|
|
# if OK, procede the file cleanup
|
|
# else return an error message
|
|
|
|
message = ''
|
|
EXT_ALLOWED = ['text/html']
|
|
MAX_SIZE = 10 * (1024 ** 2) # 10 Mb
|
|
temp_folder = request.registry.settings['temp_folder']
|
|
logfile_name = os.path.join(temp_folder, 'errors_log')
|
|
input_encoding = 'utf-8'
|
|
output_name = ''
|
|
|
|
# --- check the mime type
|
|
mime = magic.from_buffer(input_file.read(), mime=True)
|
|
if mime not in EXT_ALLOWED:
|
|
message = "Le format du fichier n'est pas valide. Téléchargement refusé."
|
|
add_error2log(logfile_name, input_name, message)
|
|
return message, input_encoding, output_name
|
|
|
|
# get input_file size
|
|
input_file.seek(0, 2) #seek to end
|
|
filesize = input_file.tell()
|
|
input_file.seek(0) # back to begining position
|
|
# --- check input_file size
|
|
if filesize > MAX_SIZE:
|
|
message = "La taille du fichier dépasse la limite autorisée. Téléchargement refusé."
|
|
add_error2log(logfile_name, input_name, message)
|
|
return message, input_encoding, output_name
|
|
|
|
# Finally write the data to a temporary file
|
|
input_filename = os.path.join(temp_folder, input_name)
|
|
# if temporary file exists then remove it
|
|
if os.path.exists(input_filename):
|
|
os.remove(input_filename)
|
|
|
|
# copy input file to temp folder
|
|
input_file.seek(0)
|
|
with open(input_filename, 'wb') as output_file:
|
|
shutil.copyfileobj(input_file, output_file)
|
|
|
|
# detect text coding and transform to utf-8 if needed
|
|
with open(input_filename, 'rb') as f:
|
|
raw_bytes = f.read()
|
|
detected = charset_normalizer.detect(raw_bytes)
|
|
input_encoding = (detected.get('encoding') or 'utf-8').lower()
|
|
if input_encoding != 'utf-8':
|
|
with open(input_filename, 'r', encoding=input_encoding) as f:
|
|
input_buffer = f.read()
|
|
# --- check if input_file contains tag <body>
|
|
body_lower = input_buffer.lower()
|
|
if body_lower.find('<body') == -1 or body_lower.find('</body>') == -1:
|
|
message = "Le format du fichier n'est pas valide (absence de tag <body> ou </body>). Téléchargement refusé."
|
|
add_error2log(logfile_name, input_name, message)
|
|
return message, input_encoding, output_name
|
|
|
|
# --- convert input_file encoding to utf-8
|
|
with open(input_filename, 'w', encoding='utf-8') as f:
|
|
message = "Le fichier est converti de " + input_encoding + " en utf-8."
|
|
add_error2log(logfile_name, input_name, message)
|
|
f.write(input_buffer)
|
|
|
|
|
|
# format output name by adding '-clean'
|
|
slots = input_name.rsplit('.', 1)
|
|
if len(slots) == 1:
|
|
output_name = input_name + "-clean"
|
|
else:
|
|
output_name = slots[0].strip() + "-clean." + slots[1]
|
|
|
|
# controls OK, cleanup input_filename
|
|
output_path = os.path.join(temp_folder, output_name)
|
|
message = clean_html(input_filename, output_path)
|
|
if message:
|
|
add_error2log(logfile_name, input_name, message)
|
|
else:
|
|
add_error2log(logfile_name, input_name, "*** Cleanup réussi. ***")
|
|
|
|
return message, input_encoding, output_name
|
|
|
|
def clean_html(input_file, output_file):
|
|
#
|
|
# cleanup undesirable tags in html file
|
|
#
|
|
#20241118: thn: changed chardet to charset_normalizer
|
|
#20251216: thn: changed align:"justify" to style="text-align:justify;" in <p>,
|
|
#because it was deprecated in HTML 5
|
|
#accepted id in <span, p, h1, ... h6>
|
|
#20251217: thn, replaced re.match by re.search
|
|
#20251223: thn, accepted hyphen in id. Avoid: [.:] because they are using in CSS
|
|
#20251224: check input syntax for id
|
|
#Only support <tag ....> in one line
|
|
#20260104: treated "div" the same as "span", "h1"..."h6"
|
|
#20260104: removed "span", </span>
|
|
#20260122: move the function check_input_name into the function clean_html as inner function
|
|
#20260128: removed text-align:justify in <p>
|
|
#20260129: added strikethrough tag: <s>, <del>, <ins>: insert with underline
|
|
#version: 3_5_3_5
|
|
|
|
def check_input_name(attribute):
|
|
ret_flag=False
|
|
name_regex=r'.*=\s*\"(.*)\".*'
|
|
matched=re.search(name_regex, attribute)
|
|
if matched:
|
|
if matched.group(1)[0].isalpha():
|
|
ret_flag=True
|
|
return ret_flag
|
|
|
|
|
|
encoding = "utf-8"
|
|
message = ""
|
|
|
|
try:
|
|
with open(input_file, 'r') as fi, open(output_file, 'w') as fo:
|
|
|
|
body = False
|
|
skip_tag = False
|
|
nb_lines = 0
|
|
|
|
for line_in in fi:
|
|
line_out = ""
|
|
|
|
if not body:
|
|
body_match = re.search(r'^\s*<body.*', line_in, re.IGNORECASE)
|
|
if body_match:
|
|
body = True
|
|
fo.write("<html>\n\t<head>\n\t\t<meta http-equiv=\"Content-Type\" content=\"text/html; charset={}\">\n\t\t<title>untitled</title>\n\t</head>\n".format(encoding))
|
|
|
|
if body:
|
|
nb_lines += 1
|
|
|
|
if skip_tag:
|
|
start_from = line_in.find(">")
|
|
if start_from != -1 and start_from >= 0:
|
|
skip_tag = False
|
|
start_from = start_from + 1 if start_from < len(line_in) else 0
|
|
else:
|
|
start_from = 0
|
|
|
|
if start_from >= 0:
|
|
done = False
|
|
while not done:
|
|
next_tag = line_in.find("<", start_from)
|
|
if next_tag == -1:
|
|
line_out += line_in[start_from:]
|
|
done = True
|
|
skip_tag = False
|
|
else:
|
|
if next_tag > start_from:
|
|
line_out += line_in[start_from:next_tag]
|
|
|
|
end_tag = line_in.find(">", next_tag + 1)
|
|
if end_tag == -1:
|
|
s = line_in[next_tag + 1:]
|
|
done = True
|
|
skip_tag = True
|
|
else:
|
|
s = line_in[next_tag + 1:end_tag]
|
|
|
|
if end_tag < len(line_in):
|
|
start_from = end_tag + 1
|
|
done = False
|
|
skip_tag = False
|
|
else:
|
|
done = True
|
|
skip_tag = False
|
|
|
|
tag = s.split()
|
|
|
|
if tag:
|
|
tag[0] = tag[0].lower()
|
|
id_match_regex=r'.*(id=\"[\w-]+\").*'
|
|
|
|
if tag[0] in ["table", "img", "a"]:
|
|
line_out += "<{}".format(s)
|
|
if end_tag > 0:
|
|
# si end tag trouvé, fermer
|
|
line_out += ">"
|
|
else:
|
|
# sinon contine de recopier la ligne
|
|
skip_tag = False
|
|
|
|
elif tag[0] in ["div", "p", "h1", "h2", "h3", "h4", "h5", "h6"]:
|
|
line_out += "<{}".format(tag[0])
|
|
# 26.01 : ne plus ajouter text-align dans <p>
|
|
# if tag[0] == "p":
|
|
# line_out += ' style="text-align:justify;"'
|
|
if len(tag) > 1 and not skip_tag:
|
|
matched = re.search(id_match_regex, line_in)
|
|
if matched:
|
|
if check_input_name(matched.group(1)):
|
|
line_out += ' '+ matched.group(1)
|
|
line_out += '>'
|
|
|
|
elif tag[0] in ["b", "/b", "em", "/em", "i", "/i", "li", "/li", "ol", "/ol",
|
|
"ul", "/ul", "strong", "/strong", "sub", "/sub", "sup", "/sup",
|
|
"u", "/u", "s", "/s", "del", "/del", "ins", "/ins"]:
|
|
line_out += "<{}>".format(tag[0])
|
|
elif tag[0] in ["/h1", "/h2", "/h3", "/h4", "/h5", "/h6", "/p",
|
|
"body", "/body", "/html", "/div", "/img", "/a"]:
|
|
line_out += "<{}>".format(tag[0])
|
|
elif tag[0] in ["/table", "td", "/td", "th", "/th", "tr", "/tr"]:
|
|
line_out += "<{}>".format(tag[0])
|
|
elif tag[0] in ["br", "br/", "hr"]:
|
|
line_out += "<{}>".format(tag[0])
|
|
|
|
if line_out != "":
|
|
fo.write(line_out + "\n")
|
|
|
|
except Exception as e:
|
|
message = "Error: {}".format(str(e))
|
|
|
|
return message
|
|
|
|
def add_error2log(logfile_name, input_name, message):
|
|
# ecrire message d'erreur dans le log
|
|
|
|
# Get the current date and time
|
|
now = datetime.datetime.now()
|
|
with open(logfile_name, 'a', encoding='utf-8') as file:
|
|
line = now.strftime("%d-%m-%Y %H:%M") + ' [ ' + input_name + ' ] ' + f'{message}\n'
|
|
file.write(line)
|
|
|
|
@view_config(route_name='view_log', renderer='html_cleanup:templates/view_log.jinja2')
|
|
def view_log(request):
|
|
|
|
message = ''
|
|
temp_folder = request.registry.settings['temp_folder']
|
|
logfile_name = os.path.join(temp_folder, 'errors_log')
|
|
users = ['minh','hvtien','dinh','giao','ctphuoc','TamCauDePiR','VHN','zidol','sautam']
|
|
lines = []
|
|
files_list = []
|
|
username = ''
|
|
|
|
if 'form.submitted' in request.params:
|
|
username = request.POST.get('username')
|
|
if username in users:
|
|
# lire et affiche le fichier log
|
|
try:
|
|
with open(logfile_name, "r") as log_fh:
|
|
lines = log_fh.readlines()
|
|
except Exception as e:
|
|
message = "Le fichier Journal est vide !"
|
|
pass
|
|
|
|
# lire tous les fichiers du dossier triés par date
|
|
entries = sorted(os.scandir(temp_folder), key=lambda ent: ent.stat().st_mtime)
|
|
|
|
# parcorrir tous les fichiers du dossier
|
|
for f in entries:
|
|
if f.name[0] != '.' and f.name != 'errors_log':
|
|
f_path = temp_folder + f.name
|
|
nbs = os.path.getctime(f_path) # en nombre de secondes
|
|
dtLocaltime = time.localtime(nbs) # convertir en local time
|
|
|
|
file_entry = []
|
|
file_entry.append(time.strftime("%d/%m/%Y %H:%M", dtLocaltime))
|
|
file_entry.append(f.name)
|
|
file_entry.append('%s Ko' % round(os.path.getsize(f_path) / 1024, 1))
|
|
files_list.append(file_entry)
|
|
else:
|
|
message = "Identifiant invalide !"
|
|
|
|
if 'form.erased' in request.params:
|
|
# effacer tous les fichiers du dossier temp
|
|
with os.scandir(temp_folder) as entries:
|
|
for entry in entries:
|
|
if entry.is_file():
|
|
os.unlink(entry.path)
|
|
message = "Tous les fichiers du dossier TEMP sont effacés."
|
|
|
|
|
|
return {
|
|
'page_title': "Journal de Cleanup",
|
|
'message': message,
|
|
'lines': lines,
|
|
'files_list': files_list,
|
|
}
|