-
Notifications
You must be signed in to change notification settings - Fork 3
/
Filerize.py
116 lines (88 loc) · 3.28 KB
/
Filerize.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
import asyncio
import logging
import os
import sys
import time
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer
import filetools.fstructs as fstructs
import filetools.ftools as ftools
from classify import FileClassifier
from config import globalvar as gv
from config.config import Config
from filetools.FileToText import FileToText
from filetools.fwatchdog import ListenForFiles
class ConfigHandler:
@staticmethod
def check_config() -> bool:
return os.path.exists(gv.DEFAULT_CONFIG_PATH)
@staticmethod
def create_config() -> None:
print("No config file detected...\nInitialising & creating config file...")
print("Creating config...")
Config.create(path="config.json")
print("Input destination directory paths & labels")
c = 0
while True:
pth = sum = None
pth = str(input(f"Enter directory {c+1}: "))
if not os.path.exists(pth) and pth != "q":
print("Given directory does not exist")
continue
elif pth == "q":
break
sum = str(input(f"Enter directory {c+1} summary: "))
if sum == "q":
break
c += 1
print("\n", end="")
Config.add_label(label=pth, summary=sum)
if not c:
# TODO: Delete config file
print("No directories given. Aborting...")
exit()
Config.save()
@staticmethod
def load_config() -> None:
Config.load(gv.DEFAULT_CONFIG_PATH)
def init(directory) -> fstructs.Folder:
if not ConfigHandler.check_config():
ConfigHandler.create_config()
ConfigHandler.load_config()
folder = fstructs.Folder(path=directory)
ftools.full_crawl(folder=folder)
return folder
def sort(folder: fstructs.Folder):
asyncio.run(ftools.label_all(folder=folder))
ftools.move_all(folder=folder)
def listen(folder: fstructs.Folder):
listener = ListenForFiles(dir=folder.path)
listener.run()
# async def sort(path: str):
# folder = fstructs.Folder(path=path)
# await ftools.recursive_visit(folder=folder, visit=label_files)
# ftools.recursive_move(folder=folder)
# async def label_files(folder: fstructs.Folder):
# files = [file for file in folder.files if file.ext == 'pdf']
# tasks: dict[fstructs.File, asyncio.Task] = {}
# for file in files:
# # PDF to text
# test_pdf: str = FileToText.pdf_to_text(
# f'{file.path}', CUT_STR=True, max_output_length=1000)
# # Classify text
# tasks[file] = asyncio.create_task(
# FileClassifier.classify(test_pdf, Config.labels))
# for file in tasks:
# file.label = await tasks[file]
# async def label_file(path:str):
# text:str = FileToText.docx_to_text(
# path=path,
# CUT_STR=True, max_output_length=1000)
# return await FileClassifier.classify(text, Config.labels)
if __name__ == "__main__":
Config.load(cfg_path=gv.DEFAULT_CONFIG_PATH)
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
folder = fstructs.Folder(path=r"D:\dev\hacknotts\23\Filerize\testing\src")
ftools.full_crawl(folder=folder)
asyncio.run(ftools.label_all(folder=folder))
ftools.move_all(folder=folder)