"""Парсер 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 datetime import date, datetime from pathlib import Path from typing import Any # 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 # ── 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())