33 lines
1003 B
Python
33 lines
1003 B
Python
from datetime import datetime
|
|
from parser import pipeline
|
|
|
|
import pandas as pd
|
|
import psycopg
|
|
|
|
from .config import DB_COLUMNS_MAP, STORE_NULL_BID
|
|
from .database import db_credentials
|
|
|
|
sql_statement = "".join(
|
|
("COPY records (", ", ".join(DB_COLUMNS_MAP.values()), ") FROM STDIN")
|
|
)
|
|
|
|
|
|
def job():
|
|
fetch_start = datetime.now()
|
|
print("Starting refetch job: " + fetch_start.isoformat())
|
|
|
|
parser = pipeline()
|
|
|
|
print("Rewriting db: " + datetime.now().isoformat())
|
|
|
|
with psycopg.connect(**db_credentials) as connection:
|
|
with connection.cursor() as cursor:
|
|
with cursor.copy(sql_statement) as copy:
|
|
for _, row in parser.df.iterrows():
|
|
row = row.where((pd.notnull(row)), None)
|
|
if row["building_id"] is not None or STORE_NULL_BID:
|
|
db_row = row.rename(DB_COLUMNS_MAP)
|
|
copy.write_row(db_row.to_list())
|
|
|
|
print(f"Fetched in {datetime.now() - fetch_start}\n{parser}")
|