"""
═══════════════════════════════════════════════════════════════
VillasInUAE.com — Telegram Property Monitor
Monitors 200+ UAE property Telegram channels
Extracts listings autonomously, feeds into Supabase
LEGAL: Uses official Telegram Bot API + Telethon MTProto
Official API = explicitly permitted by Telegram ToS
Only monitors PUBLIC channels (zero privacy concerns)
INSTALL:
pip install telethon supabase anthropic python-dotenv aiohttp
ENV VARS:
TELEGRAM_API_ID = from my.telegram.org (free)
TELEGRAM_API_HASH = from my.telegram.org (free)
TELEGRAM_PHONE = your registered phone number
TELEGRAM_SESSION = villasinuae_session
SUPABASE_URL = your supabase project URL
SUPABASE_SERVICE_KEY = <redacted> service role key
ANTHROPIC_API_KEY = for SIRAJ listing extraction
NOTIFICATION_WEBHOOK = webhook URL for new listing alerts
═══════════════════════════════════════════════════════════════
"""
import asyncio
import re
import json
import os
import logging
import hashlib
from datetime import datetime, timezone
from typing import Optional
from dotenv import load_dotenv
from telethon import TelegramClient, events
from telethon.tl.types import Channel, Chat, MessageMediaPhoto, MessageMediaDocument
import aiohttp
from supabase import create_client, Client
load_dotenv()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("telegram-monitor")
─────────────────────────────────────────────────────────────
SUPABASE_URL = os.environ["SUPABASE_URL"]
SUPABASE_KEY = <redacted>["SUPABASE_SERVICE_KEY"]
ANTHROPIC_KEY = <redacted>["ANTHROPIC_API_KEY"]
TG_API_ID = int(os.environ["TELEGRAM_API_ID"])
TG_API_HASH = os.environ["TELEGRAM_API_HASH"]
TG_PHONE = os.environ.get("TELEGRAM_PHONE")
TG_SESSION = os.environ.get("TELEGRAM_SESSION", "villasinuae_session")
supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY)
─────────────────────────────────────────────────────────────
UAE_PROPERTY_CHANNELS = [
# English-language Dubai channels
"dubaipropertynews",
"dubairealestatemarket",
"dubai_villas_official",
"uae_off_plan_projects",
"dubai_property_deals",
"uaerealestateinvestors",
"dubaipropertyalerts",
"luxurydubaiproperties",
"dubaihomesofficial",
"offplan_dubai",
"dubailistings",
"dubaipropertymarket",
"dubairealestategroup",
"dubaimansionsofficial",
"uaeproperty",
"dubaiestates",
"emiratesproperty",
"dubaihousingmarket",
"dubaipropertyexpert",
"investindubai",
Arabic-language channels (most active, least aggregated)
Abu Dhabi channels
RAK / Northern Emirates
Developer-specific channels
Investment & broker channels
─────────────────────────────────────────────────────────────
PROPERTY_KEYWORDS_EN = [
"villa", "townhouse", "bedroom", "br", "beds", "bhk",
"aed", "dirhams", "sqft", "sq ft", "sqm",
"for sale", "for rent", "off plan", "offplan", "ready",
"palm jumeirah", "dubai hills", "arabian ranches", "damac hills",
"sobha", "emaar", "nakheel", "tilal al ghaf", "meydan",
"handover", "payment plan", "golden visa", "freehold",
]
PROPERTY_KEYWORDS_AR = [
"فيلا", "تاون هاوس", "غرفة", "نوم", "درهم",
"للبيع", "للإيجار", "على الخريطة", "جاهز",
"بالم جميرا", "دبي هيلز", "الرانشز", "سوبها", "إعمار",
"تسليم", "خطة دفع",
]
def is_property_listing(text: str) -> bool:
if not text:
return False
text_lower = text.lower()
en_matches = sum(1 for kw in PROPERTY_KEYWORDS_EN if kw in text_lower)
ar_matches = sum(1 for kw in PROPERTY_KEYWORDS_AR if kw in text)
return en_matches >= 2 or ar_matches >= 2 or (en_matches >= 1 and ar_matches >= 1)
─────────────────────────────────────────────────────────────
EXTRACTION_PROMPT = """You are a UAE real estate data extraction engine.
Extract structured property listing data from this Telegram message.
Message may be in Arabic, English, Hindi, Urdu, Russian, or mixed.
Return ONLY valid JSON with these exact fields (use null if not found):
{
"property_type": "villa|townhouse|apartment|penthouse|null",
"bedrooms": number_or_null,
"bathrooms": number_or_null,
"area_sqft": number_or_null,
"price_aed": number_or_null (convert if in millions: 5M = 5000000),
"price_type": "sale|rent|null",
"rent_period": "yearly|monthly|null",
"community": "community name or null",
"sub_community": "sub-community name or null",
"emirate": "Dubai|Abu Dhabi|Sharjah|RAK|Ajman|Fujairah|null",
"developer": "developer name or null",
"listing_type": "off_plan|ready|null",
"contact_phone": "phone number or null",
"contact_whatsapp": "whatsapp number or null",
"payment_plan": "payment plan details or null",
"handover_date": "handover date or null",
"amenities": ["list", "of", "amenities"] or [],
"description": "clean 2-3 sentence description in English",
"original_language": "en|ar|hi|ur|ru|zh|mixed",
"confidence": "high|medium|low"
}
If this is NOT a property listing, return: {"not_listing": true}
MESSAGE:
"""
async def extract_listing_with_ai(text: str) -> Optional[dict]:
"""Use SIRAJ/Claude to extract structured data from Telegram message"""
try:
async with aiohttp.ClientSession() as session:
payload = {
"model": "claude-haiku-4-5-20251001",
"max_tokens": 600,
"messages": [{"role": "user", "content": EXTRACTION_PROMPT + text[:2000]}]
}
headers = {
"x-api-key": ANTHROPIC_KEY,
"anthropic-version": "2023-06-01",
"content-type": "application/json"
}
async with session.post(
"https://api.anthropic.com/v1/messages",
json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=15)
) as resp:
if resp.status != 200:
return None
data = await resp.json()
raw = data["content"][0]["text"].strip()
Strip markdown fences if present
parsed = json.loads(raw)
if parsed.get("not_listing"):
return None
if parsed.get("confidence") == "low":
return None
return parsed
except (json.JSONDecodeError, KeyError, aiohttp.ClientError) as e:
log.warning(f"AI extraction failed: {e}")
return None
─────────────────────────────────────────────────────────────
def make_fingerprint(listing: dict) -> str:
key_fields = [
str(listing.get("price_aed", "")),
str(listing.get("bedrooms", "")),
str(listing.get("community", "")),
str(listing.get("area_sqft", "")),
str(listing.get("contact_phone", "")),
]
fingerprint = "|".join(key_fields).lower().strip()
return hashlib.sha256(fingerprint.encode()).hexdigest()[:16]
def is_duplicate(fingerprint: str) -> bool:
try:
result = supabase.table("listings").select("id").eq(
"source_fingerprint", fingerprint
).limit(1).execute()
return len(result.data) > 0
except Exception:
return False
─────────────────────────────────────────────────────────────
async def save_listing(listing: dict, message, channel_name: str) -> Optional[str]:
fingerprint = make_fingerprint(listing)
if is_duplicate(fingerprint):
log.debug(f"Duplicate listing skipped: {fingerprint}")
return None
record = {
# Core fields
"source": "telegram",
"source_channel": channel_name,
"source_message_id": str(message.id),
"source_fingerprint": fingerprint,
Property data
Contact
Details
Platform flags
Handle photos from Telegram message
try:
result = supabase.table("listings").insert(record).execute()
listing_id = result.data[0]["id"] if result.data else None
log.info(f"✅ Saved: {listing.get('bedrooms')}BR {listing.get('property_type')} "
f"in {listing.get('community')} at AED {listing.get('price_aed'):,} "
f"from @{channel_name}")
Trigger Smart Alert matching
return listing_id
except Exception as e:
log.error(f"Supabase save failed: {e}")
return None
─────────────────────────────────────────────────────────────
async def trigger_smart_alerts(listing_id: str, listing: dict):
"""Find users with matching Smart Alerts and queue notifications"""
try:
# Get all active alerts that could match this listing
alerts_result = supabase.table("smart_alerts").select("*").eq(
"status", "active"
).execute()
matching_alerts = []
for alert in (alerts_result.data or []):
if does_listing_match_alert(listing, alert):
matching_alerts.append(alert)
if matching_alerts:
# Insert notification jobs for async processing
jobs = [{
"alert_id": a["id"],
"listing_id": listing_id,
"user_id": a["user_id"],
"channel": "email", # Will also send WhatsApp if enabled
"status": "pending",
"created_at": datetime.now(timezone.utc).isoformat(),
} for a in matching_alerts]
supabase.table("notification_queue").insert(jobs).execute()
log.info(f"Queued {len(jobs)} alert notifications for listing {listing_id}")
except Exception as e:
log.warning(f"Smart alert trigger failed: {e}")
def does_listing_match_alert(listing: dict, alert: dict) -> bool:
"""Check if listing matches an alert's criteria"""
# Price check
if alert.get("min_price") and listing.get("price_aed"):
if listing["price_aed"] < alert["min_price"]:
return False
if alert.get("max_price") and listing.get("price_aed"):
if listing["price_aed"] > alert["max_price"]:
return False
Bedrooms check
Community check (flexible — 'Dubai Hills' matches 'Dubai Hills Estate')
Property type check
Listing type check (off-plan vs ready)
return True
─────────────────────────────────────────────────────────────
async def notify_broker_of_listing(listing: dict, channel_name: str):
"""
When we find a listing from a Telegram channel, notify the broker
that their listing is now on VillasInUAE — invite them to claim their profile.
This is our primary organic broker acquisition channel.
"""
contact = listing.get("contact_whatsapp") or listing.get("contact_phone")
if not contact:
return
Clean phone number
Queue WhatsApp notification
try:
supabase.table("whatsapp_outreach_queue").insert({
"phone": phone,
"message": message,
"context": "broker_listing_claim",
"listing_community": listing.get("community"),
"status": "pending",
"created_at": datetime.now(timezone.utc).isoformat(),
}).execute()
except Exception as e:
log.warning(f"Broker outreach queue failed: {e}")
─────────────────────────────────────────────────────────────
async def discover_new_channels(client: TelegramClient) -> list:
"""
Search Telegram for new UAE property channels.
Called weekly to expand channel coverage automatically.
"""
search_terms = [
"dubai villa", "uae property", "off plan dubai",
"عقارات دبي", "فيلا دبي", "dubai real estate",
"ras al khaimah property", "abu dhabi villa",
"damac hills", "arabian ranches villa", "sobha hartland",
]
discovered = []
for term in search_terms:
try:
results = await client.get_participants(term)
for result in results[:5]:
if hasattr(result, "username") and result.username:
discovered.append(result.username)
except Exception:
pass
Store discovered channels for review
return discovered
async def get_monitored_channels(client: TelegramClient) -> list:
"""
Combine hardcoded channels + database-approved channels.
Allows adding new channels via admin dashboard without code changes.
"""
channels = list(UAE_PROPERTY_CHANNELS)
try:
db_channels = supabase.table("telegram_channels").select("username").eq(
"status", "active"
).execute()
db_usernames = [c["username"] for c in (db_channels.data or [])]
channels = list(set(channels + db_usernames))
except Exception as e:
log.warning(f"Could not fetch DB channels: {e}")
return channels
─────────────────────────────────────────────────────────────
class Stats:
def __init__(self):
self.messages_processed = 0
self.listings_found = 0
self.listings_saved = 0
self.duplicates_skipped = 0
self.channels_active = 0
self.started_at = datetime.now()
def log_summary(self):
runtime = (datetime.now() - self.started_at).total_seconds()
log.info(
f"📊 Stats: {self.messages_processed} messages | "
f"{self.listings_found} listings found | "
f"{self.listings_saved} saved | "
f"{self.duplicates_skipped} duplicates | "
f"{self.channels_active} channels | "
f"{runtime:.0f}s runtime"
)
stats = Stats()
─────────────────────────────────────────────────────────────
async def backfill_channel(client: TelegramClient, channel_username: str, days_back: int = 7):
"""Backfill historical listings from a channel on first run"""
try:
entity = await client.get_entity(channel_username)
from datetime import timedelta
since = datetime.now(timezone.utc) - timedelta(days=days_back)
count = 0
async for message in client.iter_messages(entity, offset_date=since, reverse=True, limit=500):
if not message.text:
continue
if not is_property_listing(message.text):
continue
listing = await extract_listing_with_ai(message.text)
if listing:
await save_listing(listing, message, channel_username)
count += 1
await asyncio.sleep(0.5) # Gentle rate limiting
log.info(f"Backfilled {count} listings from @{channel_username}")
except Exception as e:
log.warning(f"Backfill failed for @{channel_username}: {e}")
─────────────────────────────────────────────────────────────
async def main():
log.info("🚀 Starting VillasInUAE Telegram Monitor")
client = TelegramClient(TG_SESSION, TG_API_ID, TG_API_HASH)
await client.start(phone=TG_PHONE)
channels = await get_monitored_channels(client)
stats.channels_active = len(channels)
log.info(f"Monitoring {len(channels)} channels")
Register real-time event handler for all monitored channels
stats.messages_processed += 1
Quick keyword filter before expensive AI call
stats.listings_found += 1
Get channel name
AI extraction
Save to database
Log stats every 100 messages
Run backfill in background on startup
Weekly channel discovery
asyncio.create_task(run_backfill())
asyncio.create_task(run_channel_discovery())
log.info("✅ Real-time monitoring active")
await client.run_until_disconnected()
if __name__ == "__main__":
asyncio.run(main())