import sys import os import sqlite3 import hashlib from zipfile import ZipFile, ZIP_DEFLATED import csv from PySide6.QtWidgets import ( QApplication, QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QFileDialog, QTreeView, QMessageBox, QSplitter, QLabel, QLineEdit, QTextEdit, QFormLayout, QTabWidget, QTableWidget, QTableWidgetItem, QHeaderView ) from PySide6.QtGui import QStandardItemModel, QStandardItem from PySide6.QtCore import Qt, QThread, Signal # ---------------- CSV WORKER (QThread) ---------------- class CsvWorker(QThread): finished = Signal(str) error = Signal(str) def __init__(self, root_folder, csv_path, compute_md5): super().__init__() self.root_folder = root_folder self.csv_path = csv_path self.compute_md5 = compute_md5 def run(self): try: rows = [] for dirpath, _, filenames in os.walk(self.root_folder): for f in filenames: full = os.path.join(dirpath, f) try: size = os.path.getsize(full) modified = os.path.getmtime(full) md5 = self.compute_md5(full) rows.append([full, size, modified, md5]) except: pass with open(self.csv_path, "w", newline="", encoding="utf-8") as csvfile: writer = csv.writer(csvfile) writer.writerow(["file_path", "size_bytes", "modified_timestamp", "md5_hash"]) writer.writerows(rows) self.finished.emit(self.csv_path) except Exception as e: self.error.emit(str(e)) POPULATED_ROLE = Qt.UserRole + 1 DB_PATH = r"C:\Temp\fileCollector\DB\fileCollector.db" PREVIEW_MAX_BYTES = 256 * 1024 # 256 KB class LazyFileTree(QWidget): def __init__(self): super().__init__() self.setWindowTitle("fileCollector | Michael De La Cruz") self.resize(1400, 800) self._propagating = False self.root_folder = "C:/Users" self.worker = None self._init_db() splitter = QSplitter(Qt.Horizontal) main_layout = QHBoxLayout(self) main_layout.addWidget(splitter) # ---------------- LEFT: TREE + BUTTONS ---------------- left_widget = QWidget() left_layout = QVBoxLayout(left_widget) self.model = QStandardItemModel() self.model.setHorizontalHeaderLabels(["Files and Folders"]) self.tree = QTreeView() self.tree.setModel(self.model) self.tree.setHeaderHidden(False) left_layout.addWidget(self.tree) btns = QHBoxLayout() left_layout.addLayout(btns) self.choose_btn = QPushButton("Choose Root Folder") self.choose_btn.clicked.connect(self.choose_folder) btns.addWidget(self.choose_btn) self.zip_btn = QPushButton("Create ZIP") self.zip_btn.clicked.connect(self.create_zip) btns.addWidget(self.zip_btn) self.csv_btn = QPushButton("Export CSV (Full Listing)") self.csv_btn.clicked.connect(self.export_full_csv) btns.addWidget(self.csv_btn) splitter.addWidget(left_widget) # ---------------- RIGHT: TABS ---------------- right_widget = QWidget() right_layout = QVBoxLayout(right_widget) self.right_tabs = QTabWidget() right_layout.addWidget(self.right_tabs) # ---- Tab 1: Collection Info ---- form_tab = QWidget() form_layout = QFormLayout(form_tab) self.collector_name = QLineEdit() self.case_id = QLineEdit() self.notes = QTextEdit() form_layout.addRow(QLabel("Collection Information")) form_layout.addRow("Collector Name:", self.collector_name) form_layout.addRow("Case ID:", self.case_id) form_layout.addRow("Notes:", self.notes) self.save_btn = QPushButton("Save Collection Info") self.save_btn.clicked.connect(self.save_collection_info) form_layout.addRow(self.save_btn) self.right_tabs.addTab(form_tab, "Collection Info") # ---- Tab 2: Browser ---- browser_tab = QWidget() browser_layout = QVBoxLayout(browser_tab) self.browser_tabs = QTabWidget() browser_layout.addWidget(self.browser_tabs) # Cases tab self.cases_tab = QWidget() cases_layout = QVBoxLayout(self.cases_tab) self.cases_table = QTableWidget() self.cases_table.setColumnCount(5) self.cases_table.setHorizontalHeaderLabels( ["ID", "Case ID", "Collector", "Notes", "Created At"] ) self.cases_table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) self.cases_table.setSelectionBehavior(QTableWidget.SelectRows) self.cases_table.setEditTriggers(QTableWidget.NoEditTriggers) self.cases_table.cellClicked.connect(self.on_case_selected) cases_layout.addWidget(self.cases_table) self.browser_tabs.addTab(self.cases_tab, "Cases") # Files tab self.files_tab = QWidget() files_layout = QVBoxLayout(self.files_tab) self.files_table = QTableWidget() self.files_table.setColumnCount(3) self.files_table.setHorizontalHeaderLabels( ["ID", "File Path", "MD5 Hash"] ) self.files_table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) self.files_table.setSelectionBehavior(QTableWidget.SelectRows) self.files_table.setEditTriggers(QTableWidget.NoEditTriggers) self.files_table.cellClicked.connect(self.on_file_selected) files_layout.addWidget(self.files_table) self.browser_tabs.addTab(self.files_tab, "Files") # Preview tab self.preview_tab = QWidget() preview_layout = QVBoxLayout(self.preview_tab) self.preview_text = QTextEdit() self.preview_text.setReadOnly(True) preview_layout.addWidget(self.preview_text) self.browser_tabs.addTab(self.preview_tab, "Preview") # Refresh button self.refresh_browser_btn = QPushButton("Refresh Collections") self.refresh_browser_btn.clicked.connect(self.load_cases) browser_layout.addWidget(self.refresh_browser_btn) self.right_tabs.addTab(browser_tab, "View Collections") splitter.addWidget(right_widget) splitter.setSizes([850, 550]) # ---------------- SIGNALS ---------------- self.tree.expanded.connect(self.on_expanded) self.model.itemChanged.connect(self.on_item_changed) root_item = self._make_item("Users", self.root_folder, is_dir=True) self.model.invisibleRootItem().appendRow(root_item) self.load_cases() # ---------------- DB INIT ---------------- def _init_db(self): os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) conn = sqlite3.connect(DB_PATH) cur = conn.cursor() cur.execute(""" CREATE TABLE IF NOT EXISTS collection_info ( id INTEGER PRIMARY KEY AUTOINCREMENT, collector_name TEXT, case_id TEXT, notes TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ) """) cur.execute(""" CREATE TABLE IF NOT EXISTS collection_paths ( id INTEGER PRIMARY KEY AUTOINCREMENT, collection_id INTEGER, file_path TEXT, md5_hash TEXT, FOREIGN KEY(collection_id) REFERENCES collection_info(id) ) """) conn.commit() conn.close() # ---------------- MD5 ---------------- def compute_md5(self, filepath): hash_md5 = hashlib.md5() try: with open(filepath, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) return hash_md5.hexdigest() except: return None # ---------------- SAVE COLLECTION INFO ---------------- def save_collection_info(self): name = self.collector_name.text().strip() case = self.case_id.text().strip() notes = self.notes.toPlainText().strip() if not case: QMessageBox.warning(self, "Missing Case ID", "Case ID is required.") return conn = sqlite3.connect(DB_PATH) cur = conn.cursor() cur.execute(""" SELECT id FROM collection_info WHERE case_id = ? ORDER BY id DESC LIMIT 1 """, (case,)) row = cur.fetchone() if row: collection_id = row[0] cur.execute(""" UPDATE collection_info SET collector_name = ?, notes = ? WHERE id = ? """, (name, notes, collection_id)) else: cur.execute(""" INSERT INTO collection_info (collector_name, case_id, notes) VALUES (?, ?, ?) """, (name, case, notes)) conn.commit() conn.close() QMessageBox.information(self, "Saved", "Collection information saved/updated.") self.load_cases() # ---------------- TREE ---------------- def choose_folder(self): folder = QFileDialog.getExistingDirectory(self, "Select Folder", "C:/Users") if not folder: return self.root_folder = folder self.model.removeRows(0, self.model.rowCount()) root_item = self._make_item(os.path.basename(folder) or folder, folder, is_dir=True) self.model.invisibleRootItem().appendRow(root_item) def _make_item(self, name, full_path, is_dir): item = QStandardItem(name) item.setData(full_path, Qt.UserRole) item.setCheckable(True) if is_dir: item.setData(False, POPULATED_ROLE) item.appendRow(QStandardItem()) # placeholder return item # ---------------- CHECKBOX PROPAGATION ---------------- def on_item_changed(self, item): if self._propagating: return if not item.isCheckable(): return state = item.checkState() self._propagating = True try: self._set_children_check_state(item, state) finally: self._propagating = False def _set_children_check_state(self, item, state): for i in range(item.rowCount()): child = item.child(i) if child.isCheckable(): child.setCheckState(state) self._set_children_check_state(child, state) # ---------------- LAZY LOADING ---------------- def on_expanded(self, index): item = self.model.itemFromIndex(index) if not item or item.data(POPULATED_ROLE): return item.removeRows(0, item.rowCount()) folder = item.data(Qt.UserRole) self._populate_folder(item, folder) item.setData(True, POPULATED_ROLE) def _populate_folder(self, parent_item, folder): try: entries = sorted(os.listdir(folder)) except (PermissionError, FileNotFoundError): return for entry in entries: full_path = os.path.join(folder, entry) if os.path.islink(full_path): continue if os.path.isdir(full_path): child = self._make_item(entry, full_path, is_dir=True) parent_item.appendRow(child) else: child = self._make_item(entry, full_path, is_dir=False) parent_item.appendRow(child) # ---------------- CSV EXPORT (QThread) ---------------- def export_full_csv(self): if not self.root_folder or not os.path.exists(self.root_folder): QMessageBox.warning(self, "Invalid Root", "Root folder is not set or does not exist.") return csv_path, _ = QFileDialog.getSaveFileName(self, "Save CSV", "", "CSV Files (*.csv)") if not csv_path: return if not csv_path.lower().endswith(".csv"): csv_path += ".csv" self.csv_btn.setEnabled(False) self.worker = CsvWorker(self.root_folder, csv_path, self.compute_md5) self.worker.finished.connect(lambda path=csv_path: self._csv_export_finished(path)) self.worker.error.connect(lambda msg: QMessageBox.critical(self, "Error", msg)) self.worker.start() def _csv_export_finished(self, csv_path): self.csv_btn.setEnabled(True) QMessageBox.information(self, "CSV Created", f"Full file listing saved:\n{csv_path}") # ---------------- COLLECT CHECKED FILES ---------------- def _collect_checked(self, item): results = [] path = item.data(Qt.UserRole) if path is None: return results if item.checkState() == Qt.Checked: if os.path.isfile(path): results.append(path) elif os.path.isdir(path): for dirpath, _, filenames in os.walk(path): for f in filenames: results.append(os.path.join(dirpath, f)) for i in range(item.rowCount()): results.extend(self._collect_checked(item.child(i))) return results # ---------------- ZIP CREATION ---------------- def create_zip(self): root = self.model.invisibleRootItem().child(0) if not root: QMessageBox.warning(self, "No Folder", "Root not loaded.") return selected_files = self._collect_checked(root) unique_files = list({os.path.normpath(f): f for f in selected_files}.values()) if not unique_files: QMessageBox.warning(self, "No Files Selected", "Please check one or more files.") return zip_path, _ = QFileDialog.getSaveFileName(self, "Save ZIP", "", "ZIP Files (*.zip)") if not zip_path: return if not zip_path.lower().endswith(".zip"): zip_path += ".zip" try: with ZipFile(zip_path, "w", ZIP_DEFLATED) as zipf: for full in unique_files: rel = os.path.relpath(full, self.root_folder) zipf.write(full, rel) name = self.collector_name.text().strip() case = self.case_id.text().strip() notes = self.notes.toPlainText().strip() if not case: case = "" conn = sqlite3.connect(DB_PATH) cur = conn.cursor() cur.execute(""" SELECT id FROM collection_info WHERE case_id = ? ORDER BY id DESC LIMIT 1 """, (case,)) row = cur.fetchone() if row: collection_id = row[0] cur.execute(""" UPDATE collection_info SET collector_name = ?, notes = ? WHERE id = ? """, (name, notes, collection_id)) else: cur.execute(""" INSERT INTO collection_info (collector_name, case_id, notes) VALUES (?, ?, ?) """, (name, case, notes)) collection_id = cur.lastrowid rows = [] for f in unique_files: md5 = self.compute_md5(f) rows.append((collection_id, f, md5)) cur.executemany(""" INSERT INTO collection_paths (collection_id, file_path, md5_hash) VALUES (?, ?, ?) """, rows) conn.commit() conn.close() QMessageBox.information(self, "Done", f"ZIP created and collection paths logged:\n{zip_path}") self.load_cases() except Exception as e: QMessageBox.critical(self, "Error", str(e)) # ---------------- LOAD CASES ---------------- def load_cases(self): conn = sqlite3.connect(DB_PATH) cur = conn.cursor() cur.execute(""" SELECT id, case_id, collector_name, notes, created_at FROM collection_info ORDER BY id DESC """) rows = cur.fetchall() conn.close() self.cases_table.setRowCount(0) for r, row in enumerate(rows): self.cases_table.insertRow(r) for c, value in enumerate(row): item = QTableWidgetItem(str(value) if value is not None else "") self.cases_table.setItem(r, c, item) def on_case_selected(self, row, column): item = self.cases_table.item(row, 0) if not item: return collection_id = item.text() self.load_files_for_collection(collection_id) self.browser_tabs.setCurrentWidget(self.files_tab) # ---------------- LOAD FILES ---------------- def load_files_for_collection(self, collection_id): conn = sqlite3.connect(DB_PATH) cur = conn.cursor() cur.execute(""" SELECT id, file_path, md5_hash FROM collection_paths WHERE collection_id = ? ORDER BY id ASC """, (collection_id,)) rows = cur.fetchall() conn.close() self.files_table.setRowCount(0) for r, row in enumerate(rows): self.files_table.insertRow(r) for c, value in enumerate(row): item = QTableWidgetItem(str(value) if value is not None else "") self.files_table.setItem(r, c, item) # ---------------- FILE PREVIEW ---------------- def on_file_selected(self, row, column): item = self.files_table.item(row, 1) if not item: return file_path = item.text() self.load_file_preview(file_path) self.browser_tabs.setCurrentWidget(self.preview_tab) def load_file_preview(self, file_path): self.preview_text.clear() if not os.path.exists(file_path): self.preview_text.setPlainText(f"File not found:\n{file_path}") return try: size = os.path.getsize(file_path) with open(file_path, "rb") as f: content = f.read(PREVIEW_MAX_BYTES) is_text = b"\x00" not in content if is_text: try: text = content.decode("utf-8", errors="replace") except: text = content.decode("latin-1", errors="replace") header = ( f"Previewing first {len(content)} bytes of {size} total.\n" f"Path: {file_path}\n\n" ) self.preview_text.setPlainText(header + text) else: hex_preview = content[:256].hex(" ") header = ( f"Binary file detected.\n" f"Showing hex of first {min(len(content), 256)} bytes " f"out of {size} total.\n" f"Path: {file_path}\n\n" ) self.preview_text.setPlainText(header + hex_preview) except Exception as e: self.preview_text.setPlainText( f"Error reading file:\n{file_path}\n\n{e}" ) if __name__ == "__main__": app = QApplication(sys.argv) window = LazyFileTree() window.show() sys.exit(app.exec())