70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
from .base_dwd_task import BaseDwdTask
|
|
from loaders.facts.ticket import TicketLoader
|
|
|
|
class TicketDwdTask(BaseDwdTask):
|
|
"""
|
|
DWD Task: Process Ticket Details from ODS to Fact Tables
|
|
Source: billiards_ods.ods_ticket_detail
|
|
Targets:
|
|
- billiards.fact_order
|
|
- billiards.fact_order_goods
|
|
- billiards.fact_table_usage
|
|
- billiards.fact_assistant_service
|
|
"""
|
|
|
|
def get_task_code(self) -> str:
|
|
return "TICKET_DWD"
|
|
|
|
def execute(self) -> dict:
|
|
self.logger.info(f"Starting {self.get_task_code()} task")
|
|
|
|
# 1. Get Time Window (Incremental Load)
|
|
window_start, window_end, _ = self._get_time_window()
|
|
self.logger.info(f"Processing window: {window_start} to {window_end}")
|
|
|
|
# 2. Initialize Loader
|
|
loader = TicketLoader(self.db, logger=self.logger)
|
|
store_id = self.config.get("app.store_id")
|
|
|
|
total_inserted = 0
|
|
total_errors = 0
|
|
|
|
# 3. Iterate ODS Data
|
|
# We query ods_ticket_detail based on fetched_at
|
|
batches = self.iter_ods_rows(
|
|
table_name="billiards_ods.settlement_ticket_details",
|
|
columns=["payload", "fetched_at", "source_file", "record_index"],
|
|
start_time=window_start,
|
|
end_time=window_end
|
|
)
|
|
|
|
for batch in batches:
|
|
if not batch:
|
|
continue
|
|
|
|
# Extract payloads
|
|
tickets = []
|
|
for row in batch:
|
|
payload = self.parse_payload(row)
|
|
if payload:
|
|
tickets.append(payload)
|
|
|
|
# Process Batch
|
|
inserted, errors = loader.process_tickets(tickets, store_id)
|
|
total_inserted += inserted
|
|
total_errors += errors
|
|
|
|
# 4. Commit
|
|
self.db.commit()
|
|
|
|
self.logger.info(f"Task {self.get_task_code()} completed. Inserted: {total_inserted}, Errors: {total_errors}")
|
|
|
|
return {
|
|
"status": "success",
|
|
"inserted": total_inserted,
|
|
"errors": total_errors,
|
|
"window_start": window_start.isoformat(),
|
|
"window_end": window_end.isoformat()
|
|
}
|