# -*- coding: UTF-8 -*-
"""
logrotate.py
============
Permet la rotation des logs Apache
:Example:
>>> import logrotate
>>> logrotate.main()
"""
import os, shutil, zipfile
import datetime as dt
F = r'C:\wamp\logs' # Répertoire des logs Apache
LOG = 'logrotate.log' # Fichier de log pour les rotations
TXT = 'ROTATION DU FICHIER' # Texte affiché dans le log des rotations
TXT2 = 'SUPPRESSION DU FICHIER' # Texte affiché dans le log des rotations
TXT3 = 'COMPRESSION DU FICHIER' # Texte affiché dans le log des rotations
NBARCHIVE = 20 # Nombre de fichier de log historisé .0 .1 .2 etc etc ...
def now():
"""
Retourne la date et l'heure courante au format ISO 2018-06-28T10:30:23.816122
:return: La date et heure courante
:rtype: datetime.datetime
"""
return dt.datetime.now().isoformat()
def cleanList(LIST = [], LISTERR = []):
"""
Supprime de la liste ``LIST`` tous les fichiers
correspondant à ceux en erreur présent dans la liste ``LISTERR``
:return: Une nouvelle liste de fichiers nettoyées
:rtype: list
"""
for ERR in LISTERR:
LIST = list(filter(lambda f: not f.startswith(ERR), LIST))
return LIST
def log(FLOG, TXT, FILE, MSG):
"""
Ecrit dans le fichier ``FLOG`` les infos
``TXT``, ``FILE`` et ``MSG`` avec la date courante
"""
print('{DATETIME} {T:<25s} {FILE:<40s} --> {MSG}'.format(DATETIME=now(), FILE=FILE, T=TXT, MSG=MSG), file=FLOG)
def delFile(FILE, FLOG):
"""
Supprime le fichier ``FILE``
"""
try: os.remove(FILE)
except: log(FLOG, TXT2, FILE, 'KO')
else: log(FLOG, TXT2, FILE, 'OK')
def listLog(LIST, EXT):
"""
Extrait tous les fichiers de log avec l'extension ``EXT``
de la liste ``LIST``
:return: Une liste contenant uniquement les fichiers ayant l'extension ``EXT``
:rtype: list
"""
return list(filter(lambda f: f.endswith(EXT), LIST))
def removeUnusedFiles(FLOG):
"""
Supprime tous les fichiers qui ne sont plus concernés par des fichiers LOG
Retourne une nouvelle liste avec les fichiers supprimés en moins
:return: Une nouvelle liste avec les fichiers supprimés en moins
:rtype: list
"""
LIST = os.listdir(F)
LISTLOG = listLog(LIST, '.log')
LISTDEL = list(filter(lambda x: '.'.join(x.split('.')[:2]) not in LISTLOG, LIST))
for FILE in LISTDEL:
delFile(os.path.join(F, FILE), FLOG)
return list(set(LIST) - set(LISTDEL))
def incrementZipFile(LIST, FLOG, LISTERR = []):
"""
Incrémente les archives ZIP jusqu'à ``NBARCHIVE``
:return: Une liste de fichiers en erreur
:rtype: list
"""
LISTLOG = listLog(LIST, '.log')
LISTZIP = listLog(LIST, '.zip')
for f in LISTLOG:
WLIST = sorted(list(filter(lambda x: x.startswith(f), LISTZIP)), key=lambda y: int(y.split('.')[2]), reverse=True)
ERR = False
for FILE in WLIST:
TF = FILE.split('.')
IDX = int(TF[2])
if IDX < NBARCHIVE:
OLDFILE = os.path.join(F, FILE)
TF[2] = str(IDX + 1)
NEWFILE = os.path.join(F, '.'.join(TF))
try:
with zipfile.ZipFile(OLDFILE, mode='r', compression=zipfile.ZIP_DEFLATED) as f1:
try:
CT = f1.read(os.path.splitext(FILE)[0])
except: ERR = True
else:
BASE = os.path.splitext(os.path.basename(NEWFILE))[0]
try:
with zipfile.ZipFile(NEWFILE, mode='w', compression=zipfile.ZIP_DEFLATED) as f2:
f2.writestr(BASE, CT)
except: ERR = True
else: log(FLOG, TXT, OLDFILE, '{NEWFILE}'.format(NEWFILE=NEWFILE))
except: ERR = True
finally:
if ERR:
LISTERR.append(f)
log(FLOG, TXT, OLDFILE, '{NEWFILE} ***KO***'.format(NEWFILE=NEWFILE))
break
return LISTERR
def log_0ToZip(LIST, FLOG, LISTERR = []):
"""
On écrit tout le contenu des fichiers LOG *.log.0
dans de nouveaux fichiers ZIP *.log.1.zip
On supprime ensuite tous les fichiers *.log.0
:return: Une liste de fichiers en erreur
:rtype: list
"""
LISTLOG0 = listLog(LIST, '.log.0')
for FILE in LISTLOG0:
ERR = False
with open(os.path.join(F, FILE), mode='r', encoding='UTF-8') as f1:
BASE = FILE[:-1] + '1'
ZIP = os.path.join(F, BASE + '.zip')
with zipfile.ZipFile(ZIP, 'w', compression=zipfile.ZIP_DEFLATED) as f2:
try: f2.writestr(BASE, f1.read().encode('UTF-8'))
except:
ERR = True
LISTERR.append(FILE[:-2])
log(FLOG, TXT3, os.path.join(F, FILE), '{NEWFILE} ***KO***'.format(NEWFILE=ZIP))
else: log(FLOG, TXT3, os.path.join(F, FILE), '{NEWFILE}'.format(NEWFILE=ZIP))
if not ERR: delFile(os.path.join(F, FILE), FLOG)
return LISTERR
def logTo0(LIST, FLOG):
"""
On écrit tout le contenu des fichiers LOG *.log
dans de nouveaux fichiers numérotés *.log.0
On écrase ensuite le contenu des fichiers LOG
"""
LISTLOG = listLog(LIST, '.log')
for FILE in LISTLOG:
OLDFILE = os.path.join(F, FILE)
NEWFILE = os.path.join(F, FILE + '.0')
try:
with open(OLDFILE, mode='r', encoding='UTF-8') as f1:
with open(NEWFILE, mode='w', encoding='UTF-8') as f2:
f2.write(f1.read())
except: log(FLOG, TXT, OLDFILE, '{NEWFILE} ***KO***'.format(NEWFILE=NEWFILE))
else:
log(FLOG, TXT, OLDFILE, '{NEWFILE}'.format(NEWFILE=NEWFILE))
try:
with open(OLDFILE, mode='w', encoding='UTF-8') as f1:
f1.write('')
except: log(FLOG, TXT2, OLDFILE, 'KO')
else: log(FLOG, TXT2, OLDFILE, 'OK')
def main():
"""
Fonction principale
Pour chaque fichier de log (*.log) trouvé dans le dossier (variable ``F``)
Chaque niveau d'archive est incrémenté de 1 dans la limite du nombre d'archive
indiqué dans la variable ``NBARCHIVE``
"""
with open(os.path.join(F, LOG), mode='a', encoding='UTF-8') as FLOG:
LIST = removeUnusedFiles(FLOG)
LIST.remove(LOG) # On supprime de la liste le fichier de LOG du script
LISTERR = incrementZipFile(LIST, FLOG)
LIST = cleanList(LIST, LISTERR)
LISTERR = log_0ToZip(LIST, FLOG, LISTERR)
LIST = cleanList(LIST, LISTERR)
logTo0(LIST, FLOG)
if __name__ == '__main__':
main()