Refactor scan_properties.py to simplify script structure and remove type annotations

- Remove main() function and move code directly under `__main__` block
- Simplify imports and remove unused type hints
- Update DataFrame iteration method to use `iter_rows(named=True)`
- Streamline variable assignments and message sending logic

Author: Adriano Sanges

real-estate-etl/scan_properties.py

@@ -1,46 +1,44 @@
-from typing import List, Dict, Any
 from bs4 import BeautifulSoup
-import polars as pl
+import polars
 import duckdb
 from scraper import parse_listing
 from database import clean_properties, get_new_properties
 from dotenv import load_dotenv
 import os
 from telegram_api import send_message, format_property_message
 
-def main() -> None:
-    """Main function to scrape properties and send messages."""
-    load_dotenv()
 
-    url: str = os.getenv("scrape_url")
-    warehouse_name: str = os.getenv("warehouse_name")
-    motherduck_token: str = os.getenv("motherduck_token")
 
-    if not url or not warehouse_name or not motherduck_token:
-        raise ValueError("Environment variables for URL, warehouse name, or token are not set.")
+if __name__ == "__main__":
+
+
+    load_dotenv()
 
-    data: List[Dict[str, Any]] = parse_listing(url)
+    url = os.getenv("scrape_url")
+    warehouse_name = os.getenv("warehouse_name")
+    motherduck_token = os.getenv("motherduck_token")
 
-    polars_df: pl.DataFrame = pl.DataFrame(data)
+    data = parse_listing(url)
+
+    polars_df = polars.DataFrame(data)
 
     con = duckdb.connect(f"md:{warehouse_name}?motherduck_token={motherduck_token}")
 
     con.sql("create table if not exists main.properties as select * from polars_df")
 
     clean_properties(con)
 
-    new_properties: pl.DataFrame = get_new_properties(con)
-
-    # Format and send messages
-    messages: List[str] = [format_property_message(row) for row in new_properties.to_dicts()]
+    new_properties = get_new_properties(con)
+    # Iterate over the DataFrame and format each property
+    messages = [format_property_message(row) for row in new_properties.iter_rows(named=True)]
 
-    # Send messages in chunks of two
+# Send messages in chunks of two
     for i in range(0, len(messages), 2):
-        message_chunk: List[str] = messages[i:i+2]
-        full_message: str = "\n\n".join(message_chunk)
+        # Get the current chunk of two messages
+        message_chunk = messages[i:i+2]
+        # Join the two messages with a separator
+        full_message = "\n\n".join(message_chunk)
+        # Send the combined message
         send_message(full_message)
 
-if __name__ == "__main__":
-    main()
-