src/rag/searcher.py

import sqlite3
import pandas as pd
from datetime import date, timedelta
from src.database.queries import get_conn


def search_properties(
    min_price: float = None,
    max_price: float = None,
    bhk: int = None,
    property_type: str = None,
    location: str = None,
    furnishing: str = None,
    min_area: int = None,
    parking: bool = None,
) -> list[dict]:
    conn = get_conn()
    query = "SELECT * FROM properties WHERE available = 'Yes'"
    params = []

    if min_price is not None:
        query += " AND price_inr >= ?"
        params.append(min_price)
    if max_price is not None:
        query += " AND price_inr <= ?"
        params.append(max_price)
    if bhk is not None:
        query += " AND bhk = ?"
        params.append(bhk)
    if property_type:
        query += " AND LOWER(property_type) = ?"
        params.append(property_type.lower())
    if location:
        query += " AND LOWER(location) LIKE ?"
        params.append(f"%{location.lower()}%")
    if furnishing:
        query += " AND LOWER(furnishing) = ?"
        params.append(furnishing.lower())
    if min_area is not None:
        query += " AND area_sqft >= ?"
        params.append(min_area)
    if parking:
        query += " AND parking >= 1"

    query += " ORDER BY price_inr ASC LIMIT 10"
    df = pd.read_sql_query(query, conn, params=params)
    conn.close()
    return df.to_dict(orient="records")


def get_leases_expiring(days: int = 30) -> list[dict]:
    conn = get_conn()
    today = date.today().isoformat()
    future = (date.today() + timedelta(days=days)).isoformat()
    query = """
        SELECT p.property_id, p.title, p.bhk, p.location, p.price_inr, p.registration_no,
               l.lease_status, l.lease_end, l.monthly_rent,
               l.tenant_name, l.followup_person, l.followup_contact, l.notes
        FROM properties p
        JOIN leases l ON p.property_id = l.property_id
        WHERE l.lease_end BETWEEN ? AND ?
        ORDER BY l.lease_end ASC
    """
    df = pd.read_sql_query(query, conn, params=[today, future])
    conn.close()
    return df.to_dict(orient="records")


def get_leases_vacant_or_pending() -> list[dict]:
    conn = get_conn()
    query = """
        SELECT p.property_id, p.title, p.bhk, p.location, p.price_inr, p.registration_no,
               l.lease_status, l.lease_end, l.monthly_rent,
               l.tenant_name, l.followup_person, l.followup_contact, l.notes
        FROM properties p
        JOIN leases l ON p.property_id = l.property_id
        WHERE l.lease_status IN ('PENDING', 'EXPIRED')
        ORDER BY l.lease_end ASC
    """
    df = pd.read_sql_query(query, conn)
    conn.close()
    return df.to_dict(orient="records")


def get_all_properties() -> list[dict]:
    conn = get_conn()
    df = pd.read_sql_query("SELECT * FROM properties ORDER BY price_inr ASC", conn)
    conn.close()
    return df.to_dict(orient="records")