Compare commits
No commits in common. '28c7854ead27f077b8c0e28b29ff8e8a6a531335' and '152d9739af0969090ddb7db63ad6a93054147053' have entirely different histories.
28c7854ead
...
152d9739af
@ -1,113 +0,0 @@
|
|||||||
package at.procon.dip.domain.ted.service;
|
|
||||||
|
|
||||||
import at.procon.dip.domain.access.DocumentVisibility;
|
|
||||||
import at.procon.dip.domain.document.DocumentFamily;
|
|
||||||
import at.procon.dip.domain.document.DocumentStatus;
|
|
||||||
import at.procon.dip.domain.document.DocumentType;
|
|
||||||
import at.procon.dip.domain.document.entity.Document;
|
|
||||||
import at.procon.dip.domain.document.repository.DocumentRepository;
|
|
||||||
import at.procon.dip.domain.document.service.DocumentService;
|
|
||||||
import at.procon.dip.domain.document.service.command.CreateDocumentCommand;
|
|
||||||
import at.procon.dip.runtime.condition.ConditionalOnRuntimeMode;
|
|
||||||
import at.procon.dip.runtime.config.RuntimeMode;
|
|
||||||
import at.procon.ted.model.entity.ProcurementDocument;
|
|
||||||
import at.procon.ted.model.entity.TedDailyPackage;
|
|
||||||
import at.procon.ted.repository.TedDailyPackageRepository;
|
|
||||||
import at.procon.ted.util.HashUtils;
|
|
||||||
import java.time.OffsetDateTime;
|
|
||||||
import java.util.Optional;
|
|
||||||
import lombok.RequiredArgsConstructor;
|
|
||||||
import org.springframework.stereotype.Service;
|
|
||||||
import org.springframework.transaction.annotation.Transactional;
|
|
||||||
|
|
||||||
@Service
|
|
||||||
@ConditionalOnRuntimeMode(RuntimeMode.NEW)
|
|
||||||
@RequiredArgsConstructor
|
|
||||||
public class TedPackageDocumentService {
|
|
||||||
|
|
||||||
private static final String PACKAGE_MIME_TYPE = "application/gzip";
|
|
||||||
|
|
||||||
private final TedPackageIdentifierResolver packageIdentifierResolver;
|
|
||||||
private final TedDailyPackageRepository tedDailyPackageRepository;
|
|
||||||
private final DocumentRepository documentRepository;
|
|
||||||
private final DocumentService documentService;
|
|
||||||
|
|
||||||
@Transactional
|
|
||||||
public Optional<Document> ensurePackageDocumentForLegacyNotice(ProcurementDocument legacyDocument) {
|
|
||||||
return packageIdentifierResolver.resolveFromSourceMetadata(legacyDocument.getSourcePath(), legacyDocument.getSourceFilename())
|
|
||||||
.map(this::ensurePackageDocument);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Transactional
|
|
||||||
public Document ensurePackageDocument(String packageIdentifier) {
|
|
||||||
String businessKey = buildBusinessKey(packageIdentifier);
|
|
||||||
Document document = documentRepository.findByBusinessKey(businessKey)
|
|
||||||
.orElseGet(() -> createPackageDocument(packageIdentifier));
|
|
||||||
|
|
||||||
Optional<TedDailyPackage> packageEntity = tedDailyPackageRepository.findByPackageIdentifier(packageIdentifier);
|
|
||||||
document.setVisibility(DocumentVisibility.PUBLIC);
|
|
||||||
document.setDocumentType(DocumentType.TED_PACKAGE);
|
|
||||||
document.setDocumentFamily(DocumentFamily.PROCUREMENT);
|
|
||||||
document.setStatus(resolveStatus(packageEntity));
|
|
||||||
document.setTitle(buildTitle(packageIdentifier));
|
|
||||||
document.setSummary(buildSummary(packageIdentifier, packageEntity.orElse(null)));
|
|
||||||
document.setMimeType(PACKAGE_MIME_TYPE);
|
|
||||||
document.setBusinessKey(businessKey);
|
|
||||||
document.setDedupHash(HashUtils.computeSha256(businessKey));
|
|
||||||
return documentService.save(document);
|
|
||||||
}
|
|
||||||
|
|
||||||
private Document createPackageDocument(String packageIdentifier) {
|
|
||||||
String businessKey = buildBusinessKey(packageIdentifier);
|
|
||||||
return documentService.create(new CreateDocumentCommand(
|
|
||||||
null,
|
|
||||||
DocumentVisibility.PUBLIC,
|
|
||||||
DocumentType.TED_PACKAGE,
|
|
||||||
DocumentFamily.PROCUREMENT,
|
|
||||||
DocumentStatus.RECEIVED,
|
|
||||||
buildTitle(packageIdentifier),
|
|
||||||
null,
|
|
||||||
null,
|
|
||||||
PACKAGE_MIME_TYPE,
|
|
||||||
businessKey,
|
|
||||||
HashUtils.computeSha256(businessKey)
|
|
||||||
));
|
|
||||||
}
|
|
||||||
|
|
||||||
private DocumentStatus resolveStatus(Optional<TedDailyPackage> packageEntity) {
|
|
||||||
if (packageEntity.isEmpty()) {
|
|
||||||
return DocumentStatus.RECEIVED;
|
|
||||||
}
|
|
||||||
return switch (packageEntity.get().getDownloadStatus()) {
|
|
||||||
case COMPLETED -> DocumentStatus.CLASSIFIED;
|
|
||||||
case FAILED, NOT_FOUND -> DocumentStatus.FAILED;
|
|
||||||
default -> DocumentStatus.RECEIVED;
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
private String buildBusinessKey(String packageIdentifier) {
|
|
||||||
return "TED_PACKAGE:" + packageIdentifier;
|
|
||||||
}
|
|
||||||
|
|
||||||
private String buildTitle(String packageIdentifier) {
|
|
||||||
return packageIdentifier + ".tar.gz";
|
|
||||||
}
|
|
||||||
|
|
||||||
private String buildSummary(String packageIdentifier, TedDailyPackage packageEntity) {
|
|
||||||
if (packageEntity == null) {
|
|
||||||
return "TED daily package " + packageIdentifier;
|
|
||||||
}
|
|
||||||
return "TED daily package %s (status=%s, xmlFileCount=%s, processedCount=%s, failedCount=%s, downloadedAt=%s)".formatted(
|
|
||||||
packageIdentifier,
|
|
||||||
packageEntity.getDownloadStatus(),
|
|
||||||
packageEntity.getXmlFileCount(),
|
|
||||||
packageEntity.getProcessedCount(),
|
|
||||||
packageEntity.getFailedCount(),
|
|
||||||
formatTimestamp(packageEntity.getDownloadedAt())
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
private String formatTimestamp(OffsetDateTime value) {
|
|
||||||
return value == null ? null : value.toString();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,31 +0,0 @@
|
|||||||
package at.procon.dip.domain.ted.service;
|
|
||||||
|
|
||||||
import java.util.Optional;
|
|
||||||
import java.util.regex.Matcher;
|
|
||||||
import java.util.regex.Pattern;
|
|
||||||
import org.springframework.stereotype.Component;
|
|
||||||
import org.springframework.util.StringUtils;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Resolves a TED daily package identifier (YYYYSSSSS) from legacy source metadata.
|
|
||||||
*/
|
|
||||||
@Component
|
|
||||||
public class TedPackageIdentifierResolver {
|
|
||||||
|
|
||||||
private static final Pattern PACKAGE_IDENTIFIER_PATTERN = Pattern.compile("(?<!\\d)(20\\d{7})(?!\\d)");
|
|
||||||
|
|
||||||
public Optional<String> resolveFromSourceMetadata(String sourcePath, String sourceFilename) {
|
|
||||||
return resolve(sourcePath).or(() -> resolve(sourceFilename));
|
|
||||||
}
|
|
||||||
|
|
||||||
public Optional<String> resolve(String value) {
|
|
||||||
if (!StringUtils.hasText(value)) {
|
|
||||||
return Optional.empty();
|
|
||||||
}
|
|
||||||
Matcher matcher = PACKAGE_IDENTIFIER_PATTERN.matcher(value);
|
|
||||||
if (matcher.find()) {
|
|
||||||
return Optional.of(matcher.group(1));
|
|
||||||
}
|
|
||||||
return Optional.empty();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,35 +0,0 @@
|
|||||||
package at.procon.dip.migration.config;
|
|
||||||
|
|
||||||
import lombok.Data;
|
|
||||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
|
||||||
import org.springframework.context.annotation.Configuration;
|
|
||||||
|
|
||||||
@Configuration
|
|
||||||
@ConfigurationProperties(prefix = "dip.migration.legacy-ted")
|
|
||||||
@Data
|
|
||||||
public class LegacyTedBackfillProperties {
|
|
||||||
|
|
||||||
/** Enable the TED legacy -> DOC/projection backfill subsystem. */
|
|
||||||
private boolean enabled = false;
|
|
||||||
|
|
||||||
/** Run the backfill automatically on application startup in NEW runtime. */
|
|
||||||
private boolean startupEnabled = false;
|
|
||||||
|
|
||||||
/** Number of legacy TED documents to process per fetch batch. */
|
|
||||||
private int batchSize = 100;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Optional cap for a single run. 0 or negative means unlimited until the full legacy set is backfilled.
|
|
||||||
* Useful to migrate incrementally in controlled slices.
|
|
||||||
*/
|
|
||||||
private long maxDocumentsPerRun = 0;
|
|
||||||
|
|
||||||
/** Resume the latest STOPPED/FAILED run from its persisted cursor. */
|
|
||||||
private boolean resumeLatestIncompleteRun = true;
|
|
||||||
|
|
||||||
/** Import batch id written to DOC.doc_source rows created by the migration. */
|
|
||||||
private String importBatchId = "legacy-ted-backfill";
|
|
||||||
|
|
||||||
/** Queue embeddings for migrated TED representations after the DOC/projection backfill. */
|
|
||||||
private boolean queueEmbeddings = false;
|
|
||||||
}
|
|
||||||
@ -1,75 +0,0 @@
|
|||||||
package at.procon.dip.migration.entity;
|
|
||||||
|
|
||||||
import at.procon.dip.architecture.SchemaNames;
|
|
||||||
import jakarta.persistence.Column;
|
|
||||||
import jakarta.persistence.Entity;
|
|
||||||
import jakarta.persistence.FetchType;
|
|
||||||
import jakarta.persistence.GeneratedValue;
|
|
||||||
import jakarta.persistence.GenerationType;
|
|
||||||
import jakarta.persistence.Id;
|
|
||||||
import jakarta.persistence.Index;
|
|
||||||
import jakarta.persistence.JoinColumn;
|
|
||||||
import jakarta.persistence.ManyToOne;
|
|
||||||
import jakarta.persistence.PrePersist;
|
|
||||||
import jakarta.persistence.Table;
|
|
||||||
import java.time.OffsetDateTime;
|
|
||||||
import java.util.UUID;
|
|
||||||
import lombok.AllArgsConstructor;
|
|
||||||
import lombok.Builder;
|
|
||||||
import lombok.Getter;
|
|
||||||
import lombok.NoArgsConstructor;
|
|
||||||
import lombok.Setter;
|
|
||||||
|
|
||||||
@Entity
|
|
||||||
@Table(schema = SchemaNames.DOC, name = "doc_legacy_ted_migration_checkpoint", indexes = {
|
|
||||||
@Index(name = "idx_doc_legacy_ted_mig_ckpt_run", columnList = "run_id"),
|
|
||||||
@Index(name = "idx_doc_legacy_ted_mig_ckpt_batch", columnList = "batch_number")
|
|
||||||
})
|
|
||||||
@Getter
|
|
||||||
@Setter
|
|
||||||
@NoArgsConstructor
|
|
||||||
@AllArgsConstructor
|
|
||||||
@Builder
|
|
||||||
public class LegacyTedMigrationCheckpoint {
|
|
||||||
|
|
||||||
@Id
|
|
||||||
@GeneratedValue(strategy = GenerationType.UUID)
|
|
||||||
private UUID id;
|
|
||||||
|
|
||||||
@ManyToOne(fetch = FetchType.LAZY, optional = false)
|
|
||||||
@JoinColumn(name = "run_id", nullable = false)
|
|
||||||
private LegacyTedMigrationRun run;
|
|
||||||
|
|
||||||
@Column(name = "batch_number", nullable = false)
|
|
||||||
private int batchNumber;
|
|
||||||
|
|
||||||
@Column(name = "batch_processed_count", nullable = false)
|
|
||||||
private int batchProcessedCount;
|
|
||||||
|
|
||||||
@Column(name = "cumulative_processed_count", nullable = false)
|
|
||||||
private long cumulativeProcessedCount;
|
|
||||||
|
|
||||||
@Column(name = "last_legacy_created_at")
|
|
||||||
private OffsetDateTime lastLegacyCreatedAt;
|
|
||||||
|
|
||||||
@Column(name = "last_legacy_document_id")
|
|
||||||
private UUID lastLegacyDocumentId;
|
|
||||||
|
|
||||||
@Column(name = "last_doc_document_id")
|
|
||||||
private UUID lastDocDocumentId;
|
|
||||||
|
|
||||||
@Column(name = "last_projection_id")
|
|
||||||
private UUID lastProjectionId;
|
|
||||||
|
|
||||||
@Column(name = "note", columnDefinition = "TEXT")
|
|
||||||
private String note;
|
|
||||||
|
|
||||||
@Builder.Default
|
|
||||||
@Column(name = "created_at", nullable = false, updatable = false)
|
|
||||||
private OffsetDateTime createdAt = OffsetDateTime.now();
|
|
||||||
|
|
||||||
@PrePersist
|
|
||||||
protected void onCreate() {
|
|
||||||
createdAt = OffsetDateTime.now();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,103 +0,0 @@
|
|||||||
package at.procon.dip.migration.entity;
|
|
||||||
|
|
||||||
import at.procon.dip.architecture.SchemaNames;
|
|
||||||
import jakarta.persistence.Column;
|
|
||||||
import jakarta.persistence.Entity;
|
|
||||||
import jakarta.persistence.EnumType;
|
|
||||||
import jakarta.persistence.Enumerated;
|
|
||||||
import jakarta.persistence.GeneratedValue;
|
|
||||||
import jakarta.persistence.GenerationType;
|
|
||||||
import jakarta.persistence.Id;
|
|
||||||
import jakarta.persistence.Index;
|
|
||||||
import jakarta.persistence.PrePersist;
|
|
||||||
import jakarta.persistence.PreUpdate;
|
|
||||||
import jakarta.persistence.Table;
|
|
||||||
import java.time.OffsetDateTime;
|
|
||||||
import java.util.UUID;
|
|
||||||
import lombok.AllArgsConstructor;
|
|
||||||
import lombok.Builder;
|
|
||||||
import lombok.Getter;
|
|
||||||
import lombok.NoArgsConstructor;
|
|
||||||
import lombok.Setter;
|
|
||||||
|
|
||||||
@Entity
|
|
||||||
@Table(schema = SchemaNames.DOC, name = "doc_legacy_ted_migration_run", indexes = {
|
|
||||||
@Index(name = "idx_doc_legacy_ted_mig_run_status", columnList = "status"),
|
|
||||||
@Index(name = "idx_doc_legacy_ted_mig_run_started", columnList = "started_at DESC")
|
|
||||||
})
|
|
||||||
@Getter
|
|
||||||
@Setter
|
|
||||||
@NoArgsConstructor
|
|
||||||
@AllArgsConstructor
|
|
||||||
@Builder
|
|
||||||
public class LegacyTedMigrationRun {
|
|
||||||
|
|
||||||
@Id
|
|
||||||
@GeneratedValue(strategy = GenerationType.UUID)
|
|
||||||
private UUID id;
|
|
||||||
|
|
||||||
@Enumerated(EnumType.STRING)
|
|
||||||
@Column(name = "status", nullable = false, length = 32)
|
|
||||||
private LegacyTedMigrationRunStatus status;
|
|
||||||
|
|
||||||
@Column(name = "import_batch_id", length = 255)
|
|
||||||
private String importBatchId;
|
|
||||||
|
|
||||||
@Column(name = "queue_embeddings", nullable = false)
|
|
||||||
private boolean queueEmbeddings;
|
|
||||||
|
|
||||||
@Column(name = "batch_size", nullable = false)
|
|
||||||
private int batchSize;
|
|
||||||
|
|
||||||
@Column(name = "max_documents_per_run")
|
|
||||||
private Long maxDocumentsPerRun;
|
|
||||||
|
|
||||||
@Column(name = "processed_count", nullable = false)
|
|
||||||
@Builder.Default
|
|
||||||
private long processedCount = 0;
|
|
||||||
|
|
||||||
@Column(name = "success_count", nullable = false)
|
|
||||||
@Builder.Default
|
|
||||||
private long successCount = 0;
|
|
||||||
|
|
||||||
@Column(name = "failed_count", nullable = false)
|
|
||||||
@Builder.Default
|
|
||||||
private long failedCount = 0;
|
|
||||||
|
|
||||||
@Column(name = "last_legacy_created_at")
|
|
||||||
private OffsetDateTime lastLegacyCreatedAt;
|
|
||||||
|
|
||||||
@Column(name = "last_legacy_document_id")
|
|
||||||
private UUID lastLegacyDocumentId;
|
|
||||||
|
|
||||||
@Column(name = "last_doc_document_id")
|
|
||||||
private UUID lastDocDocumentId;
|
|
||||||
|
|
||||||
@Column(name = "last_projection_id")
|
|
||||||
private UUID lastProjectionId;
|
|
||||||
|
|
||||||
@Column(name = "last_error", columnDefinition = "TEXT")
|
|
||||||
private String lastError;
|
|
||||||
|
|
||||||
@Builder.Default
|
|
||||||
@Column(name = "started_at", nullable = false, updatable = false)
|
|
||||||
private OffsetDateTime startedAt = OffsetDateTime.now();
|
|
||||||
|
|
||||||
@Builder.Default
|
|
||||||
@Column(name = "updated_at", nullable = false)
|
|
||||||
private OffsetDateTime updatedAt = OffsetDateTime.now();
|
|
||||||
|
|
||||||
@Column(name = "completed_at")
|
|
||||||
private OffsetDateTime completedAt;
|
|
||||||
|
|
||||||
@PrePersist
|
|
||||||
protected void onCreate() {
|
|
||||||
startedAt = startedAt == null ? OffsetDateTime.now() : startedAt;
|
|
||||||
updatedAt = OffsetDateTime.now();
|
|
||||||
}
|
|
||||||
|
|
||||||
@PreUpdate
|
|
||||||
protected void onUpdate() {
|
|
||||||
updatedAt = OffsetDateTime.now();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,8 +0,0 @@
|
|||||||
package at.procon.dip.migration.entity;
|
|
||||||
|
|
||||||
public enum LegacyTedMigrationRunStatus {
|
|
||||||
RUNNING,
|
|
||||||
STOPPED,
|
|
||||||
COMPLETED,
|
|
||||||
FAILED
|
|
||||||
}
|
|
||||||
@ -1,10 +0,0 @@
|
|||||||
package at.procon.dip.migration.repository;
|
|
||||||
|
|
||||||
import at.procon.dip.migration.entity.LegacyTedMigrationCheckpoint;
|
|
||||||
import java.util.UUID;
|
|
||||||
import org.springframework.data.jpa.repository.JpaRepository;
|
|
||||||
|
|
||||||
public interface LegacyTedMigrationCheckpointRepository extends JpaRepository<LegacyTedMigrationCheckpoint, UUID> {
|
|
||||||
|
|
||||||
long countByRun_Id(UUID runId);
|
|
||||||
}
|
|
||||||
@ -1,23 +0,0 @@
|
|||||||
package at.procon.dip.migration.repository;
|
|
||||||
|
|
||||||
import at.procon.dip.migration.entity.LegacyTedMigrationRun;
|
|
||||||
import at.procon.dip.migration.entity.LegacyTedMigrationRunStatus;
|
|
||||||
import java.util.Collection;
|
|
||||||
import java.util.List;
|
|
||||||
import java.util.UUID;
|
|
||||||
import org.springframework.data.domain.Pageable;
|
|
||||||
import org.springframework.data.jpa.repository.JpaRepository;
|
|
||||||
import org.springframework.data.jpa.repository.Query;
|
|
||||||
import org.springframework.data.repository.query.Param;
|
|
||||||
|
|
||||||
public interface LegacyTedMigrationRunRepository extends JpaRepository<LegacyTedMigrationRun, UUID> {
|
|
||||||
|
|
||||||
@Query("""
|
|
||||||
select r
|
|
||||||
from LegacyTedMigrationRun r
|
|
||||||
where r.status in :statuses
|
|
||||||
order by r.startedAt desc
|
|
||||||
""")
|
|
||||||
List<LegacyTedMigrationRun> findLatestByStatuses(@Param("statuses") Collection<LegacyTedMigrationRunStatus> statuses,
|
|
||||||
Pageable pageable);
|
|
||||||
}
|
|
||||||
@ -1,197 +0,0 @@
|
|||||||
package at.procon.dip.migration.service;
|
|
||||||
|
|
||||||
import at.procon.dip.migration.config.LegacyTedBackfillProperties;
|
|
||||||
import at.procon.dip.migration.entity.LegacyTedMigrationCheckpoint;
|
|
||||||
import at.procon.dip.migration.entity.LegacyTedMigrationRun;
|
|
||||||
import at.procon.dip.migration.entity.LegacyTedMigrationRunStatus;
|
|
||||||
import at.procon.dip.migration.repository.LegacyTedMigrationCheckpointRepository;
|
|
||||||
import at.procon.dip.migration.repository.LegacyTedMigrationRunRepository;
|
|
||||||
import at.procon.dip.runtime.condition.ConditionalOnRuntimeMode;
|
|
||||||
import at.procon.dip.runtime.config.RuntimeMode;
|
|
||||||
import at.procon.ted.repository.LegacyTedMigrationCursor;
|
|
||||||
import at.procon.ted.repository.ProcurementDocumentRepository;
|
|
||||||
import java.time.OffsetDateTime;
|
|
||||||
import java.time.ZoneOffset;
|
|
||||||
import java.util.EnumSet;
|
|
||||||
import java.util.List;
|
|
||||||
import java.util.UUID;
|
|
||||||
import lombok.RequiredArgsConstructor;
|
|
||||||
import lombok.extern.slf4j.Slf4j;
|
|
||||||
import org.springframework.data.domain.PageRequest;
|
|
||||||
import org.springframework.stereotype.Service;
|
|
||||||
import org.springframework.transaction.annotation.Transactional;
|
|
||||||
|
|
||||||
@Service
|
|
||||||
@ConditionalOnRuntimeMode(RuntimeMode.NEW)
|
|
||||||
@RequiredArgsConstructor
|
|
||||||
@Slf4j
|
|
||||||
public class LegacyTedBackfillMigrationService {
|
|
||||||
|
|
||||||
private final LegacyTedBackfillProperties properties;
|
|
||||||
private final ProcurementDocumentRepository procurementDocumentRepository;
|
|
||||||
private final LegacyTedMigrationRunRepository runRepository;
|
|
||||||
private final LegacyTedMigrationCheckpointRepository checkpointRepository;
|
|
||||||
private final LegacyTedBackfillWorker worker;
|
|
||||||
|
|
||||||
public UUID runBackfill() {
|
|
||||||
if (!properties.isEnabled()) {
|
|
||||||
log.info("Legacy TED backfill is disabled");
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
|
|
||||||
LegacyTedMigrationRun run = resolveRun();
|
|
||||||
log.info("Starting legacy TED -> DOC/projection backfill run {} (batchSize={}, maxDocumentsPerRun={}, queueEmbeddings={})",
|
|
||||||
run.getId(), run.getBatchSize(), run.getMaxDocumentsPerRun(), run.isQueueEmbeddings());
|
|
||||||
|
|
||||||
long existingCheckpointCount = checkpointRepository.countByRun_Id(run.getId());
|
|
||||||
int batchNumber = existingCheckpointCount >= Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) existingCheckpointCount;
|
|
||||||
long processedInThisInvocation = 0;
|
|
||||||
|
|
||||||
while (true) {
|
|
||||||
int limit = effectiveBatchLimit(run, processedInThisInvocation);
|
|
||||||
if (limit <= 0) {
|
|
||||||
markStopped(run, "Stopped because maxDocumentsPerRun was reached");
|
|
||||||
createCheckpoint(run, ++batchNumber, 0, "Invocation limit reached");
|
|
||||||
return run.getId();
|
|
||||||
}
|
|
||||||
|
|
||||||
List<LegacyTedMigrationCursor> batch = loadNextBatch(run, limit);
|
|
||||||
|
|
||||||
if (batch.isEmpty()) {
|
|
||||||
markCompleted(run);
|
|
||||||
createCheckpoint(run, ++batchNumber, 0, "Backfill completed - no more legacy TED documents after current cursor");
|
|
||||||
return run.getId();
|
|
||||||
}
|
|
||||||
|
|
||||||
int processedInBatch = 0;
|
|
||||||
for (LegacyTedMigrationCursor cursor : batch) {
|
|
||||||
try {
|
|
||||||
LegacyTedBackfillWorker.BackfillOutcome outcome = worker.backfill(
|
|
||||||
cursor.getId(),
|
|
||||||
run.getImportBatchId(),
|
|
||||||
run.isQueueEmbeddings()
|
|
||||||
);
|
|
||||||
advanceRun(run, cursor, outcome);
|
|
||||||
processedInBatch++;
|
|
||||||
processedInThisInvocation++;
|
|
||||||
} catch (RuntimeException ex) {
|
|
||||||
markFailed(run, cursor, ex);
|
|
||||||
createCheckpoint(run, ++batchNumber, processedInBatch,
|
|
||||||
"Failed at legacy document %s: %s".formatted(cursor.getId(), ex.getMessage()));
|
|
||||||
throw ex;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
createCheckpoint(run, ++batchNumber, processedInBatch,
|
|
||||||
"Processed %d legacy TED documents in batch".formatted(processedInBatch));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
protected List<LegacyTedMigrationCursor> loadNextBatch(LegacyTedMigrationRun run, int limit) {
|
|
||||||
if (run.getLastLegacyCreatedAt() == null || run.getLastLegacyDocumentId() == null) {
|
|
||||||
return procurementDocumentRepository.findFirstMigrationBatch(limit);
|
|
||||||
}
|
|
||||||
return procurementDocumentRepository.findNextMigrationBatch(
|
|
||||||
run.getLastLegacyCreatedAt(),
|
|
||||||
run.getLastLegacyDocumentId(),
|
|
||||||
limit
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Transactional
|
|
||||||
protected LegacyTedMigrationRun resolveRun() {
|
|
||||||
if (properties.isResumeLatestIncompleteRun()) {
|
|
||||||
List<LegacyTedMigrationRun> resumable = runRepository.findLatestByStatuses(
|
|
||||||
EnumSet.of(LegacyTedMigrationRunStatus.RUNNING, LegacyTedMigrationRunStatus.STOPPED, LegacyTedMigrationRunStatus.FAILED),
|
|
||||||
PageRequest.of(0, 1)
|
|
||||||
);
|
|
||||||
if (!resumable.isEmpty()) {
|
|
||||||
LegacyTedMigrationRun run = resumable.get(0);
|
|
||||||
run.setStatus(LegacyTedMigrationRunStatus.RUNNING);
|
|
||||||
run.setLastError(null);
|
|
||||||
run.setCompletedAt(null);
|
|
||||||
return runRepository.save(run);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return runRepository.save(LegacyTedMigrationRun.builder()
|
|
||||||
.status(LegacyTedMigrationRunStatus.RUNNING)
|
|
||||||
.importBatchId(properties.getImportBatchId())
|
|
||||||
.queueEmbeddings(properties.isQueueEmbeddings())
|
|
||||||
.batchSize(Math.max(1, properties.getBatchSize()))
|
|
||||||
.maxDocumentsPerRun(properties.getMaxDocumentsPerRun() > 0 ? properties.getMaxDocumentsPerRun() : null)
|
|
||||||
.build());
|
|
||||||
}
|
|
||||||
|
|
||||||
@Transactional
|
|
||||||
protected void advanceRun(LegacyTedMigrationRun run,
|
|
||||||
LegacyTedMigrationCursor cursor,
|
|
||||||
LegacyTedBackfillWorker.BackfillOutcome outcome) {
|
|
||||||
run.setStatus(LegacyTedMigrationRunStatus.RUNNING);
|
|
||||||
run.setProcessedCount(run.getProcessedCount() + 1);
|
|
||||||
run.setSuccessCount(run.getSuccessCount() + 1);
|
|
||||||
run.setLastLegacyCreatedAt(cursor.getCreatedAt() != null
|
|
||||||
? cursor.getCreatedAt().atOffset(ZoneOffset.UTC)
|
|
||||||
: null);
|
|
||||||
run.setLastLegacyDocumentId(cursor.getId());
|
|
||||||
run.setLastDocDocumentId(outcome.documentId());
|
|
||||||
run.setLastProjectionId(outcome.projectionId());
|
|
||||||
run.setLastError(null);
|
|
||||||
runRepository.save(run);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Transactional
|
|
||||||
protected void createCheckpoint(LegacyTedMigrationRun run, int batchNumber, int processedInBatch, String note) {
|
|
||||||
checkpointRepository.save(LegacyTedMigrationCheckpoint.builder()
|
|
||||||
.run(run)
|
|
||||||
.batchNumber(batchNumber)
|
|
||||||
.batchProcessedCount(processedInBatch)
|
|
||||||
.cumulativeProcessedCount(run.getProcessedCount())
|
|
||||||
.lastLegacyCreatedAt(run.getLastLegacyCreatedAt())
|
|
||||||
.lastLegacyDocumentId(run.getLastLegacyDocumentId())
|
|
||||||
.lastDocDocumentId(run.getLastDocDocumentId())
|
|
||||||
.lastProjectionId(run.getLastProjectionId())
|
|
||||||
.note(note)
|
|
||||||
.build());
|
|
||||||
}
|
|
||||||
|
|
||||||
@Transactional
|
|
||||||
protected void markCompleted(LegacyTedMigrationRun run) {
|
|
||||||
run.setStatus(LegacyTedMigrationRunStatus.COMPLETED);
|
|
||||||
run.setCompletedAt(OffsetDateTime.now());
|
|
||||||
run.setLastError(null);
|
|
||||||
runRepository.save(run);
|
|
||||||
log.info("Legacy TED backfill run {} completed successfully - processed {} documents",
|
|
||||||
run.getId(), run.getProcessedCount());
|
|
||||||
}
|
|
||||||
|
|
||||||
@Transactional
|
|
||||||
protected void markStopped(LegacyTedMigrationRun run, String message) {
|
|
||||||
run.setStatus(LegacyTedMigrationRunStatus.STOPPED);
|
|
||||||
run.setLastError(message);
|
|
||||||
runRepository.save(run);
|
|
||||||
log.info("Legacy TED backfill run {} stopped after {} processed documents: {}",
|
|
||||||
run.getId(), run.getProcessedCount(), message);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Transactional
|
|
||||||
protected void markFailed(LegacyTedMigrationRun run, LegacyTedMigrationCursor cursor, RuntimeException ex) {
|
|
||||||
run.setStatus(LegacyTedMigrationRunStatus.FAILED);
|
|
||||||
run.setFailedCount(run.getFailedCount() + 1);
|
|
||||||
run.setLastError(ex.getMessage());
|
|
||||||
runRepository.save(run);
|
|
||||||
log.error("Legacy TED backfill run {} failed at legacy document {}: {}",
|
|
||||||
run.getId(), cursor.getId(), ex.getMessage(), ex);
|
|
||||||
}
|
|
||||||
|
|
||||||
private int effectiveBatchLimit(LegacyTedMigrationRun run, long processedInThisInvocation) {
|
|
||||||
long maxPerRun = run.getMaxDocumentsPerRun() != null ? run.getMaxDocumentsPerRun() : 0L;
|
|
||||||
if (maxPerRun <= 0) {
|
|
||||||
return Math.max(1, run.getBatchSize());
|
|
||||||
}
|
|
||||||
long remaining = maxPerRun - processedInThisInvocation;
|
|
||||||
if (remaining <= 0) {
|
|
||||||
return 0;
|
|
||||||
}
|
|
||||||
return (int) Math.max(1L, Math.min(run.getBatchSize(), remaining));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,390 +0,0 @@
|
|||||||
package at.procon.dip.migration.service;
|
|
||||||
|
|
||||||
import at.procon.dip.classification.spi.DetectionResult;
|
|
||||||
import at.procon.dip.domain.document.ContentRole;
|
|
||||||
import at.procon.dip.domain.document.DocumentFamily;
|
|
||||||
import at.procon.dip.domain.document.DocumentStatus;
|
|
||||||
import at.procon.dip.domain.document.DocumentType;
|
|
||||||
import at.procon.dip.domain.document.RelationType;
|
|
||||||
import at.procon.dip.domain.document.RepresentationType;
|
|
||||||
import at.procon.dip.domain.document.SourceType;
|
|
||||||
import at.procon.dip.domain.document.StorageType;
|
|
||||||
import at.procon.dip.domain.document.entity.Document;
|
|
||||||
import at.procon.dip.domain.document.entity.DocumentContent;
|
|
||||||
import at.procon.dip.domain.document.entity.DocumentSource;
|
|
||||||
import at.procon.dip.domain.document.entity.DocumentTextRepresentation;
|
|
||||||
import at.procon.dip.domain.document.repository.DocumentContentRepository;
|
|
||||||
import at.procon.dip.domain.document.repository.DocumentSourceRepository;
|
|
||||||
import at.procon.dip.domain.document.repository.DocumentTextRepresentationRepository;
|
|
||||||
import at.procon.dip.domain.document.service.DocumentContentService;
|
|
||||||
import at.procon.dip.domain.document.service.DocumentRelationService;
|
|
||||||
import at.procon.dip.domain.document.service.DocumentRepresentationService;
|
|
||||||
import at.procon.dip.domain.document.service.DocumentService;
|
|
||||||
import at.procon.dip.domain.document.service.command.AddDocumentContentCommand;
|
|
||||||
import at.procon.dip.domain.document.service.command.AddDocumentSourceCommand;
|
|
||||||
import at.procon.dip.domain.document.service.command.AddDocumentTextRepresentationCommand;
|
|
||||||
import at.procon.dip.domain.document.service.command.CreateDocumentRelationCommand;
|
|
||||||
import at.procon.dip.domain.ted.service.TedGenericDocumentRootService;
|
|
||||||
import at.procon.dip.domain.ted.service.TedNoticeProjectionService;
|
|
||||||
import at.procon.dip.domain.ted.service.TedPackageDocumentService;
|
|
||||||
import at.procon.dip.embedding.config.EmbeddingProperties;
|
|
||||||
import at.procon.dip.embedding.service.RepresentationEmbeddingOrchestrator;
|
|
||||||
import at.procon.dip.extraction.spi.ExtractedStructuredPayload;
|
|
||||||
import at.procon.dip.extraction.spi.ExtractionResult;
|
|
||||||
import at.procon.dip.ingestion.spi.OriginalContentStoragePolicy;
|
|
||||||
import at.procon.dip.ingestion.spi.SourceDescriptor;
|
|
||||||
import at.procon.dip.normalization.service.TextRepresentationBuildService;
|
|
||||||
import at.procon.dip.normalization.spi.RepresentationBuildRequest;
|
|
||||||
import at.procon.dip.normalization.spi.TextRepresentationDraft;
|
|
||||||
import at.procon.dip.runtime.condition.ConditionalOnRuntimeMode;
|
|
||||||
import at.procon.dip.runtime.config.RuntimeMode;
|
|
||||||
import at.procon.dip.search.service.DocumentLexicalIndexService;
|
|
||||||
import at.procon.ted.model.entity.ProcurementDocument;
|
|
||||||
import at.procon.ted.repository.ProcurementDocumentRepository;
|
|
||||||
import at.procon.ted.util.HashUtils;
|
|
||||||
import java.nio.charset.StandardCharsets;
|
|
||||||
import java.time.OffsetDateTime;
|
|
||||||
import java.util.ArrayList;
|
|
||||||
import java.util.Comparator;
|
|
||||||
import java.util.LinkedHashMap;
|
|
||||||
import java.util.List;
|
|
||||||
import java.util.Map;
|
|
||||||
import java.util.Objects;
|
|
||||||
import java.util.UUID;
|
|
||||||
import lombok.RequiredArgsConstructor;
|
|
||||||
import lombok.extern.slf4j.Slf4j;
|
|
||||||
import org.springframework.stereotype.Service;
|
|
||||||
import org.springframework.transaction.annotation.Propagation;
|
|
||||||
import org.springframework.transaction.annotation.Transactional;
|
|
||||||
import org.springframework.util.StringUtils;
|
|
||||||
|
|
||||||
@Service
|
|
||||||
@ConditionalOnRuntimeMode(RuntimeMode.NEW)
|
|
||||||
@RequiredArgsConstructor
|
|
||||||
@Slf4j
|
|
||||||
public class LegacyTedBackfillWorker {
|
|
||||||
|
|
||||||
private static final String XML_MIME_TYPE = "application/xml";
|
|
||||||
private static final String TEXT_MIME_TYPE = "text/plain";
|
|
||||||
private static final String CHARSET_UTF8 = StandardCharsets.UTF_8.name();
|
|
||||||
|
|
||||||
private final ProcurementDocumentRepository procurementDocumentRepository;
|
|
||||||
private final TedGenericDocumentRootService tedGenericDocumentRootService;
|
|
||||||
private final TedPackageDocumentService tedPackageDocumentService;
|
|
||||||
private final TedNoticeProjectionService tedNoticeProjectionService;
|
|
||||||
private final DocumentService documentService;
|
|
||||||
private final DocumentSourceRepository sourceRepository;
|
|
||||||
private final at.procon.dip.domain.document.service.DocumentSourceService sourceService;
|
|
||||||
private final DocumentContentRepository contentRepository;
|
|
||||||
private final DocumentContentService documentContentService;
|
|
||||||
private final DocumentTextRepresentationRepository representationRepository;
|
|
||||||
private final DocumentRepresentationService documentRepresentationService;
|
|
||||||
private final DocumentLexicalIndexService lexicalIndexService;
|
|
||||||
private final DocumentRelationService documentRelationService;
|
|
||||||
private final TextRepresentationBuildService textRepresentationBuildService;
|
|
||||||
private final RepresentationEmbeddingOrchestrator embeddingOrchestrator;
|
|
||||||
private final EmbeddingProperties embeddingProperties;
|
|
||||||
|
|
||||||
@Transactional(propagation = Propagation.REQUIRES_NEW)
|
|
||||||
public BackfillOutcome backfill(UUID legacyProcurementDocumentId, String importBatchId, boolean queueEmbeddings) {
|
|
||||||
ProcurementDocument legacyDocument = procurementDocumentRepository.findById(legacyProcurementDocumentId)
|
|
||||||
.orElseThrow(() -> new IllegalArgumentException("Unknown legacy TED procurement document id: " + legacyProcurementDocumentId));
|
|
||||||
|
|
||||||
Document document = tedGenericDocumentRootService.ensureGenericTedDocument(legacyDocument);
|
|
||||||
ensureSource(document, legacyDocument, importBatchId);
|
|
||||||
DocumentContent originalContent = ensureOriginalXmlContent(document, legacyDocument);
|
|
||||||
DocumentContent normalizedTextContent = ensureNormalizedTextContent(document, legacyDocument);
|
|
||||||
|
|
||||||
List<TextRepresentationDraft> drafts = buildDrafts(legacyDocument);
|
|
||||||
List<DocumentTextRepresentation> savedRepresentations = ensureRepresentations(document, originalContent, normalizedTextContent, drafts);
|
|
||||||
|
|
||||||
tedPackageDocumentService.ensurePackageDocumentForLegacyNotice(legacyDocument)
|
|
||||||
.ifPresent(packageDocument -> documentRelationService.ensureRelation(new CreateDocumentRelationCommand(
|
|
||||||
packageDocument.getId(),
|
|
||||||
document.getId(),
|
|
||||||
RelationType.CONTAINS,
|
|
||||||
null,
|
|
||||||
legacyDocument.getSourcePath()
|
|
||||||
)));
|
|
||||||
|
|
||||||
UUID projectionId = tedNoticeProjectionService.registerOrRefreshProjection(legacyDocument, document.getId());
|
|
||||||
documentService.updateStatus(document.getId(), DocumentStatus.REPRESENTED);
|
|
||||||
|
|
||||||
if (queueEmbeddings) {
|
|
||||||
queueEmbeddings(document.getId(), savedRepresentations);
|
|
||||||
}
|
|
||||||
|
|
||||||
return new BackfillOutcome(document.getId(), projectionId);
|
|
||||||
}
|
|
||||||
|
|
||||||
private void ensureSource(Document document, ProcurementDocument legacyDocument, String importBatchId) {
|
|
||||||
String externalSourceId = legacyDocument.getId().toString();
|
|
||||||
DocumentSource existing = sourceRepository.findByDocument_Id(document.getId()).stream()
|
|
||||||
.filter(candidate -> candidate.getSourceType() == SourceType.MIGRATION || Objects.equals(candidate.getExternalSourceId(), externalSourceId))
|
|
||||||
.findFirst()
|
|
||||||
.orElse(null);
|
|
||||||
if (existing == null) {
|
|
||||||
sourceService.addSource(new AddDocumentSourceCommand(
|
|
||||||
document.getId(),
|
|
||||||
SourceType.MIGRATION,
|
|
||||||
externalSourceId,
|
|
||||||
legacyDocument.getSourcePath(),
|
|
||||||
legacyDocument.getSourceFilename(),
|
|
||||||
null,
|
|
||||||
importBatchId,
|
|
||||||
legacyDocument.getCreatedAt() != null ? legacyDocument.getCreatedAt() : OffsetDateTime.now()
|
|
||||||
));
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
existing.setSourceType(SourceType.MIGRATION);
|
|
||||||
existing.setExternalSourceId(externalSourceId);
|
|
||||||
existing.setSourceUri(legacyDocument.getSourcePath());
|
|
||||||
existing.setSourceFilename(legacyDocument.getSourceFilename());
|
|
||||||
existing.setImportBatchId(importBatchId);
|
|
||||||
existing.setReceivedAt(legacyDocument.getCreatedAt() != null ? legacyDocument.getCreatedAt() : existing.getReceivedAt());
|
|
||||||
sourceRepository.save(existing);
|
|
||||||
}
|
|
||||||
|
|
||||||
private DocumentContent ensureOriginalXmlContent(Document document, ProcurementDocument legacyDocument) {
|
|
||||||
String xml = legacyDocument.getXmlDocument();
|
|
||||||
long sizeBytes = legacyDocument.getFileSizeBytes() != null
|
|
||||||
? legacyDocument.getFileSizeBytes()
|
|
||||||
: (xml == null ? 0L : (long) xml.getBytes(StandardCharsets.UTF_8).length);
|
|
||||||
return ensureContent(document, ContentRole.ORIGINAL, XML_MIME_TYPE, xml, legacyDocument.getDocumentHash(), sizeBytes);
|
|
||||||
}
|
|
||||||
|
|
||||||
private DocumentContent ensureNormalizedTextContent(Document document, ProcurementDocument legacyDocument) {
|
|
||||||
String normalizedText = resolveNormalizedText(legacyDocument);
|
|
||||||
String contentHash = HashUtils.computeSha256(legacyDocument.getDocumentHash() + ":NORMALIZED_TEXT:" + normalizedText);
|
|
||||||
return ensureContent(document, ContentRole.NORMALIZED_TEXT, TEXT_MIME_TYPE, normalizedText, contentHash, (long) normalizedText.length());
|
|
||||||
}
|
|
||||||
|
|
||||||
private DocumentContent ensureContent(Document document,
|
|
||||||
ContentRole role,
|
|
||||||
String mimeType,
|
|
||||||
String text,
|
|
||||||
String contentHash,
|
|
||||||
Long sizeBytes) {
|
|
||||||
List<DocumentContent> existing = contentRepository.findByDocument_IdAndContentRole(document.getId(), role);
|
|
||||||
if (!existing.isEmpty()) {
|
|
||||||
DocumentContent content = existing.get(0);
|
|
||||||
content.setStorageType(StorageType.DB_TEXT);
|
|
||||||
content.setMimeType(mimeType);
|
|
||||||
content.setCharsetName(CHARSET_UTF8);
|
|
||||||
content.setTextContent(text);
|
|
||||||
content.setBinaryContent(null);
|
|
||||||
content.setBinaryRef(null);
|
|
||||||
content.setContentHash(contentHash);
|
|
||||||
content.setSizeBytes(sizeBytes);
|
|
||||||
return contentRepository.save(content);
|
|
||||||
}
|
|
||||||
|
|
||||||
return documentContentService.addContent(new AddDocumentContentCommand(
|
|
||||||
document.getId(),
|
|
||||||
role,
|
|
||||||
StorageType.DB_TEXT,
|
|
||||||
mimeType,
|
|
||||||
CHARSET_UTF8,
|
|
||||||
text,
|
|
||||||
null,
|
|
||||||
null,
|
|
||||||
contentHash,
|
|
||||||
sizeBytes
|
|
||||||
));
|
|
||||||
}
|
|
||||||
|
|
||||||
private List<TextRepresentationDraft> buildDrafts(ProcurementDocument legacyDocument) {
|
|
||||||
String normalizedText = resolveNormalizedText(legacyDocument);
|
|
||||||
|
|
||||||
Map<ContentRole, String> derivedText = new LinkedHashMap<>();
|
|
||||||
derivedText.put(ContentRole.NORMALIZED_TEXT, normalizedText);
|
|
||||||
|
|
||||||
SourceDescriptor sourceDescriptor = new SourceDescriptor(
|
|
||||||
null,
|
|
||||||
SourceType.MIGRATION,
|
|
||||||
legacyDocument.getId().toString(),
|
|
||||||
legacyDocument.getSourcePath(),
|
|
||||||
legacyDocument.getSourceFilename(),
|
|
||||||
XML_MIME_TYPE,
|
|
||||||
null,
|
|
||||||
legacyDocument.getXmlDocument(),
|
|
||||||
legacyDocument.getCreatedAt(),
|
|
||||||
OriginalContentStoragePolicy.DEFAULT,
|
|
||||||
Map.of("legacyProcurementDocumentId", legacyDocument.getId().toString())
|
|
||||||
);
|
|
||||||
|
|
||||||
DetectionResult detectionResult = new DetectionResult(
|
|
||||||
DocumentType.TED_NOTICE,
|
|
||||||
DocumentFamily.PROCUREMENT,
|
|
||||||
XML_MIME_TYPE,
|
|
||||||
legacyDocument.getLanguageCode(),
|
|
||||||
Map.of()
|
|
||||||
);
|
|
||||||
|
|
||||||
ExtractionResult extractionResult = new ExtractionResult(
|
|
||||||
derivedText,
|
|
||||||
List.of(new ExtractedStructuredPayload("ted-notice", buildStructuredAttributes(legacyDocument))),
|
|
||||||
List.of()
|
|
||||||
);
|
|
||||||
|
|
||||||
return textRepresentationBuildService.build(new RepresentationBuildRequest(
|
|
||||||
sourceDescriptor,
|
|
||||||
detectionResult,
|
|
||||||
extractionResult
|
|
||||||
));
|
|
||||||
}
|
|
||||||
|
|
||||||
private List<DocumentTextRepresentation> ensureRepresentations(Document document,
|
|
||||||
DocumentContent originalContent,
|
|
||||||
DocumentContent normalizedTextContent,
|
|
||||||
List<TextRepresentationDraft> drafts) {
|
|
||||||
if (drafts == null || drafts.isEmpty()) {
|
|
||||||
return List.of();
|
|
||||||
}
|
|
||||||
|
|
||||||
List<DocumentTextRepresentation> existing = new ArrayList<>(representationRepository.findByDocument_Id(document.getId()));
|
|
||||||
existing.sort(Comparator.comparing(DocumentTextRepresentation::getCreatedAt, Comparator.nullsLast(Comparator.naturalOrder())));
|
|
||||||
|
|
||||||
List<DocumentTextRepresentation> saved = new ArrayList<>();
|
|
||||||
for (TextRepresentationDraft draft : drafts) {
|
|
||||||
if (!StringUtils.hasText(draft.textBody())) {
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
|
|
||||||
DocumentTextRepresentation representation = findMatchingRepresentation(existing, draft)
|
|
||||||
.orElseGet(DocumentTextRepresentation::new);
|
|
||||||
|
|
||||||
DocumentContent linkedContent = draft.sourceContentRole() == ContentRole.NORMALIZED_TEXT
|
|
||||||
? normalizedTextContent
|
|
||||||
: originalContent;
|
|
||||||
|
|
||||||
boolean isNew = representation.getId() == null;
|
|
||||||
representation.setDocument(document);
|
|
||||||
representation.setContent(linkedContent);
|
|
||||||
representation.setRepresentationType(draft.representationType());
|
|
||||||
representation.setBuilderKey(draft.builderKey());
|
|
||||||
representation.setLanguageCode(draft.languageCode());
|
|
||||||
representation.setTokenCount(draft.textBody().length());
|
|
||||||
representation.setCharCount(draft.textBody().length());
|
|
||||||
representation.setChunkIndex(draft.chunkIndex());
|
|
||||||
representation.setChunkStartOffset(draft.chunkStartOffset());
|
|
||||||
representation.setChunkEndOffset(draft.chunkEndOffset());
|
|
||||||
representation.setPrimaryRepresentation(draft.primary());
|
|
||||||
representation.setTextBody(draft.textBody());
|
|
||||||
|
|
||||||
DocumentTextRepresentation savedRepresentation;
|
|
||||||
if (isNew) {
|
|
||||||
savedRepresentation = documentRepresentationService.addRepresentation(new AddDocumentTextRepresentationCommand(
|
|
||||||
document.getId(),
|
|
||||||
linkedContent == null ? null : linkedContent.getId(),
|
|
||||||
draft.representationType(),
|
|
||||||
draft.builderKey(),
|
|
||||||
draft.languageCode(),
|
|
||||||
draft.textBody().length(),
|
|
||||||
draft.chunkIndex(),
|
|
||||||
draft.chunkStartOffset(),
|
|
||||||
draft.chunkEndOffset(),
|
|
||||||
draft.primary(),
|
|
||||||
draft.textBody()
|
|
||||||
));
|
|
||||||
existing.add(savedRepresentation);
|
|
||||||
} else {
|
|
||||||
savedRepresentation = representationRepository.save(representation);
|
|
||||||
lexicalIndexService.indexRepresentation(savedRepresentation.getId());
|
|
||||||
}
|
|
||||||
saved.add(savedRepresentation);
|
|
||||||
}
|
|
||||||
return saved;
|
|
||||||
}
|
|
||||||
|
|
||||||
private java.util.Optional<DocumentTextRepresentation> findMatchingRepresentation(List<DocumentTextRepresentation> existing,
|
|
||||||
TextRepresentationDraft draft) {
|
|
||||||
return existing.stream()
|
|
||||||
.filter(candidate -> candidate.getRepresentationType() == draft.representationType())
|
|
||||||
.filter(candidate -> Objects.equals(candidate.getChunkIndex(), draft.chunkIndex()))
|
|
||||||
.filter(candidate -> Objects.equals(candidate.isPrimaryRepresentation(), draft.primary()))
|
|
||||||
.findFirst();
|
|
||||||
}
|
|
||||||
|
|
||||||
private void queueEmbeddings(UUID documentId, List<DocumentTextRepresentation> representations) {
|
|
||||||
if (!embeddingProperties.isEnabled() || !StringUtils.hasText(embeddingProperties.getDefaultDocumentModel())) {
|
|
||||||
log.debug("Skipping embedding queue for migrated document {} because no default document model is configured", documentId);
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
for (DocumentTextRepresentation representation : representations) {
|
|
||||||
RepresentationType type = representation.getRepresentationType();
|
|
||||||
boolean queue = switch (type) {
|
|
||||||
case SEMANTIC_TEXT -> true;
|
|
||||||
case CHUNK -> true;
|
|
||||||
default -> false;
|
|
||||||
};
|
|
||||||
if (!queue) {
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
embeddingOrchestrator.enqueueRepresentation(documentId, representation.getId(), embeddingProperties.getDefaultDocumentModel());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private String resolveNormalizedText(ProcurementDocument legacyDocument) {
|
|
||||||
if (StringUtils.hasText(legacyDocument.getTextContent())) {
|
|
||||||
return legacyDocument.getTextContent().trim();
|
|
||||||
}
|
|
||||||
|
|
||||||
StringBuilder sb = new StringBuilder();
|
|
||||||
appendLine(sb, legacyDocument.getProjectTitle());
|
|
||||||
appendBlank(sb, legacyDocument.getProjectDescription());
|
|
||||||
appendLine(sb, legacyDocument.getBuyerName());
|
|
||||||
appendLine(sb, joinArray(legacyDocument.getCpvCodes()));
|
|
||||||
appendLine(sb, joinArray(legacyDocument.getNutsCodes()));
|
|
||||||
String fallback = sb.toString().trim();
|
|
||||||
return StringUtils.hasText(fallback) ? fallback : legacyDocument.getDocumentHash();
|
|
||||||
}
|
|
||||||
|
|
||||||
private Map<String, Object> buildStructuredAttributes(ProcurementDocument legacyDocument) {
|
|
||||||
Map<String, Object> attrs = new LinkedHashMap<>();
|
|
||||||
putIfText(attrs, "title", legacyDocument.getProjectTitle());
|
|
||||||
putIfText(attrs, "description", legacyDocument.getProjectDescription());
|
|
||||||
putIfText(attrs, "buyerName", legacyDocument.getBuyerName());
|
|
||||||
putIfText(attrs, "cpvCodes", joinArray(legacyDocument.getCpvCodes()));
|
|
||||||
putIfText(attrs, "nutsCodes", joinArray(legacyDocument.getNutsCodes()));
|
|
||||||
putIfText(attrs, "publicationId", legacyDocument.getPublicationId());
|
|
||||||
return attrs;
|
|
||||||
}
|
|
||||||
|
|
||||||
private void putIfText(Map<String, Object> target, String key, String value) {
|
|
||||||
if (StringUtils.hasText(value)) {
|
|
||||||
target.put(key, value);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private String joinArray(String[] values) {
|
|
||||||
if (values == null || values.length == 0) {
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
return String.join(", ", values);
|
|
||||||
}
|
|
||||||
|
|
||||||
private void appendLine(StringBuilder sb, String value) {
|
|
||||||
if (StringUtils.hasText(value)) {
|
|
||||||
if (sb.length() > 0) {
|
|
||||||
sb.append('\n');
|
|
||||||
}
|
|
||||||
sb.append(value.trim());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private void appendBlank(StringBuilder sb, String value) {
|
|
||||||
if (StringUtils.hasText(value)) {
|
|
||||||
if (sb.length() > 0) {
|
|
||||||
sb.append("\n\n");
|
|
||||||
}
|
|
||||||
sb.append(value.trim());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
public record BackfillOutcome(UUID documentId, UUID projectionId) {
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
@ -1,31 +0,0 @@
|
|||||||
package at.procon.dip.migration.startup;
|
|
||||||
|
|
||||||
import at.procon.dip.migration.config.LegacyTedBackfillProperties;
|
|
||||||
import at.procon.dip.migration.service.LegacyTedBackfillMigrationService;
|
|
||||||
import at.procon.dip.runtime.condition.ConditionalOnRuntimeMode;
|
|
||||||
import at.procon.dip.runtime.config.RuntimeMode;
|
|
||||||
import lombok.RequiredArgsConstructor;
|
|
||||||
import lombok.extern.slf4j.Slf4j;
|
|
||||||
import org.springframework.boot.ApplicationArguments;
|
|
||||||
import org.springframework.boot.ApplicationRunner;
|
|
||||||
import org.springframework.stereotype.Component;
|
|
||||||
|
|
||||||
@Component
|
|
||||||
@ConditionalOnRuntimeMode(RuntimeMode.NEW)
|
|
||||||
@RequiredArgsConstructor
|
|
||||||
@Slf4j
|
|
||||||
public class LegacyTedBackfillStartupRunner implements ApplicationRunner {
|
|
||||||
|
|
||||||
private final LegacyTedBackfillProperties properties;
|
|
||||||
private final LegacyTedBackfillMigrationService migrationService;
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void run(ApplicationArguments args) {
|
|
||||||
if (!properties.isEnabled() || !properties.isStartupEnabled()) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
log.info("Startup-triggered legacy TED backfill is enabled");
|
|
||||||
migrationService.runBackfill();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@ -1,9 +0,0 @@
|
|||||||
package at.procon.ted.repository;
|
|
||||||
|
|
||||||
import java.time.Instant;
|
|
||||||
import java.util.UUID;
|
|
||||||
|
|
||||||
public interface LegacyTedMigrationCursor {
|
|
||||||
UUID getId();
|
|
||||||
Instant getCreatedAt();
|
|
||||||
}
|
|
||||||
@ -1,21 +0,0 @@
|
|||||||
-- Store the TED daily package identifier directly on the Phase 3 TED notice projection.
|
|
||||||
-- This makes migration, audit, and repair flows package-aware without having to derive the
|
|
||||||
-- package membership from source paths at query time.
|
|
||||||
|
|
||||||
SET search_path TO TED, DOC, public;
|
|
||||||
|
|
||||||
ALTER TABLE IF EXISTS TED.ted_notice_projection
|
|
||||||
ADD COLUMN IF NOT EXISTS package_identifier VARCHAR(20);
|
|
||||||
|
|
||||||
CREATE INDEX IF NOT EXISTS idx_ted_notice_projection_package_identifier
|
|
||||||
ON TED.ted_notice_projection(package_identifier);
|
|
||||||
|
|
||||||
-- Backfill from the linked legacy TED document's source path when the path contains the
|
|
||||||
-- extracted package directory, e.g. .../extracted/202600003/123.xml
|
|
||||||
UPDATE TED.ted_notice_projection projection
|
|
||||||
SET package_identifier = substring(legacy.source_path from '(?:^|[\\/])((?:19|20)[0-9]{7})(?:[\\/]|$)')
|
|
||||||
FROM TED.procurement_document legacy
|
|
||||||
WHERE projection.legacy_procurement_document_id = legacy.id
|
|
||||||
AND projection.package_identifier IS NULL
|
|
||||||
AND legacy.source_path IS NOT NULL
|
|
||||||
AND substring(legacy.source_path from '(?:^|[\\/])((?:19|20)[0-9]{7})(?:[\\/]|$)') IS NOT NULL;
|
|
||||||
@ -1,75 +0,0 @@
|
|||||||
SET search_path TO TED, DOC, public;
|
|
||||||
|
|
||||||
ALTER TABLE IF EXISTS TED.ted_notice_projection
|
|
||||||
ADD COLUMN IF NOT EXISTS package_identifier VARCHAR(32);
|
|
||||||
|
|
||||||
CREATE INDEX IF NOT EXISTS idx_ted_notice_projection_package_identifier
|
|
||||||
ON TED.ted_notice_projection(package_identifier);
|
|
||||||
|
|
||||||
ALTER TABLE IF EXISTS TED.organization
|
|
||||||
ALTER COLUMN city TYPE TEXT;
|
|
||||||
|
|
||||||
ALTER TABLE IF EXISTS TED.organization
|
|
||||||
ALTER COLUMN company_id TYPE TEXT;
|
|
||||||
|
|
||||||
ALTER TABLE IF EXISTS TED.procurement_document
|
|
||||||
ALTER COLUMN buyer_city TYPE TEXT;
|
|
||||||
|
|
||||||
ALTER TABLE IF EXISTS TED.ted_notice_organization
|
|
||||||
ALTER COLUMN city TYPE TEXT;
|
|
||||||
|
|
||||||
ALTER TABLE IF EXISTS TED.ted_notice_projection
|
|
||||||
ALTER COLUMN buyer_city TYPE TEXT;
|
|
||||||
|
|
||||||
UPDATE TED.ted_notice_projection p
|
|
||||||
SET package_identifier = substring(coalesce(d.source_path, d.source_filename) from '(20[0-9]{2}[0-9]{5})')
|
|
||||||
FROM TED.procurement_document d
|
|
||||||
WHERE p.legacy_procurement_document_id = d.id
|
|
||||||
AND p.package_identifier IS NULL
|
|
||||||
AND substring(coalesce(d.source_path, d.source_filename) from '(20[0-9]{2}[0-9]{5})') IS NOT NULL;
|
|
||||||
|
|
||||||
SET search_path TO DOC, public;
|
|
||||||
|
|
||||||
CREATE TABLE IF NOT EXISTS DOC.doc_legacy_ted_migration_run (
|
|
||||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
|
||||||
status VARCHAR(32) NOT NULL,
|
|
||||||
import_batch_id VARCHAR(255),
|
|
||||||
queue_embeddings BOOLEAN NOT NULL DEFAULT FALSE,
|
|
||||||
batch_size INTEGER NOT NULL DEFAULT 100,
|
|
||||||
max_documents_per_run BIGINT,
|
|
||||||
processed_count BIGINT NOT NULL DEFAULT 0,
|
|
||||||
success_count BIGINT NOT NULL DEFAULT 0,
|
|
||||||
failed_count BIGINT NOT NULL DEFAULT 0,
|
|
||||||
last_legacy_created_at TIMESTAMP WITH TIME ZONE,
|
|
||||||
last_legacy_document_id UUID,
|
|
||||||
last_doc_document_id UUID,
|
|
||||||
last_projection_id UUID,
|
|
||||||
last_error TEXT,
|
|
||||||
started_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
||||||
updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
||||||
completed_at TIMESTAMP WITH TIME ZONE
|
|
||||||
);
|
|
||||||
|
|
||||||
CREATE INDEX IF NOT EXISTS idx_doc_legacy_ted_mig_run_status
|
|
||||||
ON DOC.doc_legacy_ted_migration_run(status);
|
|
||||||
CREATE INDEX IF NOT EXISTS idx_doc_legacy_ted_mig_run_started
|
|
||||||
ON DOC.doc_legacy_ted_migration_run(started_at DESC);
|
|
||||||
|
|
||||||
CREATE TABLE IF NOT EXISTS DOC.doc_legacy_ted_migration_checkpoint (
|
|
||||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
|
||||||
run_id UUID NOT NULL REFERENCES DOC.doc_legacy_ted_migration_run(id) ON DELETE CASCADE,
|
|
||||||
batch_number INTEGER NOT NULL,
|
|
||||||
batch_processed_count INTEGER NOT NULL DEFAULT 0,
|
|
||||||
cumulative_processed_count BIGINT NOT NULL DEFAULT 0,
|
|
||||||
last_legacy_created_at TIMESTAMP WITH TIME ZONE,
|
|
||||||
last_legacy_document_id UUID,
|
|
||||||
last_doc_document_id UUID,
|
|
||||||
last_projection_id UUID,
|
|
||||||
note TEXT,
|
|
||||||
created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
||||||
);
|
|
||||||
|
|
||||||
CREATE INDEX IF NOT EXISTS idx_doc_legacy_ted_mig_ckpt_run
|
|
||||||
ON DOC.doc_legacy_ted_migration_checkpoint(run_id);
|
|
||||||
CREATE INDEX IF NOT EXISTS idx_doc_legacy_ted_mig_ckpt_batch
|
|
||||||
ON DOC.doc_legacy_ted_migration_checkpoint(batch_number);
|
|
||||||
@ -1,95 +0,0 @@
|
|||||||
SET search_path TO TED, DOC, public;
|
|
||||||
|
|
||||||
WITH legacy_package_map AS (
|
|
||||||
SELECT
|
|
||||||
d.id AS legacy_procurement_document_id,
|
|
||||||
p.document_id AS child_document_id,
|
|
||||||
substring(coalesce(d.source_path, d.source_filename) from '(20[0-9]{7})') AS package_identifier
|
|
||||||
FROM TED.procurement_document d
|
|
||||||
JOIN TED.ted_notice_projection p
|
|
||||||
ON p.legacy_procurement_document_id = d.id
|
|
||||||
WHERE substring(coalesce(d.source_path, d.source_filename) from '(20[0-9]{7})') IS NOT NULL
|
|
||||||
), package_documents AS (
|
|
||||||
SELECT DISTINCT
|
|
||||||
l.package_identifier,
|
|
||||||
'TED:package:' || l.package_identifier AS business_key,
|
|
||||||
encode(digest('TED:package:' || l.package_identifier, 'sha256'), 'hex') AS dedup_hash
|
|
||||||
FROM legacy_package_map l
|
|
||||||
)
|
|
||||||
INSERT INTO DOC.doc_document (
|
|
||||||
id, visibility, document_type, document_family, status, title, summary, mime_type, business_key, dedup_hash
|
|
||||||
)
|
|
||||||
SELECT
|
|
||||||
gen_random_uuid(),
|
|
||||||
'PUBLIC',
|
|
||||||
'TED_PACKAGE',
|
|
||||||
'PROCUREMENT',
|
|
||||||
CASE
|
|
||||||
WHEN pkg.download_status = 'COMPLETED' THEN 'CLASSIFIED'
|
|
||||||
WHEN pkg.download_status IN ('FAILED', 'NOT_FOUND') THEN 'FAILED'
|
|
||||||
ELSE 'RECEIVED'
|
|
||||||
END,
|
|
||||||
'TED Daily Package ' || pd.package_identifier,
|
|
||||||
CASE
|
|
||||||
WHEN pkg.package_identifier IS NULL THEN 'TED daily package ' || pd.package_identifier
|
|
||||||
ELSE 'TED daily package ' || pd.package_identifier ||
|
|
||||||
' (status=' || coalesce(pkg.download_status::text, 'UNKNOWN') ||
|
|
||||||
', xmlFileCount=' || coalesce(pkg.xml_file_count::text, 'null') ||
|
|
||||||
', processedCount=' || coalesce(pkg.processed_count::text, 'null') ||
|
|
||||||
', failedCount=' || coalesce(pkg.failed_count::text, 'null') || ')'
|
|
||||||
END,
|
|
||||||
'application/gzip',
|
|
||||||
pd.business_key,
|
|
||||||
pd.dedup_hash
|
|
||||||
FROM package_documents pd
|
|
||||||
LEFT JOIN TED.ted_daily_package pkg
|
|
||||||
ON pkg.package_identifier = pd.package_identifier
|
|
||||||
WHERE NOT EXISTS (
|
|
||||||
SELECT 1
|
|
||||||
FROM DOC.doc_document existing
|
|
||||||
WHERE existing.business_key = pd.business_key
|
|
||||||
);
|
|
||||||
|
|
||||||
UPDATE TED.ted_notice_projection p
|
|
||||||
SET package_identifier = substring(coalesce(d.source_path, d.source_filename) from '(20[0-9]{7})')
|
|
||||||
FROM TED.procurement_document d
|
|
||||||
WHERE p.legacy_procurement_document_id = d.id
|
|
||||||
AND substring(coalesce(d.source_path, d.source_filename) from '(20[0-9]{7})') IS NOT NULL
|
|
||||||
AND p.package_identifier IS DISTINCT FROM substring(coalesce(d.source_path, d.source_filename) from '(20[0-9]{7})');
|
|
||||||
|
|
||||||
WITH legacy_package_map AS (
|
|
||||||
SELECT
|
|
||||||
p.document_id AS child_document_id,
|
|
||||||
substring(coalesce(d.source_path, d.source_filename) from '(20[0-9]{7})') AS package_identifier
|
|
||||||
FROM TED.procurement_document d
|
|
||||||
JOIN TED.ted_notice_projection p
|
|
||||||
ON p.legacy_procurement_document_id = d.id
|
|
||||||
WHERE substring(coalesce(d.source_path, d.source_filename) from '(20[0-9]{7})') IS NOT NULL
|
|
||||||
), package_documents AS (
|
|
||||||
SELECT
|
|
||||||
doc.id AS parent_document_id,
|
|
||||||
substring(doc.business_key from '(20[0-9]{7})') AS package_identifier
|
|
||||||
FROM DOC.doc_document doc
|
|
||||||
WHERE doc.document_type = 'TED_PACKAGE'
|
|
||||||
AND doc.business_key LIKE 'TED:package:%'
|
|
||||||
)
|
|
||||||
INSERT INTO DOC.doc_relation (
|
|
||||||
id, parent_document_id, child_document_id, relation_type, sort_order, relation_metadata
|
|
||||||
)
|
|
||||||
SELECT
|
|
||||||
gen_random_uuid(),
|
|
||||||
pkg.parent_document_id,
|
|
||||||
l.child_document_id,
|
|
||||||
'CONTAINS',
|
|
||||||
NULL,
|
|
||||||
'packageIdentifier=' || l.package_identifier
|
|
||||||
FROM legacy_package_map l
|
|
||||||
JOIN package_documents pkg
|
|
||||||
ON pkg.package_identifier = l.package_identifier
|
|
||||||
WHERE NOT EXISTS (
|
|
||||||
SELECT 1
|
|
||||||
FROM DOC.doc_relation rel
|
|
||||||
WHERE rel.parent_document_id = pkg.parent_document_id
|
|
||||||
AND rel.child_document_id = l.child_document_id
|
|
||||||
AND rel.relation_type = 'CONTAINS'
|
|
||||||
);
|
|
||||||
Loading…
Reference in New Issue