""" SQLite database operations for the SEAO scraper. Handles schema creation, upserts, and queries with deduplication by item_id. """ import sqlite3 import logging from datetime import datetime, timezone from contextlib import contextmanager logger = logging.getLogger(__name__) CREATE_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS contracts ( id INTEGER PRIMARY KEY AUTOINCREMENT, item_id TEXT UNIQUE NOT NULL, contract_number TEXT, reference_number TEXT, description TEXT, type_category TEXT, city TEXT, status TEXT, publication_date TEXT, closing_date TEXT, winning_bidder TEXT, contract_price TEXT, scraped_at TEXT DEFAULT (datetime('now')), last_seen_at TEXT, detail_scraped_at TEXT, source_url TEXT ); """ CREATE_RUNS_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS scrape_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, started_at TEXT NOT NULL, finished_at TEXT, date_from TEXT, date_to TEXT, status TEXT NOT NULL, scraped INTEGER DEFAULT 0, inserted_count INTEGER DEFAULT 0, skipped_count INTEGER DEFAULT 0, sync_failed INTEGER DEFAULT 0, error_message TEXT ); """ CREATE_INDEXES_SQL = [ "CREATE INDEX IF NOT EXISTS idx_item_id ON contracts(item_id);", "CREATE INDEX IF NOT EXISTS idx_status ON contracts(status);", "CREATE INDEX IF NOT EXISTS idx_publication_date ON contracts(publication_date);", "CREATE INDEX IF NOT EXISTS idx_city ON contracts(city);", ] INSERT_SQL = """ INSERT INTO contracts ( item_id, contract_number, reference_number, description, type_category, city, status, publication_date, closing_date, winning_bidder, contract_price, scraped_at, last_seen_at, detail_scraped_at, source_url ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(item_id) DO UPDATE SET contract_number=excluded.contract_number, reference_number=excluded.reference_number, description=excluded.description, type_category=excluded.type_category, city=excluded.city, status=excluded.status, publication_date=excluded.publication_date, closing_date=excluded.closing_date, winning_bidder=CASE WHEN excluded.winning_bidder <> '' THEN excluded.winning_bidder ELSE contracts.winning_bidder END, contract_price=CASE WHEN excluded.contract_price <> '' THEN excluded.contract_price ELSE contracts.contract_price END, scraped_at=excluded.scraped_at, last_seen_at=excluded.last_seen_at, detail_scraped_at=COALESCE(excluded.detail_scraped_at, contracts.detail_scraped_at), source_url=COALESCE(excluded.source_url, contracts.source_url); """ CREATE_SUBMISSIONS_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS submissions ( id INTEGER PRIMARY KEY AUTOINCREMENT, contract_item_id TEXT NOT NULL, supplier_name TEXT, supplier_address TEXT, neq TEXT, contact TEXT, bid_amount REAL, is_winner INTEGER DEFAULT 0, contract_date TEXT, contract_publication_date TEXT, contract_amount REAL, final_publication_date TEXT, final_end_date TEXT, renewal_options_exercised INTEGER DEFAULT 0, additional_acquisition_options INTEGER DEFAULT 0, total_paid_amount REAL, supplementary_info TEXT, source_tab TEXT, scraped_at TEXT DEFAULT (datetime('now')), FOREIGN KEY (contract_item_id) REFERENCES contracts(item_id) ); """ CREATE_SUBMISSIONS_INDEXES_SQL = [ "CREATE INDEX IF NOT EXISTS idx_sub_contract_item_id ON submissions(contract_item_id);", "CREATE INDEX IF NOT EXISTS idx_sub_supplier_name ON submissions(supplier_name);", "CREATE INDEX IF NOT EXISTS idx_sub_is_winner ON submissions(is_winner);", "CREATE UNIQUE INDEX IF NOT EXISTS idx_sub_unique ON submissions(contract_item_id, neq, supplier_name);", ] INSERT_SUBMISSION_SQL = """ INSERT OR REPLACE INTO submissions ( contract_item_id, supplier_name, supplier_address, neq, contact, bid_amount, is_winner, contract_date, contract_publication_date, contract_amount, final_publication_date, final_end_date, renewal_options_exercised, additional_acquisition_options, total_paid_amount, supplementary_info, source_tab, scraped_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?); """ DELETE_SUBMISSIONS_SQL = """ DELETE FROM submissions WHERE contract_item_id = ?; """ SELECT_SUBMISSIONS_SQL = """ SELECT * FROM submissions WHERE contract_item_id = ? ORDER BY bid_amount ASC; """ class Database: """SQLite database wrapper for contract storage.""" def __init__(self, db_path: str): """ Initialize database connection and create schema. Args: db_path: Path to the SQLite database file. """ import os self.db_path = db_path # Ensure parent directory exists db_dir = os.path.dirname(db_path) if db_dir and not os.path.exists(db_dir): os.makedirs(db_dir, exist_ok=True) logger.info(f"Created directory: {db_dir}") self._init_db() logger.info(f"Database initialized at {db_path}") @contextmanager def _get_connection(self): """Context manager for database connections with WAL mode.""" conn = sqlite3.connect(self.db_path) conn.execute("PRAGMA journal_mode=WAL;") conn.execute("PRAGMA foreign_keys=ON;") try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() def _init_db(self): """Create tables and indexes if they don't exist.""" with self._get_connection() as conn: conn.execute(CREATE_TABLE_SQL) conn.execute(CREATE_RUNS_TABLE_SQL) columns = {row[1] for row in conn.execute("PRAGMA table_info(contracts)")} for name, definition in {"last_seen_at": "TEXT", "detail_scraped_at": "TEXT", "source_url": "TEXT"}.items(): if name not in columns: conn.execute(f"ALTER TABLE contracts ADD COLUMN {name} {definition}") for index_sql in CREATE_INDEXES_SQL: conn.execute(index_sql) conn.execute(CREATE_SUBMISSIONS_TABLE_SQL) for index_sql in CREATE_SUBMISSIONS_INDEXES_SQL: conn.execute(index_sql) def insert_contract(self, contract: dict) -> bool: """ Insert a contract if its item_id doesn't already exist. Args: contract: Dictionary with contract data. Returns: True if inserted, False if already exists (skipped). """ scraped_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") with self._get_connection() as conn: exists = conn.execute("SELECT 1 FROM contracts WHERE item_id = ?", (contract.get("item_id"),)).fetchone() conn.execute( INSERT_SQL, ( contract.get("item_id"), contract.get("contract_number"), contract.get("reference_number"), contract.get("description"), contract.get("type_category"), contract.get("city"), contract.get("status"), contract.get("publication_date"), contract.get("closing_date"), contract.get("winning_bidder"), contract.get("contract_price"), scraped_at, scraped_at, contract.get("detail_scraped_at") or None, contract.get("source_url") or None, ), ) inserted = exists is None if inserted: logger.debug(f"Inserted: {contract.get('item_id')}") else: logger.debug(f"Skipped (duplicate): {contract.get('item_id')}") return inserted def insert_many(self, contracts: list[dict]) -> tuple[int, int]: """ Insert multiple contracts. Skips duplicates. Args: contracts: List of contract dictionaries. Returns: Tuple of (inserted_count, skipped_count). """ inserted = 0 skipped = 0 for contract in contracts: if self.insert_contract(contract): inserted += 1 else: skipped += 1 return inserted, skipped def start_run(self, date_from: str, date_to: str) -> int: with self._get_connection() as conn: cursor = conn.execute("INSERT INTO scrape_runs (started_at, date_from, date_to, status) VALUES (?, ?, ?, 'running')", (datetime.now(timezone.utc).isoformat(), date_from, date_to)) return cursor.lastrowid def finish_run(self, run_id: int, status: str, scraped: int = 0, inserted: int = 0, skipped: int = 0, sync_failed: int = 0, error_message: str = "") -> None: with self._get_connection() as conn: conn.execute("UPDATE scrape_runs SET finished_at=?, status=?, scraped=?, inserted_count=?, skipped_count=?, sync_failed=?, error_message=? WHERE id=?", (datetime.now(timezone.utc).isoformat(), status, scraped, inserted, skipped, sync_failed, error_message[:2000], run_id)) def contract_exists(self, item_id: str) -> bool: """Check if a contract with the given item_id already exists.""" with self._get_connection() as conn: cursor = conn.execute( "SELECT 1 FROM contracts WHERE item_id = ? LIMIT 1;", (item_id,), ) return cursor.fetchone() is not None def get_stats(self) -> dict: """Get database statistics.""" with self._get_connection() as conn: total = conn.execute( "SELECT COUNT(*) FROM contracts;" ).fetchone()[0] by_status = conn.execute( "SELECT status, COUNT(*) FROM contracts GROUP BY status ORDER BY COUNT(*) DESC;" ).fetchall() return { "total_contracts": total, "by_status": dict(by_status), } def get_recent_contracts(self, limit: int = 10) -> list[dict]: """Get most recently scraped contracts.""" with self._get_connection() as conn: conn.row_factory = sqlite3.Row rows = conn.execute( "SELECT * FROM contracts ORDER BY scraped_at DESC LIMIT ?;", (limit,), ).fetchall() return [dict(row) for row in rows] def get_all_contracts_with_submissions(self) -> list[dict]: """Return the complete local dataset in the API sync payload shape.""" with self._get_connection() as conn: conn.row_factory = sqlite3.Row contracts = [dict(row) for row in conn.execute( "SELECT * FROM contracts ORDER BY id" ).fetchall()] submissions = [dict(row) for row in conn.execute( "SELECT * FROM submissions ORDER BY contract_item_id, id" ).fetchall()] submissions_by_contract: dict[str, list[dict]] = {} for submission in submissions: submission.pop("id", None) submissions_by_contract.setdefault(submission["contract_item_id"], []).append(submission) for contract in contracts: contract.pop("id", None) contract["submissions"] = submissions_by_contract.get(contract["item_id"], []) return contracts def upsert_submissions(self, contract_item_id: str, submissions: list[dict]) -> tuple[int, int]: """ Replace all submissions for a contract with the new list. Args: contract_item_id: The item_id of the contract. submissions: List of submission dicts. Returns: Tuple of (inserted_count, deleted_count). """ scraped_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") with self._get_connection() as conn: cursor = conn.execute(DELETE_SUBMISSIONS_SQL, (contract_item_id,)) deleted = cursor.rowcount inserted = 0 for sub in submissions: conn.execute( INSERT_SUBMISSION_SQL, ( contract_item_id, sub.get("supplier_name", ""), sub.get("supplier_address", ""), sub.get("neq", ""), sub.get("contact", ""), sub.get("bid_amount"), 1 if sub.get("is_winner", 0) else 0, sub.get("contract_date", ""), sub.get("contract_publication_date", ""), sub.get("contract_amount"), sub.get("final_publication_date", ""), sub.get("final_end_date", ""), sub.get("renewal_options_exercised", 0), sub.get("additional_acquisition_options", 0), sub.get("total_paid_amount"), sub.get("supplementary_info", ""), sub.get("source_tab", "ouverture"), scraped_at, ), ) inserted += 1 logger.debug( f"Submissions for {contract_item_id}: {inserted} inserted, {deleted} deleted" ) return inserted, deleted def get_submissions(self, contract_item_id: str) -> list[dict]: """Get all submissions for a given contract.""" with self._get_connection() as conn: conn.row_factory = sqlite3.Row rows = conn.execute( SELECT_SUBMISSIONS_SQL, (contract_item_id,) ).fetchall() return [dict(row) for row in rows] def get_submission_stats(self) -> dict: """Get submission statistics across all contracts.""" with self._get_connection() as conn: total_submissions = conn.execute( "SELECT COUNT(*) FROM submissions;" ).fetchone()[0] total_winners = conn.execute( "SELECT COUNT(*) FROM submissions WHERE is_winner = 1;" ).fetchone()[0] contracts_with_subs = conn.execute( "SELECT COUNT(DISTINCT contract_item_id) FROM submissions;" ).fetchone()[0] return { "total_submissions": total_submissions, "total_winners": total_winners, "contracts_with_submissions": contracts_with_subs, }