122 lines
4.1 KiB
Python
122 lines
4.1 KiB
Python
"""Celery task: еженедельная синхронизация реестра КРТ-площадок ЕКБ (#1060).
|
||
|
||
Fetch → parse → UPSERT в ekb_krt_sites (ON CONFLICT по site_name).
|
||
SAVEPOINT per-row: одна упавшая запись не откатывает всю транзакцию.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
|
||
from sqlalchemy import text
|
||
|
||
from app.core.db import SessionLocal
|
||
from app.services.scrapers.ekb_krt_registry import (
|
||
KRT_URL,
|
||
KRTSite,
|
||
fetch_krt_page,
|
||
parse_krt_page,
|
||
)
|
||
from app.workers.celery_app import celery_app
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_UPSERT_SQL = text("""
|
||
INSERT INTO ekb_krt_sites (
|
||
site_name, district, krt_kind,
|
||
developer_name, contract_number, contract_date,
|
||
auction_start_price, auction_final_price,
|
||
gradpotential_sqm, area_ha, mkd_count,
|
||
paga_decisions, raw_block, source_url, fetched_at
|
||
)
|
||
VALUES (
|
||
:site_name, :district, :krt_kind,
|
||
:developer_name, :contract_number, CAST(:contract_date AS date),
|
||
CAST(:auction_start_price AS numeric), CAST(:auction_final_price AS numeric),
|
||
CAST(:gradpotential_sqm AS numeric), CAST(:area_ha AS numeric),
|
||
CAST(:mkd_count AS integer),
|
||
CAST(:paga_decisions AS jsonb), :raw_block, :source_url, now()
|
||
)
|
||
ON CONFLICT (site_name) DO UPDATE SET
|
||
district = EXCLUDED.district,
|
||
krt_kind = EXCLUDED.krt_kind,
|
||
developer_name = EXCLUDED.developer_name,
|
||
contract_number = EXCLUDED.contract_number,
|
||
contract_date = EXCLUDED.contract_date,
|
||
auction_start_price = EXCLUDED.auction_start_price,
|
||
auction_final_price = EXCLUDED.auction_final_price,
|
||
gradpotential_sqm = EXCLUDED.gradpotential_sqm,
|
||
area_ha = EXCLUDED.area_ha,
|
||
mkd_count = EXCLUDED.mkd_count,
|
||
paga_decisions = EXCLUDED.paga_decisions,
|
||
raw_block = EXCLUDED.raw_block,
|
||
source_url = EXCLUDED.source_url,
|
||
fetched_at = now()
|
||
""")
|
||
|
||
|
||
def _site_to_params(site: KRTSite) -> dict:
|
||
return {
|
||
"site_name": site.site_name,
|
||
"district": site.district,
|
||
"krt_kind": site.krt_kind,
|
||
"developer_name": site.developer_name,
|
||
"contract_number": site.contract_number,
|
||
"contract_date": site.contract_date.isoformat() if site.contract_date else None,
|
||
"auction_start_price": site.auction_start_price,
|
||
"auction_final_price": site.auction_final_price,
|
||
"gradpotential_sqm": site.gradpotential_sqm,
|
||
"area_ha": site.area_ha,
|
||
"mkd_count": site.mkd_count,
|
||
"paga_decisions": json.dumps(site.paga_decisions, ensure_ascii=False),
|
||
"raw_block": site.raw_block,
|
||
"source_url": site.source_url,
|
||
}
|
||
|
||
|
||
@celery_app.task(name="tasks.ekb_krt_sync.sync_krt_sites")
|
||
def sync_krt_sites() -> dict[str, int]:
|
||
"""Fetch + parse + UPSERT реестра КРТ-площадок ЕКБ.
|
||
|
||
Returns: {"sites": N} — число обработанных площадок.
|
||
"""
|
||
logger.info("ekb_krt_sync: start (url=%s)", KRT_URL)
|
||
|
||
html = fetch_krt_page()
|
||
sites = parse_krt_page(html)
|
||
|
||
if not sites:
|
||
logger.warning("ekb_krt_sync: parse returned 0 sites — страница изменилась?")
|
||
return {"sites": 0}
|
||
|
||
upserted = 0
|
||
failed = 0
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
for site in sites:
|
||
try:
|
||
with db.begin_nested():
|
||
db.execute(_UPSERT_SQL, _site_to_params(site))
|
||
upserted += 1
|
||
except Exception as exc:
|
||
failed += 1
|
||
logger.warning(
|
||
"ekb_krt_sync: UPSERT failed for '%s': %s",
|
||
site.site_name[:60],
|
||
exc,
|
||
)
|
||
db.commit()
|
||
except Exception:
|
||
db.rollback()
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
logger.info(
|
||
"ekb_krt_sync: done — upserted=%d failed=%d",
|
||
upserted,
|
||
failed,
|
||
)
|
||
return {"sites": upserted}
|