581 lines
20 KiB
Python
581 lines
20 KiB
Python
"""Output formatting for JSON, Markdown, and table formats."""
|
|
|
|
import json
|
|
from typing import Any, Literal
|
|
|
|
from .models import FinnAd, EiendomUnit, SimilarUnit
|
|
|
|
|
|
OutputFormat = Literal["json", "markdown", "table"]
|
|
|
|
|
|
def _validate_format(fmt: OutputFormat) -> None:
|
|
"""Raise ValueError if format is not supported."""
|
|
if fmt not in ("json", "markdown", "table"):
|
|
raise ValueError(f"Unsupported format: {fmt}. Supported: json, markdown, table")
|
|
|
|
|
|
# ============================================================================
|
|
# Ad renderers
|
|
# ============================================================================
|
|
|
|
|
|
def render_ad(payload: dict[str, Any], fmt: OutputFormat) -> str:
|
|
"""Render a single FINN ad (FinnAd model dump)."""
|
|
_validate_format(fmt)
|
|
|
|
if fmt == "json":
|
|
return json.dumps(payload, indent=2, default=str)
|
|
elif fmt == "markdown":
|
|
return _render_ad_markdown(payload)
|
|
else: # table
|
|
return _render_ad_markdown(payload) # For single ad, markdown is better than table
|
|
|
|
|
|
def _render_ad_markdown(ad: dict[str, Any]) -> str:
|
|
"""Render ad as markdown."""
|
|
lines = [
|
|
f"# {ad.get('title', 'FINN Listing')}",
|
|
"",
|
|
f"**Finnkode:** {ad.get('finnkode')} ",
|
|
f"**URL:** {ad.get('url')}",
|
|
"",
|
|
"## Details",
|
|
f"- **Address:** {ad.get('address')}",
|
|
f"- **Property Type:** {ad.get('property_type')}",
|
|
f"- **Ownership Type:** {ad.get('ownership_type')}",
|
|
f"- **Area:** {ad.get('area_m2')} m²",
|
|
f"- **Rooms:** {ad.get('rooms')} ({ad.get('bedrooms')} bedrooms)",
|
|
f"- **Floor:** {ad.get('floor')}",
|
|
f"- **Construction Year:** {ad.get('construction_year')}",
|
|
"",
|
|
"## Pricing",
|
|
f"- **Asking Price:** {ad.get('asking_price'):,}"
|
|
if ad.get("asking_price")
|
|
else "- **Asking Price:** N/A",
|
|
f"- **Total Price:** {ad.get('total_price'):,}"
|
|
if ad.get("total_price")
|
|
else "- **Total Price:** N/A",
|
|
f"- **Common Costs:** {ad.get('common_costs'):,}/month"
|
|
if ad.get("common_costs")
|
|
else "- **Common Costs:** N/A",
|
|
f"- **Shared Debt:** {ad.get('shared_debt'):,}"
|
|
if ad.get("shared_debt")
|
|
else "- **Shared Debt:** N/A",
|
|
"",
|
|
"## Features",
|
|
f"- **Heating:** {ad.get('heating')}",
|
|
f"- **Energy Rating:** {ad.get('energy_rating')}",
|
|
f"- **Balcony:** {'Yes' if ad.get('has_balcony') else 'No'}",
|
|
f"- **Terrace:** {'Yes' if ad.get('has_terrace') else 'No'}",
|
|
f"- **Elevator:** {'Yes' if ad.get('has_elevator') else 'No'}",
|
|
f"- **Parking:** {'Yes' if ad.get('has_parking') else 'No'}",
|
|
f"- **Garage:** {'Yes' if ad.get('has_garage') else 'No'}",
|
|
"",
|
|
"## Listing Info",
|
|
f"- **Broker:** {ad.get('broker_company')} ({ad.get('broker_name')})",
|
|
f"- **First Seen:** {ad.get('first_seen_at')}",
|
|
f"- **Last Seen:** {ad.get('last_seen_at')}",
|
|
]
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ============================================================================
|
|
# Unit renderers (Eiendom.no)
|
|
# ============================================================================
|
|
|
|
|
|
def render_unit(payload: dict[str, Any], fmt: OutputFormat) -> str:
|
|
"""Render an Eiendom.no unit."""
|
|
_validate_format(fmt)
|
|
|
|
if fmt == "json":
|
|
return json.dumps(payload, indent=2, default=str)
|
|
else:
|
|
return _render_unit_markdown(payload)
|
|
|
|
|
|
def _render_unit_markdown(unit: dict[str, Any]) -> str:
|
|
"""Render unit as markdown."""
|
|
lines = [
|
|
f"# Eiendom.no Unit: {unit.get('address', 'Unknown')}",
|
|
"",
|
|
f"**Unit Code:** {unit.get('unit_code')}",
|
|
f"**Coordinates:** {unit.get('lat')}, {unit.get('lng')}",
|
|
"",
|
|
"## Property Details",
|
|
f"- **Type:** {unit.get('property_type')}",
|
|
f"- **Rooms:** {unit.get('rooms')}",
|
|
f"- **Floor:** {unit.get('floor')}",
|
|
f"- **Construction Year:** {unit.get('construction_year')}",
|
|
f"- **Usable Area:** {unit.get('usable_area')} m²",
|
|
"",
|
|
"## Market Data",
|
|
f"- **Estimated Selling Price:** {unit.get('estimated_selling_price'):,}"
|
|
if unit.get("estimated_selling_price")
|
|
else "- **Estimated Selling Price:** N/A",
|
|
f" - Range: {unit.get('estimated_selling_price_lower'):,} - {unit.get('estimated_selling_price_upper'):,}"
|
|
if unit.get("estimated_selling_price_lower")
|
|
else "",
|
|
f"- **Listing Price:** {unit.get('listing_price'):,}"
|
|
if unit.get("listing_price")
|
|
else "- **Listing Price:** N/A",
|
|
f"- **Listing Price/m²:** {unit.get('listing_sqm_price'):,} NOK/m²"
|
|
if unit.get("listing_sqm_price")
|
|
else "",
|
|
f"- **Common Costs:** {unit.get('common_costs'):,}/month"
|
|
if unit.get("common_costs")
|
|
else "- **Common Costs:** N/A",
|
|
f"- **Days on Market:** {unit.get('days_on_market')}",
|
|
f"- **Market Placement Score:** {unit.get('market_placement_score')}",
|
|
f"- **Sale Status:** {unit.get('sale_status')}",
|
|
"",
|
|
f"**Fetched at:** {unit.get('fetched_at')}",
|
|
]
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ============================================================================
|
|
# Unit images renderer
|
|
# ============================================================================
|
|
|
|
|
|
def render_unit_images(payload: dict[str, Any], fmt: OutputFormat) -> str:
|
|
"""Render unit images for visual assessment."""
|
|
_validate_format(fmt)
|
|
|
|
if fmt == "json":
|
|
return json.dumps(payload, indent=2, default=str)
|
|
else:
|
|
return _render_unit_images_markdown(payload)
|
|
|
|
|
|
def _render_unit_images_markdown(data: dict[str, Any]) -> str:
|
|
"""Render unit images as markdown with image references for Claude."""
|
|
unit_code = data.get("unit_code", "Unknown")
|
|
address = data.get("address", "Unknown")
|
|
images = data.get("unit_images") or []
|
|
|
|
lines = [
|
|
f"# Unit Images: {address}",
|
|
"",
|
|
f"**Unit Code:** {unit_code}",
|
|
f"**Total Photos:** {len(images)}",
|
|
"",
|
|
"## Property Photos",
|
|
"",
|
|
]
|
|
|
|
if not images:
|
|
lines.append("No images available for this unit.")
|
|
else:
|
|
lines.append("Below are the property images for visual assessment:")
|
|
lines.append("")
|
|
for i, img_url in enumerate(images, 1):
|
|
lines.append(f"### Photo {i}")
|
|
lines.append(f"")
|
|
lines.append("")
|
|
|
|
lines.append("---")
|
|
lines.append("")
|
|
lines.append("**Analysis Notes:**")
|
|
lines.append("Review the above photos to assess:")
|
|
lines.append("- View quality (street, landscape, water, etc.)")
|
|
lines.append("- Space and layout (openness, ceiling height, etc.)")
|
|
lines.append("- Lighting and window placement")
|
|
lines.append("- Condition and maintenance state")
|
|
lines.append("- Kitchen and bathroom features")
|
|
lines.append("- Overall atmosphere and livability")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ============================================================================
|
|
# Shortlist renderer
|
|
# ============================================================================
|
|
|
|
|
|
def render_shortlist(payload: dict[str, Any], fmt: OutputFormat) -> str:
|
|
"""Render a shortlist of analyzed listings."""
|
|
_validate_format(fmt)
|
|
|
|
if fmt == "json":
|
|
return json.dumps(payload, indent=2, default=str)
|
|
elif fmt == "markdown":
|
|
return _render_shortlist_markdown(payload)
|
|
else: # table
|
|
return _render_shortlist_table(payload)
|
|
|
|
|
|
def _render_shortlist_markdown(data: dict[str, Any]) -> str:
|
|
"""Render shortlist as markdown."""
|
|
shortlist = data.get("shortlist", [])
|
|
lines = [
|
|
"# Shortlist",
|
|
f"Found {len(shortlist)} listings",
|
|
"",
|
|
]
|
|
for i, ad in enumerate(shortlist, 1):
|
|
lines.extend(
|
|
[
|
|
f"## {i}. {ad.get('title', 'Unknown')}",
|
|
f"**Finnkode:** {ad.get('finnkode')}",
|
|
f"**Price:** {ad.get('asking_price'):,} NOK"
|
|
if ad.get("asking_price")
|
|
else "**Price:** N/A",
|
|
f"**Area:** {ad.get('area_m2')} m²",
|
|
f"**Address:** {ad.get('address')}",
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _render_shortlist_table(data: dict[str, Any]) -> str:
|
|
"""Render shortlist as table."""
|
|
shortlist = data.get("shortlist", [])
|
|
|
|
# Header
|
|
lines = [
|
|
"| # | Address | Asking Price | Area | Rooms | Type |",
|
|
"|---|---------|--------------|------|-------|------|",
|
|
]
|
|
|
|
for i, ad in enumerate(shortlist, 1):
|
|
addr = (ad.get("address") or "")[:40]
|
|
price = f"{ad.get('asking_price'):,}" if ad.get("asking_price") else "N/A"
|
|
area = f"{ad.get('area_m2')} m²" if ad.get("area_m2") else "N/A"
|
|
rooms = ad.get("rooms") or "N/A"
|
|
ptype = ad.get("property_type") or "N/A"
|
|
lines.append(f"| {i} | {addr} | {price} | {area} | {rooms} | {ptype} |")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ============================================================================
|
|
# Comparison renderer
|
|
# ============================================================================
|
|
|
|
|
|
def render_comparison(payload: dict[str, Any], fmt: OutputFormat) -> str:
|
|
"""Render a side-by-side comparison of listings."""
|
|
_validate_format(fmt)
|
|
|
|
if fmt == "json":
|
|
return json.dumps(payload, indent=2, default=str)
|
|
elif fmt == "markdown":
|
|
return _render_comparison_markdown(payload)
|
|
else: # table
|
|
return _render_comparison_table(payload)
|
|
|
|
|
|
def _render_comparison_markdown(data: dict[str, Any]) -> str:
|
|
"""Render comparison as markdown."""
|
|
listings = data.get("listings", [])
|
|
lines = [
|
|
"# Listing Comparison",
|
|
f"Comparing {len(listings)} listings",
|
|
"",
|
|
]
|
|
for i, ad in enumerate(listings, 1):
|
|
lines.extend(
|
|
[
|
|
f"## {i}. {ad.get('title', 'Unknown')}",
|
|
f"- **Address:** {ad.get('address')}",
|
|
f"- **Price:** {ad.get('asking_price'):,} NOK"
|
|
if ad.get("asking_price")
|
|
else "- **Price:** N/A",
|
|
f"- **Area:** {ad.get('area_m2')} m²",
|
|
f"- **Price/m²:** {ad.get('asking_price') // ad.get('area_m2') if ad.get('asking_price') and ad.get('area_m2') else 'N/A'} NOK/m²",
|
|
f"- **Rooms:** {ad.get('rooms')}",
|
|
f"- **Common Costs:** {ad.get('common_costs'):,}/month"
|
|
if ad.get("common_costs")
|
|
else "- **Common Costs:** N/A",
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _render_comparison_table(data: dict[str, Any]) -> str:
|
|
"""Render comparison as table."""
|
|
listings = data.get("listings", [])
|
|
|
|
lines = [
|
|
"| Address | Asking Price | Area | Price/m² | Rooms | Common Costs |",
|
|
"|---------|--------------|------|----------|-------|--------------|",
|
|
]
|
|
|
|
for ad in listings:
|
|
addr = (ad.get("address") or "")[:30]
|
|
price = f"{ad.get('asking_price'):,}" if ad.get("asking_price") else "N/A"
|
|
area = f"{ad.get('area_m2')}" if ad.get("area_m2") else "N/A"
|
|
price_sqm = (
|
|
f"{ad.get('asking_price') // ad.get('area_m2')}"
|
|
if ad.get("asking_price") and ad.get("area_m2")
|
|
else "N/A"
|
|
)
|
|
rooms = ad.get("rooms") or "N/A"
|
|
costs = f"{ad.get('common_costs'):,}" if ad.get("common_costs") else "N/A"
|
|
lines.append(f"| {addr} | {price} | {area} m² | {price_sqm} | {rooms} | {costs} |")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ============================================================================
|
|
# Similar units renderer
|
|
# ============================================================================
|
|
|
|
|
|
def render_similar_units(payload: dict[str, Any], fmt: OutputFormat) -> str:
|
|
"""Render similar/comparable units."""
|
|
_validate_format(fmt)
|
|
|
|
if fmt == "json":
|
|
return json.dumps(payload, indent=2, default=str)
|
|
elif fmt == "markdown":
|
|
return _render_similar_units_markdown(payload)
|
|
else: # table
|
|
return _render_similar_units_table(payload)
|
|
|
|
|
|
def _render_similar_units_markdown(data: dict[str, Any]) -> str:
|
|
"""Render similar units as markdown."""
|
|
similar = data.get("similar_units", [])
|
|
lines = [
|
|
"# Similar/Comparable Units",
|
|
f"Found {len(similar)} comparable listings",
|
|
"",
|
|
]
|
|
for i, unit in enumerate(similar, 1):
|
|
lines.extend(
|
|
[
|
|
f"## {i}. {unit.get('address', 'Unknown')}",
|
|
f"- **Listing Price:** {unit.get('listing_price'):,} NOK"
|
|
if unit.get("listing_price")
|
|
else "- **Listing Price:** N/A",
|
|
f"- **Selling Price:** {unit.get('selling_price'):,} NOK"
|
|
if unit.get("selling_price")
|
|
else "- **Selling Price:** N/A",
|
|
f"- **Area:** {unit.get('usable_area')} m²"
|
|
if unit.get("usable_area")
|
|
else "- **Area:** N/A",
|
|
f"- **Price/m²:** {unit.get('sqm_price'):,} NOK/m²"
|
|
if unit.get("sqm_price")
|
|
else "- **Price/m²:** N/A",
|
|
f"- **Days on Market:** {unit.get('days_on_market')}",
|
|
f"- **Status:** {unit.get('sale_status')}",
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _render_similar_units_table(data: dict[str, Any]) -> str:
|
|
"""Render similar units as table."""
|
|
similar = data.get("similar_units", [])
|
|
|
|
lines = [
|
|
"| Address | Listing Price | Selling Price | Area | Price/m² | Days | Status |",
|
|
"|---------|---------------|---------------|------|----------|------|--------|",
|
|
]
|
|
|
|
for unit in similar:
|
|
addr = (unit.get("address") or "")[:30]
|
|
list_price = f"{unit.get('listing_price'):,}" if unit.get("listing_price") else "N/A"
|
|
sell_price = f"{unit.get('selling_price'):,}" if unit.get("selling_price") else "N/A"
|
|
area = f"{unit.get('usable_area')}" if unit.get("usable_area") else "N/A"
|
|
price_sqm = f"{unit.get('sqm_price'):,}" if unit.get("sqm_price") else "N/A"
|
|
days = unit.get("days_on_market") or "N/A"
|
|
status = unit.get("sale_status") or "N/A"
|
|
lines.append(
|
|
f"| {addr} | {list_price} | {sell_price} | {area} m² | {price_sqm} | {days} | {status} |"
|
|
)
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ============================================================================
|
|
# Diff renderer
|
|
# ============================================================================
|
|
|
|
|
|
def render_diff(payload: dict[str, Any], fmt: OutputFormat) -> str:
|
|
"""Render diff of new/removed/changed listings."""
|
|
_validate_format(fmt)
|
|
|
|
if fmt == "json":
|
|
return json.dumps(payload, indent=2, default=str)
|
|
elif fmt == "markdown":
|
|
return _render_diff_markdown(payload)
|
|
else: # table
|
|
return _render_diff_table(payload)
|
|
|
|
|
|
def _render_diff_markdown(data: dict[str, Any]) -> str:
|
|
"""Render diff as markdown."""
|
|
new = data.get("new_ads", [])
|
|
removed = data.get("removed_ads", [])
|
|
changed = data.get("changed_ads", [])
|
|
|
|
lines = [
|
|
"# Listing Diff",
|
|
"",
|
|
]
|
|
|
|
if new:
|
|
lines.extend(
|
|
[
|
|
f"## 🆕 New Listings ({len(new)})",
|
|
"",
|
|
]
|
|
)
|
|
for ad in new:
|
|
lines.extend(
|
|
[
|
|
f"- {ad.get('title', 'Unknown')} ({ad.get('address')})",
|
|
f" - Price: {ad.get('asking_price'):,} NOK"
|
|
if ad.get("asking_price")
|
|
else " - Price: N/A",
|
|
"",
|
|
]
|
|
)
|
|
|
|
if removed:
|
|
lines.extend(
|
|
[
|
|
f"## ❌ Removed Listings ({len(removed)})",
|
|
"",
|
|
]
|
|
)
|
|
for ad in removed:
|
|
lines.extend(
|
|
[
|
|
f"- {ad.get('title', 'Unknown')} ({ad.get('address')})",
|
|
"",
|
|
]
|
|
)
|
|
|
|
if changed:
|
|
lines.extend(
|
|
[
|
|
f"## 🔄 Changed Listings ({len(changed)})",
|
|
"",
|
|
]
|
|
)
|
|
for change in changed:
|
|
lines.extend(
|
|
[
|
|
f"- {change.get('title', 'Unknown')} ({change.get('address')})",
|
|
]
|
|
)
|
|
diffs = change.get("diffs", {})
|
|
for key, (old, new_val) in diffs.items():
|
|
lines.append(f" - {key}: {old} → {new_val}")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _render_diff_table(data: dict[str, Any]) -> str:
|
|
"""Render diff as table."""
|
|
new = data.get("new_ads", [])
|
|
removed = data.get("removed_ads", [])
|
|
changed = data.get("changed_ads", [])
|
|
|
|
lines = [
|
|
"| Type | Address | Price | Status |",
|
|
"|------|---------|-------|--------|",
|
|
]
|
|
|
|
for ad in new:
|
|
addr = (ad.get("address") or "")[:30]
|
|
price = f"{ad.get('asking_price'):,}" if ad.get("asking_price") else "N/A"
|
|
lines.append(f"| 🆕 New | {addr} | {price} | - |")
|
|
|
|
for ad in removed:
|
|
addr = (ad.get("address") or "")[:30]
|
|
lines.append(f"| ❌ Removed | {addr} | - | - |")
|
|
|
|
for change in changed:
|
|
addr = (change.get("address") or "")[:30]
|
|
diffs = change.get("diffs", {})
|
|
changed_fields = ", ".join(diffs.keys())
|
|
lines.append(f"| 🔄 Changed | {addr} | - | {changed_fields} |")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ============================================================================
|
|
# Score breakdown renderer
|
|
# ============================================================================
|
|
|
|
|
|
def render_score_breakdown(payload: dict[str, Any], fmt: OutputFormat) -> str:
|
|
"""Render score breakdown details."""
|
|
_validate_format(fmt)
|
|
|
|
if fmt == "json":
|
|
return json.dumps(payload, indent=2, default=str)
|
|
else:
|
|
return _render_score_breakdown_markdown(payload)
|
|
|
|
|
|
def _render_score_breakdown_markdown(data: dict[str, Any]) -> str:
|
|
"""Render score breakdown as markdown."""
|
|
scores = data.get("scores", {})
|
|
lines = [
|
|
"# Score Breakdown",
|
|
"",
|
|
]
|
|
|
|
for key, value in scores.items():
|
|
lines.append(f"- **{key}:** {value}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ============================================================================
|
|
# Cache stats renderer
|
|
# ============================================================================
|
|
|
|
|
|
def render_cache_stats(payload: dict[str, Any], fmt: OutputFormat) -> str:
|
|
"""Render cache statistics."""
|
|
_validate_format(fmt)
|
|
|
|
if fmt == "json":
|
|
return json.dumps(payload, indent=2, default=str)
|
|
elif fmt == "markdown":
|
|
return _render_cache_stats_markdown(payload)
|
|
else: # table
|
|
return _render_cache_stats_table(payload)
|
|
|
|
|
|
def _render_cache_stats_markdown(data: dict[str, Any]) -> str:
|
|
"""Render cache stats as markdown."""
|
|
lines = [
|
|
"# Cache Statistics",
|
|
"",
|
|
f"- **Total FINN Ads:** {data.get('total_finn_ads', 0)}",
|
|
f"- **Total Eiendom Units:** {data.get('total_eiendom_units', 0)}",
|
|
f"- **Total Search Runs:** {data.get('total_search_runs', 0)}",
|
|
f"- **Cache Size:** {data.get('cache_size_mb', 0):.2f} MB",
|
|
f"- **Last Updated:** {data.get('last_updated')}",
|
|
]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _render_cache_stats_table(data: dict[str, Any]) -> str:
|
|
"""Render cache stats as table."""
|
|
lines = [
|
|
"| Metric | Value |",
|
|
"|--------|-------|",
|
|
f"| Total FINN Ads | {data.get('total_finn_ads', 0)} |",
|
|
f"| Total Eiendom Units | {data.get('total_eiendom_units', 0)} |",
|
|
f"| Total Search Runs | {data.get('total_search_runs', 0)} |",
|
|
f"| Cache Size | {data.get('cache_size_mb', 0):.2f} MB |",
|
|
f"| Last Updated | {data.get('last_updated')} |",
|
|
]
|
|
return "\n".join(lines)
|