'commit'
This commit is contained in:
107
Model/StationProfile.py
Normal file
107
Model/StationProfile.py
Normal file
@@ -0,0 +1,107 @@
|
||||
from sqlalchemy import text
|
||||
from datetime import datetime
|
||||
|
||||
class StationProfile:
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
async def backfill_address_if_missing(self, session, station_hash, station_name, address):
|
||||
sql = """
|
||||
UPDATE t_station_profile_scd
|
||||
SET address = :address,
|
||||
station_name = :station_name
|
||||
WHERE station_hash = :station_hash
|
||||
AND (address IS NULL OR address = '')
|
||||
"""
|
||||
result = await session.execute(text(sql), {
|
||||
"station_hash": station_hash,
|
||||
"station_name": station_name,
|
||||
"address": address
|
||||
})
|
||||
try:
|
||||
return int(result.rowcount or 0)
|
||||
except Exception:
|
||||
return 0
|
||||
|
||||
async def save(self, session, id, station_hash, operator, station_name, address=None, coord_x=None, coord_y=None, valid_start_time=None):
|
||||
if valid_start_time is None:
|
||||
valid_start_time = datetime.now()
|
||||
|
||||
# 1. Check current record
|
||||
select_sql = """
|
||||
SELECT operator, station_name, address, coord_x, coord_y
|
||||
FROM t_station_profile_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:
|
||||
if address is not None and address != "" and (current_row.address is None or current_row.address == ""):
|
||||
await self.backfill_address_if_missing(
|
||||
session=session,
|
||||
station_hash=station_hash,
|
||||
station_name=station_name,
|
||||
address=address
|
||||
)
|
||||
return
|
||||
|
||||
# Merge logic: if new value is None, use old value
|
||||
# But we must respect if new value is explicitly provided (even empty string?)
|
||||
# Usually for crawler:
|
||||
# - station_name: always updated
|
||||
# - address: if new is None, use old.
|
||||
# - coord: if new is None, use old.
|
||||
|
||||
merged_address = address if address is not None else current_row.address
|
||||
merged_coord_x = coord_x if coord_x is not None else current_row.coord_x
|
||||
merged_coord_y = coord_y if coord_y is not None else current_row.coord_y
|
||||
|
||||
# Use merged values for comparison and insertion
|
||||
address = merged_address
|
||||
coord_x = merged_coord_x
|
||||
coord_y = merged_coord_y
|
||||
|
||||
# Check if changed
|
||||
# Need to handle potential float/decimal diffs for coords if needed,
|
||||
# but usually they are float in DB and float in python.
|
||||
|
||||
is_same = (
|
||||
current_row.operator == operator and
|
||||
current_row.station_name == station_name and
|
||||
current_row.address == address and
|
||||
current_row.coord_x == coord_x and
|
||||
current_row.coord_y == coord_y
|
||||
)
|
||||
|
||||
if is_same:
|
||||
return
|
||||
|
||||
# Expire old record
|
||||
update_sql = """
|
||||
UPDATE t_station_profile_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 (with potentially merged values)
|
||||
sql = """
|
||||
INSERT INTO t_station_profile_scd
|
||||
(id, station_hash, operator, station_name, address, coord_x, coord_y, valid_start_time, is_current)
|
||||
VALUES
|
||||
(:id, :station_hash, :operator, :station_name, :address, :coord_x, :coord_y, :valid_start_time, 1)
|
||||
"""
|
||||
await session.execute(text(sql), {
|
||||
"id": id,
|
||||
"station_hash": station_hash,
|
||||
"operator": operator,
|
||||
"station_name": station_name,
|
||||
"address": address,
|
||||
"coord_x": coord_x,
|
||||
"coord_y": coord_y,
|
||||
"valid_start_time": valid_start_time
|
||||
})
|
||||
Reference in New Issue
Block a user