from sqlalchemy import text from datetime import datetime import json class StationPriceSchedule: def __init__(self): pass async def save(self, session, id, station_hash, schedule_json, valid_start_time=None): if valid_start_time is None: valid_start_time = datetime.now() # Ensure json is string with consistent formatting schedule_json_str = None if schedule_json is not None: if isinstance(schedule_json, (dict, list)): schedule_json_str = json.dumps(schedule_json, ensure_ascii=False, sort_keys=True) else: schedule_json_str = schedule_json # 1. Check current record select_sql = """ SELECT schedule_json FROM t_station_price_schedule_scd WHERE station_hash = :station_hash AND is_current = 1 """ result = await session.execute(text(select_sql), {"station_hash": station_hash}) current_row = result.fetchone() if current_row: row_json = current_row.schedule_json # Normalize row_json for comparison if row_json: try: if isinstance(row_json, str): row_json_obj = json.loads(row_json) row_json = json.dumps(row_json_obj, ensure_ascii=False, sort_keys=True) except Exception: pass if row_json == schedule_json_str: # No change, skip insert return # Expire old record update_sql = """ UPDATE t_station_price_schedule_scd SET is_current = 0, valid_end_time = :valid_start_time WHERE station_hash = :station_hash AND is_current = 1 """ await session.execute(text(update_sql), { "valid_start_time": valid_start_time, "station_hash": station_hash }) # 2. Insert new record sql = """ INSERT INTO t_station_price_schedule_scd (id, station_hash, schedule_json, valid_start_time, is_current) VALUES (:id, :station_hash, :schedule_json, :valid_start_time, 1) """ await session.execute(text(sql), { "id": id, "station_hash": station_hash, "schedule_json": schedule_json_str, "valid_start_time": valid_start_time })