db.py

"""Database layer for Knowledge Engine using SQLite + FTS5."""

import sqlite3
import os
import json
from datetime import datetime

DB_PATH = os.path.join(os.path.dirname(__file__), "knowledge.db")


def get_conn():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA foreign_keys=ON")
    return conn


def init_db():
    conn = get_conn()
    c = conn.cursor()

    # Sources table — where data comes from
    c.execute("""
        CREATE TABLE IF NOT EXISTS sources (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            type TEXT NOT NULL,           -- rss, web, api, manual
            name TEXT NOT NULL,
            url TEXT,
            config TEXT DEFAULT '{}',     -- JSON config for scraper
            enabled INTEGER DEFAULT 1,
            last_scraped TEXT,
            created_at TEXT DEFAULT (datetime('now'))
        )
    """)

    # Documents table — individual pieces of knowledge
    c.execute("""
        CREATE TABLE IF NOT EXISTS documents (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            source_id INTEGER REFERENCES sources(id),
            title TEXT,
            content TEXT NOT NULL,
            url TEXT,
            author TEXT,
            keywords TEXT DEFAULT '[]',   -- JSON array of extracted keywords
            relevance_score REAL DEFAULT 0.5,  -- 0.0 to 1.0, adjusted by feedback
            feedback_count INTEGER DEFAULT 0,
            positive_feedback INTEGER DEFAULT 0,
            hash TEXT UNIQUE,             -- deduplicate
            created_at TEXT DEFAULT (datetime('now')),
            updated_at TEXT DEFAULT (datetime('now'))
        )
    """)

    # FTS5 virtual table for full-text search with BM25
    c.execute("""
        CREATE VIRTUAL TABLE IF NOT EXISTS documents_fts USING fts5(
            title,
            content,
            keywords,
            content='documents',
            content_rowid='id',
            tokenize='porter unicode61'
        )
    """)

    # Triggers to keep FTS in sync
    c.execute("""
        CREATE TRIGGER IF NOT EXISTS documents_ai AFTER INSERT ON documents BEGIN
            INSERT INTO documents_fts(rowid, title, content, keywords)
            VALUES (new.id, new.title, new.content, new.keywords);
        END
    """)

    c.execute("""
        CREATE TRIGGER IF NOT EXISTS documents_ad AFTER DELETE ON documents BEGIN
            INSERT INTO documents_fts(documents_fts, rowid, title, content, keywords)
            VALUES ('delete', old.id, old.title, old.content, old.keywords);
        END
    """)

    c.execute("""
        CREATE TRIGGER IF NOT EXISTS documents_au AFTER UPDATE ON documents BEGIN
            INSERT INTO documents_fts(documents_fts, rowid, title, content, keywords)
            VALUES ('delete', old.id, old.title, old.content, old.keywords);
            INSERT INTO documents_fts(rowid, title, content, keywords)
            VALUES (new.id, new.title, new.content, new.keywords);
        END
    """)

    # Feedback log — tracks every feedback event for learning
    c.execute("""
        CREATE TABLE IF NOT EXISTS feedback_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            document_id INTEGER REFERENCES documents(id),
            query TEXT,
            relevant INTEGER NOT NULL,    -- 1 = relevant, 0 = not
            created_at TEXT DEFAULT (datetime('now'))
        )
    """)

    # Query log — tracks what gets searched
    c.execute("""
        CREATE TABLE IF NOT EXISTS query_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            query TEXT NOT NULL,
            result_count INTEGER,
            created_at TEXT DEFAULT (datetime('now'))
        )
    """)

    # Tags table for categorization
    c.execute("""
        CREATE TABLE IF NOT EXISTS tags (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT UNIQUE NOT NULL,
            weight REAL DEFAULT 1.0       -- adjustable importance
        )
    """)

    c.execute("""
        CREATE TABLE IF NOT EXISTS document_tags (
            document_id INTEGER REFERENCES documents(id),
            tag_id INTEGER REFERENCES tags(id),
            PRIMARY KEY (document_id, tag_id)
        )
    """)

    # Properties table — cheap BC properties under threshold
    c.execute("""
        CREATE TABLE IF NOT EXISTS properties (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            price INTEGER,                  -- in CAD dollars
            location TEXT,                  -- city/area
            region TEXT,                    -- BC region: kootenay, cariboo, north, etc.
            latitude REAL,
            longitude REAL,
            property_type TEXT,             -- land, cabin, mobile, house, lot, recreational
            size_acres REAL,
            description TEXT,
            listing_url TEXT UNIQUE,
            source TEXT,                    -- craigslist, kijiji, facebook, usedvic, etc.
            contact TEXT,
            posted_date TEXT,
            scraped_at TEXT DEFAULT (datetime('now')),
            is_hidden INTEGER DEFAULT 0,    -- flagged as obscure/hard-to-find
            access TEXT,                    -- road access, water access, fly-in, remote
            services TEXT,                  -- power, well, septic, off-grid
            notes TEXT,
            score REAL DEFAULT 0.0,         -- attractiveness score
            raw_html TEXT,
            hash TEXT UNIQUE
        )
    """)

    c.execute("""
        CREATE INDEX IF NOT EXISTS idx_properties_price
        ON properties(price)
    """)
    c.execute("""
        CREATE INDEX IF NOT EXISTS idx_properties_region
        ON properties(region)
    """)
    c.execute("""
        CREATE INDEX IF NOT EXISTS idx_properties_type
        ON properties(property_type)
    """)

    # FTS5 for property search
    c.execute("""
        CREATE VIRTUAL TABLE IF NOT EXISTS properties_fts USING fts5(
            title,
            description,
            location,
            region,
            notes,
            content='properties',
            content_rowid='id',
            tokenize='porter unicode61'
        )
    """)

    c.execute("""
        CREATE TRIGGER IF NOT EXISTS properties_ai AFTER INSERT ON properties BEGIN
            INSERT INTO properties_fts(rowid, title, description, location, region, notes)
            VALUES (new.id, new.title, new.description, new.location, new.region, new.notes);
        END
    """)

    c.execute("""
        CREATE TRIGGER IF NOT EXISTS properties_ad AFTER DELETE ON properties BEGIN
            INSERT INTO properties_fts(properties_fts, rowid, title, description, location, region, notes)
            VALUES ('delete', old.id, old.title, old.description, old.location, old.region, old.notes);
        END
    """)

    c.execute("""
        CREATE TRIGGER IF NOT EXISTS properties_au AFTER UPDATE ON properties BEGIN
            INSERT INTO properties_fts(properties_fts, rowid, title, description, location, region, notes)
            VALUES ('delete', old.id, old.title, old.description, old.location, old.region, old.notes);
            INSERT INTO properties_fts(rowid, title, description, location, region, notes)
            VALUES (new.id, new.title, new.description, new.location, new.region, new.notes);
        END
    """)

    conn.commit()
    conn.close()

    run_migrations()


def run_migrations():
    """Apply any pending SQL files in migrations/ in numeric order."""
    import re
    mig_dir = os.path.join(os.path.dirname(__file__), "migrations")
    if not os.path.isdir(mig_dir):
        return

    conn = get_conn()
    conn.execute("""
        CREATE TABLE IF NOT EXISTS schema_version (
            version INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            applied_at TEXT DEFAULT (datetime('now'))
        )
    """)
    conn.commit()
    applied = {row[0] for row in conn.execute("SELECT version FROM schema_version").fetchall()}

    files = sorted(f for f in os.listdir(mig_dir) if f.endswith(".sql"))
    for fname in files:
        m = re.match(r"^(\d+)_(.+)\.sql$", fname)
        if not m:
            continue
        version = int(m.group(1))
        name = m.group(2)
        if version in applied:
            continue
        path = os.path.join(mig_dir, fname)
        with open(path, "r") as f:
            sql = f.read()
        try:
            conn.executescript(sql)
            conn.execute(
                "INSERT INTO schema_version (version, name) VALUES (?, ?)",
                (version, name),
            )
            conn.commit()
            print(f"[migrate] applied {fname}")
        except Exception as e:
            conn.rollback()
            conn.close()
            raise RuntimeError(f"Migration {fname} failed: {e}")
    conn.close()


def insert_property(data: dict) -> int | None:
    """Insert a property listing. Returns new id or None if duplicate."""
    import hashlib
    conn = get_conn()
    try:
        # Hash for dedup (title + price + location)
        h = hashlib.sha256(
            f"{data.get('title','')}{data.get('price','')}{data.get('location','')}".encode()
        ).hexdigest()
        cur = conn.execute(
            """INSERT OR IGNORE INTO properties
               (title, price, location, region, latitude, longitude,
                property_type, size_acres, description, listing_url, source,
                contact, posted_date, is_hidden, access, services, notes,
                score, raw_html, hash)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (
                data.get("title", "Unknown"),
                data.get("price"),
                data.get("location"),
                data.get("region"),
                data.get("latitude"),
                data.get("longitude"),
                data.get("property_type"),
                data.get("size_acres"),
                data.get("description"),
                data.get("listing_url"),
                data.get("source"),
                data.get("contact"),
                data.get("posted_date"),
                1 if data.get("is_hidden") else 0,
                data.get("access"),
                data.get("services"),
                data.get("notes"),
                data.get("score", 0.0),
                data.get("raw_html"),
                h,
            ),
        )
        conn.commit()
        if cur.rowcount == 0:
            return None
        return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    except sqlite3.IntegrityError:
        return None
    finally:
        conn.close()


def search_properties(query=None, max_price=30000, region=None, ptype=None, limit=50):
    """Search properties by price, region, type, and optional text."""
    conn = get_conn()
    if query:
        # FTS search with filters
        rows = conn.execute("""
            SELECT p.*
            FROM properties_fts fts
            JOIN properties p ON p.id = fts.rowid
            WHERE properties_fts MATCH ?
              AND (p.price IS NULL OR p.price <= ?)
              AND (? IS NULL OR p.region = ?)
              AND (? IS NULL OR p.property_type = ?)
            ORDER BY p.score DESC, p.price ASC
            LIMIT ?
        """, (query, max_price, region, region, ptype, ptype, limit)).fetchall()
    else:
        rows = conn.execute("""
            SELECT * FROM properties
            WHERE (price IS NULL OR price <= ?)
              AND (? IS NULL OR region = ?)
              AND (? IS NULL OR property_type = ?)
            ORDER BY score DESC, price ASC
            LIMIT ?
        """, (max_price, region, region, ptype, ptype, limit)).fetchall()
    conn.close()
    return rows


def property_stats():
    conn = get_conn()
    s = {
        "total": conn.execute("SELECT COUNT(*) FROM properties").fetchone()[0],
        "under_30k": conn.execute("SELECT COUNT(*) FROM properties WHERE price <= 30000").fetchone()[0],
        "under_10k": conn.execute("SELECT COUNT(*) FROM properties WHERE price <= 10000").fetchone()[0],
        "hidden": conn.execute("SELECT COUNT(*) FROM properties WHERE is_hidden = 1").fetchone()[0],
        "avg_price": conn.execute("SELECT COALESCE(AVG(price), 0) FROM properties WHERE price > 0").fetchone()[0],
        "by_region": dict(conn.execute("SELECT region, COUNT(*) FROM properties GROUP BY region").fetchall()),
        "by_type": dict(conn.execute("SELECT property_type, COUNT(*) FROM properties GROUP BY property_type").fetchall()),
        "by_source": dict(conn.execute("SELECT source, COUNT(*) FROM properties GROUP BY source").fetchall()),
    }
    conn.close()
    return s


def insert_document(source_id, title, content, url=None, author=None, keywords=None, doc_hash=None):
    conn = get_conn()
    try:
        conn.execute(
            """INSERT OR IGNORE INTO documents
               (source_id, title, content, url, author, keywords, hash)
               VALUES (?, ?, ?, ?, ?, ?, ?)""",
            (source_id, title, content, url, author,
             json.dumps(keywords or []), doc_hash)
        )
        conn.commit()
        return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    except sqlite3.IntegrityError:
        return None
    finally:
        conn.close()


def search(query, limit=20, min_relevance=0.0):
    """BM25 full-text search with relevance score boosting."""
    conn = get_conn()
    results = conn.execute("""
        SELECT d.*,
               bm25(documents_fts, 5.0, 1.0, 3.0) as bm25_score
        FROM documents_fts fts
        JOIN documents d ON d.id = fts.rowid
        WHERE documents_fts MATCH ?
          AND d.relevance_score >= ?
        ORDER BY (bm25(documents_fts, 5.0, 1.0, 3.0) * (0.5 + d.relevance_score))
        LIMIT ?
    """, (query, min_relevance, limit)).fetchall()

    # Log the query
    conn.execute(
        "INSERT INTO query_log (query, result_count) VALUES (?, ?)",
        (query, len(results))
    )
    conn.commit()
    conn.close()
    return results


def record_feedback(document_id, query, relevant):
    """Record feedback and adjust document relevance score."""
    conn = get_conn()

    # Log the feedback
    conn.execute(
        "INSERT INTO feedback_log (document_id, query, relevant) VALUES (?, ?, ?)",
        (document_id, query, 1 if relevant else 0)
    )

    # Update document relevance using exponential moving average
    doc = conn.execute(
        "SELECT relevance_score, feedback_count, positive_feedback FROM documents WHERE id = ?",
        (document_id,)
    ).fetchone()

    if doc:
        count = doc["feedback_count"] + 1
        positive = doc["positive_feedback"] + (1 if relevant else 0)
        # EMA with decay — recent feedback matters more
        alpha = 0.3
        new_score = alpha * (1.0 if relevant else 0.0) + (1 - alpha) * doc["relevance_score"]

        conn.execute(
            """UPDATE documents
               SET relevance_score = ?, feedback_count = ?, positive_feedback = ?,
                   updated_at = datetime('now')
               WHERE id = ?""",
            (new_score, count, positive, document_id)
        )

    conn.commit()
    conn.close()


def add_source(type_, name, url=None, config=None):
    conn = get_conn()
    conn.execute(
        "INSERT INTO sources (type, name, url, config) VALUES (?, ?, ?, ?)",
        (type_, name, url, json.dumps(config or {}))
    )
    conn.commit()
    conn.close()


def get_sources(enabled_only=True):
    conn = get_conn()
    q = "SELECT * FROM sources"
    if enabled_only:
        q += " WHERE enabled = 1"
    results = conn.execute(q).fetchall()
    conn.close()
    return results


def get_stats():
    conn = get_conn()
    stats = {
        "documents": conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0],
        "sources": conn.execute("SELECT COUNT(*) FROM sources").fetchone()[0],
        "feedback_events": conn.execute("SELECT COUNT(*) FROM feedback_log").fetchone()[0],
        "queries": conn.execute("SELECT COUNT(*) FROM query_log").fetchone()[0],
        "avg_relevance": conn.execute(
            "SELECT COALESCE(AVG(relevance_score), 0) FROM documents"
        ).fetchone()[0],
    }
    conn.close()
    return stats


def add_graph_edge(from_id, to_id, relationship, weight=1.0, created_by='system'):
    """Add a directed edge between two documents in the knowledge graph."""
    conn = get_conn()
    try:
        conn.execute(
            """INSERT OR IGNORE INTO knowledge_graph
               (from_doc_id, to_doc_id, relationship, weight, created_by)
               VALUES (?, ?, ?, ?, ?)""",
            (from_id, to_id, relationship, weight, created_by)
        )
        conn.commit()
    finally:
        conn.close()


def get_connections(doc_id, direction='both'):
    """Return documents connected to doc_id. direction: 'from', 'to', or 'both'."""
    conn = get_conn()
    try:
        results = []
        if direction in ('from', 'both'):
            rows = conn.execute("""
                SELECT kg.*, d.title, d.url
                FROM knowledge_graph kg
                JOIN documents d ON d.id = kg.to_doc_id
                WHERE kg.from_doc_id = ?
                ORDER BY kg.weight DESC
            """, (doc_id,)).fetchall()
            results.extend(rows)
        if direction in ('to', 'both'):
            rows = conn.execute("""
                SELECT kg.*, d.title, d.url
                FROM knowledge_graph kg
                JOIN documents d ON d.id = kg.from_doc_id
                WHERE kg.to_doc_id = ?
                ORDER BY kg.weight DESC
            """, (doc_id,)).fetchall()
            results.extend(rows)
        return results
    finally:
        conn.close()


def add_domain(name, parent_id=None, description=None, target_doc_count=10, priority=1.0):
    """Add a knowledge domain. Returns the new domain id."""
    conn = get_conn()
    try:
        conn.execute(
            """INSERT OR IGNORE INTO knowledge_domains
               (name, parent_id, description, target_doc_count, priority)
               VALUES (?, ?, ?, ?, ?)""",
            (name, parent_id, description, target_doc_count, priority)
        )
        conn.commit()
        return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    finally:
        conn.close()


def assign_domain(doc_id, domain_id, confidence=1.0):
    """Assign a document to a knowledge domain."""
    conn = get_conn()
    try:
        conn.execute(
            """INSERT OR REPLACE INTO document_domains
               (document_id, domain_id, confidence)
               VALUES (?, ?, ?)""",
            (doc_id, domain_id, confidence)
        )
        conn.commit()
    finally:
        conn.close()


def get_domain_coverage():
    """Return each domain with current doc count vs target."""
    conn = get_conn()
    try:
        rows = conn.execute("""
            SELECT kd.id, kd.name, kd.parent_id, kd.description,
                   kd.target_doc_count, kd.priority,
                   COUNT(dd.document_id) AS doc_count
            FROM knowledge_domains kd
            LEFT JOIN document_domains dd ON dd.domain_id = kd.id
            GROUP BY kd.id
            ORDER BY kd.priority DESC
        """).fetchall()
        return rows
    finally:
        conn.close()


def add_research_plan(domain_id, topic, reason, priority=1.0, assigned_to='gemma'):
    """Create a research plan. Returns the new plan id."""
    conn = get_conn()
    try:
        conn.execute(
            """INSERT INTO research_plans
               (domain_id, topic, reason, priority, assigned_to)
               VALUES (?, ?, ?, ?, ?)""",
            (domain_id, topic, reason, priority, assigned_to)
        )
        conn.commit()
        return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    finally:
        conn.close()


def get_pending_research(limit=10):
    """Return pending research plans ordered by priority."""
    conn = get_conn()
    try:
        rows = conn.execute("""
            SELECT rp.*, kd.name AS domain_name
            FROM research_plans rp
            LEFT JOIN knowledge_domains kd ON kd.id = rp.domain_id
            WHERE rp.status = 'pending'
            ORDER BY rp.priority DESC
            LIMIT ?
        """, (limit,)).fetchall()
        return rows
    finally:
        conn.close()


def complete_research(plan_id, doc_id):
    """Mark a research plan as completed and link the resulting document."""
    conn = get_conn()
    try:
        conn.execute(
            """UPDATE research_plans
               SET status = 'completed', result_doc_id = ?,
                   completed_at = datetime('now')
               WHERE id = ?""",
            (doc_id, plan_id)
        )
        conn.commit()
    finally:
        conn.close()


if __name__ == "__main__":
    init_db()
    print("Database initialized successfully.")