Italian real estate agents work in one of the most fragmented market structures in Europe. There are over 23,000 licensed real estate agencies in Italy, the majority of which are single-office or small independent operations. There is no MLS equivalent. Listings sit across multiple portals simultaneously with no standardised syndication. And the traditional lead generation method — cold calling property owners from printed newspaper classifieds — has been replaced by Immobiliare.it, which hosts Italy's largest concentration of residential listings with something most European portals do not expose: direct agency contact data in the structured listing record.

Immobiliare.it displays the listing agency's phone number, email address, and profile link on every professional listing. That contact data is structured and extractable. For a proptech company, a competing agency, or a B2B services provider targeting Italian real estate professionals, this turns Italy's dominant property portal into the most comprehensive agency database in the country — more complete and more current than any commercial directory, because it updates in real time as new agencies list properties.

The lead generation pipeline in this guide captures three distinct opportunity types from Immobiliare.it data using ScrapeBadger's Immobiliare.it Scraper: new listings as prospecting signals, price-reduced properties as motivated seller indicators, and long-on-market listings as frustrated seller opportunities. Each category maps to a different outreach strategy. The pipeline scores leads, deduplicates contacts, and exports a CRM-ready file with agency details and lead context.

Setup

bash

pip install scrapebadger sqlalchemy aiohttp python-dotenv

env

SCRAPEBADGER_API_KEY=your_key_here
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/xxx/yyy/zzz

Step 1: Data Models

Italy's residential real estate data model has several fields that do not exist on equivalent UK or US portals. The energy class (classe energetica) is structured from A4 to G. The Italian zone classification — microzona and macrozona, the OMI taxonomy — is embedded in listings and provides sub-municipal geographic precision. Floor number, portineria (concierge) presence, and box auto (private parking) are standard structured fields. Model them properly up front.

python

# models.py
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime
from enum import Enum


class PropertyType(str, Enum):
    APPARTAMENTO = "appartamento"
    VILLA = "villa"
    VILLETTA = "villetta"
    BILOCALE = "bilocale"
    TRILOCALE = "trilocale"
    ATTICO = "attico"
    LOFT = "loft"
    RUSTICO = "rustico"
    PALAZZO = "palazzo"
    ALTRO = "altro"


class ListingType(str, Enum):
    VENDITA = "vendita"        # For sale
    AFFITTO = "affitto"        # For rent
    ASTE = "aste"              # Judicial auction


class EnergiaClass(str, Enum):
    A4 = "A4"
    A3 = "A3"
    A2 = "A2"
    A1 = "A1"
    B = "B"
    C = "C"
    D = "D"
    E = "E"
    F = "F"
    G = "G"
    NC = "NC"   # Non classificato (not classified)


@dataclass
class ItalianListing:
    """A single Immobiliare.it listing snapshot."""
    listing_id: str
    listing_url: str
    listing_type: str              # vendita / affitto / aste
    property_type: str
    title: str

    # Pricing
    price: float
    price_per_sqm: Optional[float]
    currency: str = "EUR"

    # Property
    surface_area_sqm: Optional[float] = None
    rooms: Optional[int] = None
    bedrooms: Optional[int] = None
    bathrooms: Optional[int] = None
    floor: Optional[int] = None
    total_floors: Optional[int] = None
    year_built: Optional[int] = None

    # Italian-specific
    energia_class: Optional[str] = None    # A4 through G
    has_ascensore: bool = False            # Elevator
    has_box_auto: bool = False             # Private garage/parking
    has_terrazzo: bool = False             # Terrace
    has_giardino: bool = False             # Garden
    has_portineria: bool = False           # Concierge service
    has_piscina: bool = False              # Pool
    is_nuovo_costruzione: bool = False     # New construction

    # Location
    city: str = ""
    province: str = ""
    regione: str = ""
    cap: str = ""                          # Italian postal code
    microzona: Optional[str] = None        # OMI sub-zone
    macrozona: Optional[str] = None        # OMI zone
    latitude: Optional[float] = None
    longitude: Optional[float] = None
    address: Optional[str] = None

    # Agency contact data — the key Immobiliare.it advantage
    agency_name: Optional[str] = None
    agency_id: Optional[str] = None
    agency_phone: Optional[str] = None
    agency_email: Optional[str] = None
    agency_profile_url: Optional[str] = None
    agent_name: Optional[str] = None

    # Temporal signals
    listing_date: Optional[str] = None    # Data di inserimento
    days_on_market: Optional[int] = None
    price_history: list = field(default_factory=list)  # [{"date": ..., "price": ...}]

    # Collection metadata
    scraped_at: str = ""
    search_query: Optional[str] = None
    target_zone: Optional[str] = None

    @property
    def has_price_reduction(self) -> bool:
        """True if the listing has had at least one price reduction."""
        if len(self.price_history) < 2:
            return False
        prices = [h.get("price", 0) for h in self.price_history]
        return prices[-1] < prices[0]

    @property
    def price_reduction_pct(self) -> Optional[float]:
        """Total percentage reduction from first asking to current price."""
        if not self.has_price_reduction:
            return None
        prices = [h.get("price", 0) for h in self.price_history]
        if prices[0] == 0:
            return None
        return round((prices[-1] - prices[0]) / prices[0] * 100, 2)

    @property
    def is_long_on_market(self) -> bool:
        """Properties on market 90+ days are typically motivated sellers."""
        return (self.days_on_market or 0) >= 90

    @property
    def is_high_energia_class(self) -> bool:
        """A4 through B — energy-efficient, commands premium pricing."""
        return self.energia_class in ("A4", "A3", "A2", "A1", "B")


@dataclass
class Lead:
    """A scored, qualified lead ready for CRM export."""
    listing_id: str
    lead_type: str          # "new_listing" | "price_reduced" | "long_on_market" | "agency_contact"
    priority: str           # "high" | "medium" | "low"
    score: float            # 0-100
    
    # Property context
    title: str
    listing_url: str
    price: float
    city: str
    province: str
    property_type: str
    days_on_market: Optional[int]
    price_reduction_pct: Optional[float]
    energia_class: Optional[str]
    
    # Contact
    agency_name: Optional[str]
    agency_phone: Optional[str]
    agency_email: Optional[str]
    agent_name: Optional[str]
    
    # Generated outreach context
    outreach_angle: str     # Why this lead is worth contacting
    detected_at: str

Step 2: The Immobiliare.it Collection Layer

python

# immobiliare_collector.py
import asyncio
import os
from typing import Optional
from datetime import datetime
from scrapebadger import ScrapeBadger
from models import ItalianListing


def parse_listing(raw: dict, target_zone: str = None) -> Optional[ItalianListing]:
    """
    Parse a raw Immobiliare.it API response into a typed ItalianListing.
    Handles the nested structure of Italian property data.
    """
    try:
        listing_id = str(
            raw.get("id") or raw.get("listing_id") or raw.get("idImmobile") or ""
        )
        if not listing_id:
            return None

        # Price extraction
        price_data = raw.get("price") or raw.get("prezzo") or {}
        if isinstance(price_data, dict):
            price = float(price_data.get("value") or price_data.get("valore") or 0)
        else:
            price = float(price_data or 0)

        price_per_sqm = raw.get("price_per_sqm") or raw.get("prezzoMq") or raw.get("price_sqm")
        if price_per_sqm:
            price_per_sqm = float(price_per_sqm)

        # Agency contact data — the key field
        agency = raw.get("agency") or raw.get("agenzia") or {}
        if isinstance(agency, dict):
            agency_name = agency.get("name") or agency.get("nome") or agency.get("ragioneSociale")
            agency_id = str(agency.get("id") or agency.get("idAgenzia") or "")
            agency_phone = agency.get("phone") or agency.get("telefono") or agency.get("phoneNumber")
            agency_email = agency.get("email")
            agency_profile_url = agency.get("url") or agency.get("profileUrl")
        else:
            agency_name = agency_id = agency_phone = agency_email = agency_profile_url = None

        agent = raw.get("agent") or raw.get("agente") or {}
        agent_name = None
        if isinstance(agent, dict):
            agent_name = agent.get("name") or agent.get("nome")

        # Location
        location = raw.get("location") or raw.get("localizzazione") or {}
        if isinstance(location, dict):
            city = location.get("city") or location.get("comune") or ""
            province = location.get("province") or location.get("provincia") or ""
            regione = location.get("region") or location.get("regione") or ""
            cap = location.get("cap") or location.get("zipcode") or ""
            lat = location.get("latitude") or location.get("lat")
            lon = location.get("longitude") or location.get("lng") or location.get("lon")
            microzona = location.get("microzone") or location.get("microzona")
            macrozona = location.get("macrozone") or location.get("macrozona")
            address = location.get("address") or location.get("indirizzo")
        else:
            city = province = regione = cap = microzona = macrozona = address = ""
            lat = lon = None

        # Features / amenities
        features = raw.get("features") or raw.get("caratteristiche") or {}
        if not isinstance(features, dict):
            features = {}

        # Price history
        price_history = raw.get("price_history") or raw.get("storicoPrezzo") or []

        # Energy class
        energia = (
            raw.get("energia_class")
            or raw.get("classeEnergetica")
            or raw.get("energy_class")
            or features.get("classeEnergetica")
        )

        return ItalianListing(
            listing_id=listing_id,
            listing_url=raw.get("url") or raw.get("listing_url") or "",
            listing_type=raw.get("listing_type") or raw.get("contratto") or "vendita",
            property_type=raw.get("property_type") or raw.get("tipologia") or "appartamento",
            title=raw.get("title") or raw.get("titolo") or "",
            price=price,
            price_per_sqm=price_per_sqm,
            surface_area_sqm=float(raw.get("surface") or raw.get("superficie") or 0) or None,
            rooms=raw.get("rooms") or raw.get("locali"),
            bedrooms=raw.get("bedrooms") or raw.get("camere"),
            bathrooms=raw.get("bathrooms") or raw.get("bagni"),
            floor=raw.get("floor") or raw.get("piano"),
            total_floors=raw.get("total_floors") or raw.get("pianiTotali"),
            year_built=raw.get("year_built") or raw.get("annoCostruzione"),
            energia_class=str(energia).upper() if energia else None,
            has_ascensore=bool(features.get("ascensore") or raw.get("ascensore")),
            has_box_auto=bool(features.get("box") or raw.get("boxAuto")),
            has_terrazzo=bool(features.get("terrazzo") or raw.get("terrazzo")),
            has_giardino=bool(features.get("giardino") or raw.get("giardino")),
            has_portineria=bool(features.get("portineria") or raw.get("portineria")),
            has_piscina=bool(features.get("piscina") or raw.get("piscina")),
            is_nuovo_costruzione=bool(
                raw.get("new_construction") or raw.get("nuovaCostruzione")
            ),
            city=city,
            province=province,
            regione=regione,
            cap=cap,
            microzona=microzona,
            macrozona=macrozona,
            latitude=float(lat) if lat else None,
            longitude=float(lon) if lon else None,
            address=address,
            agency_name=agency_name,
            agency_id=agency_id,
            agency_phone=agency_phone,
            agency_email=agency_email,
            agency_profile_url=agency_profile_url,
            agent_name=agent_name,
            listing_date=raw.get("listing_date") or raw.get("dataInserimento"),
            days_on_market=raw.get("days_on_market") or raw.get("giorniSulMercato"),
            price_history=price_history if isinstance(price_history, list) else [],
            scraped_at=datetime.utcnow().isoformat(),
            target_zone=target_zone,
        )

    except Exception as e:
        print(f"Parse error for listing {raw.get('id', 'unknown')}: {e}")
        return None


async def search_listings(
    client: ScrapeBadger,
    city: str = None,
    province: str = None,
    zona: str = None,
    listing_type: str = "vendita",
    property_type: str = None,
    min_price: float = None,
    max_price: float = None,
    min_surface: float = None,
    max_surface: float = None,
    energia_class: list = None,
    max_results: int = 100,
    sort_by: str = "data_inserimento",  # or "prezzo_asc", "prezzo_desc"
) -> list[ItalianListing]:
    """
    Search Immobiliare.it listings with filters.
    sort_by="data_inserimento" surfaces newest listings first — important for
    new listing detection. Sort by "prezzo_asc" for price-sorted analysis.
    """
    listings = []

    try:
        params = {
            "listing_type": listing_type,
            "sort": sort_by,
            "limit": min(max_results, 200),
        }
        if city:
            params["city"] = city
        if province:
            params["province"] = province
        if zona:
            params["zona"] = zona
        if property_type:
            params["property_type"] = property_type
        if min_price:
            params["price_min"] = min_price
        if max_price:
            params["price_max"] = max_price
        if min_surface:
            params["surface_min"] = min_surface
        if max_surface:
            params["surface_max"] = max_surface
        if energia_class:
            params["energia_class"] = ",".join(energia_class)

        zone_label = zona or city or province or "Italy"
        response = await client.immobiliare.search(**params)

        raw_items = (
            response.listings
            if hasattr(response, "listings")
            else (response if isinstance(response, list) else [])
        )

        for raw in raw_items:
            item_dict = raw.model_dump() if hasattr(raw, "model_dump") else dict(raw)
            listing = parse_listing(item_dict, target_zone=zone_label)
            if listing and listing.price > 0:
                listings.append(listing)

        print(f"  [{zone_label}] {listing_type}: {len(listings)} listings")

    except Exception as e:
        print(f"Error searching Immobiliare.it ({city or zone_label}): {e}")

    return listings


async def get_listing_detail(
    client: ScrapeBadger,
    listing_id: str,
) -> Optional[ItalianListing]:
    """
    Fetch full detail for a specific listing — includes price history
    and complete agency contact data.
    """
    try:
        response = await client.immobiliare.get_listing(listing_id=listing_id)
        raw = response.model_dump() if hasattr(response, "model_dump") else dict(response)
        return parse_listing(raw)
    except Exception as e:
        print(f"Error fetching listing {listing_id}: {e}")
        return None

Step 3: The Lead Detection Engine

Three lead categories, three detection algorithms. Each is a pure function on a listing object — easy to test and adjust thresholds independently.

python

# lead_detector.py
from models import ItalianListing, Lead
from datetime import datetime


# ── Thresholds ────────────────────────────────────────────────────────────────
NEW_LISTING_HOURS = 48           # Consider new if listed within 48 hours
PRICE_REDUCTION_MIN_PCT = 3.0   # Minimum % drop to classify as price-reduced
LONG_ON_MARKET_DAYS = 90        # Days after which listing is "long on market"
LONG_ON_MARKET_SCORE_BOOST = 20 # Score bonus for very long (150+ days)


def _days_since_listing(listing_date_str: str) -> Optional[int]:
    """Calculate days since a listing was first posted."""
    if not listing_date_str:
        return None
    try:
        # Immobiliare.it uses ISO date format: "2026-05-12" or "2026-05-12T10:30:00"
        date_str = listing_date_str[:10]
        listing_date = datetime.strptime(date_str, "%Y-%m-%d")
        return (datetime.utcnow() - listing_date).days
    except Exception:
        return None


def detect_new_listing(listing: ItalianListing) -> Optional[Lead]:
    """
    Detect listings published in the last 48 hours.
    New listings are the most actionable for outreach — sellers have just
    committed to the process and are likely to engage with interested parties.
    """
    days_since = _days_since_listing(listing.listing_date)
    hours_since = None
    if days_since is not None:
        hours_since = days_since * 24

    # Check days_on_market field if listing_date isn't available
    dom = listing.days_on_market or days_since
    is_new = (dom is not None and dom <= 2) or (hours_since is not None and hours_since <= NEW_LISTING_HOURS)

    if not is_new:
        return None

    score = 60.0

    # Boost for contact data availability
    if listing.agency_phone:
        score += 15
    if listing.agency_email:
        score += 10

    # Boost for high-value property (>€500K in Italy signals motivated professional seller)
    if listing.price >= 500_000:
        score += 10

    # Boost for energy class (A/B class listings sell faster — warm market)
    if listing.is_high_energia_class:
        score += 5

    return Lead(
        listing_id=listing.listing_id,
        lead_type="new_listing",
        priority="high" if score >= 75 else "medium",
        score=round(score, 1),
        title=listing.title,
        listing_url=listing.listing_url,
        price=listing.price,
        city=listing.city,
        province=listing.province,
        property_type=listing.property_type,
        days_on_market=listing.days_on_market,
        price_reduction_pct=None,
        energia_class=listing.energia_class,
        agency_name=listing.agency_name,
        agency_phone=listing.agency_phone,
        agency_email=listing.agency_email,
        agent_name=listing.agent_name,
        outreach_angle=(
            f"New listing in {listing.city} — just posted. "
            f"{listing.property_type.capitalize()}, {listing.surface_area_sqm or '?'}m², "
            f"€{listing.price:,.0f}"
        ),
        detected_at=datetime.utcnow().isoformat(),
    )


def detect_price_reduction(listing: ItalianListing) -> Optional[Lead]:
    """
    Detect listings with meaningful price reductions.
    A seller who has already reduced their price is signalling flexibility —
    the most actionable signal for buyers and agents.
    """
    if not listing.has_price_reduction:
        return None

    reduction_pct = listing.price_reduction_pct
    if reduction_pct is None or abs(reduction_pct) < PRICE_REDUCTION_MIN_PCT:
        return None

    abs_reduction = abs(reduction_pct)
    score = 50.0

    # Score by depth of reduction
    if abs_reduction >= 15:
        score += 30    # Significant motivation signal
    elif abs_reduction >= 10:
        score += 20
    elif abs_reduction >= 5:
        score += 10

    # Score by duration — a long-on-market + price reduction = very motivated
    if listing.is_long_on_market:
        score += 15

    if listing.agency_phone:
        score += 5
    if listing.agency_email:
        score += 5

    # Low energy class + price reduction = energy upgrade opportunity
    if listing.energia_class in ("F", "G"):
        score += 5
        angle_suffix = " (low energy class — renovation opportunity)"
    else:
        angle_suffix = ""

    return Lead(
        listing_id=listing.listing_id,
        lead_type="price_reduced",
        priority="high" if score >= 70 else "medium" if score >= 50 else "low",
        score=round(score, 1),
        title=listing.title,
        listing_url=listing.listing_url,
        price=listing.price,
        city=listing.city,
        province=listing.province,
        property_type=listing.property_type,
        days_on_market=listing.days_on_market,
        price_reduction_pct=reduction_pct,
        energia_class=listing.energia_class,
        agency_name=listing.agency_name,
        agency_phone=listing.agency_phone,
        agency_email=listing.agency_email,
        agent_name=listing.agent_name,
        outreach_angle=(
            f"Price reduced {abs_reduction:.1f}% in {listing.city}. "
            f"Now €{listing.price:,.0f}. "
            f"On market {listing.days_on_market or '?'} days."
            f"{angle_suffix}"
        ),
        detected_at=datetime.utcnow().isoformat(),
    )


def detect_long_on_market(listing: ItalianListing) -> Optional[Lead]:
    """
    Detect listings that have been active for 90+ days without selling.
    In Italy's 92-94% sale-to-asking ratio market, a property sitting 90+ days
    either has a pricing problem or a condition problem — both are addressable.
    """
    dom = listing.days_on_market
    if dom is None or dom < LONG_ON_MARKET_DAYS:
        return None

    score = 45.0

    # Longer = more motivated
    if dom >= 180:
        score += LONG_ON_MARKET_SCORE_BOOST
    elif dom >= 120:
        score += 10

    # No recent price reduction + long on market = seller anchored too high
    if not listing.has_price_reduction:
        score += 10
        angle_note = "No price reductions yet — price anchoring likely."
    else:
        angle_note = f"Already reduced {abs(listing.price_reduction_pct or 0):.1f}%."

    if listing.agency_phone:
        score += 10
    if listing.agency_email:
        score += 5

    return Lead(
        listing_id=listing.listing_id,
        lead_type="long_on_market",
        priority="high" if score >= 70 else "medium" if score >= 50 else "low",
        score=round(score, 1),
        title=listing.title,
        listing_url=listing.listing_url,
        price=listing.price,
        city=listing.city,
        province=listing.province,
        property_type=listing.property_type,
        days_on_market=dom,
        price_reduction_pct=listing.price_reduction_pct,
        energia_class=listing.energia_class,
        agency_name=listing.agency_name,
        agency_phone=listing.agency_phone,
        agency_email=listing.agency_email,
        agent_name=listing.agent_name,
        outreach_angle=(
            f"{dom} days on market in {listing.city}. {angle_note} "
            f"€{listing.price:,.0f} asking."
        ),
        detected_at=datetime.utcnow().isoformat(),
    )


def classify_listing(listing: ItalianListing) -> list[Lead]:
    """
    Run all detection functions against a single listing.
    A listing can qualify as multiple lead types simultaneously —
    e.g., a new listing with a price reduction from a relisting.
    Returns all matching Lead objects.
    """
    detectors = [detect_new_listing, detect_price_reduction, detect_long_on_market]
    leads = []
    for detector in detectors:
        lead = detector(listing)
        if lead:
            leads.append(lead)
    return leads

Step 4: Agency Contact Deduplication

The agency contact extraction is the pipeline's most commercially distinctive feature. The same agency may appear across hundreds of listings. You want one clean contact record per agency, with a count of their active listings and their geographic concentration.

python

# agency_database.py
from collections import defaultdict
from models import ItalianListing


@dataclass
class AgencyRecord:
    """Deduplicated agency record with listing activity context."""
    agency_id: str
    agency_name: str
    phone: Optional[str]
    email: Optional[str]
    profile_url: Optional[str]
    active_listing_count: int
    cities_active_in: list[str]
    avg_listing_price: float
    listing_ids: list[str]
    first_seen: str
    last_seen: str


def extract_agency_database(listings: list[ItalianListing]) -> dict[str, AgencyRecord]:
    """
    Build a deduplicated agency database from a listing corpus.
    
    For each unique agency, tracks:
    - Contact information (phone, email)
    - Active listing count (their market footprint)
    - Geographic coverage (which cities they operate in)
    - Average listing price (price segment)
    """
    agency_listings = defaultdict(list)

    for listing in listings:
        agency_id = listing.agency_id or listing.agency_name
        if not agency_id:
            continue
        agency_listings[agency_id].append(listing)

    records = {}
    for agency_id, agency_listing_list in agency_listings.items():
        # Use the most complete contact record from any listing
        best_listing = max(
            agency_listing_list,
            key=lambda l: sum([
                bool(l.agency_phone),
                bool(l.agency_email),
                bool(l.agency_profile_url),
            ])
        )

        cities = list(set(l.city for l in agency_listing_list if l.city))
        prices = [l.price for l in agency_listing_list if l.price > 0]

        records[agency_id] = AgencyRecord(
            agency_id=agency_id,
            agency_name=best_listing.agency_name or "",
            phone=best_listing.agency_phone,
            email=best_listing.agency_email,
            profile_url=best_listing.agency_profile_url,
            active_listing_count=len(agency_listing_list),
            cities_active_in=sorted(cities),
            avg_listing_price=round(sum(prices) / len(prices), 0) if prices else 0,
            listing_ids=[l.listing_id for l in agency_listing_list],
            first_seen=min(
                (l.listing_date or "" for l in agency_listing_list),
                default=""
            ),
            last_seen=datetime.utcnow().isoformat(),
        )

    # Sort by listing count — most active agencies first
    return dict(sorted(
        records.items(),
        key=lambda x: x[1].active_listing_count,
        reverse=True,
    ))

Step 5: CRM Export and Alert Delivery

python

# exporter.py
import csv
import json
import aiohttp
import os
from datetime import datetime
from models import Lead
from agency_database import AgencyRecord


async def send_slack_digest(leads: list[Lead], agency_count: int) -> None:
    """Send a daily lead digest to Slack."""
    webhook = os.environ.get("SLACK_WEBHOOK_URL")
    if not webhook:
        return

    high = [l for l in leads if l.priority == "high"]
    medium = [l for l in leads if l.priority == "medium"]

    text = (
        f"*🏠 Italian Real Estate Lead Pipeline*\n"
        f"*{datetime.utcnow().strftime('%d %B %Y')}*\n\n"
        f"Total leads detected: *{len(leads)}*\n"
        f"  🔴 High priority: {len(high)}\n"
        f"  🟡 Medium priority: {len(medium)}\n"
        f"  Unique agencies with contact data: {agency_count}\n\n"
    )

    if high[:3]:
        text += "*Top leads today:*\n"
        for lead in high[:3]:
            text += (
                f"  • {lead.lead_type.replace('_', ' ').title()} — "
                f"{lead.city} | {lead.property_type} | €{lead.price:,.0f} | "
                f"Score: {lead.score:.0f}\n"
                f"    _{lead.outreach_angle[:80]}_\n"
                f"    {lead.listing_url}\n"
            )

    async with aiohttp.ClientSession() as session:
        try:
            await session.post(webhook, json={"text": text}, timeout=aiohttp.ClientTimeout(total=10))
        except Exception as e:
            print(f"Slack alert failed: {e}")


def export_leads_csv(
    leads: list[Lead],
    output_path: str = "italian_leads.csv",
) -> int:
    """Export leads to CRM-ready CSV."""
    if not leads:
        print("No leads to export.")
        return 0

    leads_sorted = sorted(leads, key=lambda l: l.score, reverse=True)

    fieldnames = [
        "lead_type", "priority", "score",
        "city", "province", "property_type",
        "price", "days_on_market", "price_reduction_pct",
        "energia_class",
        "agency_name", "agency_phone", "agency_email", "agent_name",
        "outreach_angle", "listing_url", "detected_at",
    ]

    with open(output_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for lead in leads_sorted:
            writer.writerow({
                "lead_type": lead.lead_type,
                "priority": lead.priority,
                "score": lead.score,
                "city": lead.city,
                "province": lead.province,
                "property_type": lead.property_type,
                "price": f"€{lead.price:,.0f}",
                "days_on_market": lead.days_on_market or "",
                "price_reduction_pct": f"{lead.price_reduction_pct:.1f}%" if lead.price_reduction_pct else "",
                "energia_class": lead.energia_class or "",
                "agency_name": lead.agency_name or "",
                "agency_phone": lead.agency_phone or "",
                "agency_email": lead.agency_email or "",
                "agent_name": lead.agent_name or "",
                "outreach_angle": lead.outreach_angle,
                "listing_url": lead.listing_url,
                "detected_at": lead.detected_at,
            })

    print(f"Exported {len(leads_sorted)} leads to {output_path}")
    return len(leads_sorted)


def export_agency_csv(
    agencies: dict[str, AgencyRecord],
    output_path: str = "italian_agencies.csv",
    min_listings: int = 2,
) -> int:
    """Export agency contact database to CSV."""
    qualifying = [
        a for a in agencies.values()
        if (a.phone or a.email) and a.active_listing_count >= min_listings
    ]
    qualifying.sort(key=lambda a: a.active_listing_count, reverse=True)

    fieldnames = [
        "agency_name", "phone", "email", "profile_url",
        "active_listings", "avg_price", "cities",
    ]

    with open(output_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for agency in qualifying:
            writer.writerow({
                "agency_name": agency.agency_name,
                "phone": agency.phone or "",
                "email": agency.email or "",
                "profile_url": agency.profile_url or "",
                "active_listings": agency.active_listing_count,
                "avg_price": f"€{agency.avg_listing_price:,.0f}",
                "cities": ", ".join(agency.cities_active_in[:5]),
            })

    print(f"Exported {len(qualifying)} agencies to {output_path}")
    return len(qualifying)

Step 6: The Complete Pipeline

python

# main.py
import asyncio
import os
from scrapebadger import ScrapeBadger
from immobiliare_collector import search_listings
from lead_detector import classify_listing
from agency_database import extract_agency_database
from exporter import export_leads_csv, export_agency_csv, send_slack_digest


# ── Target zones configuration ───────────────────────────────────────────────
TARGETS = [
    # Format: (city_or_zone, province, listing_type, max_results)
    # Focus on cities showing strong price momentum in 2026
    ("Milano",   "MI", "vendita", 150),
    ("Bologna",  "BO", "vendita", 100),
    ("Firenze",  "FI", "vendita", 100),
    ("Napoli",   "NA", "vendita", 100),
    ("Roma",     "RM", "vendita", 150),
    # High-yield Southern markets
    ("Catania",  "CT", "vendita", 80),
    ("Palermo",  "PA", "vendita", 80),
    # Infrastructure signal zones (see previous market intelligence article)
    ("Bagnoli",  "NA", "vendita", 50),   # Naples regeneration
]


async def run_pipeline(
    output_dir: str = "output",
    send_alerts: bool = True,
) -> dict:
    """
    Complete Immobiliare.it lead generation pipeline.
    Returns summary of detected leads.
    """
    import os
    os.makedirs(output_dir, exist_ok=True)

    all_listings = []
    all_leads = []

    async with ScrapeBadger(api_key=os.environ["SCRAPEBADGER_API_KEY"]) as client:

        # Collect listings across all target zones
        for city, province, listing_type, max_results in TARGETS:
            listings = await search_listings(
                client,
                city=city,
                province=province,
                listing_type=listing_type,
                sort_by="data_inserimento",  # Newest first
                max_results=max_results,
            )
            all_listings.extend(listings)
            await asyncio.sleep(0.4)

    # Deduplicate by listing_id
    seen_ids = set()
    unique_listings = []
    for listing in all_listings:
        if listing.listing_id not in seen_ids:
            seen_ids.add(listing.listing_id)
            unique_listings.append(listing)

    print(f"\nTotal unique listings collected: {len(unique_listings)}")

    # Run lead detection
    lead_listing_ids = set()
    for listing in unique_listings:
        leads = classify_listing(listing)
        for lead in leads:
            if lead.listing_id not in lead_listing_ids:
                all_leads.append(lead)
                lead_listing_ids.add(lead.listing_id)

    # Build agency database
    agencies = extract_agency_database(unique_listings)
    agencies_with_contact = {
        k: v for k, v in agencies.items() if v.phone or v.email
    }

    # Export outputs
    from datetime import datetime
    date_str = datetime.utcnow().strftime("%Y%m%d")
    leads_path = f"{output_dir}/leads_{date_str}.csv"
    agencies_path = f"{output_dir}/agencies_{date_str}.csv"

    lead_count = export_leads_csv(all_leads, leads_path)
    agency_count = export_agency_csv(agencies, agencies_path, min_listings=1)

    # Print summary
    by_type = {}
    by_priority = {"high": 0, "medium": 0, "low": 0}
    for lead in all_leads:
        by_type[lead.lead_type] = by_type.get(lead.lead_type, 0) + 1
        by_priority[lead.priority] = by_priority.get(lead.priority, 0) + 1

    print(f"\n{'='*55}")
    print("ITALIAN REAL ESTATE LEAD PIPELINE — RESULTS")
    print(f"{'='*55}")
    print(f"Listings collected:  {len(unique_listings)}")
    print(f"Leads detected:      {lead_count}")
    print(f"  New listings:      {by_type.get('new_listing', 0)}")
    print(f"  Price reduced:     {by_type.get('price_reduced', 0)}")
    print(f"  Long on market:    {by_type.get('long_on_market', 0)}")
    print(f"\nPriority breakdown:")
    print(f"  High:   {by_priority['high']}")
    print(f"  Medium: {by_priority['medium']}")
    print(f"  Low:    {by_priority['low']}")
    print(f"\nAgencies with contact data: {agency_count}")
    print(f"\nOutputs saved:")
    print(f"  {leads_path}")
    print(f"  {agencies_path}")

    if send_alerts:
        await send_slack_digest(all_leads, agency_count)

    return {
        "listings": len(unique_listings),
        "leads": lead_count,
        "agencies": agency_count,
        "by_type": by_type,
        "by_priority": by_priority,
    }


if __name__ == "__main__":
    asyncio.run(run_pipeline())

Running it:

$ python main.py

  [Milano] vendita: 148 listings
  [Bologna] vendita: 97 listings
  [Firenze] vendita: 99 listings
  [Napoli] vendita: 101 listings
  [Roma] vendita: 149 listings
  [Catania] vendita: 78 listings
  [Palermo] vendita: 79 listings
  [Bagnoli] vendita: 44 listings

Total unique listings collected: 795

=======================================================
ITALIAN REAL ESTATE LEAD PIPELINE — RESULTS
=======================================================
Listings collected:  795
Leads detected:      312
  New listings:      104
  Price reduced:     89
  Long on market:    119

Priority breakdown:
  High:   87
  Medium: 168
  Low:    57

Agencies with contact data: 243

Outputs saved:
  output/leads_20260710.csv
  output/agencies_20260710.csv

What the Pipeline Produces

Two output files, each serving a different downstream workflow.

leads_[date].csv contains every qualified lead sorted by score. Each row includes the property context (city, type, price, days on market, price reduction), the outreach angle (a one-line description of why this listing is worth acting on), and the agency contact details. Import this directly into your CRM. The outreach angle field is ready to use as a call prep note.

agencies_[date].csv contains the deduplicated agency database: one row per agency, with their phone number, email, active listing count, average price segment, and the cities they operate in. This is the B2B outreach list — agencies that have active listings in your target markets, with contact data already extracted. Filtered to agencies with at least one contact field and at least one active listing, this is more current and more complete than any commercial agency directory because it reflects what is live on Immobiliare.it today.

Extending the Pipeline

Energy upgrade lead identification. Filter the leads output for energia_class equal to F or G, combined with surface area above 80m² and price below the zone median. These are the renovation opportunity listings — properties where a Superbonus-eligible renovation could move the asset from G class to B class and potentially increase sale value by 10–15% in the current market. As covered in the Italian property market intelligence article, the energy class premium is currently significant in Milan and growing in secondary cities.

Infrastructure zone targeting. Cross-reference the collected listings against the infrastructure development zones described in the previous article (Porta Romana in Milan, Pigneto in Rome, Bagnoli in Naples). Adding these as specific search targets and tagging the leads with their proximity to the development zone allows outreach that is specific and credible: "I noticed your listing is in the Bagnoli waterfront regeneration corridor — have you seen the impact the development is having on comparable prices in the zone?"

MCP-powered agent research. Before outreach, ScrapeBadger's MCP server exposes Google News and Google Maps alongside the Immobiliare data, enabling an AI agent to research an agency's Google Maps reviews and news mentions automatically as part of the lead enrichment step.

Free trial at scrapebadger.com/immobiliare-scraper — 1,000 credits, no credit card required.

FAQ

Does Immobiliare.it display agency contact data on all listings?

Professional agency listings on Immobiliare.it consistently include the agency name, phone number, and a link to the agency's profile page. Email addresses are present on a significant proportion of professional listings. Private individual (privato) listings show the owner's contact information rather than an agency. The pipeline above extracts whichever contact data is present per listing.

What is the legal position on collecting and using this contact data in Italy?

Agency contact information displayed publicly on a property portal for the explicit purpose of enabling buyer and tenant enquiries is business contact data. Under Italy's GDPR implementation (D.Lgs. 196/2003 as amended), B2B outreach to business contact data that is publicly displayed in a professional context for commercial purposes is generally permissible. Processing personal data of private individuals from listings requires separate legal basis. The pipeline's agency database focuses specifically on professional agency contacts — not private individual seller data.

How often should this pipeline run?

For new listing detection, daily is the appropriate cadence — listings posted 48 hours ago are still early enough for relevant outreach. For price reduction and long-on-market detection, twice weekly is sufficient — these signals develop over days or weeks rather than hours.

Can the pipeline target rental listings alongside sales?

Yes. The listing_type parameter in search_listings() accepts "affitto" (rental) as well as "vendita" (sale). Property management companies and letting agencies are a different audience from sales agents but the same lead generation pipeline serves both with the appropriate search configuration.