Track error registry entries and reprocess option

config/app.yaml

@@ -64,3 +64,5 @@ logging:
 registry:
   # Path to processed files registry
   file: "out/processed_registry.json"
+  # Allow reprocessing files that previously failed
+  allow_reprocess_errors: false

src/config.py

@@ -232,6 +232,10 @@ class Config:
         if "registry" in config:
             registry = cls._require_dict(config, "registry", path)
             cls._require_str(registry, "registry.file", path, allow_empty=False)
+            if "allow_reprocess_errors" in registry:
+                cls._require_bool(
+                    registry, "registry.allow_reprocess_errors", path
+                )
 
     @classmethod
     def _validate_brand_aliases(cls, data: Dict[str, Any], path: Path) -> None:

src/processor.py

@@ -54,6 +54,10 @@ class FileProcessor:
         
         registry_file = Path(config.get('registry.file', 'out/processed_registry.json'))
         self.registry = Registry(registry_file)
+        self.allow_reprocess_errors = config.get(
+            'registry.allow_reprocess_errors',
+            False,
+        )
         
         self.reporter = Reporter(self.out_dir / 'run_report.ndjson')
         
@@ -84,39 +88,60 @@ class FileProcessor:
             file_hash = compute_file_hash(file_path)
             
             # Check if already processed
-            if self.registry.is_processed(file_hash):
-                self.logger.info(
-                    f"File already processed (hash: {file_hash[:8]}), skipping",
-                    extra={'file': filename, 'sha': file_hash[:8], 'status': 'skipped'}
-                )
-                
-                # Move to processed directory anyway
-                existing_entry = self.registry.get_entry(file_hash)
-                n_records = existing_entry.get('n_records', 0) if existing_entry else 0
-                
-                processing_time = time.time() - start_time
-                self.reporter.write_report(
-                    filename=filename,
-                    file_hash=file_hash,
-                    status='skipped_duplicate',
-                    n_rows=n_records,
-                    n_added=0,
-                    n_skipped=n_records,
-                    n_conflicts=0,
-                    processing_time=processing_time
-                )
-                
-                # Move to processed with existing name
-                dest_name = existing_entry.get('processed_name', filename) if existing_entry else filename
-                dest_path = self.processed_dir / dest_name
-                
-                if dest_path.exists():
-                    # Add timestamp to avoid collision
-                    dest_path = self.processed_dir / f"{dest_path.stem}_dup{dest_path.suffix}"
-                
-                shutil.move(str(file_path), str(dest_path))
-                
-                return 'skipped_duplicate', n_records
+            existing_entry = self.registry.get_entry(file_hash)
+            if existing_entry:
+                existing_status = existing_entry.get('status')
+                if existing_status == 'error' and self.allow_reprocess_errors:
+                    self.logger.info(
+                        f"Reprocessing previously failed file (hash: {file_hash[:8]})",
+                        extra={
+                            'file': filename,
+                            'sha': file_hash[:8],
+                            'status': 'reprocess_error',
+                        },
+                    )
+                else:
+                    self.logger.info(
+                        f"File already processed (hash: {file_hash[:8]}), skipping",
+                        extra={
+                            'file': filename,
+                            'sha': file_hash[:8],
+                            'status': 'skipped',
+                        },
+                    )
+                    
+                    # Move to processed directory anyway
+                    n_records = existing_entry.get('n_records', 0)
+                    
+                    processing_time = time.time() - start_time
+                    self.reporter.write_report(
+                        filename=filename,
+                        file_hash=file_hash,
+                        status='skipped_duplicate',
+                        n_rows=n_records,
+                        n_added=0,
+                        n_skipped=n_records,
+                        n_conflicts=0,
+                        processing_time=processing_time
+                    )
+                    
+                    # Move to processed with existing name
+                    dest_name = existing_entry.get(
+                        'processed_name',
+                        filename,
+                    )
+                    dest_path = self.processed_dir / dest_name
+                    
+                    if dest_path.exists():
+                        # Add timestamp to avoid collision
+                        dest_path = (
+                            self.processed_dir
+                            / f"{dest_path.stem}_dup{dest_path.suffix}"
+                        )
+                    
+                    shutil.move(str(file_path), str(dest_path))
+                    
+                    return 'skipped_duplicate', n_records
             
             # Detect file type
             file_type = detect_file_type(file_path)
@@ -205,6 +230,7 @@ class FileProcessor:
             # Handle error
             error_msg = str(e)
             self.logger.error(f"Error processing {filename}: {error_msg}", exc_info=True)
+            error_code = 'PARSE_ERROR'
             
             # Move to error directory
             try:
@@ -217,10 +243,21 @@ class FileProcessor:
                 n_records=0,
                 file_hash=file_hash,
                 is_error=True,
-                error_code='PARSE_ERROR'
+                error_code=error_code
             )
             
             dest_path = self.error_dir / error_name
+
+            if file_hash != 'unknown':
+                self.registry.add_entry(
+                    file_hash=file_hash,
+                    original_name=filename,
+                    processed_name=error_name,
+                    n_records=0,
+                    status='error',
+                    error_code=error_code,
+                    error_message=error_msg,
+                )
             
             try:
                 shutil.move(str(file_path), str(dest_path))

src/registry.py

@@ -60,7 +60,9 @@ class Registry:
         original_name: str,
         processed_name: str,
         n_records: int,
-        status: str = 'success'
+        status: str = 'success',
+        error_code: str | None = None,
+        error_message: str | None = None,
     ) -> None:
         """Add entry to registry.
         
@@ -70,13 +72,20 @@ class Registry:
             processed_name: Processed filename
             n_records: Number of records processed
             status: Processing status
+            error_code: Optional error code
+            error_message: Optional error message
         """
-        self._data[file_hash] = {
+        entry = {
             'original_name': original_name,
             'processed_name': processed_name,
             'n_records': n_records,
             'status': status,
         }
+        if error_code:
+            entry['error_code'] = error_code
+        if error_message:
+            entry['error_message'] = error_message
+        self._data[file_hash] = entry
         self._save()
     
     def get_entry(self, file_hash: str) -> Optional[Dict]: