# -*- coding: utf-8 -*- from qgis.core import ( QgsProcessing, QgsProcessingAlgorithm, QgsProcessingParameterVectorLayer, QgsProcessingParameterField, QgsProcessingParameterFolderDestination, QgsProcessingParameterString, QgsProcessingContext, QgsProcessingFeedback, QgsVectorLayer, QgsVectorFileWriter, QgsFeatureRequest, QgsExpression ) import os class ExcelSplitterByField(QgsProcessingAlgorithm): """ Splits an Excel layer into multiple Excel files based on a unique field value. """ INPUT_LAYER = 'INPUT_LAYER' SPLIT_FIELD = 'SPLIT_FIELD' OUTPUT_FOLDER = 'OUTPUT_FOLDER' def initAlgorithm(self, config=None): self.addParameter(QgsProcessingParameterVectorLayer( self.INPUT_LAYER, 'Input Excel Layer (or any Vector Layer)', types=[QgsProcessing.TypeVector] )) self.addParameter(QgsProcessingParameterField( self.SPLIT_FIELD, 'Field to split by', parentLayerParameterName=self.INPUT_LAYER )) self.addParameter(QgsProcessingParameterString( 'SHEET_NAME', 'Sheet Name (Optional - if input is a file with multiple sheets)', optional=True, defaultValue='' )) self.addParameter(QgsProcessingParameterFolderDestination( self.OUTPUT_FOLDER, 'Output Directory to save extracted files' )) def processAlgorithm(self, parameters, context: QgsProcessingContext, feedback: QgsProcessingFeedback): layer = self.parameterAsVectorLayer(parameters, self.INPUT_LAYER, context) if not layer: raise Exception("Invalid input layer.") sheet_name = self.parameterAsString(parameters, 'SHEET_NAME', context).strip() # If a sheet name is specified, try to reload the layer with that sheet source = layer.source() path = source.split('|')[0].replace('\\', '/') if path.lower().endswith(('.xlsx', '.xls', '.ods', '.gpkg')): from osgeo import ogr ds = ogr.Open(path) if ds: layers_in_file = [ds.GetLayer(i).GetName() for i in range(ds.GetLayerCount())] feedback.pushInfo(f"Detected layers/sheets in file: {', '.join(layers_in_file)}") # Identify sheets to merge (SSU sheets) target_sheets = [] if sheet_name: if sheet_name in layers_in_file: target_sheets = [sheet_name] else: feedback.pushWarning(f"Specific sheet '{sheet_name}' not found. Falling back to SSU merge.") if not target_sheets: # Auto-detect all SSU sheets target_sheets = [s for s in layers_in_file if 'SSU' in s.upper()] if target_sheets: feedback.pushInfo(f"Merging {len(target_sheets)} sheets: {', '.join(target_sheets)}") # Load first sheet base_uri = f"{path}|layername={target_sheets[0]}" base_layer = QgsVectorLayer(base_uri, target_sheets[0], "ogr") if not base_layer.isValid(): feedback.pushWarning(f"Failed to load base sheet '{target_sheets[0]}'. Trying by index...") base_uri = f"{path}|layerid={layers_in_file.index(target_sheets[0])}" base_layer = QgsVectorLayer(base_uri, target_sheets[0], "ogr") if base_layer.isValid(): # Create a memory layer copy to allow merging layer = base_layer.materialize(QgsFeatureRequest()) if len(target_sheets) > 1: layer.startEditing() for sname in target_sheets[1:]: next_uri = f"{path}|layername={sname}" next_lyr = QgsVectorLayer(next_uri, sname, "ogr") if not next_lyr.isValid(): next_uri = f"{path}|layerid={layers_in_file.index(sname)}" next_lyr = QgsVectorLayer(next_uri, sname, "ogr") if next_lyr.isValid(): feedback.pushInfo(f" -> Adding {next_lyr.featureCount()} features from {sname}") # Note: This assumes fields match or are compatible layer.addFeatures(next_lyr.getFeatures()) else: feedback.pushWarning(f" -> Could not load sheet {sname}, skipping.") layer.commitChanges() feedback.pushInfo(f"Merge complete. Total features for extraction: {layer.featureCount()}") else: feedback.reportError("Could not load any valid sheets for merging.") else: feedback.pushWarning("No SSU sheets found to merge. Using default input layer.") else: feedback.pushWarning(f"Could not open file {path} to list sheets.") field_name = self.parameterAsString(parameters, self.SPLIT_FIELD, context) if layer.fields().indexFromName(field_name) == -1: raise Exception(f"Field '{field_name}' not found in the selected layer/file. Available fields: {', '.join([f.name() for f in layer.fields()])}") output_folder = self.parameterAsString(parameters, self.OUTPUT_FOLDER, context) if not os.path.exists(output_folder): os.makedirs(output_folder, exist_ok=True) # Optimized unique value extraction request = QgsFeatureRequest().setFlags(QgsFeatureRequest.NoGeometry).setSubsetOfAttributes([field_name], layer.fields()) unique_values = set() for f in layer.getFeatures(request): unique_values.add(f[field_name]) feedback.pushInfo(f"Found {len(unique_values)} unique values in field '{field_name}'") success_count = 0 for val in sorted(list(unique_values), key=lambda x: str(x)): if feedback.isCanceled(): break # Sanitize filename if val in [None, 'NULL', 'null', '']: val_str = "EMPTY_OR_NULL" expression = f'"{field_name}" IS NULL OR "{field_name}" = \'\'' else: # Normalize float to int if it's a whole number if isinstance(val, float) and val.is_integer(): val = int(val) val_str = str(val).strip() # Remove invalid filename characters for char in r'<>:"/\|?*': val_str = val_str.replace(char, '_') # Handle numeric vs string in expression quoted_val = QgsExpression.quotedValue(val) expression = f'"{field_name}" = {quoted_val}' feedback.pushInfo(f"Processing group: {val_str}") output_path = os.path.join(output_folder, f"{val_str}.xlsx") # Skip if input file is the same as output file to prevent lock/hang if 'path' in locals() and os.path.normcase(os.path.abspath(path)) == os.path.normcase(os.path.abspath(output_path)): feedback.pushInfo(f" - Input file is already the target file ({val_str}.xlsx), skipping split to prevent overwrite lock.") continue # Create a subset of features using materialize request = QgsFeatureRequest().setFilterExpression(expression) subset_layer = layer.materialize(request) subset_layer.setName("Sample SSU") if subset_layer.featureCount() == 0: feedback.pushInfo(f" - No features found for {val_str}, skipping.") continue # Configure Excel Export options options = QgsVectorFileWriter.SaveVectorOptions() options.driverName = "XLSX" options.fileEncoding = "UTF-8" options.layerName = "Sample SSU" # Write to file res = QgsVectorFileWriter.writeAsVectorFormatV3( subset_layer, output_path, context.transformContext(), options ) error, error_msg = res[0], res[1] if error != QgsVectorFileWriter.NoError: feedback.reportError(f" [ERROR] Failed to save {val_str}: {error_msg}") else: feedback.pushInfo(f" [OK] Saved to: {os.path.basename(output_path)}") success_count += 1 feedback.pushInfo(f"\n[DONE] Successfully created {success_count} Excel files in: {output_folder}") return {self.OUTPUT_FOLDER: output_folder} def name(self): return 'excel_splitter_by_field' def displayName(self): return 'Split Excel/Layer by Field' def group(self): return 'Excel Tools' def groupId(self): return 'excel_tools' def createInstance(self): return ExcelSplitterByField() def shortHelpString(self): return ("This algorithm takes a vector layer (e.g., a loaded Excel sheet) and splits " "it into multiple smaller Excel (.xlsx) files based on the unique values " "of a selected field. Each file will be named after the field value.")