Sync from GitHub 7acd8038042b484e9650a6acfffda15038b5f1f3

DEPLOYMENT_CHECKLIST.md

@@ -70,7 +70,7 @@ Cron auf demselben Host:
 - [x] EU-Cluster bereitgestellt, `.env`: `WEAVIATE_CLUSTER_URL` + `WEAVIATE_API_KEY` gesetzt
 - [x] `python main.py --weaviate checkhealth` → Connection ✓ OK
 - [x] `python main.py --weaviate init` → Collections `hsg_rag_content_de`/`_en` angelegt
-- [ ] **Datenimport abgeschlossen:** `python main.py --scrape --full_scrape`
+- [ ] **Datenimport abgeschlossen:** `python main.py --scrape full`
       (läuft; danach Objekt-Counts in beiden Collections plausibel prüfen — EN/embax nicht unterrepräsentiert)
 - [ ] `python main.py --weaviate checkhealth` → beide Collections ✓ OK
 - [ ] Stichprobe: Query "Was macht die HSG besonders?" liefert echte Chunks (keine `QUERY_EXCEPTION_MESSAGE`)
@@ -109,8 +109,8 @@ Cron auf demselben Host:
       ```
 - [ ] **Scraping-Refresh** — als Cron (robuster als Dauerprozess):
       ```
-      0 3 * * 1-6 cd <repo> && ./venv/bin/python main.py --scrape              >> logs/scrape.log 2>&1
-      0 2 * * 0   cd <repo> && ./venv/bin/python main.py --scrape --full_scrape >> logs/scrape.log 2>&1
+      0 3 * * 1-6 cd <repo> && ./venv/bin/python main.py --scrape simple >> logs/scrape.log 2>&1
+      0 2 * * 0   cd <repo> && ./venv/bin/python main.py --scrape full   >> logs/scrape.log 2>&1
       ```
       (Alternative: `python tools/scraping.py --init_sched` als systemd-Service — APScheduler intern.)
 - [ ] **Alert-Chain einmal testen:** Preis in `data/database/programme_facts.json` ändern →

README.md

@@ -154,8 +154,8 @@ python main.py --help
 Useful operational commands:
 
 ```bash
-python main.py --scrape
-python main.py --scrape --full_scrape
+python main.py --scrape simple
+python main.py --scrape full
 python main.py --imports path/to/file1 path/to/file2
 python main.py --weaviate checkhealth
 python main.py --weaviate init
@@ -167,7 +167,7 @@ Embedding model changes require a Weaviate collection rebuild and re-import:
 
 ```bash
 python main.py --weaviate redo
-python main.py --scrape
+python main.py --scrape full
 # plus python main.py --imports ... for any local source files you maintain
 ```
 

docs/deploy_readiness_checklist.md

@@ -75,7 +75,7 @@ This checklist reflects the GitHub `main` state at commit `c0462c1b7c5074af682ec
 - Verify admissions handover path
 - Verify booking widget visibility
 - If imports are part of rollout:
-  - verify `--scrape`
+  - verify `--scrape simple` and `--scrape full`
   - verify `--imports`
 - If admin operations are required:
   - verify the DB app

docs/weaviate_database_setup.md

@@ -25,7 +25,7 @@ When changing embedding model, tokenizer, or vector dimensions, rebuild the coll
 
 ```bash
 python main.py --weaviate redo
-python main.py --scrape
+python main.py --scrape full
 ```
 
 Run `python main.py --imports ...` afterward for any local documents that are

main.py

@@ -14,18 +14,18 @@ def logging_startup():
     return get_logger('main_module')
 
 
-def run_scraper(full_scrape: bool = False) -> None:
+def run_scraper(scrape_type: str) -> None:
     """
     Run the scraper to collect program data.
 
     Args:
-        full_scrape: Whether to ignore timestamps and scrape all pages.
+        scrape_type: Whether to ignore timestamps and scrape all pages.
     """
     from src.pipeline.pipeline import ImportPipeline
     logger = logging_startup()
 
     logger.info("Running scraper...")
-    ImportPipeline().scrape_website(scrape_all=full_scrape)
+    ImportPipeline().scrape_website(scrape_all=True if scrape_type == 'full' else False)
     logger.info("Scraping completed.")
 
 
@@ -89,10 +89,8 @@ def parse_args():
     parser = argparse.ArgumentParser(description="University of St. Gallen Executive Education RAG Chatbot")
 
     # Add arguments
-    parser.add_argument("--scrape", action="store_true",
+    parser.add_argument("--scrape", type=str, choices=['simple', 'full'],
                         help="Scrapes the data from the HSG website and imports it into the database")
-    parser.add_argument("--full_scrape", action="store_true",
-                        help="When used with --scrape, ignores timestamps and scrapes all pages")
     parser.add_argument("--imports", nargs="+", help="Runs the data importing pipeline for the provided files")
 
     parser.add_argument("--weaviate", type=str, choices=['init', 'delete', 'redo', 'checkhealth', 'backup', 'restore'],
@@ -118,7 +116,7 @@ def main():
 
     # Run the specified components
     if args.scrape:
-        run_scraper(args.full_scrape)
+        run_scraper(args.scrape)
 
     if args.imports:
         run_importer(args.imports)

requirements.txt

@@ -30,7 +30,7 @@ beautifulsoup4>=4.14.3
 fake-useragent>=1.5.1
 
 # Weaviate Vector DB 
-weaviate-client>=4.16.9
+weaviate-client>=4.16.9,<5
 PyYAML>=6.0
 
 # Scheduling

src/database/weavservice.py

@@ -1,7 +1,9 @@
 from functools import reduce
 import weaviate as wvt
-import datetime, os
+import datetime, json, os
+from dataclasses import dataclass
 from threading import Event, Lock, RLock, Thread
+from typing import Any
 
 from time import perf_counter, sleep
 from weaviate.classes.config import Configure, Property, DataType
@@ -10,6 +12,7 @@ from weaviate.collections.collection import Collection
 from weaviate.classes.init import AdditionalConfig, Timeout
 from weaviate.classes.query import Filter
 from weaviate.config import AdditionalConfig
+from weaviate.util import generate_uuid5
 
 from ..utils.logging import get_logger
 from ..config import config
@@ -19,6 +22,7 @@ logger = get_logger("weaviate_service")
 
 _get_collection_name = lambda lang: f'{config.weaviate.WEAVIATE_COLLECTION_BASENAME}_{lang}'
 _collection_names = [_get_collection_name(lang) for lang in config.get('AVAILABLE_LANGUAGES')]
+_IMPORT_ID_VERSION = 1
 
 
 def _default_properties() -> list[Property]:
@@ -32,6 +36,26 @@ def _default_properties() -> list[Property]:
     ]
 
 
+class ImportBatchError(RuntimeError):
+    """Raised when Weaviate rejects any object in a strict batch import."""
+
+
+@dataclass(frozen=True)
+class PreparedImportRow:
+    uuid: str
+    properties: dict[str, Any]
+    vector: dict[str, list[float]]
+
+
+@dataclass
+class SourceReconciliationSummary:
+    source: str
+    expected: int = 0
+    retained: int = 0
+    inserted: int = 0
+    deleted: int = 0
+
+
 class WeaviateService:
     """
     Provides an interface for interacting with the Weaviate vector database.
@@ -323,74 +347,270 @@ class WeaviateService:
         return client.collections.use(collection_name), collection_name
 
 
-    def batch_import(self, data_rows: list, lang: str) -> list:
-        """
-        Perform a batch import of multiple objects into the current collection.
+    @staticmethod
+    def _deterministic_object_uuid(data_row: dict[str, Any], lang: str) -> str:
+        identity = {
+            "version": _IMPORT_ID_VERSION,
+            "lang": lang,
+            "source": data_row.get("source", ""),
+            "document_id": data_row.get("document_id", ""),
+            "chunk_id": data_row.get("chunk_id", ""),
+            "programs": sorted(data_row.get("programs") or []),
+        }
+        return generate_uuid5(
+            json.dumps(identity, sort_keys=True, separators=(",", ":")),
+            namespace="hsg-rag-import",
+        )
 
-        Args:
-            data_rows (list): List of dictionaries representing the data rows to import.
-            lang (str, optional): Language collection to use. If not provided, uses the current one.
 
-        Returns:
-            list[dict]: List of failed imports with error details, if any.
+    def prepare_batch_import(
+        self,
+        data_rows: list[dict[str, Any]],
+        lang: str,
+    ) -> list[PreparedImportRow]:
+        """Generate all embeddings and deterministic UUIDs without writing to Weaviate."""
+        prepared = []
+        batch_size = max(1, config.processing.EMBEDDING_BATCH_SIZE)
+
+        for start_idx in range(0, len(data_rows), batch_size):
+            batch_rows = list(
+                enumerate(
+                    data_rows[start_idx:start_idx + batch_size],
+                    start=start_idx,
+                )
+            )
+            vectors = self._embed_batch_vectors(batch_rows)
+            prepared.extend(
+                PreparedImportRow(
+                    uuid=self._deterministic_object_uuid(data_row, lang),
+                    properties=data_row,
+                    vector=vector,
+                )
+                for (_, data_row), vector in zip(batch_rows, vectors)
+            )
+
+        return prepared
+
+
+    def _write_prepared_rows(
+        self,
+        prepared_rows: list[PreparedImportRow],
+        lang: str,
+    ) -> None:
+        if not prepared_rows:
+            return
 
-        Raises:
-            If no active collection is available or a connection error was catched.
-        """
         collection, collection_name = self._select_collection(lang)
         if collection is None:
-            logger.error("No working collection selected!")
-            return []
-
-        import_errors = []
-        logger.info(f"Batch importing {len(data_rows)} rows into {collection_name}")
+            raise RuntimeError(f"No working collection selected for language '{lang}'")
 
+        logger.info(f"Batch importing {len(prepared_rows)} rows into {collection_name}")
         batch_size = max(1, config.processing.EMBEDDING_BATCH_SIZE)
         max_attempts = 2
 
-        def _import_batch(batch_rows: list[tuple[int, dict]]) -> None:
-            vectors = self._embed_batch_vectors(batch_rows)
-            with collection.batch.fixed_size(batch_size=batch_size, concurrent_requests=1) as batch:
-                for (idx, data_row), vector in zip(batch_rows, vectors):
-                    try:
-                        batch.add_object(properties=data_row, vector=vector)
-                    except Exception as e:
-                        import_errors.append({'index': idx, 'chunk_id': data_row['chunk_id'], 'error': str(e)})
-
-                    if idx % 20 == 0 and idx > 0:
-                        if batch.number_errors > 0:
-                            logger.info(f"Failed imports at index {idx}: {batch.number_errors}")
-
-        try:
-            with self._client_lock:
-                for start_idx in range(0, len(data_rows), batch_size):
-                    batch_rows = list(enumerate(data_rows[start_idx:start_idx + batch_size], start=start_idx))
-                    for attempt in range(1, max_attempts + 1):
-                        try:
-                            _import_batch(batch_rows)
-                            break
-                        except Exception as e:
-                            if attempt == max_attempts:
-                                raise e
+        with self._client_lock:
+            for start_idx in range(0, len(prepared_rows), batch_size):
+                rows = prepared_rows[start_idx:start_idx + batch_size]
+                last_error = None
 
+                for attempt in range(1, max_attempts + 1):
+                    try:
+                        failed_before = list(
+                            getattr(collection.batch, "failed_objects", [])
+                        )
+                        with collection.batch.fixed_size(
+                            batch_size=batch_size,
+                            concurrent_requests=1,
+                        ) as batch:
+                            for row in rows:
+                                batch.add_object(
+                                    uuid=row.uuid,
+                                    properties=row.properties,
+                                    vector=row.vector,
+                                )
+
+                        failed_after = list(
+                            getattr(collection.batch, "failed_objects", [])
+                        )
+                        new_failures = (
+                            failed_after[len(failed_before):]
+                            if len(failed_after) >= len(failed_before)
+                            else failed_after
+                        )
+                        error_count = max(
+                            getattr(batch, "number_errors", 0),
+                            len(new_failures),
+                        )
+                        if error_count:
+                            details = "; ".join(str(error) for error in new_failures[:3])
+                            raise ImportBatchError(
+                                f"Weaviate rejected {error_count} object(s)"
+                                + (f": {details}" if details else "")
+                            )
+                        last_error = None
+                        break
+                    except Exception as error:
+                        last_error = error
+                        if attempt < max_attempts:
                             logger.warning(
                                 "Batch import failed for rows %s-%s; retrying once: %s",
                                 start_idx,
-                                start_idx + len(batch_rows) - 1,
-                                e,
+                                start_idx + len(rows) - 1,
+                                error,
                             )
                             sleep(1)
 
-            self._last_query_time = perf_counter()
-            logger.info(f"Batch import finished. Total errors: {len(import_errors)}")
-            
+                if last_error is not None:
+                    raise ImportBatchError(
+                        f"Failed importing rows {start_idx}-{start_idx + len(rows) - 1} "
+                        f"into {collection_name}: {last_error}"
+                    ) from last_error
+
+        self._last_query_time = perf_counter()
+
+
+    def batch_import(self, data_rows: list, lang: str) -> list:
+        """
+        Perform a strict, deterministic batch import into the current collection.
+
+        Args:
+            data_rows (list): List of dictionaries representing the data rows to import.
+            lang (str, optional): Language collection to use. If not provided, uses the current one.
+
+        Returns:
+            An empty list on success, retained for compatibility.
+
+        Raises:
+            ImportBatchError: If any synchronous or asynchronous object import fails.
+        """
+        try:
+            prepared_rows = self.prepare_batch_import(data_rows, lang)
+            self._write_prepared_rows(prepared_rows, lang)
+            logger.info("Batch import finished successfully")
         except Exception as e:
             if 'connection' in str(e).lower():
                 logger.error(f"Connection error during batch import: {e}")
                 self._client = None
             raise e
 
-        return import_errors
+        return []
+
+
+    def _collect_source_objects(
+        self,
+        lang: str,
+        source: str,
+    ) -> dict[str, dict[str, Any]]:
+        collection, _ = self._select_collection(lang)
+        if collection is None:
+            raise RuntimeError(f"No working collection selected for language '{lang}'")
+
+        objects = {}
+        offset = 0
+        page_size = 100
+        filters = Filter.by_property("source").equal(source)
+
+        while True:
+            with self._client_lock:
+                response = collection.query.fetch_objects(
+                    limit=page_size,
+                    offset=offset,
+                    filters=filters,
+                    return_properties=["source", "chunk_id", "document_id", "programs"],
+                )
+            page = response.objects
+            for obj in page:
+                objects[str(obj.uuid)] = obj.properties or {}
+
+            if len(page) < page_size:
+                break
+            offset += page_size
+
+        return objects
+
+
+    def source_object_count(self, source: str) -> int:
+        return sum(
+            len(self._collect_source_objects(lang, source))
+            for lang in config.get('AVAILABLE_LANGUAGES')
+        )
+
+
+    def reconcile_source(
+        self,
+        source: str,
+        rows_by_language: dict[str, list[dict[str, Any]]],
+    ) -> SourceReconciliationSummary:
+        """Insert and verify a complete source replacement before deleting old objects."""
+        languages = config.get('AVAILABLE_LANGUAGES')
+        prepared_by_language = {
+            lang: self.prepare_batch_import(rows_by_language.get(lang, []), lang)
+            for lang in languages
+        }
+        expected_ids = {
+            lang: {row.uuid for row in prepared_by_language[lang]}
+            for lang in languages
+        }
+
+        existing_before = {
+            lang: self._collect_source_objects(lang, source)
+            for lang in languages
+        }
+        summary = SourceReconciliationSummary(
+            source=source,
+            expected=sum(len(ids) for ids in expected_ids.values()),
+            retained=sum(
+                len(expected_ids[lang] & set(existing_before[lang]))
+                for lang in languages
+            ),
+        )
+
+        for lang in languages:
+            missing_rows = [
+                row for row in prepared_by_language[lang]
+                if row.uuid not in existing_before[lang]
+            ]
+            self._write_prepared_rows(missing_rows, lang)
+            summary.inserted += len(missing_rows)
+
+        verified = {
+            lang: self._collect_source_objects(lang, source)
+            for lang in languages
+        }
+        for lang in languages:
+            missing_ids = expected_ids[lang] - set(verified[lang])
+            if missing_ids:
+                raise ImportBatchError(
+                    f"Verification failed for source '{source}' in language '{lang}': "
+                    f"{len(missing_ids)} expected object(s) are missing"
+                )
+
+        for lang in languages:
+            obsolete_ids = set(verified[lang]) - expected_ids[lang]
+            if not obsolete_ids:
+                continue
+
+            collection, collection_name = self._select_collection(lang)
+            if collection is None:
+                raise RuntimeError(f"No working collection selected for language '{lang}'")
+
+            for object_uuid in obsolete_ids:
+                if not collection.data.delete_by_id(object_uuid):
+                    raise RuntimeError(
+                        f"Failed deleting obsolete object {object_uuid} from {collection_name}"
+                    )
+                summary.deleted += 1
+
+        self._last_query_time = perf_counter()
+        logger.info(
+            "Reconciled source %s: expected=%s retained=%s inserted=%s deleted=%s",
+            source,
+            summary.expected,
+            summary.retained,
+            summary.inserted,
+            summary.deleted,
+        )
+        return summary
     
 
     @staticmethod

src/pipeline/pipeline.py

@@ -1,8 +1,9 @@
 from .utils import *
 from .processors import *
 from ..scraping.scraper import Scraper
+from ..scraping.types import ScrapeManifest
 
-from ..database.weavservice  import WeaviateService
+from ..database.weavservice import SourceReconciliationSummary, WeaviateService
 from ..utils.logging import get_logger
 from ..utils.tools import call_with_exponential_backoff
 from ..config import config
@@ -13,8 +14,7 @@ implogger  = get_logger("import_pipeline")
 
 class ImportPipeline:
     """
-    Main pipeline class responsible for importing website and local documents
-    into the database with deduplication and language-based organization.
+    Import website and local documents with source-level safe replacement.
     """
 
     def __init__(
@@ -23,33 +23,43 @@ class ImportPipeline:
         deduplication_callback = None,
     ) -> None:
         """
-        Initialize the import pipeline with optional callbacks for logging and deduplication.
-
-        This sets up the processors for websites and documents and recieves existing chunk IDs 
-        from the database for deduplication purposes.
+        Initialize processors and optional UI callbacks.
 
         Args:
             logging_callback (callable, optional): A callback function for logging progress.
                 Defaults to a placeholder if not provided.
-            deduplication_callback (callable, optional): A callback function for handling
-                deduplication decisions. Defaults to a placeholder if not provided.
+            deduplication_callback (callable, optional): For existing local sources,
+                returns True to replace the source or False to append without cleanup.
         """
         self._logging_callback = logging_callback or logging_callback_placeholder
-        self._deduplication_callback = deduplication_callback or deduplication_callback_placeholder
+        self._deduplication_callback = deduplication_callback
         self._docprocessor = DocumentProcessor()
         self._service      = WeaviateService()
-        self._ids          = self._service._collect_chunk_ids()
         
         implogger.info('Import pipeline initialization finished!')
     
 
-    def import_from_scraper(self, scraper_chunks: dict[str, dict]) -> None:
-        for lang, chunks in scraper_chunks.items():
-            if not chunks: continue
-
-            sources = list(set([chunk.get('source', '') for chunk in chunks]))
-            self._service.delete_chunks(lang, property_filters={'source': sources})
-            self._service.batch_import(data_rows=chunks, lang=lang)
+    def import_from_scraper(
+        self,
+        manifest: ScrapeManifest,
+    ) -> list[SourceReconciliationSummary]:
+        rows_by_source = {
+            source: {
+                lang: []
+                for lang in config.get('AVAILABLE_LANGUAGES')
+            }
+            for source in manifest.processed_sources
+        }
+        for lang, chunks in manifest.chunks_by_language.items():
+            for chunk in chunks:
+                source = chunk.get('source', '')
+                if source in rows_by_source:
+                    rows_by_source[source].setdefault(lang, []).append(chunk)
+
+        return [
+            self._service.reconcile_source(source, rows_by_source[source])
+            for source in manifest.processed_sources
+        ]
 
 
     def scrape_website(self, target_urls: list[str] | None = None, scrape_all: bool = False) -> None:
@@ -63,14 +73,14 @@ class ImportPipeline:
                 scraper = Scraper(scrape_all=scrape_all)
                 for target_url in target_urls:
                     self._logging_callback(f"Scraping target {target_url}...", 0)
-                    scraped_chunks = scraper.scrape_target(target_url)
-                    if not scraped_chunks:
+                    manifest = scraper.scrape_target(target_url)
+                    if not manifest:
                         self._logging_callback(f"No importable chunks scraped from {target_url}.", 100)
-                    else:
-                        self._logging_callback(f"Importing scraped chunks from {target_url}...", 90)
+                        continue
 
-                    self.import_from_scraper(scraped_chunks)
-                    scraper.delete_temp_merged_chunks(target_url)
+                    self._logging_callback(f"Importing scraped chunks from {target_url}...", 90)
+                    self.import_from_scraper(manifest)
+                    scraper.commit_scrape(manifest)
                     self._logging_callback(f"Finished scraping import for {target_url}.", 100)
             except Exception as e:
                 implogger.error(f"Scraping task was interrupted: {e}")
@@ -93,13 +103,14 @@ class ImportPipeline:
         scraper = Scraper(scrape_all=scrape_all)
         for url in urls:
             self._logging_callback(f"Scraping URL {url}...", 0)
-            scraped_chunks = scraper.scrape_target(url)
-            if not scraped_chunks:
+            manifest = scraper.scrape_target(url)
+            if not manifest:
                 self._logging_callback(f"Failed to scrape URL {url}!", 100, failed=True)
                 continue
 
             self._logging_callback(f"Importing scraped chunks from {url}...", 90)
-            self.import_from_scraper(scraped_chunks)
+            self.import_from_scraper(manifest)
+            scraper.commit_scrape(manifest)
             self._logging_callback(f"Stored scraped chunks for {url}.", 100)
                  
         
@@ -122,102 +133,85 @@ class ImportPipeline:
             reset_collections (bool, optional): If True, reset the database collections before importing.
                 Defaults to False.
         """
-        chunks = self._pipeline(paths, self._docprocessor, reset_collections)
+        results = self._process_documents(paths, fail_fast=reset_collections)
 
         if reset_collections:
-            self._logging_callback('Resetting database collections...', 60)
+            chunks_by_language = {
+                lang: []
+                for lang in config.get('AVAILABLE_LANGUAGES')
+            }
+            for result in results:
+                chunks_by_language[result.lang].extend(result.chunks)
+
+            prepared_by_language = {
+                lang: self._service.prepare_batch_import(chunks, lang)
+                for lang, chunks in chunks_by_language.items()
+            }
+            self._logging_callback(
+                'Resetting database collections (destructive operation)...',
+                60,
+            )
             self._service._reset_collections()
-        
-        self._logging_callback('Importing document chunks to database...', 90)
-        for lang, ch in chunks.items():
-            self._service.batch_import(data_rows=ch, lang=lang) 
+            for lang, prepared in prepared_by_language.items():
+                self._service._write_prepared_rows(prepared, lang)
+        else:
+            for result in results:
+                self._import_document_result(result)
 
         self._import_urls_via_scraper(urls, scrape_all=True)
 
         self._logging_callback(
-            f'Successfully imported {sum([len(ch) for ch in chunks.values()])} document chunks!',
+            f'Successfully imported {sum(len(result.chunks) for result in results)} document chunks!',
             100
         )
 
 
-    def _pipeline(
-            self, 
-            sources: list[str], 
-            processor: ProcessorBase,
-            reset_collections: bool,
-        ) -> dict:
-        """
-        Internal pipeline to process a list of sources using a given processor.
-
-        Handles processing, deduplication (if not resetting), and organizes unique chunks by language.
-        If no new unique data is found, logs a warning and returns empty chunks.
-
-        Args:
-            sources (list[str]): List of sources (paths or URLs) to process.
-            processor (ProcessorBase): The processor instance to use for handling sources.
-            reset_collections (bool): If True, skip deduplication.
-
-        Returns:
-            dict: A dictionary mapping languages to lists of unique chunk dictionaries.
-        """
-        unique_chunks = {lang: [] for lang in config.get('AVAILABLE_LANGUAGES')}
-
-        sources = [s for s in (sources or []) if s != ""]
-        if not sources:
-            return unique_chunks
-        
-        for source in sources:
+    def _process_documents(
+        self,
+        sources: list[str] | None,
+        fail_fast: bool = False,
+    ) -> list[ProcessingResult]:
+        results = []
+        for source in [s for s in (sources or []) if s]:
             self._logging_callback(f'Starting pipeline for {source}...', 0)
-            result = processor.process(source)
+            result = self._docprocessor.process(source)
 
             if not result.chunks:
                 implogger.error(f"Failed to process {source}!")
                 self._logging_callback(f"Failed to process {source}!", 100, result, failed=True)
+                if fail_fast:
+                    raise RuntimeError(
+                        f"Aborting destructive collection reset because '{source}' "
+                        "did not produce importable chunks"
+                    )
                 continue
-            
-            if not reset_collections:
-                self._deduplicate(result)
 
-            self._logging_callback(f'Storing chunks for {source}...', 100, result)
-            unique_chunks[result.lang].extend(result.chunks)
+            results.append(result)
+            self._logging_callback(f'Prepared chunks for {source}.', 70, result)
 
-        if all([len(chunks) == 0 for chunks in unique_chunks.values()]):
+        if sources and not results:
             self._logging_callback('No new data could be extracted from these sources!', 100)
-            implogger.warning(f"File(s) provided for the insertion do not contain any unique information.")
+            implogger.warning("Provided files did not contain importable information.")
 
-        return unique_chunks
-        
+        return results
 
-    def _deduplicate(self, result: ProcessingResult) -> ProcessingResult:
-        """
-        Remove duplicate chunks based on chunks that are already stored in the database.
 
-        If all chunks are duplicates, invokes the deduplication callback to decide whether
-        to delete existing duplicates and reimport. Otherwise, returns only unique chunks.
-
-        Args:
-            result (ProcessingResult): The processing result containing document chunks.
+    def _import_document_result(
+        self,
+        result: ProcessingResult,
+    ) -> None:
+        existing_count = self._service.source_object_count(result.source)
+        should_replace = True
+        if existing_count and self._deduplication_callback:
+            should_replace = self._deduplication_callback(result.source, existing_count)
+
+        if existing_count and should_replace:
+            rows_by_language = {
+                lang: []
+                for lang in config.get('AVAILABLE_LANGUAGES')
+            }
+            rows_by_language[result.lang] = result.chunks
+            self._service.reconcile_source(result.source, rows_by_language)
+            return
 
-        Returns:
-            list[dict]: List of unique chunk dictionaries (or all if reimporting duplicates).
-        """
-        self._logging_callback('Performing deduplication...', 80)
-        unique_chunks = []
-        duplicate_ids = []
-        for chunk in result.chunks:
-            chunk_id = chunk['chunk_id']
-            if chunk_id in self._ids:
-                duplicate_ids.append(chunk_id)
-            else: 
-                unique_chunks.append(chunk)
-        
-        implogger.info(f"Found {len(duplicate_ids)} already existing IDs in {len(result.chunks)} collected chunks")
-        if duplicate_ids: 
-            implogger.info(f"Duplicates found! Calling deduplication callback...")
-            if self._deduplication_callback(result.source, len(duplicate_ids)):
-                implogger.info('Duplicated chunks will be reimported as new...')
-                self._service._delete_by_id(duplicate_ids) 
-                return result
-        
-        result.chunks = unique_chunks
-        return result
+        self._service.batch_import(result.chunks, result.lang)

src/scraping/scraper.py

@@ -69,7 +69,7 @@ class Scraper:
         os.makedirs(self._path.EXTRACTED_TEXT_OUTPUT, exist_ok=True)
 
 
-    def scrape_target(self, target_url: str) -> list[ChunkMetadata]:
+    def scrape_target(self, target_url: str) -> ScrapeManifest:
         if getattr(self, "_content_cleaner", None) is None or hasattr(self, "_scrape_all"):
             self._content_cleaner = ContentCleaner(getattr(self, "_scrape_all", True))
 
@@ -77,7 +77,7 @@ class Scraper:
         analyzed_domain = self._analyze_domain(target_url)
         if not analyzed_domain:
             logger.error(f"Failed to scrape target URL {target_url}")
-            return {}
+            return ScrapeManifest.empty(target_url)
 
         sitemap_urls = analyzed_domain.urls
         self._save_results(self._path.URLS_OUTPUT, 'sitemap_urls', sitemap_urls, target_url)
@@ -106,10 +106,14 @@ class Scraper:
         # Step 4: Load temp chunks first so resume works even when there are no new documents.
         temp_filename      = self._get_temp_chunks_filename(target_url)
         temp_merged_chunks = self._load_data(self._path.TEMP_CHUNKS_OUTPUT, temp_filename)
+        self._pending_timestamps = self._load_data(
+            self._path.TEMP_CHUNKS_OUTPUT,
+            self._get_temp_timestamps_filename(target_url),
+        )
 
         if not documents and not temp_merged_chunks:
             logger.info(f"No new content was scraped from the target URL {target_url}")
-            return {}
+            return ScrapeManifest.empty(target_url)
 
         tagged_documents = []
         # Step 5: Analyze the converted URLs
@@ -136,7 +140,11 @@ class Scraper:
 
         logger.info(f"Scraping finished for target URL '{target_url}'")
 
-        return chunk_metadatas.get('final', chunk_metadatas['merged'])
+        return ScrapeManifest(
+            target_url=target_url,
+            chunks_by_language=chunk_metadatas.get('final', {}),
+            processed_sources=sorted(chunk_metadatas.get('processed_sources', [])),
+        )
 
 
     def _analyze_domain(self, target_url: str) -> DomainAnalysisReport | None:
@@ -351,13 +359,17 @@ class Scraper:
 
         raw_chunks     = []
         deleted_chunks = []
-        merged_chunks, final_chunks = self._read_temp_chunks(temp_chunks, tagged_documents)
+        merged_chunks, final_chunks, restored_sources = self._read_temp_chunks(
+            temp_chunks,
+            tagged_documents,
+        )
         # Temp snapshots must only carry (a) URLs newly chunked in this session
         # (added by _store_temp_chunks) and (b) URLs whose restore failed and
         # which must survive for the next session. Successfully restored URLs
         # are finalized now — keeping them in temp would re-restore them on the
         # next resume and duplicate their chunks.
         self._active_temp_chunks = dict(getattr(self, '_unrestorable_temp_chunks', {}))
+        processed_sources = set(restored_sources)
 
         program_counter = self._build_program_counter_from_merged_chunks(merged_chunks)
 
@@ -368,6 +380,7 @@ class Scraper:
             program  = entry.tags.program
             language = entry.tags.language
             url = document.name
+            processed_sources.add(url)
             url_filename = self._normalizer.url_to_filename(url)
             
             program_counter[program] += 1
@@ -426,6 +439,7 @@ class Scraper:
             'merged':  merged_chunks,
             'deleted': deleted_chunks,
             'final':   final_chunks,
+            'processed_sources': processed_sources,
         }
     
 
@@ -433,7 +447,7 @@ class Scraper:
         self, 
         temp_chunks: dict[str, list[ChunkMetadata]], 
         tagged_documents: list[TaggedDocument]
-    ) -> set[list, list[dict]]:
+    ) -> tuple[list, dict[str, list[dict]], set[str]]:
         loaded_temp_chunks = temp_chunks.copy()
         prepared_temp_chunks = {lang: [] for lang in config.get('AVAILABLE_LANGUAGES', ['en', 'de'])}
 
@@ -443,6 +457,7 @@ class Scraper:
                 del loaded_temp_chunks[url]
         
         restored_temp_chunks = []
+        restored_sources = set()
         # URLs whose chunks could not be finalized; they must stay in the temp
         # store so the next session can re-scrape them (read in _collect_chunks).
         self._unrestorable_temp_chunks: dict[str, list[ChunkMetadata]] = {}
@@ -465,24 +480,32 @@ class Scraper:
                     prepared_temp_chunks[lang].extend(prepared_chunks[lang])
 
             restored_temp_chunks.extend(chunks)
+            restored_sources.add(url)
             incupd_logger.info(f"Restored {len(chunks)} chunks for URL {url} from temp")
         
-        return restored_temp_chunks, prepared_temp_chunks
+        return restored_temp_chunks, prepared_temp_chunks, restored_sources
 
 
     def _store_temp_chunks(self, target_url: str, url: str, chunks: list[ChunkMetadata]) -> None:
-        self._url_timestamps[url] = self._url_temp_timestamps[url]
-        
         temp_chunks = getattr(self, '_active_temp_chunks', {}).copy()
         temp_chunks[url] = chunks
         self._active_temp_chunks = temp_chunks
 
+        pending_timestamps = getattr(self, '_pending_timestamps', {}).copy()
+        if url in self._url_temp_timestamps:
+            pending_timestamps[url] = self._url_temp_timestamps[url]
+        self._pending_timestamps = pending_timestamps
+
         self._save_results(
             self._path.TEMP_CHUNKS_OUTPUT,
             self._get_temp_chunks_filename(target_url),
             _TempChunkSnapshot(temp_chunks),
         )
-        self._save_results(self._path.SCRAPING_OUTPUT, 'url_timestamps', self._url_timestamps)
+        self._save_results(
+            self._path.TEMP_CHUNKS_OUTPUT,
+            self._get_temp_timestamps_filename(target_url),
+            pending_timestamps,
+        )
         
         incupd_logger.info(f"Stored {len(chunks)} chunks in temp for URL {url}")
 
@@ -530,14 +553,70 @@ class Scraper:
         return self._normalizer.url_to_filename(target_url) + '_merged_chunks'
 
 
-    def delete_temp_merged_chunks(self, target_url: str) -> None:
-        temp_path = os.path.join(
+    def _get_temp_timestamps_filename(self, target_url: str) -> str:
+        return self._normalizer.url_to_filename(target_url) + '_pending_timestamps'
+
+
+    def commit_scrape(self, manifest: ScrapeManifest) -> None:
+        """Commit scraper checkpoints after all database sources are reconciled."""
+        processed_sources = set(manifest.processed_sources)
+        pending_timestamps = getattr(self, '_pending_timestamps', {})
+        for url in processed_sources:
+            timestamp = pending_timestamps.get(url)
+            if timestamp:
+                self._url_timestamps[url] = timestamp
+
+        self._save_results(self._path.SCRAPING_OUTPUT, 'url_timestamps', self._url_timestamps)
+
+        remaining_chunks = {
+            url: chunks
+            for url, chunks in getattr(self, '_active_temp_chunks', {}).items()
+            if url not in processed_sources
+        }
+        chunks_path = os.path.join(
             self._path.TEMP_CHUNKS_OUTPUT,
-            self._get_temp_chunks_filename(target_url) + '.json'
+            self._get_temp_chunks_filename(manifest.target_url) + '.json',
         )
-        if os.path.exists(temp_path):
-            os.remove(temp_path)
-            incupd_logger.info(f"Deleted temp merged chunks file '{temp_path}'")
+        if os.path.exists(chunks_path):
+            os.remove(chunks_path)
+        if remaining_chunks:
+            self._save_results(
+                self._path.TEMP_CHUNKS_OUTPUT,
+                self._get_temp_chunks_filename(manifest.target_url),
+                _TempChunkSnapshot(remaining_chunks),
+            )
+
+        remaining_timestamps = {
+            url: timestamp
+            for url, timestamp in pending_timestamps.items()
+            if url not in processed_sources
+        }
+        timestamps_path = os.path.join(
+            self._path.TEMP_CHUNKS_OUTPUT,
+            self._get_temp_timestamps_filename(manifest.target_url) + '.json',
+        )
+        if remaining_timestamps:
+            self._save_results(
+                self._path.TEMP_CHUNKS_OUTPUT,
+                self._get_temp_timestamps_filename(manifest.target_url),
+                remaining_timestamps,
+            )
+        elif os.path.exists(timestamps_path):
+            os.remove(timestamps_path)
+
+        self._active_temp_chunks = remaining_chunks
+        self._pending_timestamps = remaining_timestamps
+
+
+    def delete_temp_merged_chunks(self, target_url: str) -> None:
+        for filename in (
+            self._get_temp_chunks_filename(target_url),
+            self._get_temp_timestamps_filename(target_url),
+        ):
+            temp_path = os.path.join(self._path.TEMP_CHUNKS_OUTPUT, filename + '.json')
+            if os.path.exists(temp_path):
+                os.remove(temp_path)
+                incupd_logger.info(f"Deleted scraper temp file '{temp_path}'")
 
 
     def _get_etag(self, url: str) -> str | None:
@@ -628,15 +707,17 @@ class Scraper:
         logger.info(f"Fetching head for URL '{url}'...")
 
         etag = self._get_etag(url)
-        response = call_with_exponential_backoff(fetch_head, args=(url, etag), delay=crawl_delay)
-        if response['status'] == 'FAIL':
-            logger.warning(f"Failed to fetch head for URL {url}: {response['last_error']}! Skipping...")
-            return ScrapingResult(status=ScrapingStatus.REJECTED)
-
-        fetch_result = response['result']
-        validation = self._is_fetch_valid(url, visited_urls, fetch_result) 
-        if validation != ScrapingStatus.OK:
-            return ScrapingResult(status=validation)
+        try:
+            fetch_result = fetch_head(url, etag)
+        except Exception as head_error:
+            logger.warning(
+                f"HEAD request failed for URL {url}: {head_error}. "
+                "Falling back to a GET request..."
+            )
+        else:
+            validation = self._is_fetch_valid(url, visited_urls, fetch_result)
+            if validation != ScrapingStatus.OK:
+                return ScrapingResult(status=validation)
 
         response = call_with_exponential_backoff(fetch_url, args=(url, etag), delay=crawl_delay)
         if response['status'] == 'FAIL':
@@ -714,6 +795,12 @@ class Scraper:
                 for url, ts in results.items():
                     results_dict[url] = dataclass_to_dict(ts)
 
+            case _ if filename.endswith('_pending_timestamps'):
+                results_dict = {
+                    url: dataclass_to_dict(timestamp)
+                    for url, timestamp in results.items()
+                }
+
             case 'url_priorities':
                 for prio, urls in results.items():
                     prev = set(results_dict.get(prio, []))
@@ -769,6 +856,12 @@ class Scraper:
                     incupd_logger.debug(f"Loaded {len(loaded_data)} temp merged chunks")
                     return loaded_data
 
+                case _ if filename.endswith('_pending_timestamps'):
+                    for url, timestamp in loaded_data.items():
+                        loaded_data[url] = dict_to_dataclass(timestamp, UrlTimestamps)
+                    incupd_logger.debug(f"Loaded {len(loaded_data)} pending timestamps")
+                    return loaded_data
+
                 case _:
                     incupd_logger.info(f"Loaded data '{filename}'")
                     return loaded_data

src/scraping/types.py

@@ -1,6 +1,7 @@
 from dataclasses import asdict, dataclass, is_dataclass 
 from datetime import datetime
 from enum import Enum
+from typing import Any
 
 from docling_core.types.doc.document import DoclingDocument
 
@@ -99,6 +100,22 @@ class DocumentAnalysisReport:
     tagged_documents: list[TaggedDocument]
 
 
+@dataclass
+class ScrapeManifest:
+    """Prepared database rows and authoritative sources from one scrape target."""
+
+    target_url: str
+    chunks_by_language: dict[str, list[dict[str, Any]]]
+    processed_sources: list[str]
+
+    def __bool__(self) -> bool:
+        return bool(self.processed_sources)
+
+    @classmethod
+    def empty(cls, target_url: str) -> "ScrapeManifest":
+        return cls(target_url=target_url, chunks_by_language={}, processed_sources=[])
+
+
 def dataclass_to_dict(obj) -> dict:
     if not is_dataclass(obj): return obj
     return asdict(obj, dict_factory=lambda items: {

src/scraping/utils.py

@@ -160,8 +160,8 @@ def fetch_head(url: str, etag: str | None = None) -> FetchResult:
             etag          = response.headers.get('ETag')
         )
     except Exception as e:
-        logger.exception(f"Head fetch failed: {url}")
-        raise e 
+        logger.warning(f"Head fetch failed for '{url}': {e}")
+        raise
 
 
 def fetch_url(url: str, etag: str | None = None) -> dict:

tests/scraping/test_scraping.py

@@ -1,7 +1,8 @@
 import pytest, os
 from datetime import datetime, timedelta
+from types import SimpleNamespace
 
-from src.scraping.types import UrlTimestamps
+from src.scraping.types import FetchResult, ScrapingStatus, UrlTimestamps
 from src.scraping.url_normalizer import UrlNormalizer
 from src.scraping.scraper import Scraper
 from src.utils.logging import init_logging
@@ -71,6 +72,67 @@ class TestScrapingErrors:
         for url in scraper._url_timestamps:
             assert scraper._is_url_prioritized(url) == expected[url]
 
+    def test_head_timeout_falls_back_to_get(self, tmp_path, monkeypatch):
+        url = "https://embax.ch/contact/"
+        get_calls = []
+        scraper = Scraper.__new__(Scraper)
+        scraper._scrape_all = True
+        scraper._url_timestamps = {}
+        scraper._url_priorities = {}
+        scraper._normalizer = SimpleNamespace(
+            is_url_blacklisted=lambda _: False,
+            url_to_filename=lambda _: "embax-contact",
+        )
+        scraper._content_cleaner = SimpleNamespace(
+            clean_mobile_content=lambda html: html,
+            extract_urls=lambda document: [],
+            collect_repetitive_content=lambda document: None,
+        )
+        scraper._processor = SimpleNamespace(
+            process=lambda final_url, html: SimpleNamespace(name=final_url),
+            convert_to_txt=lambda document: "contact page",
+        )
+
+        def fail_head(*_):
+            raise TimeoutError("HEAD timed out")
+
+        def fetch_get(request_url, etag):
+            get_calls.append((request_url, etag))
+            return FetchResult(
+                final_url=request_url,
+                last_modified=None,
+                etag=None,
+                text="<html><body>Contact</body></html>",
+                page_hash="hash",
+            )
+
+        def no_wait_backoff(func, args=(), **_):
+            return {
+                "result": func(*args),
+                "retries": 0,
+                "last_error": None,
+                "status": "OK",
+            }
+
+        monkeypatch.setattr("src.scraping.scraper.fetch_head", fail_head)
+        monkeypatch.setattr("src.scraping.scraper.fetch_url", fetch_get)
+        monkeypatch.setattr(
+            "src.scraping.scraper.call_with_exponential_backoff",
+            no_wait_backoff,
+        )
+        monkeypatch.setattr(config.paths, "RAW_HTML_OUTPUT", str(tmp_path))
+        monkeypatch.setattr(config.paths, "RAW_TEXT_OUTPUT", str(tmp_path))
+
+        result = scraper._scrape_page(
+            url=url,
+            crawl_delay=0,
+            visited_urls=set(),
+        )
+
+        assert result.status == ScrapingStatus.OK
+        assert result.final_url == url
+        assert get_calls == [(url, None)]
+
 
 if __name__ == "__main__":
     pytest.main([__file__, '-v'])

tests/scraping/test_scraping_resume.py

@@ -5,7 +5,7 @@ from types import SimpleNamespace
 import pytest
 
 from src.scraping.scraper import Scraper
-from src.scraping.types import ChunkMetadata, UrlTimestamps
+from src.scraping.types import ChunkMetadata, ScrapeManifest, UrlTimestamps
 
 
 class DummyNormalizer:
@@ -188,7 +188,8 @@ class TestTempBehaviorComplete:
             'https://target.example/page-1',
             'https://target.example/page-2',
         ]
-        assert set(scraper._url_timestamps) == {
+        assert scraper._url_timestamps == {}
+        assert set(scraper._pending_timestamps) == {
             'https://target.example/page-1',
             'https://target.example/page-2',
         }
@@ -274,13 +275,18 @@ class TestTempBehaviorComplete:
                 'merged': merged,
                 'deleted': [],
                 'final': {'en': merged, 'de': []},
+                'processed_sources': {'https://target.example/page-1'},
             }
 
         scraper._collect_chunks = fake_collect_chunks
 
         result = scraper.scrape_target(target_url)
 
-        assert [chunk.source_url for chunk in result['en']] == ['https://target.example/page-1']
+        assert isinstance(result, ScrapeManifest)
+        assert [chunk.source_url for chunk in result.chunks_by_language['en']] == [
+            'https://target.example/page-1'
+        ]
+        assert result.processed_sources == ['https://target.example/page-1']
         assert len(collect_calls) == 1
         assert collect_calls[0]['tagged_documents'] == []
         assert collect_calls[0]['existing_merged_chunks'] == {'https://target.example/page-1': [existing_chunk]}
@@ -315,7 +321,40 @@ class TestTempBehaviorComplete:
 
         result = scraper.scrape_target(target_url)
 
-        assert result == {}
+        assert isinstance(result, ScrapeManifest)
+        assert not result
         saved_filenames = [call['filename'] for call in scraper._save_calls]
         assert 'merged_chunk_metadata' not in saved_filenames
         assert scraper._content_cleaner.perform_calls == []
+
+    def test_commit_scrape_persists_timestamps_and_removes_temp_files(self, scraper):
+        target_url = 'https://target.example/'
+        source_url = 'https://target.example/page-1'
+        timestamp = _make_timestamp()
+        scraper._pending_timestamps = {source_url: timestamp}
+
+        for filename in (
+            scraper._get_temp_chunks_filename(target_url),
+            scraper._get_temp_timestamps_filename(target_url),
+        ):
+            path = os.path.join(scraper._path.TEMP_CHUNKS_OUTPUT, filename + '.json')
+            with open(path, 'w', encoding='utf-8') as f:
+                f.write('{}')
+
+        scraper.commit_scrape(ScrapeManifest(
+            target_url=target_url,
+            chunks_by_language={'en': [], 'de': []},
+            processed_sources=[source_url],
+        ))
+
+        assert scraper._url_timestamps[source_url] == timestamp
+        assert any(
+            call['filename'] == 'url_timestamps'
+            for call in scraper._save_calls
+        )
+        for filename in (
+            scraper._get_temp_chunks_filename(target_url),
+            scraper._get_temp_timestamps_filename(target_url),
+        ):
+            path = os.path.join(scraper._path.TEMP_CHUNKS_OUTPUT, filename + '.json')
+            assert not os.path.exists(path)

tests/test_master_transfer_integrations.py

@@ -168,8 +168,8 @@ class FakeBatchContext:
     def __exit__(self, exc_type, exc, tb):
         return False
 
-    def add_object(self, properties, vector=None):
-        self.added.append({"properties": properties, "vector": vector})
+    def add_object(self, properties, vector=None, uuid=None):
+        self.added.append({"properties": properties, "vector": vector, "uuid": uuid})
 
 
 class FakeBatchFactory:
@@ -199,6 +199,7 @@ def test_batch_import_embeds_rows_and_writes_named_vectors(monkeypatch):
     assert errors == []
     assert service._embedding_client.document_inputs == [["First chunk"]]
     assert batch_context.added[0]["vector"] == {"test_vectors": [0.7, 0.8, 0.9]}
+    assert batch_context.added[0]["uuid"]
 
 
 def test_query_embeds_once_and_passes_vector_to_hybrid(monkeypatch):

tests/test_safe_import_reconciliation.py

@@ -0,0 +1,369 @@
+from types import SimpleNamespace
+from threading import RLock
+
+import pytest
+
+from src.database.weavservice import (
+    ImportBatchError,
+    PreparedImportRow,
+    WeaviateService,
+)
+from src.pipeline.pipeline import ImportPipeline
+from src.pipeline.utils import ProcessingResult
+from src.scraping.types import ScrapeManifest
+
+
+class FakeData:
+    def __init__(self, state, lang):
+        self.state = state
+        self.lang = lang
+
+    def delete_by_id(self, object_uuid):
+        self.state[self.lang].pop(str(object_uuid), None)
+        return True
+
+
+def _reconciliation_service(initial_state):
+    service = object.__new__(WeaviateService)
+    service._last_query_time = 0
+    state = {
+        lang: dict(objects)
+        for lang, objects in initial_state.items()
+    }
+
+    def prepare(rows, lang):
+        return [
+            PreparedImportRow(
+                uuid=row["uuid"],
+                properties=row,
+                vector={"test": [1.0]},
+            )
+            for row in rows
+        ]
+
+    def collect(lang, source):
+        return {
+            object_uuid: properties
+            for object_uuid, properties in state[lang].items()
+            if properties.get("source") == source
+        }
+
+    def write(rows, lang):
+        for row in rows:
+            state[lang][row.uuid] = row.properties
+
+    service.prepare_batch_import = prepare
+    service._collect_source_objects = collect
+    service._write_prepared_rows = write
+    service._select_collection = lambda lang: (
+        SimpleNamespace(data=FakeData(state, lang)),
+        f"collection_{lang}",
+    )
+    return service, state
+
+
+def test_collect_source_objects_uses_offset_with_filters():
+    source = "https://example.test/page"
+    pages = [
+        [
+            SimpleNamespace(
+                uuid=f"uuid-{index}",
+                properties={"source": source, "chunk_id": str(index)},
+            )
+            for index in range(start, min(start + 100, 205))
+        ]
+        for start in (0, 100, 200)
+    ]
+    calls = []
+
+    def fetch_objects(**kwargs):
+        calls.append(kwargs)
+        return SimpleNamespace(objects=pages[kwargs["offset"] // 100])
+
+    service = object.__new__(WeaviateService)
+    service._client_lock = RLock()
+    service._select_collection = lambda lang: (
+        SimpleNamespace(query=SimpleNamespace(fetch_objects=fetch_objects)),
+        "test_collection",
+    )
+
+    objects = service._collect_source_objects("en", source)
+
+    assert len(objects) == 205
+    assert [call["offset"] for call in calls] == [0, 100, 200]
+    assert all("after" not in call for call in calls)
+    assert all(call["filters"] is not None for call in calls)
+
+
+def test_reconcile_migrates_legacy_uuid_and_language():
+    source = "https://example.test/page"
+    service, state = _reconciliation_service({
+        "en": {"legacy-random": {"source": source}},
+        "de": {},
+    })
+
+    summary = service.reconcile_source(
+        source,
+        {
+            "en": [],
+            "de": [{"uuid": "deterministic-new", "source": source}],
+        },
+    )
+
+    assert state["en"] == {}
+    assert set(state["de"]) == {"deterministic-new"}
+    assert summary.inserted == 1
+    assert summary.deleted == 1
+
+
+def test_reconcile_zero_chunks_removes_old_source():
+    source = "https://example.test/removed"
+    service, state = _reconciliation_service({
+        "en": {"old": {"source": source}},
+        "de": {},
+    })
+
+    service.reconcile_source(source, {"en": [], "de": []})
+
+    assert state == {"en": {}, "de": {}}
+
+
+def test_reconcile_insert_failure_preserves_old_objects():
+    source = "https://example.test/page"
+    service, state = _reconciliation_service({
+        "en": {"old": {"source": source}},
+        "de": {},
+    })
+    service._write_prepared_rows = lambda rows, lang: (
+        (_ for _ in ()).throw(ImportBatchError("embedding or insert failed"))
+        if rows else None
+    )
+
+    with pytest.raises(ImportBatchError):
+        service.reconcile_source(
+            source,
+            {"en": [{"uuid": "new", "source": source}], "de": []},
+        )
+
+    assert set(state["en"]) == {"old"}
+
+
+def test_reconcile_verification_failure_prevents_cleanup():
+    source = "https://example.test/page"
+    service, state = _reconciliation_service({
+        "en": {"old": {"source": source}},
+        "de": {},
+    })
+    service._write_prepared_rows = lambda rows, lang: None
+
+    with pytest.raises(ImportBatchError, match="Verification failed"):
+        service.reconcile_source(
+            source,
+            {"en": [{"uuid": "missing", "source": source}], "de": []},
+        )
+
+    assert set(state["en"]) == {"old"}
+
+
+def test_reconcile_retries_after_partial_insert_without_duplicates():
+    source = "https://example.test/page"
+    service, state = _reconciliation_service({
+        "en": {"old": {"source": source}},
+        "de": {},
+    })
+    normal_write = service._write_prepared_rows
+    failed_once = False
+
+    def partial_write(rows, lang):
+        nonlocal failed_once
+        if rows and not failed_once:
+            failed_once = True
+            normal_write(rows[:1], lang)
+            raise ImportBatchError("connection dropped after a partial insert")
+        normal_write(rows, lang)
+
+    service._write_prepared_rows = partial_write
+    rows = {
+        "en": [
+            {"uuid": "new-1", "source": source},
+            {"uuid": "new-2", "source": source},
+        ],
+        "de": [],
+    }
+
+    with pytest.raises(ImportBatchError):
+        service.reconcile_source(source, rows)
+
+    summary = service.reconcile_source(source, rows)
+
+    assert set(state["en"]) == {"new-1", "new-2"}
+    assert summary.inserted == 1
+    assert summary.retained == 1
+
+
+def test_reconcile_retries_after_cleanup_failure():
+    source = "https://example.test/page"
+    service, state = _reconciliation_service({
+        "en": {"old": {"source": source}},
+        "de": {},
+    })
+
+    class FailOnceData(FakeData):
+        failed = False
+
+        def delete_by_id(self, object_uuid):
+            if not self.failed:
+                self.failed = True
+                raise RuntimeError("delete connection dropped")
+            return super().delete_by_id(object_uuid)
+
+    failing_data = FailOnceData(state, "en")
+    service._select_collection = lambda lang: (
+        SimpleNamespace(data=failing_data if lang == "en" else FakeData(state, lang)),
+        f"collection_{lang}",
+    )
+    rows = {"en": [{"uuid": "new", "source": source}], "de": []}
+
+    with pytest.raises(RuntimeError, match="delete connection dropped"):
+        service.reconcile_source(source, rows)
+
+    assert set(state["en"]) == {"old", "new"}
+
+    summary = service.reconcile_source(source, rows)
+
+    assert set(state["en"]) == {"new"}
+    assert summary.retained == 1
+    assert summary.inserted == 0
+
+
+class AlwaysFailBatchContext:
+    number_errors = 1
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def add_object(self, **kwargs):
+        pass
+
+
+class AlwaysFailBatch:
+    def __init__(self):
+        self.failed_objects = []
+
+    def fixed_size(self, **kwargs):
+        return AlwaysFailBatchContext()
+
+
+class FailedObjectBatchContext:
+    number_errors = 0
+
+    def __init__(self, owner):
+        self.owner = owner
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        self.owner.failed_objects.append("server rejected object")
+        return False
+
+    def add_object(self, **kwargs):
+        pass
+
+
+class FailedObjectBatch:
+    def __init__(self):
+        self.failed_objects = []
+
+    def fixed_size(self, **kwargs):
+        return FailedObjectBatchContext(self)
+
+
+def test_strict_batch_raises_on_asynchronous_errors(monkeypatch):
+    service = object.__new__(WeaviateService)
+    service._client_lock = RLock()
+    service._last_query_time = 0
+    service._select_collection = lambda lang: (
+        SimpleNamespace(batch=AlwaysFailBatch()),
+        "test_collection",
+    )
+    monkeypatch.setattr("src.database.weavservice.sleep", lambda _: None)
+
+    with pytest.raises(ImportBatchError):
+        service._write_prepared_rows(
+            [
+                PreparedImportRow(
+                    uuid="68cda09a-5466-5f23-a595-3e9575af4102",
+                    properties={"source": "source", "chunk_id": "chunk"},
+                    vector={"test": [1.0]},
+                )
+            ],
+            "en",
+        )
+
+
+def test_strict_batch_checks_failed_objects(monkeypatch):
+    service = object.__new__(WeaviateService)
+    service._client_lock = RLock()
+    service._last_query_time = 0
+    service._select_collection = lambda lang: (
+        SimpleNamespace(batch=FailedObjectBatch()),
+        "test_collection",
+    )
+    monkeypatch.setattr("src.database.weavservice.sleep", lambda _: None)
+
+    with pytest.raises(ImportBatchError, match="server rejected object"):
+        service._write_prepared_rows(
+            [
+                PreparedImportRow(
+                    uuid="68cda09a-5466-5f23-a595-3e9575af4102",
+                    properties={"source": "source", "chunk_id": "chunk"},
+                    vector={"test": [1.0]},
+                )
+            ],
+            "en",
+        )
+
+
+def test_scraper_manifest_reconciles_empty_languages():
+    pipeline = object.__new__(ImportPipeline)
+    calls = []
+    pipeline._service = SimpleNamespace(
+        reconcile_source=lambda source, rows: calls.append((source, rows))
+    )
+    manifest = ScrapeManifest(
+        target_url="https://example.test",
+        chunks_by_language={"de": []},
+        processed_sources=["https://example.test/empty"],
+    )
+
+    pipeline.import_from_scraper(manifest)
+
+    source, rows = calls[0]
+    assert source == "https://example.test/empty"
+    assert rows == {"en": [], "de": []}
+
+
+def test_cli_local_import_replaces_existing_source():
+    pipeline = object.__new__(ImportPipeline)
+    calls = []
+    pipeline._deduplication_callback = None
+    pipeline._service = SimpleNamespace(
+        source_object_count=lambda source: 2,
+        reconcile_source=lambda source, rows: calls.append(("replace", source, rows)),
+        batch_import=lambda rows, lang: calls.append(("append", lang, rows)),
+    )
+    result = ProcessingResult(
+        chunks=[{"source": "document.pdf", "chunk_id": "new"}],
+        source="document.pdf",
+        lang="en",
+    )
+
+    pipeline._import_document_result(result)
+
+    assert calls[0][0] == "replace"
+    assert calls[0][2]["en"] == result.chunks
+    assert calls[0][2]["de"] == []

tools/scraping.py

@@ -21,9 +21,11 @@ def scraping_task(full_scrape: bool):
             pipeline = ImportPipeline()
 
             for target_url in config.scraping.TARGET_URLS:
-                chunks = scraper.scrape_target(target_url)
-                pipeline.import_from_scraper(chunks)
-                scraper.delete_temp_merged_chunks(target_url)
+                manifest = scraper.scrape_target(target_url)
+                if not manifest:
+                    continue
+                pipeline.import_from_scraper(manifest)
+                scraper.commit_scrape(manifest)
 
             logger.info("Scraper task finished gracefully")
         except Exception as e: