723 lines
25 KiB
Python
723 lines
25 KiB
Python
"""Atomic PostgreSQL activation for deterministic Booking Excel imports."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from typing import Any, Callable, Mapping, Optional
|
|
|
|
from arr_ingestion.contracts import ArtifactRef
|
|
|
|
from .excel import (
|
|
PROCESSOR_NAME,
|
|
PROCESSOR_VERSION,
|
|
RESULT_SCHEMA_VERSION,
|
|
XLSX_MIME,
|
|
ExcelDocument,
|
|
build_excel_parse_result,
|
|
canonical_json_bytes,
|
|
rule_set_sha256,
|
|
)
|
|
|
|
|
|
TARGET_DATABASE = "booking_test"
|
|
DATABASE_ENV = "ARR_DATABASE_URL"
|
|
ARTIFACT_STORAGE_PROVIDER = "oss"
|
|
ARTIFACT_BUCKET_ALIAS = "arr-private"
|
|
|
|
|
|
class BookingExcelRepositoryError(RuntimeError):
|
|
def __init__(self, code: str, safe_message: str, status: int = 503):
|
|
super().__init__(safe_message)
|
|
self.code = code
|
|
self.safe_message = safe_message
|
|
self.status = status
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DatabaseConfig:
|
|
dsn: str
|
|
|
|
@classmethod
|
|
def from_environment(cls) -> "DatabaseConfig":
|
|
dsn = os.environ.get(DATABASE_ENV, "").strip()
|
|
if not dsn:
|
|
raise BookingExcelRepositoryError(
|
|
"BOOKING_EXCEL_DATABASE_CONFIG_MISSING",
|
|
"Booking Excel 数据库配置缺失",
|
|
)
|
|
return cls(dsn=dsn)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BookingSourceSummary:
|
|
source_batch_id: int
|
|
source_kind: str
|
|
filename: Optional[str]
|
|
source_rows: int
|
|
worksheet_count: int
|
|
distinct_group_codes: int
|
|
room_quantity: int
|
|
activated_at: Optional[datetime]
|
|
disposition: str = "current"
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return {
|
|
"source_batch_id": self.source_batch_id,
|
|
"source_type": (
|
|
"excel" if self.source_kind == "booking_excel" else "historical"
|
|
),
|
|
"filename": self.filename if self.source_kind == "booking_excel" else None,
|
|
"source_rows": self.source_rows,
|
|
"worksheet_count": self.worksheet_count,
|
|
"distinct_group_codes": self.distinct_group_codes,
|
|
"room_quantity": self.room_quantity,
|
|
"activated_at": (
|
|
self.activated_at.isoformat() if self.activated_at is not None else None
|
|
),
|
|
"disposition": self.disposition,
|
|
}
|
|
|
|
|
|
CURRENT_SOURCE_SQL = """
|
|
SELECT
|
|
batch.id,
|
|
batch.source_kind,
|
|
CASE
|
|
WHEN batch.source_kind = 'booking_excel' THEN artifact.original_filename
|
|
ELSE NULL
|
|
END AS uploaded_filename,
|
|
count(source.id)::bigint AS source_rows,
|
|
count(DISTINCT source.source_worksheet)::bigint AS worksheet_count,
|
|
count(DISTINCT source.group_code_key)::bigint AS distinct_group_codes,
|
|
COALESCE(sum(source.no_of_rooms), 0)::bigint AS room_quantity,
|
|
active_source.activated_at
|
|
FROM booking.current_source_batch AS active_source
|
|
JOIN booking.source_batches AS batch
|
|
ON batch.id = active_source.source_batch_id
|
|
JOIN ingestion.artifacts AS artifact
|
|
ON artifact.id = batch.source_artifact_id
|
|
LEFT JOIN booking.source_rows AS source
|
|
ON source.source_batch_id = batch.id
|
|
WHERE active_source.singleton
|
|
AND batch.batch_status = 'accepted'
|
|
GROUP BY
|
|
batch.id,
|
|
batch.source_kind,
|
|
artifact.original_filename,
|
|
active_source.activated_at
|
|
""".strip()
|
|
|
|
|
|
BATCH_SUMMARY_SQL = """
|
|
SELECT
|
|
batch.id,
|
|
batch.source_kind,
|
|
CASE
|
|
WHEN batch.source_kind = 'booking_excel' THEN artifact.original_filename
|
|
ELSE NULL
|
|
END AS uploaded_filename,
|
|
count(source.id)::bigint AS source_rows,
|
|
count(DISTINCT source.source_worksheet)::bigint AS worksheet_count,
|
|
count(DISTINCT source.group_code_key)::bigint AS distinct_group_codes,
|
|
COALESCE(sum(source.no_of_rooms), 0)::bigint AS room_quantity,
|
|
active_source.activated_at
|
|
FROM booking.source_batches AS batch
|
|
JOIN ingestion.artifacts AS artifact
|
|
ON artifact.id = batch.source_artifact_id
|
|
LEFT JOIN booking.source_rows AS source
|
|
ON source.source_batch_id = batch.id
|
|
LEFT JOIN booking.current_source_batch AS active_source
|
|
ON active_source.source_batch_id = batch.id
|
|
AND active_source.singleton
|
|
WHERE batch.id = %s
|
|
AND batch.batch_status = 'accepted'
|
|
GROUP BY
|
|
batch.id,
|
|
batch.source_kind,
|
|
artifact.original_filename,
|
|
active_source.activated_at
|
|
""".strip()
|
|
|
|
|
|
def _default_connect(dsn: str) -> Any:
|
|
try:
|
|
import psycopg # type: ignore[import-not-found]
|
|
except ImportError:
|
|
raise BookingExcelRepositoryError(
|
|
"BOOKING_EXCEL_DATABASE_DRIVER_UNAVAILABLE",
|
|
"Booking Excel 数据库驱动不可用",
|
|
) from None
|
|
try:
|
|
return psycopg.connect(dsn, autocommit=False)
|
|
except Exception:
|
|
raise BookingExcelRepositoryError(
|
|
"BOOKING_EXCEL_DATABASE_UNAVAILABLE",
|
|
"Booking Excel 数据库暂不可用",
|
|
) from None
|
|
|
|
|
|
class PostgresBookingExcelRepository:
|
|
def __init__(
|
|
self,
|
|
config: DatabaseConfig,
|
|
connect: Optional[Callable[[str], Any]] = None,
|
|
) -> None:
|
|
self._config = config
|
|
self._connect = connect or _default_connect
|
|
|
|
def _open(self) -> Any:
|
|
try:
|
|
return self._connect(self._config.dsn)
|
|
except BookingExcelRepositoryError:
|
|
raise
|
|
except Exception:
|
|
raise BookingExcelRepositoryError(
|
|
"BOOKING_EXCEL_DATABASE_UNAVAILABLE",
|
|
"Booking Excel 数据库暂不可用",
|
|
) from None
|
|
|
|
@staticmethod
|
|
def _begin(cursor: Any, *, read_only: bool = False) -> None:
|
|
mode = "REPEATABLE READ READ ONLY" if read_only else "SERIALIZABLE"
|
|
cursor.execute(f"SET TRANSACTION ISOLATION LEVEL {mode}")
|
|
cursor.execute("SET LOCAL lock_timeout = '10s'")
|
|
cursor.execute("SET LOCAL statement_timeout = '120s'")
|
|
cursor.execute(
|
|
"SELECT current_database(), to_regclass('booking.current_source_batch')"
|
|
)
|
|
row = cursor.fetchone()
|
|
if not row or row[0] != TARGET_DATABASE or row[1] is None:
|
|
raise BookingExcelRepositoryError(
|
|
"BOOKING_EXCEL_DATABASE_NOT_READY",
|
|
"Booking Excel 数据库迁移尚未就绪",
|
|
)
|
|
|
|
@staticmethod
|
|
def _summary_from_row(
|
|
row: object,
|
|
*,
|
|
disposition: str = "current",
|
|
) -> Optional[BookingSourceSummary]:
|
|
if row is None:
|
|
return None
|
|
values = tuple(row) # type: ignore[arg-type]
|
|
return BookingSourceSummary(
|
|
source_batch_id=int(values[0]),
|
|
source_kind=str(values[1]),
|
|
filename=str(values[2]) if values[2] else None,
|
|
source_rows=int(values[3]),
|
|
worksheet_count=int(values[4]),
|
|
distinct_group_codes=int(values[5]),
|
|
room_quantity=int(values[6]),
|
|
activated_at=values[7] if isinstance(values[7], datetime) else None,
|
|
disposition=disposition,
|
|
)
|
|
|
|
def current_source(self) -> Optional[BookingSourceSummary]:
|
|
connection = self._open()
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
self._begin(cursor, read_only=True)
|
|
cursor.execute(CURRENT_SOURCE_SQL)
|
|
return self._summary_from_row(cursor.fetchone())
|
|
except BookingExcelRepositoryError:
|
|
raise
|
|
except Exception:
|
|
raise BookingExcelRepositoryError(
|
|
"BOOKING_EXCEL_DATABASE_UNAVAILABLE",
|
|
"当前 Excel 数据源暂时无法读取",
|
|
) from None
|
|
finally:
|
|
try:
|
|
connection.rollback()
|
|
finally:
|
|
connection.close()
|
|
|
|
@staticmethod
|
|
def _find_existing_batch(cursor: Any, source_sha256: str) -> Optional[int]:
|
|
cursor.execute(
|
|
"""
|
|
SELECT batch.id
|
|
FROM booking.source_batches AS batch
|
|
JOIN ingestion.artifacts AS artifact
|
|
ON artifact.id = batch.source_artifact_id
|
|
WHERE batch.source_kind = 'booking_excel'
|
|
AND batch.batch_status = 'accepted'
|
|
AND artifact.artifact_kind = 'booking_excel'
|
|
AND artifact.sha256 = %s
|
|
ORDER BY batch.id
|
|
LIMIT 1
|
|
FOR UPDATE OF batch
|
|
""",
|
|
(source_sha256,),
|
|
)
|
|
row = cursor.fetchone()
|
|
return int(row[0]) if row is not None else None
|
|
|
|
@staticmethod
|
|
def _activate(cursor: Any, source_batch_id: int) -> bool:
|
|
cursor.execute(
|
|
"""
|
|
SELECT source_batch_id
|
|
FROM booking.current_source_batch
|
|
WHERE singleton
|
|
FOR UPDATE
|
|
"""
|
|
)
|
|
current = cursor.fetchone()
|
|
already_active = current is not None and int(current[0]) == source_batch_id
|
|
if not already_active:
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO booking.current_source_batch (
|
|
singleton,
|
|
source_batch_id,
|
|
activated_at
|
|
)
|
|
VALUES (true, %s, now())
|
|
ON CONFLICT (singleton) DO UPDATE
|
|
SET source_batch_id = EXCLUDED.source_batch_id,
|
|
activated_at = EXCLUDED.activated_at
|
|
""",
|
|
(source_batch_id,),
|
|
)
|
|
return already_active
|
|
|
|
@staticmethod
|
|
def _batch_summary(
|
|
cursor: Any,
|
|
source_batch_id: int,
|
|
disposition: str,
|
|
) -> BookingSourceSummary:
|
|
cursor.execute(BATCH_SUMMARY_SQL, (source_batch_id,))
|
|
summary = PostgresBookingExcelRepository._summary_from_row(
|
|
cursor.fetchone(),
|
|
disposition=disposition,
|
|
)
|
|
if summary is None:
|
|
raise BookingExcelRepositoryError(
|
|
"BOOKING_EXCEL_DATABASE_STATE_INVALID",
|
|
"Booking Excel 数据源状态无效",
|
|
)
|
|
return summary
|
|
|
|
def activate_existing(
|
|
self,
|
|
source_sha256: str,
|
|
) -> Optional[BookingSourceSummary]:
|
|
connection = self._open()
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
self._begin(cursor)
|
|
cursor.execute(
|
|
"SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))",
|
|
("arr_booking_excel_import",),
|
|
)
|
|
source_batch_id = self._find_existing_batch(cursor, source_sha256)
|
|
if source_batch_id is None:
|
|
connection.rollback()
|
|
return None
|
|
already_active = self._activate(cursor, source_batch_id)
|
|
summary = self._batch_summary(
|
|
cursor,
|
|
source_batch_id,
|
|
"already_active" if already_active else "reactivated",
|
|
)
|
|
connection.commit()
|
|
return summary
|
|
except BookingExcelRepositoryError:
|
|
connection.rollback()
|
|
raise
|
|
except Exception:
|
|
connection.rollback()
|
|
raise BookingExcelRepositoryError(
|
|
"BOOKING_EXCEL_DATABASE_WRITE_FAILED",
|
|
"Excel 已校验，但数据源未能完成入库",
|
|
) from None
|
|
finally:
|
|
connection.close()
|
|
|
|
@staticmethod
|
|
def _ensure_artifact(cursor: Any, reference: ArtifactRef) -> int:
|
|
cursor.execute(
|
|
"""
|
|
SELECT
|
|
id,
|
|
artifact_kind,
|
|
original_filename,
|
|
sha256,
|
|
byte_size,
|
|
mime_type
|
|
FROM ingestion.artifacts
|
|
WHERE storage_provider = %s
|
|
AND bucket_alias = %s
|
|
AND object_key = %s
|
|
AND object_version_id IS NULL
|
|
FOR SHARE
|
|
""",
|
|
(
|
|
ARTIFACT_STORAGE_PROVIDER,
|
|
ARTIFACT_BUCKET_ALIAS,
|
|
reference.object_key,
|
|
),
|
|
)
|
|
row = cursor.fetchone()
|
|
if row is not None:
|
|
if (
|
|
str(row[1]) != reference.file_kind
|
|
or str(row[2]) != reference.original_filename
|
|
or str(row[3]) != reference.sha256
|
|
or int(row[4]) != reference.byte_size
|
|
or str(row[5] or "") != reference.mime_type
|
|
):
|
|
raise BookingExcelRepositoryError(
|
|
"BOOKING_EXCEL_ARTIFACT_CONFLICT",
|
|
"Excel 工件身份冲突",
|
|
409,
|
|
)
|
|
return int(row[0])
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO ingestion.artifacts (
|
|
artifact_kind,
|
|
storage_provider,
|
|
bucket_alias,
|
|
object_key,
|
|
original_filename,
|
|
sha256,
|
|
byte_size,
|
|
mime_type
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
|
RETURNING id
|
|
""",
|
|
(
|
|
reference.file_kind,
|
|
ARTIFACT_STORAGE_PROVIDER,
|
|
ARTIFACT_BUCKET_ALIAS,
|
|
reference.object_key,
|
|
reference.original_filename,
|
|
reference.sha256,
|
|
reference.byte_size,
|
|
reference.mime_type,
|
|
),
|
|
)
|
|
return int(cursor.fetchone()[0])
|
|
|
|
def import_document(
|
|
self,
|
|
job_id: str,
|
|
source: ArtifactRef,
|
|
document: ExcelDocument,
|
|
) -> BookingSourceSummary:
|
|
if (
|
|
source.role != "booking_source"
|
|
or source.file_kind != "booking_excel"
|
|
or source.mime_type != XLSX_MIME
|
|
or source.sha256 != document.source_sha256
|
|
or source.byte_size != document.source_byte_size
|
|
):
|
|
raise BookingExcelRepositoryError(
|
|
"BOOKING_EXCEL_ARTIFACT_INVALID",
|
|
"Excel 工件身份无效",
|
|
422,
|
|
)
|
|
if document.pending_item_count:
|
|
raise BookingExcelRepositoryError(
|
|
"BOOKING_EXCEL_REVIEW_REQUIRED",
|
|
"仍有房型需要人工确认或删除",
|
|
409,
|
|
)
|
|
|
|
connection = self._open()
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
self._begin(cursor)
|
|
cursor.execute(
|
|
"SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))",
|
|
("arr_booking_excel_import",),
|
|
)
|
|
existing_batch_id = self._find_existing_batch(
|
|
cursor,
|
|
document.source_sha256,
|
|
)
|
|
if existing_batch_id is not None:
|
|
already_active = self._activate(cursor, existing_batch_id)
|
|
summary = self._batch_summary(
|
|
cursor,
|
|
existing_batch_id,
|
|
"already_active" if already_active else "reactivated",
|
|
)
|
|
connection.commit()
|
|
return summary
|
|
|
|
artifact_id = self._ensure_artifact(cursor, source)
|
|
delivery = {
|
|
"source_sha256": document.source_sha256,
|
|
"source_rows": len(document.rows),
|
|
"worksheet_count": document.worksheet_count,
|
|
"distinct_group_codes": document.distinct_group_code_count,
|
|
"room_quantity": document.room_quantity,
|
|
}
|
|
delivery_bytes = canonical_json_bytes(delivery)
|
|
delivery_sha256 = hashlib.sha256(delivery_bytes).hexdigest()
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO ingestion.processing_runs (
|
|
run_key,
|
|
pipeline_type,
|
|
source_artifact_id,
|
|
run_status,
|
|
requested_processor_version,
|
|
requested_rule_set_sha256,
|
|
delivered_processor_version,
|
|
delivered_rule_set_sha256,
|
|
result_schema_version,
|
|
delivery_sha256,
|
|
delivery_json
|
|
)
|
|
VALUES (
|
|
%s,
|
|
'booking_source_import',
|
|
%s,
|
|
'validating',
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s::jsonb
|
|
)
|
|
RETURNING id
|
|
""",
|
|
(
|
|
job_id,
|
|
artifact_id,
|
|
PROCESSOR_VERSION,
|
|
rule_set_sha256(),
|
|
PROCESSOR_VERSION,
|
|
rule_set_sha256(),
|
|
RESULT_SCHEMA_VERSION,
|
|
delivery_sha256,
|
|
delivery_bytes.decode("utf-8"),
|
|
),
|
|
)
|
|
processing_run_id = int(cursor.fetchone()[0])
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO booking.source_batches (
|
|
source_artifact_id,
|
|
source_kind,
|
|
source_format_version,
|
|
batch_status,
|
|
source_rows,
|
|
accepted_rows,
|
|
failed_rows
|
|
)
|
|
VALUES (
|
|
%s,
|
|
'booking_excel',
|
|
'xlsx/1.0',
|
|
'processing',
|
|
%s,
|
|
0,
|
|
0
|
|
)
|
|
RETURNING id
|
|
""",
|
|
(artifact_id, len(document.rows)),
|
|
)
|
|
source_batch_id = int(cursor.fetchone()[0])
|
|
|
|
for row in document.rows:
|
|
row_sha256 = row.sha256()
|
|
result = build_excel_parse_result(row)
|
|
result_bytes = canonical_json_bytes(result)
|
|
result_sha256 = hashlib.sha256(result_bytes).hexdigest()
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO booking.source_rows (
|
|
source_batch_id,
|
|
source_worksheet,
|
|
source_row_no,
|
|
group_code_raw,
|
|
type_of_room_raw,
|
|
no_of_rooms,
|
|
source_row_sha256
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s)
|
|
RETURNING id
|
|
""",
|
|
(
|
|
source_batch_id,
|
|
row.worksheet,
|
|
row.row_no,
|
|
row.group_code_raw,
|
|
row.type_of_room_raw,
|
|
row.no_of_rooms,
|
|
row_sha256,
|
|
),
|
|
)
|
|
source_row_id = int(cursor.fetchone()[0])
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO booking.parse_versions (
|
|
source_row_id,
|
|
version_no,
|
|
parse_status,
|
|
result_schema_version,
|
|
processor_name,
|
|
processor_version,
|
|
rule_set_sha256,
|
|
input_row_sha256,
|
|
result_sha256,
|
|
result_json,
|
|
validated_at
|
|
)
|
|
VALUES (
|
|
%s,
|
|
1,
|
|
'accepted',
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s,
|
|
%s::jsonb,
|
|
now()
|
|
)
|
|
RETURNING id
|
|
""",
|
|
(
|
|
source_row_id,
|
|
RESULT_SCHEMA_VERSION,
|
|
PROCESSOR_NAME,
|
|
PROCESSOR_VERSION,
|
|
rule_set_sha256(),
|
|
row_sha256,
|
|
result_sha256,
|
|
result_bytes.decode("utf-8"),
|
|
),
|
|
)
|
|
parse_version_id = int(cursor.fetchone()[0])
|
|
for item in row.room_items:
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO booking.room_items (
|
|
parse_version_id,
|
|
item_no,
|
|
room_type_raw,
|
|
room_type_code,
|
|
quantity,
|
|
unit_price,
|
|
currency_code,
|
|
price_token_raw,
|
|
source_fragment
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, NULL, NULL, NULL, %s)
|
|
""",
|
|
(
|
|
parse_version_id,
|
|
item.item_no,
|
|
item.room_type_raw,
|
|
item.room_type_code,
|
|
item.quantity,
|
|
item.source_fragment,
|
|
),
|
|
)
|
|
|
|
cursor.execute(
|
|
"""
|
|
UPDATE booking.source_batches
|
|
SET batch_status = 'accepted',
|
|
accepted_rows = source_rows,
|
|
failed_rows = 0,
|
|
validated_at = now(),
|
|
finished_at = now()
|
|
WHERE id = %s
|
|
""",
|
|
(source_batch_id,),
|
|
)
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO booking.current_row_parses (
|
|
source_row_id,
|
|
parse_version_id
|
|
)
|
|
SELECT source.id, parsed.id
|
|
FROM booking.source_rows AS source
|
|
JOIN booking.parse_versions AS parsed
|
|
ON parsed.source_row_id = source.id
|
|
AND parsed.version_no = 1
|
|
AND parsed.parse_status = 'accepted'
|
|
WHERE source.source_batch_id = %s
|
|
""",
|
|
(source_batch_id,),
|
|
)
|
|
self._activate(cursor, source_batch_id)
|
|
cursor.execute(
|
|
"""
|
|
UPDATE ingestion.processing_runs
|
|
SET run_status = 'accepted',
|
|
updated_at = now(),
|
|
validated_at = now(),
|
|
finished_at = now()
|
|
WHERE id = %s
|
|
""",
|
|
(processing_run_id,),
|
|
)
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO ingestion.outbox_events (
|
|
event_key,
|
|
aggregate_type,
|
|
aggregate_id,
|
|
event_type,
|
|
payload
|
|
)
|
|
VALUES (
|
|
%s,
|
|
'booking_source_batch',
|
|
%s,
|
|
'booking.source_batch.accepted',
|
|
%s::jsonb
|
|
)
|
|
""",
|
|
(
|
|
f"booking-source-batch:{source_batch_id}:accepted",
|
|
source_batch_id,
|
|
json.dumps(
|
|
{
|
|
"source_batch_id": source_batch_id,
|
|
"source_rows": len(document.rows),
|
|
"distinct_group_codes": document.distinct_group_code_count,
|
|
},
|
|
separators=(",", ":"),
|
|
),
|
|
),
|
|
)
|
|
summary = self._batch_summary(
|
|
cursor,
|
|
source_batch_id,
|
|
"imported_and_activated",
|
|
)
|
|
connection.commit()
|
|
return summary
|
|
except BookingExcelRepositoryError:
|
|
connection.rollback()
|
|
raise
|
|
except Exception:
|
|
connection.rollback()
|
|
raise BookingExcelRepositoryError(
|
|
"BOOKING_EXCEL_DATABASE_WRITE_FAILED",
|
|
"Excel 已校验，但数据源未能完成入库",
|
|
) from None
|
|
finally:
|
|
connection.close()
|