90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
"""Import the existing research CSV files into the configured MySQL database.
|
|
|
|
This is a one-time/manual import utility. It does not schedule scraping.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
import pandas as pd
|
|
import pymysql
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from app.config import Config
|
|
|
|
|
|
BASE_DIR = Path(__file__).resolve().parents[1]
|
|
CSV_FILES = {
|
|
"rawit": next(BASE_DIR.glob("harga_cabai_rawit_*.csv")),
|
|
"merah": next(BASE_DIR.glob("harga_cabe_merah_besar_*.csv")),
|
|
"keriting": next(BASE_DIR.glob("harga_cabai_merah_keriting_*.csv")),
|
|
}
|
|
MARKETS = (
|
|
"pasar_tanjung",
|
|
"pasar_kreongan",
|
|
"pasar_tegal_besar",
|
|
"pasar_mangli",
|
|
"pasar_wirolegi",
|
|
)
|
|
|
|
|
|
def normalized_market_column(market: str) -> str:
|
|
return "Pasar_" + "_".join(part.title() for part in market.removeprefix("pasar_").split("_"))
|
|
|
|
|
|
def main() -> None:
|
|
connection = pymysql.connect(
|
|
host=Config.MYSQL_HOST,
|
|
port=Config.MYSQL_PORT,
|
|
database=Config.MYSQL_DATABASE,
|
|
user=Config.MYSQL_USER,
|
|
password=Config.MYSQL_PASSWORD,
|
|
connect_timeout=Config.MYSQL_CONNECT_TIMEOUT,
|
|
)
|
|
inserted = 0
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("SELECT id, slug FROM commodities")
|
|
commodities = {slug: row_id for row_id, slug in cursor.fetchall()}
|
|
cursor.execute("SELECT id, slug FROM markets")
|
|
markets = {slug: row_id for row_id, slug in cursor.fetchall()}
|
|
|
|
for commodity, csv_path in CSV_FILES.items():
|
|
frame = pd.read_csv(csv_path)
|
|
for _, row in frame.iterrows():
|
|
price_date = pd.to_datetime(row["Tanggal"], errors="coerce")
|
|
if pd.isna(price_date):
|
|
continue
|
|
for market in MARKETS:
|
|
value = pd.to_numeric(row.get(normalized_market_column(market)), errors="coerce")
|
|
if pd.isna(value) or float(value) <= 0:
|
|
continue
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO price_history
|
|
(commodity_id, market_id, price_date, price, source)
|
|
VALUES (%s, %s, %s, %s, 'csv_import')
|
|
ON DUPLICATE KEY UPDATE
|
|
price = VALUES(price), source = VALUES(source)
|
|
""",
|
|
(
|
|
commodities[commodity],
|
|
markets[market],
|
|
price_date.date(),
|
|
float(value) * 1000,
|
|
),
|
|
)
|
|
inserted += cursor.rowcount
|
|
connection.commit()
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
finally:
|
|
connection.close()
|
|
print(f"Import selesai. Baris diproses: {inserted}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|