import os
import pandas as pd
import numpy as np
import copy
import warnings
[docs]
class FileSelecter():
def __init__(self,rawfolder:str,storagedir:str,filewiki:str='filewiki.csv',recursive=False,depth=-1,fields:list=None):
self.clear()
self.set_rawfolder(rawfolder)
self.set_storagedir(storagedir)
self.set_filewiki(filewiki)
self.set_start_characters()
self.set_recursive(recursive,depth)
self.set_file_wiki_fields(fields)
self.read_files()
self.invalid_files()
[docs]
def set_recursive(self,recursive:bool=False,depth:int=-1):
if depth < 0:
self.depth=None
self.recursive=recursive
elif depth == 0:
self.depth=None
self.recursive=False
else:
self.depth=depth
self.recursive=recursive
[docs]
def set_rawfolder(self,folder:str):
if folder[-1]!='/':
folder+='/'
self.rawfolder = folder
[docs]
def set_storagedir(self,folder:str):
if folder[-1]!='/':
folder+='/'
self.storagedir = folder
[docs]
def set_filewiki(self,filename:str):
self.filewiki = filename
[docs]
def set_start_characters(self,ch=['.','~']):
self.start_characters = ch
[docs]
def set_file_wiki_fields(self,fields:list=None):
self.fields=fields
[docs]
def get_rawfolder(self):
return self.rawfolder
[docs]
def get_recursive(self):
return self.recursive
[docs]
def get_depth(self):
return self.depth
[docs]
def get_storagedir(self):
return self.storagedir
[docs]
def get_filewiki(self):
return self.filewiki
[docs]
def get_files(self):
return self.files
[docs]
def get_file_wiki_fields(self):
return self.fields
[docs]
def sort(self,ll:list):
ll.sort()
return ll
[docs]
def clear(self):
self.rawfolder = None
self.storagedir = None
self.filewiki = None
self.files = None
self.filetable=None
[docs]
def read_files_in_folder(self,folder):
files=[]
dirs = []
for ff in os.listdir(folder):
if os.path.isfile(folder+ff):
files.append(ff)
elif os.path.isdir(folder+ff):
dirs.append(ff+'/')
files = self.exclude_files(files)
dirs = self.exclude_files(dirs)
files=self.sort(files)
return files,dirs
[docs]
def read_files(self):
files,dirs = self.read_files_in_folder(self.get_rawfolder())
self.add_files(files)
if self.get_recursive():
dep = 0
stop=False
while(not stop):
temp_dirs=[]
for dd in dirs:
ffiles,ddirs = self.read_files_in_folder(self.get_rawfolder()+dd)
ffiles = [dd+ff for ff in ffiles]
ddirs = [dd+d for d in ddirs]
self.add_files(ffiles)
temp_dirs+=ddirs
dirs=temp_dirs
dep+=1
if len(dirs)==0:
stop=True
if self.get_depth() is not None:
if dep>=self.get_depth():
stop=True
[docs]
def exclude_files(self,files:list):
exclude = []
for i,ff in enumerate(files):
if self.exclude_condition(ff):
exclude.append(i)
files = [ff for i,ff in enumerate(files) if i not in exclude]
return files
[docs]
def exclude_condition(self,ff:str):
exclude = False
if ff[0] in self.start_characters:
exclude=True
return exclude
[docs]
def add_files(self,files:list):
if len(files)>0:
if self.files is None:
self.files = {ff:True for ff in files}
else:
for ff in files:
self.files[ff] = True
if len(self.files)==0:
self.files=None
[docs]
def check_files(self,exclude:list):
indexes = []
for ex in exclude:
if isinstance(ex,int):
indexes.append(ex)
if isinstance(ex,str):
[indexes.append(i) for i,ff in enumerate(self.files) if ex in ff]
return indexes
[docs]
def invalid_files(self,exclude:list=[]):
for ff in self.files:
self.files[ff] = not self.invalid_condition(ff)
[self.invalid(i) for i in self.check_files(exclude)]
[docs]
def invalid(self,idx:int):
self.files[list(self.files.keys())[idx]]=False
[docs]
def invalid_condition(self,file:str):
if 'Telegram' in file:
return True
return False
[docs]
def get_valid_files(self):
return [ff for ff in self.files if self.files[ff]]
[docs]
def make_file_wiki(self):
fields = self.get_file_wiki_fields()
if fields is not None:
if 'Rawfile Name' not in fields:
fields = ['Rawfile Name']+fields
else:
fields=['Rawfile Name']
self.filetable = pd.DataFrame(columns=fields)
self.filetable['Rawfile Name'] = self.get_valid_files()
[docs]
def check_file_existance(self,folder:str,file:str):
if file in os.listdir(folder):
return True
return False
[docs]
def save_file_wiki(self,mode='a'):
if self.filetable ==None:
self.make_file_wiki()
tmp_file = self.get_filewiki()+"_tmp_diana"
if self.check_file_existance(self.get_storagedir(),tmp_file):
warnings.warn(f"Temporary file still present, check and delete the file!")
return False
filewiki_exists = self.check_file_existance(self.get_storagedir(),self.get_filewiki())
filetable = copy.deepcopy(self.filetable)
if filewiki_exists and mode =='a':
old_filetable = pd.read_csv(self.get_storagedir()+self.get_filewiki())
common = pd.merge(old_filetable['Rawfile Name'],self.filetable['Rawfile Name'],how='inner')
for com in common['Rawfile Name']:
mask = (filetable['Rawfile Name'] != com)
filetable = filetable[mask]
filetable = pd.concat((old_filetable,filetable),ignore_index=True)
#save tmp file
filetable.to_csv(self.get_storagedir()+tmp_file,sep=',',mode='w',index=False)
if not self.check_file_existance(self.get_storagedir(),tmp_file):
return False
filetable.to_csv(self.get_storagedir()+self.get_filewiki(),sep=',',mode='w',index=False)
os.remove(self.get_storagedir()+tmp_file)
return True
def __str__(self):
firstline = 'FileSelecter'
firstlen = len(firstline)//2
firstline = ' '*firstlen + firstline+' '*(len(firstline)-firstlen)
linesep = '-'*len(firstline)
ot = [firstline,linesep]
if self.get_rawfolder() is not None:
ot.append("Raw Folder: "+self.get_rawfolder())
if self.get_storagedir() is not None:
ot.append("Storage Folder: "+self.get_storagedir())
if self.get_filewiki() is not None:
ot.append("Filewiki: "+self.get_filewiki())
if self.get_files() is not None:
if self.recursive:
ot.append("Files (recursive): ")
else:
ot.append("Files: ")
for i,ff in enumerate(self.get_files()):
if self.get_files()[ff]:
ot.append(f"\t{i}) {ff}")
else:
ot.append(f"\t{i})[INVALID] {ff}")
return '\n'.join(ot)
[docs]
class WikiManager():
def __init__(self,storagedir:str=None,filewiki:str='filewiki.csv'):
self.clear()
self.set_storagedir(storagedir)
self.set_filewiki(filewiki)
self.load_saved_wiki()
[docs]
def clear(self):
self.set_storagedir()
self.set_filewiki()
self.filetable=None
[docs]
def set_storagedir(self,folder:str=None):
if folder is not None:
if folder[-1]!='/':
folder+='/'
self.storagedir = folder
[docs]
def get_storagedir(self):
return self.storagedir
[docs]
def get_filewiki(self):
return self.filewiki
[docs]
def set_filewiki(self,filename:str=None):
self.filewiki = filename
[docs]
def load_saved_wiki(self):
self.filetable = pd.read_csv(self.get_storagedir()+self.get_filewiki())
[docs]
def get_saved_wiki(self):
self.load_saved_wiki()
return copy.deepcopy(self.filetable)
[docs]
def read_column(self,colname:str):
return list(self.filetable[colname])
[docs]
def add_column(self,colname:str):
trial=0
newcolname=colname
while(newcolname in list(self.filetable.columns)):
trial+=1
newcolname=colname+f"_{trial}"
self.write_column(newcolname,[None]*len(self.filetable))
[docs]
def write_column(self,colname:str,content:list):
self.filetable[colname]=content
[docs]
def check_file_existance(self,folder:str,file:str):
if file in os.listdir(folder):
return True
return False
[docs]
def save_file_wiki(self):
if self.filetable is None:
return False
tmp_file = self.get_filewiki()+"_tmp_diana"
if self.check_file_existance(self.get_storagedir(),tmp_file):
warnings.warn(f"Temporary file still present, check and delete the file!")
return False
#save tmp file
self.filetable.to_csv(self.get_storagedir()+tmp_file,sep=',',mode='w',index=False)
if not self.check_file_existance(self.get_storagedir(),tmp_file):
return False
self.filetable.to_csv(self.get_storagedir()+self.get_filewiki(),sep=',',mode='w',index=False)
os.remove(self.get_storagedir()+tmp_file)
return True