import asyncio import re import os from datetime import datetime from dotenv import load_dotenv load_dotenv(os.path.join(os.path.dirname(__file__), '../.env')) import mysql.connector import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) from keyword_tracker import is_keyword_done, mark_keyword_done def get_db_connection(): return mysql.connector.connect( host=os.getenv('DB_HOST'), port=int(os.getenv('DB_PORT', 3306)), user=os.getenv('DB_USER'), password=os.getenv('DB_PASSWORD'), database=os.getenv('DB_NAME'), ) from playwright.async_api import async_playwright SOURCE = 'www.publicprocurement.be' ITEMS_PER_PAGE = 25 MAX_PAGES = int(os.getenv('PUBLICPROCUREMENT_MAX_PAGES', 3)) SEARCH_LINKS_FILE = os.path.join(os.path.dirname(__file__), 'search_links.md') def load_search_urls(): """Parse URLs from search_links.md.""" urls = [] with open(SEARCH_LINKS_FILE, 'r') as f: for line in f: line = line.strip() if line.startswith('- http'): urls.append(line[2:].strip()) return urls def parse_datetime(value): if not value: return None formats = [ '%d/%m/%Y %H:%M:%S', '%d/%m/%Y %H:%M', '%d/%m/%Y', '%Y-%m-%dT%H:%M:%S', '%Y-%m-%d %H:%M:%S', '%Y-%m-%d', '%d-%m-%Y', '%d %B %Y', '%B %d, %Y', ] for fmt in formats: try: return datetime.strptime(value.strip(), fmt).strftime('%Y-%m-%d %H:%M:%S') except ValueError: continue return None def insert_tender(cursor, tender): cursor.execute( "SELECT id FROM tenders WHERE source = %s AND reference_number = %s", (tender['source'], tender['reference_number']) ) if cursor.fetchone(): return None sql = """ INSERT INTO tenders (source, source_id, title, reference_number, description, organization, url, publication_type, closing_date, detail, created_at, updated_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) """ cursor.execute(sql, ( tender['source'], tender['reference_number'], tender['title'], tender['reference_number'], tender['description'], tender['organization'], tender['url'], tender['publication_type'], tender['closing_date'], )) return cursor.lastrowid def insert_tender_detail(cursor, tender_id, detail): cursor.execute("SELECT id FROM tender_details WHERE tender_id = %s", (tender_id,)) existing = cursor.fetchone() if existing: sql = """ UPDATE tender_details SET reference_number = %s, contracting_authority_name = %s, category = %s, notice_type = %s, specific_procedure = %s, framework_agreement = %s, updated_at = CURRENT_TIMESTAMP WHERE tender_id = %s """ cursor.execute(sql, ( detail['reference_number'], detail['contracting_authority_name'], detail['category'], detail['notice_type'], detail['specific_procedure'], detail['framework_agreement'], tender_id, )) else: sql = """ INSERT INTO tender_details (tender_id, reference_number, contracting_authority_name, category, notice_type, specific_procedure, framework_agreement, created_at, updated_at) VALUES (%s, %s, %s, %s, %s, %s, %s, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) """ cursor.execute(sql, ( tender_id, detail['reference_number'], detail['contracting_authority_name'], detail['category'], detail['notice_type'], detail['specific_procedure'], detail['framework_agreement'], )) async def scrape_search_url(page, base_search_url, page_num): """Scrape one page of results from a shortLink search URL with pagination.""" # Append page/itemsPerPage params to the shortLink URL url = f"{base_search_url}&page={page_num}&itemsPerPage={ITEMS_PER_PAGE}&sortBy[]&sortDesc%5B%5D=false" print(f" Fetching page {page_num}: {url}") await page.goto(url, wait_until='networkidle', timeout=60000) try: await page.wait_for_selector('div.v-data-iterator', timeout=20000) except Exception: print(f" No results panel found on page {page_num}, stopping.") return [] raw_records = await page.evaluate(""" () => { function getVal(container, titleVal) { const els = container.querySelectorAll('[title]'); for (const el of els) { if (el.getAttribute('title').trim().toLowerCase() === titleVal.trim().toLowerCase()) { const text = el.innerText.trim(); if (text.toLowerCase() === titleVal.trim().toLowerCase()) { const next = el.nextElementSibling; return next ? next.innerText.trim() : ''; } return text; } } return ''; } const rows = Array.from(document.querySelectorAll('div.v-data-iterator div.v-row')) .filter(row => row.querySelector('a.v-card--link')); return rows.map(row => { const linkEl = row.querySelector('a.v-card--link'); const titleEl = row.querySelector('h2.page-header--title'); const descEl = row.querySelector('div.text-body-18'); let href = linkEl ? linkEl.getAttribute('href') : ''; if (href && !href.startsWith('http')) { href = 'https://www.publicprocurement.be' + href; } return { title: titleEl ? titleEl.innerText.trim() : '', url: href, description: descEl ? descEl.innerText.trim() : '', reference_number: getVal(row, 'Reference number'), organization: getVal(row, 'Organisation'), category: getVal(row, 'Category (CPV code)'), notice_type: getVal(row, "Nature(s)"), specific_procedure: getVal(row, 'Procedure'), framework_agreement:getVal(row, 'Legal Framework'), publication_date: getVal(row, 'Publication date'), dispatch_date: getVal(row, 'Dispatch date'), }; }); } """) if not raw_records: print(f" No cards found on page {page_num}.") return [] print(f" Found {len(raw_records)} cards on page {page_num}.") records = [] for r in raw_records: publication_type = parse_datetime(r['publication_date']) closing_date_dt = parse_datetime(r['dispatch_date']) closing_date = closing_date_dt[:10] if closing_date_dt else None tender = { 'source': SOURCE, 'source_id': r['reference_number'], 'title': r['title'], 'reference_number': r['reference_number'], 'description': r['description'], 'organization': r['organization'], 'url': r['url'], 'publication_type': publication_type, 'closing_date': closing_date, } detail = { 'reference_number': r['reference_number'], 'contracting_authority_name':r['organization'], 'category': r['category'], 'notice_type': r['notice_type'], 'specific_procedure': r['specific_procedure'], 'framework_agreement': r['framework_agreement'], } records.append((tender, detail)) return records async def main(): search_urls = load_search_urls() if not search_urls: print("No URLs found in search_links.md. Exiting.") return print(f"Loaded {len(search_urls)} search URL(s) from search_links.md.") conn = get_db_connection() cursor = conn.cursor() async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context( user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ' '(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36' ) page = await context.new_page() total_inserted = 0 total_skipped = 0 for idx, search_url in enumerate(search_urls, 1): print(f"\n[{idx}/{len(search_urls)}] Scraping: {search_url}") if is_keyword_done(conn, search_url, SOURCE): print(f" [SKIP] Already ran today: {search_url}") continue for page_num in range(1, MAX_PAGES + 1): records = await scrape_search_url(page, search_url, page_num) if not records: print(f" No records on page {page_num}, moving to next URL.") break for tender, detail in records: if not tender['reference_number']: continue tender_id = insert_tender(cursor, tender) if tender_id: insert_tender_detail(cursor, tender_id, detail) total_inserted += 1 else: total_skipped += 1 conn.commit() print(f" Page {page_num}: {len(records)} records. Inserted: {total_inserted}, Skipped: {total_skipped}") mark_keyword_done(conn, search_url, SOURCE) print(f" [SAVED] Marked URL done: {search_url}") await browser.close() cursor.close() conn.close() print(f"\nDone. Total Inserted: {total_inserted}, Total Skipped (duplicates): {total_skipped}") if __name__ == '__main__': asyncio.run(main())