154 lines
7 KiB
Bash
154 lines
7 KiB
Bash
#!/usr/bin/env bash
|
||
# Loads all 7 quarters of dataset_СДЕЛКИ into rosreestr_deals via staging.
|
||
# Q3 2024 uses ';' separator, all later quarters use '~'.
|
||
#
|
||
# Usage:
|
||
# bash 02_load_all_quarters.sh # local test container (gd_test_pg)
|
||
# PG_HOST=localhost PG_PORT=15432 PGPASSWORD=... bash 02_load_all_quarters.sh # remote (via SSH tunnel)
|
||
set -euo pipefail
|
||
|
||
if [[ -n "${PG_HOST:-}" ]]; then
|
||
# Remote / TCP mode (e.g. prod via SSH tunnel)
|
||
PG_PORT="${PG_PORT:-5432}"
|
||
PG_USER="${PG_USER:-gendesign}"
|
||
PG_DB="${PG_DB:-gendesign}"
|
||
PSQL="docker run --rm -i -e PGPASSWORD postgres:16-alpine psql -h host.docker.internal -p $PG_PORT -U $PG_USER -d $PG_DB -v ON_ERROR_STOP=1"
|
||
PSQL_PIPE="docker run --rm -i -e PGPASSWORD postgres:16-alpine psql -h host.docker.internal -p $PG_PORT -U $PG_USER -d $PG_DB -v ON_ERROR_STOP=1"
|
||
else
|
||
# Local docker exec into the test container
|
||
PG_CONTAINER="${PG_CONTAINER:-gd_test_pg}"
|
||
PSQL="docker exec -i $PG_CONTAINER psql -U gendesign -d gendesign -v ON_ERROR_STOP=1"
|
||
PSQL_PIPE="$PSQL"
|
||
fi
|
||
|
||
cd "$(dirname "$0")/../raw"
|
||
|
||
# Имя CSV-файла внутри zip может варьироваться (rosreestr менял схему в Q4 2024).
|
||
# Скрипт автоматически пробует обе именные конвенции.
|
||
# Если файла нет — квартал пропускается с warning (не fail). Это позволяет
|
||
# догружать 2023 / 2024Q1-Q2 / новые кварталы постепенно по мере выкачки.
|
||
declare -a JOBS=(
|
||
# Формат CSV изменился в Q4 2024 (delim ; → ~, имя r_all → r-r_01-92).
|
||
# 2024Q1+Q2 уже опубликованы по новой схеме (r-r_01-92_y_YYYY_q_N), но
|
||
# delimiter мог не успеть перейти — попробуй сначала ~, при ошибках смени на ;.
|
||
# 2023 в виде квартальных CSV НЕ ПУБЛИКУЕТСЯ (архив до 2023г. содержит только
|
||
# JSON-снимки от ноября-декабря 2021 в /data-sets/Архив до 2023г. включительно/).
|
||
"2024Q1:2024-01-01:dataset_СДЕЛКИ_r-r_01-92_y_2024_q_1.csv:;"
|
||
"2024Q2:2024-04-01:dataset_СДЕЛКИ_r-r_01-92_y_2024_q_2.csv:;"
|
||
"2024Q3:2024-07-01:dataset_СДЕЛКИ_r_all_q_3.csv:;"
|
||
"2024Q4:2024-10-01:dataset_СДЕЛКИ_r-r_01-92_y_2024_q_4.csv:~"
|
||
"2025Q1:2025-01-01:dataset_СДЕЛКИ_r-r_01-92_y_2025_q_1.csv:~"
|
||
"2025Q2:2025-04-01:dataset_СДЕЛКИ_r-r_01-92_y_2025_q_2.csv:~"
|
||
"2025Q3:2025-07-01:dataset_СДЕЛКИ_r-r_01-92_y_2025_q_3.csv:~"
|
||
"2025Q4:2025-10-01:dataset_СДЕЛКИ_r-r_01-92_y_2025_q_4.csv:~"
|
||
"2026Q1:2026-01-01:dataset_СДЕЛКИ_r-r_01-92_y_2026_q_1.csv:~"
|
||
)
|
||
|
||
# Возможные имена zip-ов в data/raw/ (можно класть как есть, скрипт распакует).
|
||
# Имя в data/raw/ — sdelki_YYYYqN.csv.zip ИЛИ оригинальное dataset_СДЕЛКИ_*.csv.zip.
|
||
|
||
resolve_csv() {
|
||
# Принимает comma-separated список кандидатов имён csv. Возвращает первый
|
||
# существующий (распакованный) файл. Если ничего нет — пытается распаковать
|
||
# из zip с известными именами. Возвращает пустую строку если файл не найден.
|
||
local CANDIDATES="$1"
|
||
local SRC_Q="$2"
|
||
IFS=',' read -ra NAMES <<< "$CANDIDATES"
|
||
for name in "${NAMES[@]}"; do
|
||
if [[ -f "$name" ]]; then
|
||
echo "$name"
|
||
return 0
|
||
fi
|
||
done
|
||
# Try unzip from sdelki_YYYYqN.csv.zip (lowercase quarter).
|
||
local LOWER=$(echo "$SRC_Q" | tr '[:upper:]' '[:lower:]')
|
||
local ZIP_LOCAL="sdelki_${LOWER}.csv.zip"
|
||
if [[ -f "$ZIP_LOCAL" ]]; then
|
||
unzip -o -q "$ZIP_LOCAL" >&2 || true
|
||
for name in "${NAMES[@]}"; do
|
||
if [[ -f "$name" ]]; then echo "$name"; return 0; fi
|
||
done
|
||
fi
|
||
# Try original rosreestr-named zips.
|
||
for name in "${NAMES[@]}"; do
|
||
if [[ -f "${name}.zip" ]]; then
|
||
unzip -o -q "${name}.zip" >&2 || true
|
||
[[ -f "$name" ]] && { echo "$name"; return 0; }
|
||
fi
|
||
done
|
||
echo ""
|
||
return 1
|
||
}
|
||
|
||
for job in "${JOBS[@]}"; do
|
||
IFS=':' read -r SRC_Q PERIOD FILE_CANDIDATES SEP <<< "$job"
|
||
FILE=$(resolve_csv "$FILE_CANDIDATES" "$SRC_Q")
|
||
if [[ -z "$FILE" ]]; then
|
||
echo "=== $SRC_Q SKIP — нет CSV в data/raw/ (искал: $FILE_CANDIDATES; sdelki_${SRC_Q,,}.csv.zip)"
|
||
continue
|
||
fi
|
||
# Idempotency: пропускаем если уже загружен с тем же source_quarter.
|
||
EXISTING=$($PSQL -t -A -c "SELECT COUNT(*) FROM rosreestr_deals WHERE source_quarter = '$SRC_Q'" || echo 0)
|
||
if [[ "${EXISTING:-0}" -gt 0 ]]; then
|
||
echo "=== $SRC_Q SKIP — уже загружен ($EXISTING строк). Удалить: DELETE FROM rosreestr_deals WHERE source_quarter='$SRC_Q'"
|
||
continue
|
||
fi
|
||
echo "=== $SRC_Q ($FILE, sep='$SEP') ==="
|
||
$PSQL -c "TRUNCATE rosreestr_deals_staging;"
|
||
start=$(date +%s)
|
||
cat "$FILE" | $PSQL -c "\\copy rosreestr_deals_staging FROM stdin (FORMAT csv, HEADER true, DELIMITER '$SEP', QUOTE '\"', NULL '')"
|
||
$PSQL <<SQL
|
||
INSERT INTO rosreestr_deals (
|
||
source_quarter, period_start_date,
|
||
region_code, okato, district, city, quarter_cad_number, street,
|
||
realestate_type_code, wall_material_code, purpose_code,
|
||
year_build, floor, area,
|
||
doc_type, deal_price, currency, deal_count
|
||
)
|
||
SELECT
|
||
'$SRC_Q', period_start_date,
|
||
region_code, NULLIF(okato, '-'), NULLIF(district, ''), NULLIF(city, ''),
|
||
quarter_cad_number, NULLIF(street, ''),
|
||
NULLIF(realestate_type_code, ''),
|
||
NULLIF(wall_material_code, ''),
|
||
NULLIF(purpose_code, ''),
|
||
-- year_build can contain garbage like '09.07.1988' — keep only plausible 4-digit years
|
||
CASE WHEN year_build ~ '^[12][0-9]{3}$' THEN year_build::INT ELSE NULL END,
|
||
NULLIF(floor, ''),
|
||
-- discard absurd area/price (data-quality outliers from source)
|
||
CASE WHEN area > 0 AND area < 1e8 THEN area ELSE NULL END,
|
||
NULLIF(doc_type, ''),
|
||
CASE WHEN deal_price > 0 AND deal_price < 1e15 THEN deal_price ELSE NULL END,
|
||
COALESCE(NULLIF(currency, ''), 'рубль'),
|
||
COALESCE(deal_count, 1)
|
||
FROM rosreestr_deals_staging
|
||
WHERE region_code IS NOT NULL
|
||
AND quarter_cad_number IS NOT NULL
|
||
AND quarter_cad_number <> ''
|
||
AND doc_type IS NOT NULL
|
||
AND doc_type <> ''
|
||
AND period_start_date IS NOT NULL;
|
||
SQL
|
||
end=$(date +%s)
|
||
echo " loaded in $((end-start))s"
|
||
done
|
||
|
||
$PSQL <<'SQL'
|
||
SELECT
|
||
source_quarter,
|
||
COUNT(*) AS rows_loaded,
|
||
SUM(deal_count) AS deals,
|
||
SUM(deal_count) FILTER (WHERE region_code = 66) AS sverdl_deals,
|
||
pg_size_pretty(pg_relation_size('rosreestr_deals_'||lower(source_quarter))) AS partition_size
|
||
FROM rosreestr_deals
|
||
GROUP BY source_quarter
|
||
ORDER BY source_quarter;
|
||
|
||
SELECT
|
||
'TOTAL' AS quarter,
|
||
COUNT(*) AS rows,
|
||
SUM(deal_count) AS deals,
|
||
SUM(deal_count) FILTER (WHERE region_code = 66) AS sverdl_deals,
|
||
pg_size_pretty(pg_total_relation_size('rosreestr_deals')) AS total_size
|
||
FROM rosreestr_deals;
|
||
SQL
|