import asyncio import re import sys import os from datetime import datetime from dotenv import load_dotenv load_dotenv(os.path.join(os.path.dirname(__file__), '../.env')) import mysql.connector 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' FETCH_BATCH = 200 # rows loaded from DB per iteration 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 parse_date(value): dt = parse_datetime(value) return dt[:10] if dt else None def extract_timezone(value): """Split 'DD/MM/YYYY HH:MM CET' into ('DD/MM/YYYY HH:MM', 'CET'). Returns (clean_value, timezone) — timezone is None if not found.""" if not value: return value, None parts = value.strip().split() if len(parts) >= 2 and re.match(r'^[A-Z]{2,5}$', parts[-1]): return ' '.join(parts[:-1]), parts[-1] return value, None async def scrape_detail_page(page, url): """Visit a detail page and extract fields by div[title] attribute.""" print(f" Visiting: {url}") try: await page.goto(url, wait_until='networkidle', timeout=60000) except Exception as e: print(f" ERROR loading page: {e}") return None # Wait for content try: await page.wait_for_selector('[title]', timeout=20000) except Exception: print(f" WARNING: no titled elements found") data = await page.evaluate(""" () => { function getVal(titleVal) { const els = document.querySelectorAll('[title]'); for (const el of els) { if (el.getAttribute('title').trim().toLowerCase() === titleVal.trim().toLowerCase()) { const text = (el.innerText || el.textContent || '').trim(); // If the element text equals the title label, value is in next sibling if (text.toLowerCase() === titleVal.trim().toLowerCase()) { const next = el.nextElementSibling; return next ? (next.innerText || next.textContent || '').trim() : ''; } return text; } } return null; } return { number_of_lots: getVal('Number of lots'), general_purpose: getVal('General purpose'), specific_purpose: getVal('Specific purpose'), initially_dispatched_at: getVal('Initially dispatched at'), last_dispatched_at: getVal('Last dispatched at'), status: getVal('Status'), published_on_ted: getVal('Published on TED'), legal_basis: getVal('Legal basis'), questions_from: getVal('Questions can be submitted from'), questions_until: getVal('Questions can be submitted until'), procedure: getVal('Procedure'), special_purchasing_technique:getVal('Special purchasing technique'), submission_deadline: getVal('Submission deadline'), }; } """) return data def upsert_tender_detail(cursor, tender_id, data): cursor.execute("SELECT id FROM tender_details WHERE tender_id = %s", (tender_id,)) existing = cursor.fetchone() raw_deadline = data.get('submission_deadline') clean_deadline, deadline_tz = extract_timezone(raw_deadline) fields = { 'divided_into_lots': data.get('number_of_lots'), 'summary': data.get('general_purpose'), 'subject_name': data.get('specific_purpose'), 'open_date': parse_date(data.get('initially_dispatched_at')), 'amendment_date': parse_date(data.get('last_dispatched_at')), 'notice_type': data.get('status'), 'eu_identifier': data.get('published_on_ted'), 'trade_agreements': data.get('legal_basis'), 'submission_start': parse_datetime(data.get('questions_from')), 'submission_end': parse_datetime(data.get('questions_until')), 'procedure_type': data.get('procedure'), 'specific_procedure': data.get('special_purchasing_technique'), 'deadline': parse_datetime(clean_deadline), 'deadline_timezone': deadline_tz, } if existing: set_clause = ', '.join(f"`{col}` = %s" for col in fields) sql = f"UPDATE tender_details SET {set_clause}, updated_at = CURRENT_TIMESTAMP WHERE tender_id = %s" cursor.execute(sql, list(fields.values()) + [tender_id]) else: cols = ', '.join(f"`{col}`" for col in fields) placeholders = ', '.join('%s' for _ in fields) sql = f"INSERT INTO tender_details (tender_id, {cols}, created_at, updated_at) VALUES (%s, {placeholders}, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)" cursor.execute(sql, [tender_id] + list(fields.values())) async def main(): 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() offset = 0 processed = 0 while True: cursor.execute( "SELECT id, url FROM tenders WHERE source = %s AND detail = 0 ORDER BY id LIMIT %s OFFSET %s", (SOURCE, FETCH_BATCH, offset) ) rows = cursor.fetchall() if not rows: break print(f"Fetched {len(rows)} tenders (offset={offset}).") for tender_id, url in rows: processed += 1 print(f"[{processed}] tender_id={tender_id}") if not url: print(" No URL, skipping.") continue data = await scrape_detail_page(page, url) if data is None: print(" Failed to scrape, skipping.") continue upsert_tender_detail(cursor, tender_id, data) clean_dl, _ = extract_timezone(data.get('submission_deadline')) cursor.execute( "UPDATE tenders SET detail = 1, closing_date = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s", (parse_date(clean_dl), tender_id) ) conn.commit() print(f" Done. detail=1 set.") offset += len(rows) await browser.close() cursor.close() conn.close() print(f"\nFinished. Processed {processed} tenders.") if __name__ == '__main__': asyncio.run(main())