"""Парсер Objective raw → normalized слой. Читает payload (либо из БД objective_raw_reports.raw_id, либо из локального JSON-файла для smoke-теста) и раскладывает по таблицам: - corp_sum (Сводные/Корпуса) → objective_corpus_room_month - lots_pf (Поквартирные/Лоты) → objective_lots (UPSERT) + objective_lots_history (INSERT) Запуск: cd backend # Smoke по локальным файлам (без БД): uv run python -X utf8 ../data/sql/70_parse_objective_raw.py \ --file-corp-sum ../data/raw/objective_test/20260507_162110/corp_sum.json \ --file-lots-pf ../data/raw/objective_test/20260507_162110/lots_pf.json \ --dry-run # Из БД по raw_id (нужен schema 68 + DATABASE_URL): uv run python -X utf8 ../data/sql/70_parse_objective_raw.py --raw-id 123 # Все unparsed raw из последнего run: uv run python -X utf8 ../data/sql/70_parse_objective_raw.py --latest-run Что считается «unparsed»: rows_extracted IS NULL ИЛИ rows_extracted = 0 в objective_raw_reports. Идемпотентность: - corpus_room_month: UNIQUE (report_month, group, project, corpus, room_bucket) → ON CONFLICT DO UPDATE (перезатирает старое значение метрик; это правильно т.к. отчёт за месяц может уточняться неделя-к-неделе пока месяц не закрыт) - lots: UNIQUE objective_lot_id → ON CONFLICT DO UPDATE - lots_history: UNIQUE (objective_lot_id, snapshot_date) → ON CONFLICT DO NOTHING (один снимок per день — повторный sync в тот же день не создаёт дубль) """ from __future__ import annotations import argparse import json import logging import re import sys from collections.abc import Iterator from datetime import date, datetime from pathlib import Path from typing import Any, BinaryIO # repo-root → /backend sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent / "backend")) from sqlalchemy import text # noqa: E402 from app.core.db import SessionLocal # noqa: E402 logger = logging.getLogger("objective_parser") logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") # ── helpers ───────────────────────────────────────────────────────────────── _RU_MONTHS = { "январь": 1, "февраль": 2, "март": 3, "апрель": 4, "май": 5, "июнь": 6, "июль": 7, "август": 8, "сентябрь": 9, "октябрь": 10, "ноябрь": 11, "декабрь": 12, } def _parse_ru_month(s: str | None) -> date | None: """'апрель-2026' → date(2026, 4, 1). Также допускает 'Апрель 2026'.""" if not s: return None parts = re.split(r"[-\s]+", s.strip().lower()) if len(parts) != 2: return None month_name, year_str = parts month = _RU_MONTHS.get(month_name) if not month: return None try: return date(int(year_str), month, 1) except (ValueError, TypeError): return None def _parse_iso_date(s: str | None) -> date | None: if not s: return None # Объектив отдаёт YYYY-MM-DD try: return date.fromisoformat(s[:10]) except (ValueError, TypeError): return None def _to_int(v: Any) -> int | None: if v is None or v == "": return None try: return int(v) except (ValueError, TypeError): return None def _to_num(v: Any) -> float | None: if v is None or v == "": return None try: return float(v) except (ValueError, TypeError): return None def _normalize_rooms(s: str | None) -> int | None: """'студия' → 0, '1' → 1, … 'неопределенный тип' → None.""" if not s: return None s_low = s.strip().lower() if "студ" in s_low: return 0 # Извлекаем первое число m = re.match(r"^(\d+)", s_low) if m: n = int(m.group(1)) return min(n, 5) # 5+ всё агрегируем под 5 return None def _parse_pct(s: str | None) -> int | None: """'17%' → 17.""" if s is None: return None if isinstance(s, (int, float)): return int(s) m = re.match(r"^\s*(\d+)", str(s)) return int(m.group(1)) if m else None def _parse_bool_yes(v: Any) -> bool | None: """'да' → True, 'нет' → False.""" if v is None or v == "": return None s = str(v).strip().lower() if s in ("да", "true", "1", "yes"): return True if s in ("нет", "false", "0", "no"): return False return None # ── corp_sum (Сводные/Корпуса) → objective_corpus_room_month ──────────────── def _row_to_corp_room_month(row: dict, group_name: str, raw_id: int | None) -> dict: rooms_raw = row.get("Количество комнат (Данные Объектива)") return { "report_month": _parse_ru_month(row.get("Месяц")), "group_name": group_name, "project_name": row.get("Проект"), "developer": row.get("Девелопер"), "district": row.get("Район"), "class": row.get("Класс"), "corpus_name": row.get("Корпус"), "room_bucket": rooms_raw, "rooms_int": _normalize_rooms(rooms_raw), "sales_start_date": _parse_iso_date(row.get("Старт продаж")), "plan_completion_date": _parse_iso_date(row.get("Планируемая дата ввода")), "actual_completion_date": _parse_iso_date(row.get("Фактическая дата ввода")), "months_in_realization": _to_int(row.get("Месяцев в реализации")), "lots_pd_count": _to_int(row.get("Лотов по ПД, шт.")), "lots_pd_area": _to_num(row.get("Площадь по ПД, м2.")), # ДДУ "deals_ddu_count": _to_int(row.get("Количество в сделках (ДДУ), шт.")), "deals_ddu_count_priced": _to_int(row.get("Количество лотов в сделках (ДДУ с ценами), шт.")), "deals_ddu_vol_m2": _to_num(row.get("Объем реализации (ДДУ), м2.")), "deals_ddu_vol_m2_priced": _to_num(row.get("Объем реализации (ДДУ с ценами), м2.")), "deals_ddu_sum_mln_rub": _to_num(row.get("Суммарная стоимость лотов (ДДУ с ценами), млн.Р")), "deals_ddu_avg_price_thousand_rub_per_m2": _to_num( row.get("Средняя цена м2 лота в сделках (ДДУ), тыс.Р/м2") ), "deals_ddu_avg_area_m2": _to_num(row.get("Средняя площадь лота в сделках (ДДУ), м2")), # ДКП "deals_dkp_count": _to_int(row.get("Количество в сделках (ДКП), шт.")), "deals_dkp_count_priced": _to_int(row.get("Количество лотов в сделках (ДКП с ценами), шт.")), "deals_dkp_vol_m2": _to_num(row.get("Объем реализации (ДКП), м2.")), "deals_dkp_vol_m2_priced": _to_num(row.get("Объем реализации (ДКП с ценами), м2.")), "deals_dkp_sum_mln_rub": _to_num(row.get("Суммарная стоимость лотов (ДКП с ценами), млн.Р")), "deals_dkp_avg_price_thousand_rub_per_m2": _to_num( row.get("Средняя цена м2 лота в сделках (ДКП), тыс.Р/м2") ), "deals_dkp_avg_area_m2": _to_num(row.get("Средняя площадь лота в сделках (ДКП), м2")), # Всего "deals_total_count": _to_int(row.get("Количество в сделках (всего), шт.")), "deals_total_count_priced": _to_int(row.get("Количество лотов в сделках (с ценами), шт.")), "deals_total_vol_m2": _to_num(row.get("Объем реализации (всего), м2.")), "deals_total_vol_m2_priced": _to_num(row.get("Объем реализации (с ценами), м2.")), "deals_total_sum_mln_rub": _to_num(row.get("Суммарная стоимость лотов (с ценами), млн.Р")), "deals_total_avg_price_thousand_rub_per_m2": _to_num( row.get("Средняя цена м2 лота в сделках, тыс.Р/м2") ), "deals_total_avg_area_m2": _to_num(row.get("Средняя площадь лота в сделках, м2")), # Предложение "offer_count": _to_int(row.get("Объем предложения, шт.")), "offer_area_m2": _to_num(row.get("Объем предложения, м2.")), "offer_sum_mln_rub": _to_num(row.get("Суммарная стоимость лотов в продаже, млн.Р")), "offer_avg_price_thousand_rub_per_m2": _to_num( row.get("Средняя цена м2 лота в продаже, тыс.Р/м2") ), # Audit "raw_id": raw_id, "raw_props": json.dumps(row, ensure_ascii=False), } _CRM_UPSERT = text( """ INSERT INTO objective_corpus_room_month ( report_month, group_name, project_name, developer, district, class, corpus_name, room_bucket, rooms_int, sales_start_date, plan_completion_date, actual_completion_date, months_in_realization, lots_pd_count, lots_pd_area, deals_ddu_count, deals_ddu_count_priced, deals_ddu_vol_m2, deals_ddu_vol_m2_priced, deals_ddu_sum_mln_rub, deals_ddu_avg_price_thousand_rub_per_m2, deals_ddu_avg_area_m2, deals_dkp_count, deals_dkp_count_priced, deals_dkp_vol_m2, deals_dkp_vol_m2_priced, deals_dkp_sum_mln_rub, deals_dkp_avg_price_thousand_rub_per_m2, deals_dkp_avg_area_m2, deals_total_count, deals_total_count_priced, deals_total_vol_m2, deals_total_vol_m2_priced, deals_total_sum_mln_rub, deals_total_avg_price_thousand_rub_per_m2, deals_total_avg_area_m2, offer_count, offer_area_m2, offer_sum_mln_rub, offer_avg_price_thousand_rub_per_m2, raw_id, raw_props ) VALUES ( :report_month, :group_name, :project_name, :developer, :district, :class, :corpus_name, :room_bucket, :rooms_int, :sales_start_date, :plan_completion_date, :actual_completion_date, :months_in_realization, :lots_pd_count, :lots_pd_area, :deals_ddu_count, :deals_ddu_count_priced, :deals_ddu_vol_m2, :deals_ddu_vol_m2_priced, :deals_ddu_sum_mln_rub, :deals_ddu_avg_price_thousand_rub_per_m2, :deals_ddu_avg_area_m2, :deals_dkp_count, :deals_dkp_count_priced, :deals_dkp_vol_m2, :deals_dkp_vol_m2_priced, :deals_dkp_sum_mln_rub, :deals_dkp_avg_price_thousand_rub_per_m2, :deals_dkp_avg_area_m2, :deals_total_count, :deals_total_count_priced, :deals_total_vol_m2, :deals_total_vol_m2_priced, :deals_total_sum_mln_rub, :deals_total_avg_price_thousand_rub_per_m2, :deals_total_avg_area_m2, :offer_count, :offer_area_m2, :offer_sum_mln_rub, :offer_avg_price_thousand_rub_per_m2, :raw_id, CAST(:raw_props AS jsonb) ) ON CONFLICT (report_month, group_name, project_name, corpus_name, room_bucket) DO UPDATE SET developer = EXCLUDED.developer, district = EXCLUDED.district, class = EXCLUDED.class, rooms_int = EXCLUDED.rooms_int, sales_start_date = EXCLUDED.sales_start_date, plan_completion_date = EXCLUDED.plan_completion_date, actual_completion_date = EXCLUDED.actual_completion_date, months_in_realization = EXCLUDED.months_in_realization, lots_pd_count = EXCLUDED.lots_pd_count, lots_pd_area = EXCLUDED.lots_pd_area, deals_ddu_count = EXCLUDED.deals_ddu_count, deals_ddu_count_priced = EXCLUDED.deals_ddu_count_priced, deals_ddu_vol_m2 = EXCLUDED.deals_ddu_vol_m2, deals_ddu_vol_m2_priced = EXCLUDED.deals_ddu_vol_m2_priced, deals_ddu_sum_mln_rub = EXCLUDED.deals_ddu_sum_mln_rub, deals_ddu_avg_price_thousand_rub_per_m2 = EXCLUDED.deals_ddu_avg_price_thousand_rub_per_m2, deals_ddu_avg_area_m2 = EXCLUDED.deals_ddu_avg_area_m2, deals_dkp_count = EXCLUDED.deals_dkp_count, deals_dkp_count_priced = EXCLUDED.deals_dkp_count_priced, deals_dkp_vol_m2 = EXCLUDED.deals_dkp_vol_m2, deals_dkp_vol_m2_priced = EXCLUDED.deals_dkp_vol_m2_priced, deals_dkp_sum_mln_rub = EXCLUDED.deals_dkp_sum_mln_rub, deals_dkp_avg_price_thousand_rub_per_m2 = EXCLUDED.deals_dkp_avg_price_thousand_rub_per_m2, deals_dkp_avg_area_m2 = EXCLUDED.deals_dkp_avg_area_m2, deals_total_count = EXCLUDED.deals_total_count, deals_total_count_priced = EXCLUDED.deals_total_count_priced, deals_total_vol_m2 = EXCLUDED.deals_total_vol_m2, deals_total_vol_m2_priced = EXCLUDED.deals_total_vol_m2_priced, deals_total_sum_mln_rub = EXCLUDED.deals_total_sum_mln_rub, deals_total_avg_price_thousand_rub_per_m2 = EXCLUDED.deals_total_avg_price_thousand_rub_per_m2, deals_total_avg_area_m2 = EXCLUDED.deals_total_avg_area_m2, offer_count = EXCLUDED.offer_count, offer_area_m2 = EXCLUDED.offer_area_m2, offer_sum_mln_rub = EXCLUDED.offer_sum_mln_rub, offer_avg_price_thousand_rub_per_m2 = EXCLUDED.offer_avg_price_thousand_rub_per_m2, raw_id = EXCLUDED.raw_id, raw_props = EXCLUDED.raw_props, fetched_at = NOW() """ ) # ── lots_pf (Поквартирные/Лоты) → objective_lots + objective_lots_history ─── def _row_to_lot(row: dict, raw_id: int | None, snapshot_date: date) -> dict: rooms_obj = row.get("Количество комнат(Данные объектива)") return { "objective_lot_id": _to_int(row.get("Id")), "objective_project_id": _to_int(row.get("Id проекта")), "project_name": row.get("Проект"), "developer": row.get("Девелопер"), "city": row.get("Город"), "district": row.get("Район"), "corpus_name": row.get("Корпус"), "address": row.get("Адрес"), "class": row.get("Класс"), "section": row.get("Секция"), "floor": _to_int(row.get("Этаж")), "lot_number": row.get("Номер лота"), "premise_kind": row.get("Вид помещения"), "sales_start_date": _parse_iso_date(row.get("Старт продаж")), "plan_completion_date": _parse_iso_date(row.get("Планируемая дата ввода")), "actual_completion_date": _parse_iso_date(row.get("Фактическая дата ввода")), "readiness_pct": _parse_pct(row.get("Готовность")), "construction_stage": row.get("Стадия строительства"), "finishing": row.get("Отделка по корпусу"), "status": row.get("Статус"), "is_sold": _parse_bool_yes(row.get("Продано")), "rooms_dev_site": row.get("Количество комнат(Сайт девелопера)"), "rooms_pd": row.get("Количество комнат(ПД)"), "rooms_objective": rooms_obj, "rooms_int": _normalize_rooms(rooms_obj), "area_dev_site": _to_num(row.get("Площадь, м2(Сайт девелопера)")), "area_pd": _to_num(row.get("Площадь, м2(ПД)")), "price_calculated_total_rub": _to_num(row.get("Расчетный бюджет лота, Р")), "price_per_m2_rub": _to_num(row.get("Цена за м2, Р")), "price_offer_total_rub": _to_num(row.get("Цена предложения, Р")), "price_delta_rub": _to_num(row.get("Дельта цена, Р")), "price_delta_pct": _to_num(row.get("Дельта цена, %")), "pricing_method": row.get("Способ определения цены"), "price_set_date": _parse_iso_date(row.get("Дата установки цены")), "price_actual_date": _parse_iso_date(row.get("Дата актуальности цены")), "contract_date": _parse_iso_date(row.get("Дата договора")), "registration_date": _parse_iso_date(row.get("Дата регистрации")), "contract_type": row.get("Тип сделки"), "buyer_type": row.get("Тип покупателя"), "registration_number": row.get("Номер регистрации"), "encumbrance_type": row.get("Тип обременения"), "bank_name": row.get("Банк"), "encumbrance_start_date": _parse_iso_date(row.get("Дата начала обременения")), "egrn_actual_date": _parse_iso_date(row.get("Дата актуальности данных из ЕГРН")), "raw_id": raw_id, "raw_props": json.dumps(row, ensure_ascii=False), "snapshot_date": snapshot_date, } _LOTS_UPSERT = text( """ INSERT INTO objective_lots ( objective_lot_id, objective_project_id, project_name, developer, city, district, corpus_name, address, class, section, floor, lot_number, premise_kind, sales_start_date, plan_completion_date, actual_completion_date, readiness_pct, construction_stage, finishing, status, is_sold, rooms_dev_site, rooms_pd, rooms_objective, rooms_int, area_dev_site, area_pd, price_calculated_total_rub, price_per_m2_rub, price_offer_total_rub, price_delta_rub, price_delta_pct, pricing_method, price_set_date, price_actual_date, contract_date, registration_date, contract_type, buyer_type, registration_number, encumbrance_type, bank_name, encumbrance_start_date, egrn_actual_date, raw_id, raw_props, snapshot_date ) VALUES ( :objective_lot_id, :objective_project_id, :project_name, :developer, :city, :district, :corpus_name, :address, :class, :section, :floor, :lot_number, :premise_kind, :sales_start_date, :plan_completion_date, :actual_completion_date, :readiness_pct, :construction_stage, :finishing, :status, :is_sold, :rooms_dev_site, :rooms_pd, :rooms_objective, :rooms_int, :area_dev_site, :area_pd, :price_calculated_total_rub, :price_per_m2_rub, :price_offer_total_rub, :price_delta_rub, :price_delta_pct, :pricing_method, :price_set_date, :price_actual_date, :contract_date, :registration_date, :contract_type, :buyer_type, :registration_number, :encumbrance_type, :bank_name, :encumbrance_start_date, :egrn_actual_date, :raw_id, CAST(:raw_props AS jsonb), :snapshot_date ) ON CONFLICT (objective_lot_id) DO UPDATE SET objective_project_id = EXCLUDED.objective_project_id, project_name = EXCLUDED.project_name, developer = EXCLUDED.developer, city = EXCLUDED.city, district = EXCLUDED.district, corpus_name = EXCLUDED.corpus_name, address = EXCLUDED.address, class = EXCLUDED.class, section = EXCLUDED.section, floor = EXCLUDED.floor, lot_number = EXCLUDED.lot_number, premise_kind = EXCLUDED.premise_kind, sales_start_date = EXCLUDED.sales_start_date, plan_completion_date = EXCLUDED.plan_completion_date, actual_completion_date = EXCLUDED.actual_completion_date, readiness_pct = EXCLUDED.readiness_pct, construction_stage = EXCLUDED.construction_stage, finishing = EXCLUDED.finishing, status = EXCLUDED.status, is_sold = EXCLUDED.is_sold, rooms_dev_site = EXCLUDED.rooms_dev_site, rooms_pd = EXCLUDED.rooms_pd, rooms_objective = EXCLUDED.rooms_objective, rooms_int = EXCLUDED.rooms_int, area_dev_site = EXCLUDED.area_dev_site, area_pd = EXCLUDED.area_pd, price_calculated_total_rub = EXCLUDED.price_calculated_total_rub, price_per_m2_rub = EXCLUDED.price_per_m2_rub, price_offer_total_rub = EXCLUDED.price_offer_total_rub, price_delta_rub = EXCLUDED.price_delta_rub, price_delta_pct = EXCLUDED.price_delta_pct, pricing_method = EXCLUDED.pricing_method, price_set_date = EXCLUDED.price_set_date, price_actual_date = EXCLUDED.price_actual_date, contract_date = EXCLUDED.contract_date, registration_date = EXCLUDED.registration_date, contract_type = EXCLUDED.contract_type, buyer_type = EXCLUDED.buyer_type, registration_number = EXCLUDED.registration_number, encumbrance_type = EXCLUDED.encumbrance_type, bank_name = EXCLUDED.bank_name, encumbrance_start_date = EXCLUDED.encumbrance_start_date, egrn_actual_date = EXCLUDED.egrn_actual_date, raw_id = EXCLUDED.raw_id, raw_props = EXCLUDED.raw_props, snapshot_date = EXCLUDED.snapshot_date, updated_at = NOW() """ ) _LOTS_HISTORY_INSERT = text( """ INSERT INTO objective_lots_history ( objective_lot_id, snapshot_date, status, is_sold, price_calculated_total_rub, price_per_m2_rub, price_offer_total_rub, price_delta_pct, area_pd, contract_date, registration_date, contract_type, bank_name, raw_id ) VALUES ( :objective_lot_id, :snapshot_date, :status, :is_sold, :price_calculated_total_rub, :price_per_m2_rub, :price_offer_total_rub, :price_delta_pct, :area_pd, :contract_date, :registration_date, :contract_type, :bank_name, :raw_id ) ON CONFLICT (objective_lot_id, snapshot_date) DO NOTHING """ ) # ── orchestration ─────────────────────────────────────────────────────────── def parse_corp_sum(payload: dict, group_name: str, raw_id: int | None, db, dry_run: bool) -> int: rows = payload.get("result") if isinstance(payload, dict) else None if not isinstance(rows, list): logger.warning("corp_sum: payload без list result, skip") return 0 n_inserted = 0 for row in rows: if not isinstance(row, dict): continue params = _row_to_corp_room_month(row, group_name, raw_id) if not params["report_month"] or not params["project_name"] or not params["corpus_name"]: logger.warning("corp_sum: пропуск row без обязательных полей: %s", {k: row.get(k) for k in ("Месяц", "Проект", "Корпус")}) continue if not dry_run: db.execute(_CRM_UPSERT, params) n_inserted += 1 return n_inserted def parse_lots_pf(payload: dict, raw_id: int | None, snapshot_date: date, db, dry_run: bool) -> tuple[int, int]: rows = payload.get("result") if isinstance(payload, dict) else None if not isinstance(rows, list): logger.warning("lots_pf: payload без list result, skip") return 0, 0 n_lots = 0 n_history = 0 for row in rows: if not isinstance(row, dict): continue params = _row_to_lot(row, raw_id, snapshot_date) if not params["objective_lot_id"]: logger.warning("lots_pf: пропуск row без Id") continue if not dry_run: db.execute(_LOTS_UPSERT, params) db.execute(_LOTS_HISTORY_INSERT, { "objective_lot_id": params["objective_lot_id"], "snapshot_date": snapshot_date, "status": params["status"], "is_sold": params["is_sold"], "price_calculated_total_rub": params["price_calculated_total_rub"], "price_per_m2_rub": params["price_per_m2_rub"], "price_offer_total_rub": params["price_offer_total_rub"], "price_delta_pct": params["price_delta_pct"], "area_pd": params["area_pd"], "contract_date": params["contract_date"], "registration_date": params["registration_date"], "contract_type": params["contract_type"], "bank_name": params["bank_name"], "raw_id": raw_id, }) n_lots += 1 n_history += 1 return n_lots, n_history def detect_kind(payload: Any) -> str | None: """Эвристика по полям первого row: 'corp_sum' | 'lots_pf' | None.""" if not isinstance(payload, dict): return None rows = payload.get("result") if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict): return None first = rows[0] if "Месяц" in first and "Корпус" in first and "Объем предложения, шт." in first: return "corp_sum" if "Id" in first and "Номер лота" in first and "Площадь, м2(ПД)" in first: return "lots_pf" return None # ── streaming support ─────────────────────────────────────────────────────── class _IterBytesReader: """File-like wrapper над iterator of bytes chunks (для ijson.items()). ijson 3.x ожидает file-like с .read(n) — но httpx.Response.iter_bytes() возвращает generator байт. Этот класс собирает chunks в буфер и отдаёт их по требованию через .read(n). """ def __init__(self, chunks: Iterator[bytes]) -> None: self._chunks = iter(chunks) self._buf = bytearray() self._eof = False def read(self, n: int = -1) -> bytes: if n < 0: # Прочитать всё что осталось for c in self._chunks: self._buf.extend(c) data = bytes(self._buf) self._buf.clear() self._eof = True return data while len(self._buf) < n and not self._eof: try: self._buf.extend(next(self._chunks)) except StopIteration: self._eof = True break out = bytes(self._buf[:n]) del self._buf[:n] return out def parse_lots_pf_stream( stream: BinaryIO | Iterator[bytes], raw_id: int | None, snapshot_date: date, db: Any, dry_run: bool, batch_size: int = 500, ) -> tuple[int, int]: """Streaming-парсер lots_pf через ijson — для больших ответов (>100 МБ). В отличие от parse_lots_pf (читает весь payload в dict), эта функция обрабатывает JSON инкрементально через ijson.items(), что позволяет парсить файлы >100 МБ без загрузки всего payload в память. Args: stream: либо file-like с .read(n), либо iterator of bytes chunks (например httpx.Response.iter_bytes(65536)). Если stream не имеет метода .read — автоматически оборачивается в _IterBytesReader. raw_id: ссылка на objective_raw_reports.raw_id (может быть None) snapshot_date: дата снимка для objective_lots_history db: SQLAlchemy Session dry_run: если True — только подсчёт, без записи в БД batch_size: коммит каждые N лотов (снижает нагрузку на память транзакции) Returns: tuple[n_lots, n_history] """ import ijson # type: ignore[import-untyped] # Если stream — iterator chunks (без .read), оборачиваем в file-like reader. if not hasattr(stream, "read"): stream = _IterBytesReader(stream) # type: ignore[arg-type] n_lots = 0 n_history = 0 batch: list[dict] = [] def _flush(batch: list[dict]) -> None: for params in batch: db.execute(_LOTS_UPSERT, params) db.execute(_LOTS_HISTORY_INSERT, { "objective_lot_id": params["objective_lot_id"], "snapshot_date": snapshot_date, "status": params["status"], "is_sold": params["is_sold"], "price_calculated_total_rub": params["price_calculated_total_rub"], "price_per_m2_rub": params["price_per_m2_rub"], "price_offer_total_rub": params["price_offer_total_rub"], "price_delta_pct": params["price_delta_pct"], "area_pd": params["area_pd"], "contract_date": params["contract_date"], "registration_date": params["registration_date"], "contract_type": params["contract_type"], "bank_name": params["bank_name"], "raw_id": raw_id, }) for row in ijson.items(stream, "result.item"): if not isinstance(row, dict): continue params = _row_to_lot(row, raw_id, snapshot_date) if not params["objective_lot_id"]: logger.warning("lots_pf_stream: пропуск row без Id") continue n_lots += 1 n_history += 1 if not dry_run: batch.append(params) if len(batch) >= batch_size: _flush(batch) db.commit() batch.clear() logger.info("lots_pf_stream: committed batch, n_lots=%d so far", n_lots) if not dry_run and batch: _flush(batch) db.commit() batch.clear() logger.info( "lots_pf_stream: done — n_lots=%d n_history=%d dry_run=%s", n_lots, n_history, dry_run, ) return n_lots, n_history # ── CLI ───────────────────────────────────────────────────────────────────── def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--file-corp-sum", type=str, help="Локальный JSON-файл с corp_sum payload") ap.add_argument("--file-lots-pf", type=str, help="Локальный JSON-файл с lots_pf payload") ap.add_argument("--raw-id", type=int, help="Парсить один объект из objective_raw_reports по raw_id") ap.add_argument("--latest-run", action="store_true", help="Парсить все unparsed raw из последнего objective_scrape_runs.run_id") ap.add_argument("--group-name", default="Екатеринбург", help="GroupName для file-режима (default: Екатеринбург)") ap.add_argument("--snapshot-date", type=str, default=None, help="Дата снимка для lots_pf (default: сегодня)") ap.add_argument("--dry-run", action="store_true", help="Не писать в БД (только parse + считать rows)") args = ap.parse_args() snap = _parse_iso_date(args.snapshot_date) or date.today() # Local-file mode (smoke без БД) if args.file_corp_sum or args.file_lots_pf: if not args.dry_run: db = SessionLocal() else: db = None # type: ignore try: if args.file_corp_sum: payload = json.loads(Path(args.file_corp_sum).read_text(encoding="utf-8")) n = parse_corp_sum(payload, args.group_name, None, db, args.dry_run) print(f"corp_sum: {n} rows {'(dry-run)' if args.dry_run else 'upserted'}") if args.file_lots_pf: payload = json.loads(Path(args.file_lots_pf).read_text(encoding="utf-8")) n_lots, n_hist = parse_lots_pf(payload, None, snap, db, args.dry_run) print(f"lots_pf: {n_lots} lots / {n_hist} history rows " f"{'(dry-run)' if args.dry_run else 'upserted'}") if not args.dry_run and db is not None: db.commit() finally: if db is not None: db.close() return 0 # DB mode — нужна реальная БД db = SessionLocal() try: if args.raw_id: raw_ids = [args.raw_id] elif args.latest_run: row = db.execute(text( "SELECT MAX(run_id) FROM objective_scrape_runs" )).scalar() if not row: print("Нет ни одного run.") return 1 raw_ids = [r[0] for r in db.execute(text( """ SELECT raw_id FROM objective_raw_reports WHERE run_id = :rid AND (rows_extracted IS NULL OR rows_extracted = 0) ORDER BY raw_id """ ), {"rid": row}).all()] if not raw_ids: print(f"В run {row} нет unparsed raw.") return 0 else: ap.error("Нужен --raw-id, --latest-run, или --file-corp-sum/--file-lots-pf") return 2 for rid in raw_ids: r = db.execute(text( """ SELECT raw_id, report_type, report_name, group_name, payload FROM objective_raw_reports WHERE raw_id = :rid """ ), {"rid": rid}).mappings().first() if not r: logger.warning("raw_id=%s не найден", rid) continue kind = detect_kind(r["payload"]) print(f"raw_id={rid} type={r['report_type']}/{r['report_name']} kind={kind}") n = 0 if kind == "corp_sum": n = parse_corp_sum(r["payload"], r["group_name"], rid, db, args.dry_run) elif kind == "lots_pf": n_lots, _ = parse_lots_pf(r["payload"], rid, snap, db, args.dry_run) n = n_lots else: logger.warning("raw_id=%s — kind неизвестен (тариф не отдал?), skip", rid) if not args.dry_run and n > 0: db.execute(text( "UPDATE objective_raw_reports SET rows_extracted = :n WHERE raw_id = :rid" ), {"n": n, "rid": rid}) if not args.dry_run: db.commit() print("DONE.") return 0 finally: db.close() if __name__ == "__main__": sys.exit(main())