ole
55d93894ac
feat(scripts): Add backfill script for content_hash in cache tables feat(scripts): Create recompute script for analysis_cache population test(tests): Implement comprehensive tests for analysis module functions fix(tests): Update CLI tests to assert errors on stderr instead of stdout fix(tests): Adjust MCP integration tests to pass context parameter correctly fix(tests): Modify service tests to return hash on save functions for consistency
484 lines
17 KiB
Python
484 lines
17 KiB
Python
"""Typer-based CLI for FINN real estate analysis."""
|
|
|
|
import asyncio
|
|
from typing import Literal, Optional
|
|
|
|
import typer
|
|
|
|
from . import formatting
|
|
from .service import (
|
|
analyze_ad as svc_analyze_ad,
|
|
analyze_ad_against_comps as svc_analyze_ad_against_comps,
|
|
analyze_search as svc_analyze_search,
|
|
build_unit_vector_for_unit_code as svc_build_unit_vector,
|
|
compare_ads as svc_compare_ads,
|
|
decode_unit_vector_to_dict as svc_decode_unit_vector,
|
|
find_similar_to_liked as svc_find_similar_to_liked,
|
|
get_new_ads_since_last_run as svc_get_new_ads_since_last_run,
|
|
get_or_fetch_ad as svc_get_or_fetch_ad,
|
|
get_or_fetch_eiendom_unit as svc_get_or_fetch_eiendom_unit,
|
|
get_or_fetch_similar_units as svc_get_or_fetch_similar_units,
|
|
get_shortlist as svc_get_shortlist,
|
|
resolve_eiendom_unit_from_finn_url as svc_resolve_eiendom_unit,
|
|
save_feedback as svc_save_feedback,
|
|
)
|
|
|
|
app = typer.Typer(help="FINN real estate analysis and Eiendom.no enrichment")
|
|
cache_app = typer.Typer(help="Cache management commands")
|
|
config_app = typer.Typer(help="Configuration commands")
|
|
|
|
app.add_typer(cache_app, name="cache")
|
|
app.add_typer(config_app, name="config")
|
|
|
|
OutputFormat = Literal["json", "markdown", "table"]
|
|
|
|
|
|
# ============================================================================
|
|
# Search and analysis commands
|
|
# ============================================================================
|
|
|
|
|
|
@app.command()
|
|
def analyze_search(
|
|
url: str = typer.Argument(..., help="FINN search URL"),
|
|
max_pages: int = typer.Option(3, help="Max pages to scrape"),
|
|
detail_limit: int = typer.Option(20, help="Max details to fetch"),
|
|
no_details: bool = typer.Option(False, help="Skip fetching listing details"),
|
|
no_eiendom: bool = typer.Option(False, help="Skip Eiendom.no enrichment"),
|
|
with_similar: bool = typer.Option(False, help="Include similar units for each listing"),
|
|
format: OutputFormat = typer.Option("json", help="Output format"),
|
|
) -> None:
|
|
"""Analyze a FINN search URL and return ranked shortlist."""
|
|
try:
|
|
result = asyncio.run(
|
|
svc_analyze_search(
|
|
url,
|
|
max_pages=max_pages,
|
|
detail_limit=detail_limit,
|
|
include_details=not no_details,
|
|
include_eiendom_no=not no_eiendom,
|
|
include_similar_units_for_shortlist=with_similar,
|
|
)
|
|
)
|
|
output = formatting.render_shortlist(result, format)
|
|
typer.echo(output)
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@app.command()
|
|
def get_ad(
|
|
finnkode: str = typer.Argument(..., help="FINN property code"),
|
|
force_refresh: bool = typer.Option(False, help="Bypass cache"),
|
|
no_eiendom: bool = typer.Option(False, help="Skip Eiendom.no enrichment"),
|
|
with_similar: bool = typer.Option(False, help="Include similar units for this property"),
|
|
format: OutputFormat = typer.Option("json", help="Output format"),
|
|
) -> None:
|
|
"""Fetch and display a single FINN listing."""
|
|
try:
|
|
ad = asyncio.run(svc_get_or_fetch_ad(finnkode, force_refresh=force_refresh))
|
|
ad_data = ad.model_dump()
|
|
output = formatting.render_ad(ad_data, format)
|
|
typer.echo(output)
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@app.command()
|
|
def enrich_ad(
|
|
finnkode: str = typer.Argument(..., help="FINN property code"),
|
|
with_similar: bool = typer.Option(False, help="Include similar units for this property"),
|
|
format: OutputFormat = typer.Option("json", help="Output format"),
|
|
) -> None:
|
|
"""Fetch and enrich a single FINN listing with Eiendom.no data."""
|
|
try:
|
|
result = asyncio.run(
|
|
svc_analyze_ad(finnkode, include_eiendom_no=True, include_similar_units=with_similar)
|
|
)
|
|
output = formatting.render_ad(result.get("ad", {}), format)
|
|
typer.echo(output)
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@app.command()
|
|
def compare(
|
|
finnkoder: list[str] = typer.Argument(..., help="FINN property codes to compare (2-10 codes)"),
|
|
no_eiendom: bool = typer.Option(False, help="Skip Eiendom.no enrichment"),
|
|
no_comps: bool = typer.Option(False, help="Skip comparable units"),
|
|
format: OutputFormat = typer.Option("json", help="Output format"),
|
|
) -> None:
|
|
"""Compare multiple FINN listings side by side."""
|
|
if len(finnkoder) < 2:
|
|
typer.echo("Error: Provide at least 2 finnkoder", err=True)
|
|
raise typer.Exit(1)
|
|
if len(finnkoder) > 10:
|
|
typer.echo("Error: Provide at most 10 finnkoder", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
try:
|
|
result = asyncio.run(
|
|
svc_compare_ads(
|
|
finnkoder,
|
|
include_eiendom_no=not no_eiendom,
|
|
include_comps=not no_comps,
|
|
)
|
|
)
|
|
output = formatting.render_comparison(result, format)
|
|
typer.echo(output)
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@app.command()
|
|
def analyze_against_comps(
|
|
finnkode: str = typer.Argument(..., help="FINN property code"),
|
|
listing_status: str = typer.Option(
|
|
"RECENTLY_SOLD", help="Comparable status (RECENTLY_SOLD, FOR_SALE, CURRENT)"
|
|
),
|
|
format: OutputFormat = typer.Option("json", help="Output format"),
|
|
) -> None:
|
|
"""Evaluate a listing against comparable recently-sold units."""
|
|
try:
|
|
result = asyncio.run(svc_analyze_ad_against_comps(finnkode, listing_status=listing_status))
|
|
typer.echo(formatting.render_comparison({"listings": [result]}, format))
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
# ============================================================================
|
|
# Eiendom.no commands
|
|
# ============================================================================
|
|
|
|
|
|
@app.command()
|
|
def resolve_unit(
|
|
url: str = typer.Argument(..., help="FINN listing URL"),
|
|
format: OutputFormat = typer.Option("json", help="Output format"),
|
|
) -> None:
|
|
"""Resolve an Eiendom.no unit_code from a FINN URL."""
|
|
try:
|
|
unit = asyncio.run(svc_resolve_eiendom_unit(url))
|
|
if unit:
|
|
typer.echo(formatting.render_unit(unit.model_dump(), format))
|
|
else:
|
|
typer.echo("Error: Could not resolve unit from URL", err=True)
|
|
raise typer.Exit(1)
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@app.command()
|
|
def get_unit(
|
|
unit_code: str = typer.Argument(..., help="Eiendom.no unit code"),
|
|
force_refresh: bool = typer.Option(False, help="Bypass cache"),
|
|
format: OutputFormat = typer.Option("json", help="Output format"),
|
|
) -> None:
|
|
"""Fetch Eiendom.no unit details."""
|
|
try:
|
|
unit = asyncio.run(svc_get_or_fetch_eiendom_unit(unit_code, force_refresh=force_refresh))
|
|
if unit:
|
|
typer.echo(formatting.render_unit(unit.model_dump(), format))
|
|
else:
|
|
typer.echo("Error: Unit not found", err=True)
|
|
raise typer.Exit(1)
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@app.command()
|
|
def build_vector(
|
|
unit_code: str = typer.Argument(..., help="Eiendom.no unit code"),
|
|
format: OutputFormat = typer.Option("json", help="Output format"),
|
|
) -> None:
|
|
"""Build a unit vector for an Eiendom.no unit."""
|
|
try:
|
|
result = asyncio.run(svc_build_unit_vector(unit_code))
|
|
typer.echo(formatting.render_ad(result, format))
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@app.command()
|
|
def decode_vector(
|
|
unit_vector: str = typer.Argument(..., help="Unit vector string (base64)"),
|
|
format: OutputFormat = typer.Option("json", help="Output format"),
|
|
) -> None:
|
|
"""Decode a unit vector to human-readable JSON."""
|
|
try:
|
|
result = svc_decode_unit_vector(unit_vector)
|
|
typer.echo(formatting.render_ad(result, format))
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@app.command()
|
|
def similar_units(
|
|
unit_code: str = typer.Argument(..., help="Eiendom.no unit code"),
|
|
status: str = typer.Option(
|
|
"RECENTLY_SOLD", help="Listing status (RECENTLY_SOLD, FOR_SALE, CURRENT)"
|
|
),
|
|
format: OutputFormat = typer.Option("json", help="Output format"),
|
|
) -> None:
|
|
"""Fetch similar/comparable units from Eiendom.no."""
|
|
try:
|
|
units = asyncio.run(svc_get_or_fetch_similar_units(unit_code, listing_status=status))
|
|
result = {"similar_units": [u.model_dump() for u in units]}
|
|
typer.echo(formatting.render_similar_units(result, format))
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@app.command()
|
|
def similar_to_liked(
|
|
finnkode: str = typer.Argument(..., help="FINN property code"),
|
|
mode: str = typer.Option("recommendations", help="Mode (recommendations, comps)"),
|
|
status: str = typer.Option(
|
|
"FOR_SALE", help="Listing status (RECENTLY_SOLD, FOR_SALE, CURRENT)"
|
|
),
|
|
format: OutputFormat = typer.Option("json", help="Output format"),
|
|
) -> None:
|
|
"""Find properties similar to one you've liked."""
|
|
try:
|
|
result = asyncio.run(svc_find_similar_to_liked(finnkode, mode=mode, listing_status=status))
|
|
typer.echo(formatting.render_similar_units(result, format))
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
# ============================================================================
|
|
# Feedback commands
|
|
# ============================================================================
|
|
|
|
|
|
@app.command()
|
|
def save_feedback(
|
|
finnkode: str = typer.Argument(..., help="FINN property code"),
|
|
verdict: str = typer.Argument(..., help="Verdict (liked, rejected, interesting, etc.)"),
|
|
notes: Optional[str] = typer.Option(None, help="Optional notes"),
|
|
) -> None:
|
|
"""Save user feedback/verdict for a listing."""
|
|
try:
|
|
result = svc_save_feedback(finnkode, verdict, notes)
|
|
typer.echo(f"Saved feedback for {finnkode}: {verdict}")
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
# ============================================================================
|
|
# History and diff commands
|
|
# ============================================================================
|
|
|
|
|
|
@app.command()
|
|
def shortlist(
|
|
run_id: Optional[int] = typer.Option(None, help="Run ID (defaults to latest)"),
|
|
limit: int = typer.Option(10, help="Max listings to return"),
|
|
format: OutputFormat = typer.Option("json", help="Output format"),
|
|
) -> None:
|
|
"""Fetch stored shortlist from a previous search run."""
|
|
try:
|
|
result = svc_get_shortlist(run_id, limit)
|
|
typer.echo(formatting.render_shortlist(result, format))
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@app.command()
|
|
def diff(
|
|
url: str = typer.Argument(..., help="FINN search URL"),
|
|
format: OutputFormat = typer.Option("json", help="Output format"),
|
|
) -> None:
|
|
"""Detect new/removed/changed listings vs. the previous run."""
|
|
try:
|
|
result = svc_get_new_ads_since_last_run(url)
|
|
typer.echo(formatting.render_diff(result, format))
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
# ============================================================================
|
|
# Cache management
|
|
# ============================================================================
|
|
|
|
|
|
@cache_app.command()
|
|
def stats() -> None:
|
|
"""Show cache statistics."""
|
|
try:
|
|
import json
|
|
import sqlite3
|
|
|
|
from .config import FINN_CACHE_PATH
|
|
|
|
conn = sqlite3.connect(str(FINN_CACHE_PATH))
|
|
cursor = conn.cursor()
|
|
|
|
# Get row counts and hash statistics for each table
|
|
tables = ["finn_ads", "eiendom_units", "similar_units", "analysis_cache", "cache_meta"]
|
|
stats = {}
|
|
|
|
for table in tables:
|
|
cursor.execute(f"SELECT COUNT(*) FROM {table}")
|
|
total = cursor.fetchone()[0]
|
|
|
|
if total == 0:
|
|
stats[table] = {"total_rows": 0}
|
|
continue
|
|
|
|
# For tables with content_hash or deps_hash
|
|
if table == "analysis_cache":
|
|
cursor.execute(f"SELECT COUNT(*) FROM {table} WHERE deps_hash IS NOT NULL")
|
|
with_hash = cursor.fetchone()[0]
|
|
elif table != "cache_meta" or True: # All have content_hash or value
|
|
cursor.execute(f"SELECT COUNT(*) FROM {table} WHERE content_hash IS NOT NULL")
|
|
with_hash = cursor.fetchone()[0]
|
|
|
|
stats[table] = {
|
|
"total_rows": total,
|
|
"rows_with_hash": with_hash,
|
|
"pct_with_hash": round(100 * with_hash / total, 1) if total > 0 else 0,
|
|
}
|
|
|
|
# Special checks for finn_ads
|
|
cursor.execute(
|
|
'SELECT COUNT(*) FROM finn_ads '
|
|
'WHERE json_extract(payload, "$.eiendom_unit_code") IS NOT NULL '
|
|
'AND json_extract(payload, "$.eiendom_unit_code") != "null"'
|
|
)
|
|
ads_with_unit_code = cursor.fetchone()[0]
|
|
if "finn_ads" in stats and stats["finn_ads"]["total_rows"] > 0:
|
|
stats["finn_ads"]["with_eiendom_unit_code"] = ads_with_unit_code
|
|
stats["finn_ads"]["pct_with_unit_code"] = round(100 * ads_with_unit_code / stats["finn_ads"]["total_rows"], 1)
|
|
|
|
# Get fetched_at date ranges
|
|
for table in ["finn_ads", "eiendom_units", "similar_units"]:
|
|
cursor.execute(f"SELECT MIN(fetched_at), MAX(fetched_at) FROM {table}")
|
|
min_date, max_date = cursor.fetchone()
|
|
if min_date and max_date:
|
|
stats[table]["oldest_fetch"] = min_date
|
|
stats[table]["newest_fetch"] = max_date
|
|
|
|
conn.close()
|
|
|
|
# Format output
|
|
typer.echo("\n=== Cache Statistics ===\n")
|
|
for table, table_stats in stats.items():
|
|
typer.echo(f"{table}:")
|
|
for key, value in table_stats.items():
|
|
typer.echo(f" {key}: {value}")
|
|
typer.echo()
|
|
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@cache_app.command()
|
|
def clear() -> None:
|
|
"""Clear all cache."""
|
|
if not typer.confirm("Are you sure you want to clear all cache?"):
|
|
raise typer.Exit()
|
|
try:
|
|
# TODO: implement cache clear via cache.py
|
|
typer.echo("Cache cleared (not yet implemented)")
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@cache_app.command()
|
|
def clear_html() -> None:
|
|
"""Clear cached HTML only."""
|
|
try:
|
|
# TODO: implement HTML cache clear via cache.py
|
|
typer.echo("HTML cache cleared (not yet implemented)")
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@cache_app.command()
|
|
def clear_json() -> None:
|
|
"""Clear cached JSON only."""
|
|
try:
|
|
# TODO: implement JSON cache clear via cache.py
|
|
typer.echo("JSON cache cleared (not yet implemented)")
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
# ============================================================================
|
|
# Config management
|
|
# ============================================================================
|
|
|
|
|
|
@config_app.command()
|
|
def show() -> None:
|
|
"""Show current configuration."""
|
|
try:
|
|
from .config import FINN_CACHE_PATH
|
|
|
|
typer.echo(f"Cache path: {FINN_CACHE_PATH}")
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
@config_app.command()
|
|
def path() -> None:
|
|
"""Show cache directory path."""
|
|
try:
|
|
from .config import FINN_CACHE_PATH
|
|
|
|
typer.echo(FINN_CACHE_PATH)
|
|
except Exception as e:
|
|
typer.echo(f"Error: {e}", err=True)
|
|
raise typer.Exit(1)
|
|
|
|
|
|
# ============================================================================
|
|
# Utility commands
|
|
# ============================================================================
|
|
|
|
|
|
@app.command()
|
|
def version() -> None:
|
|
"""Show version information."""
|
|
typer.echo("finn-eiendom 0.1.0")
|
|
|
|
|
|
@app.command()
|
|
def serve(
|
|
transport: str = typer.Option("stdio", help="Transport (stdio, http)"),
|
|
host: str = typer.Option("127.0.0.1", help="HTTP host"),
|
|
port: int = typer.Option(8010, help="HTTP port"),
|
|
) -> None:
|
|
"""Run the MCP server."""
|
|
if transport == "stdio":
|
|
from .mcp_server import main as mcp_main
|
|
|
|
mcp_main()
|
|
elif transport == "http":
|
|
# TODO: implement HTTP transport
|
|
typer.echo(f"HTTP transport not yet implemented (would run on {host}:{port})")
|
|
raise typer.Exit(1)
|
|
else:
|
|
typer.echo(f"Error: Unknown transport {transport}", err=True)
|
|
raise typer.Exit(1)
|