""" Database layer — SQLite with all tables: properties, insurance, documents, leads, query_logs, admin_logs, config """ import sqlite3, json from pathlib import Path from datetime import date, timedelta from src.config.settings import DB_PATH, DATA_DIR def get_conn(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") return conn def init_db(): DB_PATH.parent.mkdir(parents=True, exist_ok=True) conn = get_conn() conn.executescript(""" CREATE TABLE IF NOT EXISTS properties ( id INTEGER PRIMARY KEY, title TEXT, bhk INTEGER, type TEXT, location TEXT, city TEXT, price_cr REAL, price_display TEXT, area_sqft INTEGER, price_per_sqft INTEGER, floor INTEGER, total_floors INTEGER, furnishing TEXT, amenities TEXT, parking INTEGER, pool INTEGER, age_years INTEGER, status TEXT, builder TEXT, society TEXT, bedrooms INTEGER, bathrooms INTEGER, balconies INTEGER, facing TEXT, contact TEXT, description TEXT, available INTEGER DEFAULT 1, featured INTEGER DEFAULT 0, listed_days_ago INTEGER DEFAULT 0, created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS insurance ( id INTEGER PRIMARY KEY AUTOINCREMENT, property_id INTEGER, company TEXT, policy_no TEXT, status TEXT, start_date TEXT, expiry_date TEXT, premium_annual INTEGER, followup_person TEXT, followup_contact TEXT, notes TEXT, FOREIGN KEY(property_id) REFERENCES properties(id) ); CREATE TABLE IF NOT EXISTS documents ( id INTEGER PRIMARY KEY AUTOINCREMENT, property_id INTEGER, document_type TEXT, status TEXT, received_date TEXT, notes TEXT, FOREIGN KEY(property_id) REFERENCES properties(id) ); CREATE TABLE IF NOT EXISTS leads ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, phone TEXT, email TEXT, property_id INTEGER, property_title TEXT, inquiry_type TEXT, message TEXT, visit_date TEXT, time_slot TEXT, budget TEXT, status TEXT DEFAULT 'new', notes TEXT DEFAULT '', whatsapp_sent INTEGER DEFAULT 0, created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS query_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, query TEXT, intent TEXT, answer_preview TEXT, confidence REAL, timestamp TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS admin_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, action TEXT, details TEXT, timestamp TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS config ( key TEXT PRIMARY KEY, value TEXT ); """) conn.commit() # Load properties from JSON if table empty if conn.execute("SELECT COUNT(*) FROM properties").fetchone()[0] == 0: props_path = DATA_DIR / "properties.json" if props_path.exists(): props = json.loads(props_path.read_text()) for p in props: conn.execute(""" INSERT OR REPLACE INTO properties (id,title,bhk,type,location,city,price_cr,price_display,area_sqft, price_per_sqft,floor,total_floors,furnishing,amenities,parking,pool, age_years,status,builder,society,bedrooms,bathrooms,balconies, facing,contact,description,available,featured,listed_days_ago) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( p["id"],p["title"],p["bhk"],p["type"],p["location"],p["city"], p["price_cr"],p["price_display"],p["area_sqft"],p["price_per_sqft"], p["floor"],p["total_floors"],p["furnishing"], json.dumps(p["amenities"]),int(p["parking"]),int(p["pool"]), p["age_years"],p["status"],p["builder"],p["society"], p["bedrooms"],p["bathrooms"],p["balconies"],p["facing"], p["contact"],p["description"],int(p["available"]),int(p.get("featured",0)), p["listed_days_ago"] )) conn.commit() print(f"Loaded {len(props)} properties into DB") # Load insurance + documents from generate script data ins_path = DATA_DIR / "properties.json" # re-derive from same seed _seed_insurance_docs(conn) conn.close() print(f"DB ready at {DB_PATH}") def _seed_insurance_docs(conn): import random from datetime import date, timedelta random.seed(42) INS_COMPANIES = ["New India Assurance","HDFC ERGO","Bajaj Allianz","ICICI Lombard","National Insurance"] INS_STATUSES = ["ACTIVE","ACTIVE","ACTIVE","PENDING","EXPIRED"] DOCUMENT_TYPES= ["Sale Agreement","NOC from Society","Occupation Certificate", "Property Card","Index II","Stamp Duty Receipt","Possession Letter", "Title Search Report","Encumbrance Certificate","Building Plan Approval"] DOC_STATUSES = ["RECEIVED","PENDING","RECEIVED","RECEIVED","MISSING"] PERSONS = ["Rajan Sharma","Priya Ghosh","Amit Das","Sneha Patil","Rahul Mehta"] props = conn.execute("SELECT id FROM properties").fetchall() for row in props: pid = row["id"] ins_status = random.choice(INS_STATUSES) start = date.today() - timedelta(days=random.randint(10,350)) expiry = start + timedelta(days=365) conn.execute("""INSERT INTO insurance (property_id,company,policy_no,status,start_date,expiry_date, premium_annual,followup_person,followup_contact,notes) VALUES (?,?,?,?,?,?,?,?,?,?)""", (pid, random.choice(INS_COMPANIES), f"POL-{pid}-{random.randint(10000,99999)}", ins_status, start.isoformat(), expiry.isoformat(), random.randint(8000,45000), random.choice(PERSONS), f"+91-9{random.randint(100000000,999999999)}", random.choice(["Renewal reminder sent","Follow up","Paid","","On hold"]))) doc_types = random.sample(DOCUMENT_TYPES, k=random.randint(2,5)) for dt in doc_types: recv = (date.today()-timedelta(days=random.randint(0,180))).isoformat() if random.random()>0.3 else None conn.execute("INSERT INTO documents (property_id,document_type,status,received_date,notes) VALUES (?,?,?,?,?)", (pid, dt, random.choice(DOC_STATUSES), recv, "")) conn.commit() # ── Property queries ────────────────────────────────────────────────────────── def search_properties(min_price=None, max_price=None, bhk=None, prop_type=None, location=None, city=None, furnishing=None, status=None, parking=None, pool=None, min_area=None, max_area=None, builder=None, featured_only=False, limit=20) -> list[dict]: conn = get_conn() q = "SELECT * FROM properties WHERE available=1" params = [] if min_price: q += " AND price_cr>=?"; params.append(min_price) if max_price: q += " AND price_cr<=?"; params.append(max_price) if bhk: q += " AND bhk=?"; params.append(int(bhk)) if prop_type: q += " AND LOWER(type)=?"; params.append(prop_type.lower()) if location: q += " AND LOWER(location) LIKE ?"; params.append(f"%{location.lower()}%") if city: q += " AND LOWER(city)=?"; params.append(city.lower()) if furnishing: q += " AND LOWER(furnishing) LIKE ?"; params.append(f"%{furnishing.lower()}%") if status: q += " AND LOWER(status) LIKE ?"; params.append(f"%{status.lower()}%") if parking: q += " AND parking=1" if pool: q += " AND pool=1" if min_area: q += " AND area_sqft>=?"; params.append(min_area) if max_area: q += " AND area_sqft<=?"; params.append(max_area) if builder: q += " AND LOWER(builder) LIKE ?"; params.append(f"%{builder.lower()}%") if featured_only: q += " AND featured=1" q += " ORDER BY featured DESC, price_cr ASC LIMIT ?" params.append(limit) rows = conn.execute(q, params).fetchall() conn.close() return [_prop_row(r) for r in rows] def get_property(pid: int) -> dict | None: conn = get_conn() row = conn.execute("SELECT * FROM properties WHERE id=?", (pid,)).fetchone() if not row: conn.close(); return None p = _prop_row(row) ins = conn.execute("SELECT * FROM insurance WHERE property_id=?", (pid,)).fetchone() docs= conn.execute("SELECT * FROM documents WHERE property_id=?", (pid,)).fetchall() if ins: p["insurance"] = dict(ins) p["documents"] = [dict(d) for d in docs] conn.close() return p def get_all_properties(available_only=False) -> list[dict]: conn = get_conn() q = "SELECT * FROM properties" if available_only: q += " WHERE available=1" q += " ORDER BY featured DESC, price_cr ASC" rows = conn.execute(q).fetchall() conn.close() return [_prop_row(r) for r in rows] def add_property(data: dict) -> int: conn = get_conn() cur = conn.execute("""INSERT INTO properties (title,bhk,type,location,city,price_cr,price_display,area_sqft,price_per_sqft, floor,total_floors,furnishing,amenities,parking,pool,age_years,status,builder, society,bedrooms,bathrooms,balconies,facing,contact,description,available,featured) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (data["title"],data.get("bhk",2),data.get("type","Apartment"), data["location"],data.get("city","Mumbai"), data["price_cr"],f"₹{data['price_cr']} Cr", data.get("area_sqft",800), int(data["price_cr"]*100/(data.get("area_sqft",800)/100)), data.get("floor",1),data.get("total_floors",10), data.get("furnishing","Unfurnished"), json.dumps(data.get("amenities",[])), int(data.get("parking",False)),int(data.get("pool",False)), data.get("age_years",0),data.get("status","Ready to Move"), data.get("builder",""),data.get("society",""), data.get("bhk",2),data.get("bhk",2),data.get("balconies",1), data.get("facing","East"),data.get("contact",""), data.get("description",""),1,int(data.get("featured",False)))) new_id = cur.lastrowid conn.commit(); conn.close() _log("add_property", f"Added {data['title']} (ID {new_id})") return new_id def update_property(pid: int, updates: dict): allowed = ["title","price_cr","price_display","area_sqft","furnishing","status", "description","featured","available","floor","total_floors","builder","society","contact"] updates = {k:v for k,v in updates.items() if k in allowed} if not updates: return if "price_cr" in updates: updates["price_display"] = f"₹{updates['price_cr']} Cr" conn = get_conn() conn.execute(f"UPDATE properties SET {','.join(f'{k}=?' for k in updates)} WHERE id=?", [*updates.values(), pid]) conn.commit(); conn.close() _log("update_property", f"Updated {pid}: {list(updates.keys())}") def delete_property(pid: int): conn = get_conn() conn.execute("DELETE FROM properties WHERE id=?", (pid,)) conn.execute("DELETE FROM insurance WHERE property_id=?", (pid,)) conn.execute("DELETE FROM documents WHERE property_id=?", (pid,)) conn.commit(); conn.close() _log("delete_property", f"Deleted property {pid}") def _prop_row(row) -> dict: d = dict(row) try: d["amenities"] = json.loads(d.get("amenities","[]")) except: d["amenities"] = [] d["parking"] = bool(d.get("parking",0)) d["pool"] = bool(d.get("pool",0)) d["available"]= bool(d.get("available",1)) d["featured"] = bool(d.get("featured",0)) return d # ── Insurance ──────────────────────────────────────────────────────────────── def get_insurance_expiring(days=30) -> list[dict]: conn = get_conn() today = date.today().isoformat() future = (date.today()+timedelta(days=days)).isoformat() rows = conn.execute(""" SELECT p.id,p.title,p.location,p.price_display, i.company,i.policy_no,i.status,i.expiry_date, i.premium_annual,i.followup_person,i.followup_contact,i.notes FROM insurance i JOIN properties p ON i.property_id=p.id WHERE i.expiry_date BETWEEN ? AND ? ORDER BY i.expiry_date ASC""", (today,future)).fetchall() conn.close() return [dict(r) for r in rows] def get_insurance_pending() -> list[dict]: conn = get_conn() rows = conn.execute(""" SELECT p.id,p.title,p.location,p.price_display, i.company,i.policy_no,i.status,i.expiry_date, i.premium_annual,i.followup_person,i.followup_contact,i.notes FROM insurance i JOIN properties p ON i.property_id=p.id WHERE i.status IN ('PENDING','EXPIRED') ORDER BY i.expiry_date ASC""").fetchall() conn.close() return [dict(r) for r in rows] def update_insurance(property_id:int, updates:dict): conn = get_conn() allowed = ["status","expiry_date","notes","followup_person","followup_contact","premium_annual"] updates = {k:v for k,v in updates.items() if k in allowed} conn.execute(f"UPDATE insurance SET {','.join(f'{k}=?' for k in updates)} WHERE property_id=?", [*updates.values(), property_id]) conn.commit(); conn.close() # ── Documents ──────────────────────────────────────────────────────────────── def get_documents_missing() -> list[dict]: conn = get_conn() rows = conn.execute(""" SELECT p.id,p.title,p.location,d.document_type,d.status,d.received_date,d.notes FROM documents d JOIN properties p ON d.property_id=p.id WHERE d.status IN ('MISSING','PENDING') ORDER BY p.id ASC""").fetchall() conn.close() return [dict(r) for r in rows] def update_document(doc_id:int, updates:dict): conn = get_conn() allowed = ["status","received_date","notes"] updates = {k:v for k,v in updates.items() if k in allowed} conn.execute(f"UPDATE documents SET {','.join(f'{k}=?' for k in updates)} WHERE id=?", [*updates.values(), doc_id]) conn.commit(); conn.close() # ── Leads ──────────────────────────────────────────────────────────────────── def save_lead(data:dict) -> int: conn = get_conn() cur = conn.execute("""INSERT INTO leads (name,phone,email,property_id,property_title,inquiry_type, message,visit_date,time_slot,budget,status,notes,whatsapp_sent) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""", (data.get("name",""),data.get("phone",""),data.get("email",""), data.get("property_id"),data.get("property_title",""), data.get("inquiry_type","general"),data.get("message",""), data.get("visit_date",""),data.get("time_slot",""),data.get("budget",""), "new","",0)) lid = cur.lastrowid conn.commit(); conn.close() return lid def get_leads(status=None, limit=200) -> list[dict]: conn = get_conn() q = "SELECT * FROM leads" params = [] if status: q += " WHERE status=?"; params.append(status) q += " ORDER BY created_at DESC LIMIT ?"; params.append(limit) rows = conn.execute(q, params).fetchall() conn.close() return [dict(r) for r in rows] def update_lead(lid:int, updates:dict): conn = get_conn() allowed = ["status","notes","whatsapp_sent"] updates = {k:v for k,v in updates.items() if k in allowed} conn.execute(f"UPDATE leads SET {','.join(f'{k}=?' for k in updates)} WHERE id=?", [*updates.values(), lid]) conn.commit(); conn.close() # ── Analytics ──────────────────────────────────────────────────────────────── def get_analytics() -> dict: conn = get_conn() total_props = conn.execute("SELECT COUNT(*) FROM properties").fetchone()[0] avail_props = conn.execute("SELECT COUNT(*) FROM properties WHERE available=1").fetchone()[0] total_leads = conn.execute("SELECT COUNT(*) FROM leads").fetchone()[0] new_leads = conn.execute("SELECT COUNT(*) FROM leads WHERE status='new'").fetchone()[0] leads_today = conn.execute("SELECT COUNT(*) FROM leads WHERE date(created_at)=date('now')").fetchone()[0] queries_today = conn.execute("SELECT COUNT(*) FROM query_logs WHERE date(timestamp)=date('now')").fetchone()[0] avg_conf = conn.execute("SELECT AVG(confidence) FROM query_logs WHERE confidence>0").fetchone()[0] or 0 ins_expiring = conn.execute("SELECT COUNT(*) FROM insurance WHERE expiry_date<=date('now','+30 days') AND status='ACTIVE'").fetchone()[0] docs_missing = conn.execute("SELECT COUNT(*) FROM documents WHERE status IN ('MISSING','PENDING')").fetchone()[0] avg_price = conn.execute("SELECT AVG(price_cr) FROM properties WHERE available=1").fetchone()[0] or 0 by_bhk = conn.execute("SELECT bhk,COUNT(*) FROM properties WHERE available=1 GROUP BY bhk ORDER BY bhk").fetchall() by_location = conn.execute("SELECT location,COUNT(*) FROM properties WHERE available=1 GROUP BY location ORDER BY COUNT(*) DESC LIMIT 8").fetchall() by_status = conn.execute("SELECT status,COUNT(*) FROM properties WHERE available=1 GROUP BY status").fetchall() intent_dist = conn.execute("SELECT intent,COUNT(*) FROM query_logs GROUP BY intent ORDER BY COUNT(*) DESC").fetchall() lead_trend = conn.execute("SELECT date(created_at) as d,COUNT(*) as c FROM leads GROUP BY d ORDER BY d DESC LIMIT 14").fetchall() recent_leads = conn.execute("SELECT * FROM leads ORDER BY created_at DESC LIMIT 10").fetchall() conn.close() return { "total_properties": total_props, "available_properties": avail_props, "total_leads": total_leads, "new_leads": new_leads, "leads_today": leads_today, "queries_today": queries_today, "avg_confidence": round(avg_conf*100, 1), "insurance_expiring_30d": ins_expiring, "documents_missing": docs_missing, "avg_price_cr": round(avg_price, 2), "by_bhk": [{"bhk":r[0],"count":r[1]} for r in by_bhk], "by_location": [{"location":r[0],"count":r[1]} for r in by_location], "by_status": [{"status":r[0],"count":r[1]} for r in by_status], "intent_distribution": [{"intent":r[0],"count":r[1]} for r in intent_dist], "lead_trend": [{"date":r[0],"count":r[1]} for r in lead_trend], "recent_leads": [dict(r) for r in recent_leads], } # ── Helpers ────────────────────────────────────────────────────────────────── def log_query(session_id:str, query:str, intent:str, answer:str, confidence:float): try: conn = get_conn() conn.execute("INSERT INTO query_logs (session_id,query,intent,answer_preview,confidence) VALUES (?,?,?,?,?)", (session_id,query,intent,answer[:200],confidence)) conn.commit(); conn.close() except: pass def _log(action:str, details:str): try: conn = get_conn() conn.execute("INSERT INTO admin_logs (action,details) VALUES (?,?)", (action,details)) conn.commit(); conn.close() except: pass