"
)
# ── Dual price headline: «Ориентир запроса» + «Ожидаемая цена продажи» ───────
# Зеркалит web HeroSummary (#648 S5a): две равновесные цифры. Asking — медиана
# по объявлениям; expected_sold — реалистичная цена сделки (asking × ratio из
# asking_to_sold_ratios, migration 080). Graceful: при отсутствии sold (None/0 —
# старые оценки / пустой бакет ratio-таблицы) рисуем ТОЛЬКО asking-колонку.
def _discount_pct(estimate: AggregatedEstimate) -> int | None:
"""Скидка запрос→продажа в % (как web): из asking_to_sold_ratio при наличии,
иначе из самих чисел sold/median. None если sold нет или скидка < 1%."""
sold = estimate.expected_sold_price_rub
if not sold or sold <= 0:
return None
ratio = estimate.asking_to_sold_ratio
if ratio is not None and ratio > 0:
pct = round((1 - ratio) * 100)
elif estimate.median_price_rub > 0:
pct = round((1 - sold / estimate.median_price_rub) * 100)
else:
return None
return pct if pct >= 1 else None
def _price_figure_cell(
*,
label: str,
value_rub: int,
range_low: int | None,
range_high: int | None,
per_m2: int | None,
accent: str,
delta_pct: int | None = None,
) -> str:
"""Одна ценовая колонка (label + крупное число + диапазон + ₽/м²).
Каждое под-поле (диапазон / ₽/м²) рендерится только при валидном значении —
как web (guard на typeof number). delta_pct — опциональный chip «−N%»."""
delta_chip = ""
if delta_pct is not None:
delta_chip = (
f"−{delta_pct}%"
)
range_html = ""
if range_low is not None and range_high is not None and range_high > range_low:
range_html = (
f"
'
"Запрос — по чему выставлены сопоставимые квартиры в объявлениях. "
"Ожидаемая цена продажи — реалистичная цена сделки по данным ДКП "
"Росреестра, обычно ниже запроса."
"
"
)
return (
f'
{asking_cell}{sold_cell}
'
f"{explainer}"
)
# ── Page 1: Cover ────────────────────────────────────────────────────────────
def _build_cover(estimate: AggregatedEstimate, input_snapshot: dict, brand) -> str: # type: ignore[no-untyped-def,type-arg]
today = dt.date.today()
# Используем реальное поле expires_at из оценки; fallback на +30 дней если None.
expires = (
estimate.expires_at.date()
if estimate.expires_at is not None
else today + dt.timedelta(days=30)
)
report_num = _report_number(estimate.estimate_id)
# Короткий адрес (для cover): берём первую часть до запятой
full_address = input_snapshot.get("address", "—")
address_short = full_address.split(",")[0:3]
address_short = ", ".join(s.strip() for s in address_short)
address = _html.escape(address_short or full_address)
area = input_snapshot.get("area_m2", 0)
rooms = input_snapshot.get("rooms", 0)
floor = input_snapshot.get("floor") or "—"
total_floors = input_snapshot.get("total_floors") or "—"
year_built = input_snapshot.get("year_built", "—")
house_type = input_snapshot.get("house_type")
repair_state = input_snapshot.get("repair_state")
has_balcony = input_snapshot.get("has_balcony")
house_labels = {
"panel": "Панельный",
"brick": "Кирпичный",
"monolith": "Монолитный",
"monolith_brick": "Монолит-кирпич",
"other": "Другое",
}
repair_labels = {
"needs_repair": "Требуется ремонт",
"standard": "Стандартный",
"good": "Хороший",
"excellent": "Евроремонт",
}
rooms_label = (
"Студия"
if rooms == 0
else f"{rooms} комнат" + ("а" if rooms == 1 else "ы" if 2 <= rooms <= 4 else "")
)
house_label = house_labels.get(house_type, "—") if house_type else "—"
repair_label = repair_labels.get(repair_state, "Не указано") if repair_state else "Не указано"
balcony_label = "Есть" if has_balcony else "Нет" if has_balcony is False else "—"
# Active market subband — 4-118 days range (как у Брусники).
# Если есть days_on_market в analogs — берём min/max, иначе фиксированно.
days_min, days_max = _days_on_market_range(estimate.analogs)
# Deals range — если deals есть, считаем; иначе fallback к listings range
deals_low, deals_high = _deals_range(
estimate.actual_deals, fallback=(estimate.range_low_rub, estimate.range_high_rub)
)
listings_bar = _price_range_bar(
estimate.range_low_rub,
estimate.range_high_rub,
sub_label="Активный рынок с аналогичным состоянием ремонта",
days_min=days_min,
days_max=days_max,
show_days=True,
)
deals_bar = _price_range_bar(
deals_low,
deals_high,
sub_label="Диапазон цен по фактическим сделкам",
)
logo_url = _safe_logo_url(brand.logo_url)
logo_html = (
f""
if logo_url
else f""
f"{_html.escape(brand.name).upper()}"
)
disclaimer_html = ""
if brand.pdf_disclaimer:
disclaimer_html = (
f"
"
f"{_html.escape(brand.pdf_disclaimer)}
"
)
return f"""
{logo_html}
АНАЛИЗ РЫНКА И РАСЧЕТ ВЫКУПНОЙ СТОИМОСТИ КВАРТИРЫ
№ отчета
{report_num}
Дата отчета
{today.strftime("%d.%m.%Y")}
Срок действия данных
до {expires.strftime("%d.%m.%Y")}
Адрес
{address}
Год постройки
{year_built}
Тип дома
{house_label}
Этаж / этажность
{floor} / {total_floors}
Площадь
{area} м²
Планировка
{rooms_label}
Состояние ремонта
{repair_label}
Балкон / лоджия
{balcony_label}
{_dual_price_block(estimate, brand)}
Диапазон цен в объявлениях
(без учета ремонта)
{listings_bar}
Диапазон цен по фактическим сделкам
{deals_bar}
Что важно при оценке стоимости квартиры:
Цены в объявлениях — ожидания собственников
Фактические сделки проходят на 5–12% ниже, что подтверждают
Росреестр, ДомКлик и продажи агенств недвижимости
Ремонт оценивается по состоянию, а не по вложенным суммам
Инвестиции в 400 – 600 тыс. руб. повышают цену объекта всего
на 150 – 250 тыс. руб — покупатель оценивает общее состояние квартиры
Неочевидные расходы при самостоятельной продаже
При самостоятельной продаже суммарные расходы могут достигать
до 15% стоимости квартиры (торг, риелтор, нотариус, справки)
Этот отчёт онлайн: {settings.public_url}?id={estimate.estimate_id}
{disclaimer_html}
"""
def _days_on_market_range(lots: list[AnalogLot]) -> tuple[int, int]:
"""Min/max days_on_market по аналогам. Fallback 4-118 как у Брусники."""
days = [lot.days_on_market for lot in lots if lot.days_on_market is not None]
if not days:
return 4, 118
return min(days), max(days)
def _deals_range(deals: list[AnalogLot], fallback: tuple[int, int]) -> tuple[int, int]:
"""Min/max price от deals. Fallback к listings range."""
if not deals:
return fallback
prices = [d.price_rub for d in deals]
return min(prices), max(prices)
# ── Page 2: Listings (market) ────────────────────────────────────────────────
def _build_listings_page(estimate: AggregatedEstimate, input_snapshot: dict, brand) -> str: # type: ignore[no-untyped-def,type-arg]
n_total = estimate.n_analogs
# #1531: убрана строка-дубль «(с учётом ремонта)». Estimator НЕ фильтрует
# аналоги по repair_state (coverage listings.repair_state ~2%, см. estimator.py:160),
# а лишь применяет ценовой коэффициент к медиане/диапазону — поэтому отдельного
# count «с учётом ремонта» не существует, второе число было идентично n_total.
# Source logos (pseudo) — берём из estimate.sources_used (не захардкоженный список).
sources_to_show = estimate.sources_used or []
sources_html = "".join(_source_logo_pill(s) for s in sources_to_show[:5])
# Params правой колонки — параметры поиска (НЕ конкретной квартиры)
area = float(input_snapshot.get("area_m2", 0) or 0)
area_min = round(area * 0.85, 1)
area_max = round(area * 1.15, 1)
rooms = input_snapshot.get("rooms", 0)
year_built = input_snapshot.get("year_built")
year_range = f"{year_built - 2}-{year_built + 2}" if year_built else "—"
house_type = input_snapshot.get("house_type")
house_label = (
{
"panel": "Эконом до 2000 года",
"brick": "Кирпич до 2000",
"monolith": "Современный (монолит)",
"monolith_brick": "Современный (м-к)",
"other": "Любой",
}.get(house_type, "Любой")
if house_type
else "Любой"
)
repair_state = input_snapshot.get("repair_state")
repair_label = (
{
"needs_repair": "Требуется ремонт",
"standard": "Стандартный",
"good": "Хороший",
"excellent": "Евроремонт",
}.get(repair_state, "Любой")
if repair_state
else "Любой"
)
rooms_label = "Студия" if rooms == 0 else f"{rooms} комнаты"
# Полоска диапазона
days_min, days_max = _days_on_market_range(estimate.analogs)
range_bar = _price_range_bar(
estimate.range_low_rub,
estimate.range_high_rub,
sub_label="Рынок",
days_min=days_min,
days_max=days_max,
show_days=True,
)
# Топ-5 примеров (отсортированных по distance)
top5 = sorted(estimate.analogs, key=lambda x: x.distance_m or 9999)[:5]
examples_rows = _examples_rows(top5)
return f"""
{_html.escape(brand.name).upper()}
РЫНОК КВАРТИР – АНАЛОГОВ ПО ОБЪЯВЛЕНИЯМ
Количество объявлений по аналогичным объектам
{n_total} шт.
Источники данных
{sources_html}
Расположение
± 1 км от локации дома
Тип дома
{house_label}
Год постройки
{year_range}
Диапазон площади
{area_min} - {area_max} м²
Планировка
{rooms_label}
Состояние ремонта
{repair_label}
Диапазон цен в объявлениях
{range_bar}
Примеры аналогичных объектов с учётом ремонта
Адрес квартиры
Источник
Стоимость 1 м², руб.
Стоимость объекта, руб.
Срок экспозиции, дней
{examples_rows}
"""
def _examples_rows(lots: list[AnalogLot]) -> str:
if not lots:
return (
"
Нет данных
"
)
rows = []
for lot in lots:
addr = _html.escape(lot.address)
safe = _safe_url(lot.source_url)
if safe:
addr_cell = (
f"{addr} ↗"
)
else:
addr_cell = addr
ppm2_cell = _fmt_ppm2(lot.price_per_m2) if lot.price_per_m2 is not None else "—"
price_cell = _fmt_ppm2(lot.price_rub) if lot.price_rub is not None else "—"
# #1533: 0 — валидное значение (лот размещён сегодня), `0 or '—'` теряло бы его.
days_cell = lot.days_on_market if lot.days_on_market is not None else "—"
rows.append(
"
"
f"
{addr_cell}
"
f"
"
f"{_source_badge_inline(lot.source)}
"
f"
{ppm2_cell}
"
f"
{price_cell}
"
f"
{days_cell}
"
"
"
)
return "".join(rows)
# ── Page 3: Deals ────────────────────────────────────────────────────────────
def _build_deals_page(estimate: AggregatedEstimate, input_snapshot: dict, brand) -> str: # type: ignore[no-untyped-def,type-arg]
n_deals = len(estimate.actual_deals)
today = dt.date.today()
period_start = today - dt.timedelta(days=estimate.period_months * 30)
# Источники для сделок — берём из estimate.sources_used (не захардкоженный список).
# Фильтруем по известным источникам сделок; fallback к пустому (не fabricate).
_deal_source_keys = {"etazhi", "domklik", "rosreestr"}
deal_sources = [s for s in (estimate.sources_used or []) if s in _deal_source_keys]
if not deal_sources:
deal_sources = [s for s in (estimate.sources_used or [])]
sources_html = "".join(_source_logo_pill(s) for s in deal_sources[:5])
area = float(input_snapshot.get("area_m2", 0) or 0)
area_min = round(area * 0.85, 1)
area_max = round(area * 1.15, 1)
rooms = input_snapshot.get("rooms", 0)
year_built = input_snapshot.get("year_built")
year_range = f"{year_built - 2}–{year_built + 2}" if year_built else "—"
house_type = input_snapshot.get("house_type")
# #1532: маппинг идентичен странице «Листинги» (включая monolith_brick/other),
# иначе один и тот же «Тип дома» рассинхронизирован между стр.2 и стр.3.
house_label = (
{
"panel": "Эконом до 2000 года",
"brick": "Кирпич до 2000",
"monolith": "Современный (монолит)",
"monolith_brick": "Современный (м-к)",
"other": "Любой",
}.get(house_type, "Любой")
if house_type
else "Любой"
)
rooms_label = "Студия" if rooms == 0 else f"{rooms} комнаты"
deals_low, deals_high = _deals_range(
estimate.actual_deals, fallback=(estimate.range_low_rub, estimate.range_high_rub)
)
# #1366: НЕ показываем срок экспозиции для бара сделок. actual_deals — это
# зарегистрированные ДКП Росреестра без days_on_market (поле None), поэтому
# _days_on_market_range вернул бы хардкод-fallback (4, 118) — выдуманные сроки
# под заголовком «Диапазон цен по фактическим сделкам». Cover-страница так же
# намеренно не рисует days для бара сделок (show_days по умолчанию False).
range_bar = _price_range_bar(
deals_low,
deals_high,
sub_label="Рынок",
)
top5 = estimate.actual_deals[:5]
examples_rows = _examples_rows(top5)
return f"""
По данному объекту не найдено достаточного количества аналогов
для формирования надёжной рыночной оценки.
Пожалуйста, уточните параметры или обратитесь к специалисту.
"""
def generate_trade_in_pdf(
estimate: AggregatedEstimate,
input_snapshot: dict, # type: ignore[type-arg]
*,
brand=None, # type: ignore[no-untyped-def]
) -> bytes:
"""Генерирует WeasyPrint PDF в стиле Брусника.Обмен.
Если estimate.insufficient_data=True — рендерит одностраничный empty-state
вместо числового отчёта («Недостаточно данных», #697/#9).
Pages (normal):
1. Cover — № отчёта + параметры + 2 диапазона + 3 блока «Что важно»
2. Listings — рынок аналогов + источники + таблица примеров
3. Deals — фактические сделки + источники + таблица
4. Offer — Trade-In vs самопродажа + 4 преимущества
Args:
estimate: AggregatedEstimate из БД
input_snapshot: ввод пользователя (address, area_m2, ...)
brand: optional Brand для white-label
Returns:
PDF bytes для Response(media_type="application/pdf")
"""
# WeasyPrint импортируется лениво: его native-зависимости (pango/cairo) нужны
# только для финального write_pdf — HTML-билдеры (cover/listings/...) работают
# без них, что делает рендер-функции тестируемыми в окружениях без native libs.
from weasyprint import CSS, HTML
from app.services.brand import Brand
if brand is None:
brand = Brand(
slug="generic",
name="Trade-In",
logo_url=None,
primary_color="#1d4ed8",
accent_color="#f59e0b",
footer_text=None,
pdf_disclaimer=None,
)
# Sanitize brand colors before inserting into CSS/style attributes (#13).
brand = Brand(
slug=brand.slug,
name=brand.name,
logo_url=brand.logo_url,
primary_color=_safe_color(brand.primary_color, "#1d4ed8"),
accent_color=_safe_color(brand.accent_color, "#f59e0b"),
footer_text=brand.footer_text,
pdf_disclaimer=brand.pdf_disclaimer,
)
# #697/#9: insufficient data → empty-state page, NOT fabricated 0-median report.
if estimate.insufficient_data:
body_html = _build_insufficient_data_page(estimate, input_snapshot, brand)
else:
body_html = (
_build_cover(estimate, input_snapshot, brand)
+ _build_listings_page(estimate, input_snapshot, brand)
+ _build_deals_page(estimate, input_snapshot, brand)
+ _build_offer_page(estimate, brand)
)
html_str = (
''
f"Trade-In — {_html.escape(input_snapshot.get('address', ''))}"
f"{body_html}"
)
css_str = _build_css(brand=brand)
# SSRF guard (#13): кастомный url_fetcher блокирует file:// и нелоговые схемы.
# base_url=None — не resolve никакие относительные URL (все ресурсы data: или https:).
pdf_bytes = HTML(string=html_str, base_url=None).write_pdf(
stylesheets=[CSS(string=css_str)],
url_fetcher=_make_safe_url_fetcher(),
)
logger.info(
"PDF generated estimate_id=%s brand=%s size=%d insufficient_data=%s",
estimate.estimate_id,
brand.slug,
len(pdf_bytes),
estimate.insufficient_data,
)
return pdf_bytes