import sys import os import time import re import json from datetime import datetime if sys.platform == 'win32': sys.stdout.reconfigure(encoding='utf-8') from dotenv import load_dotenv load_dotenv(os.path.join(os.path.dirname(__file__), '../.env')) import mysql.connector sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) from keyword_tracker import is_keyword_done, mark_keyword_done DB_CONFIG = { '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'), } # CPV_CODES = json.loads(os.getenv('CPV_CODES', '{}')) # {code: description} # MAX_PAGES = int(os.getenv('EVERGABE_MAX_PAGES', 3)) def get_db_connection(): return mysql.connector.connect(**DB_CONFIG) def load_cpv_from_db(): cpv = {} try: conn = get_db_connection() cursor = conn.cursor() # Try to load CPV codes from managed_keywords table cursor.execute( "SELECT keyword FROM managed_keywords WHERE language_code = 'cpv'" ) rows = cursor.fetchall() if rows: for (keyword_cell,) in rows: for entry in keyword_cell.split(','): entry = entry.strip() if entry: cpv[entry] = entry else: # Debug: Show what language codes are available cursor.execute("SELECT DISTINCT language_code FROM managed_keywords") available_codes = [row[0] for row in cursor.fetchall()] print(f" Warning: No CPV codes found. Available language_codes: {available_codes}") cursor.close() conn.close() except Exception as e: print(f" Warning: Could not load CPV codes from DB: {e}") return cpv CPV_CODES = load_cpv_from_db() print(f"Loaded {len(CPV_CODES)} CPV codes from DB: {list(CPV_CODES.keys())}") # exit() MAX_PAGES = int(os.getenv('EVERGABE_MAX_PAGES', 3)) from playwright.sync_api import sync_playwright BASE_URL = 'https://www.evergabe-online.de/search.html?4' SITE_ROOT = 'https://www.evergabe-online.de' SOURCE = 'www.evergabe-online.de' WAIT_AFTER_SEARCH_S = 10 # seconds to wait after clicking search TABLE_TIMEOUT_MS = 30000 PAGE_LOAD_WAIT_MS = 4000 def parse_deadline(raw): if not raw: return None try: return datetime.strptime(raw.strip(), '%d.%m.%y, %H:%M').strftime('%Y-%m-%d') except ValueError: return None def parse_publication_date(raw): if not raw: return None try: return datetime.strptime(raw.strip(), '%d.%m.%y').strftime('%Y-%m-%d') except ValueError: return None def extract_source_id(url): m = re.search(r'[?&]id=(\d+)', url) return int(m.group(1)) if m else None def compute_status(closing_date): """Return 'Open' if closing_date is today or future, else 'Closed'.""" if not closing_date: return 'Open' try: deadline = datetime.strptime(closing_date, '%Y-%m-%d').date() return 'Open' if deadline >= datetime.today().date() else 'Closed' except ValueError: return 'Open' def insert_tender(cursor, title, url, reference_number, organization, cpv_code, closing_date=None): cursor.execute( "SELECT id FROM tenders WHERE reference_number = %s AND source = %s LIMIT 1", (reference_number, SOURCE) ) row = cursor.fetchone() if row: return None # duplicate source_id = extract_source_id(url) status = compute_status(closing_date) cursor.execute( """INSERT INTO tenders (source, source_id, title, url, reference_number, organization, keyword, detail, closing_date, status, created_at) VALUES (%s, %s, %s, %s, %s, %s, %s, 0, %s, %s, NOW())""", (SOURCE, source_id, title, url, reference_number, organization, cpv_code, closing_date, status) ) return cursor.lastrowid def insert_tender_detail(cursor, tender_id, contracting_authority_name, contracting_org_address, procedure_type, deadline, deadline_text, publication_date): cursor.execute( """INSERT INTO tender_details (tender_id, contracting_authority_name, contracting_org_address, procedure_type, deadline, deadline_text, publication_date, created_at) VALUES (%s, %s, %s, %s, %s, %s, %s, NOW())""", (tender_id, contracting_authority_name, contracting_org_address, procedure_type, deadline, deadline_text, publication_date) ) def scrape_rows(page): rows = page.eval_on_selector_all( 'table#datatable tbody tr', """trs => trs.map(tr => { const tds = Array.from(tr.querySelectorAll('td')); if (tds.length < 4) return null; const anchor = tds[0].querySelector('a'); return { title : tds[0].innerText.trim(), url : anchor ? anchor.href : '', reference_number : tds[1].innerText.trim(), organization : tds[2].innerText.trim(), address : tds[3].innerText.trim(), procedure_type : tds[4] ? tds[4].innerText.trim() : '', deadline : tds[5] ? tds[5].innerText.trim() : '', publication_date : tds[6] ? tds[6].innerText.trim() : '', }; }).filter(r => r !== null)""" ) return rows def process_rows(rows, conn, cursor, cpv_code): inserted = 0 skipped = 0 for row in rows: ref = row.get('reference_number', '').strip() if not ref: skipped += 1 continue tender_id = insert_tender( cursor, title = row['title'], url = row['url'], reference_number = ref, organization = row['organization'], cpv_code = cpv_code, closing_date = parse_deadline(row['deadline']), ) if tender_id is None: skipped += 1 continue insert_tender_detail( cursor, tender_id = tender_id, contracting_authority_name = row['organization'], contracting_org_address = row['address'], procedure_type = row['procedure_type'], deadline = parse_deadline(row['deadline']), deadline_text = row['deadline'], publication_date = parse_publication_date(row['publication_date']), ) inserted += 1 conn.commit() return inserted, skipped def is_next_disabled(page): try: classes = page.eval_on_selector('a.next.icon', "el => el.className") return 'disabled' in classes.split() except Exception: return True def scrape_cpv(pg, conn, cursor, cpv_code, cpv_label): print(f'\n CPV: {cpv_code} — {cpv_label}') # Navigate to search page pg.goto(BASE_URL, wait_until='networkidle', timeout=60000) # Expand advanced search panel if CPV field is hidden try: cpv_input = pg.locator('#cpvCode') if not cpv_input.is_visible(): print(' CPV field hidden — looking for advanced search toggle...') # Try common toggle selectors for selector in [ 'a[data-toggle="collapse"]', 'button[data-toggle="collapse"]', '.advancedSearch', '#advancedSearch', 'a:has-text("Erweiterte")', 'a:has-text("Advanced")', 'button:has-text("Erweiterte")', 'span:has-text("Erweiterte")', ]: try: toggle = pg.locator(selector).first if toggle.is_visible(): toggle.click() print(f' Clicked toggle: {selector}') time.sleep(1) break except Exception: continue except Exception: pass # Fill CPV code field try: pg.wait_for_selector('#cpvCode:not([style*="display: none"])', timeout=5000) except Exception: pass # try force-fill anyway try: pg.locator('#cpvCode').fill(cpv_code, force=True) print(f' Filled #cpvCode with: {cpv_code}') except Exception as e: print(f' ERROR: Could not fill #cpvCode field: {e}') return 0, 0 # Click search button via JS (button is hidden/proxy element) try: pg.evaluate("document.querySelector('[name=\"submitButton\"]').click()") print(f' Clicked submitButton (JS)') except Exception as e: print(f' ERROR: Could not click submitButton: {e}') return 0, 0 # Wait for results print(f' Waiting {WAIT_AFTER_SEARCH_S}s for results...') time.sleep(WAIT_AFTER_SEARCH_S) total_inserted = 0 total_skipped = 0 page_num = 1 while True: print(f' --- Page {page_num} ---') try: pg.wait_for_selector('table#datatable tbody tr', timeout=TABLE_TIMEOUT_MS) except Exception: print(' table#datatable not found — stopping.') break rows = scrape_rows(pg) print(f' Found {len(rows)} row(s)') if not rows: print(' No rows — stopping.') break ins, skp = process_rows(rows, conn, cursor, cpv_code) total_inserted += ins total_skipped += skp print(f' Inserted: {ins} | Skipped (duplicates/empty): {skp}') if page_num >= MAX_PAGES: print(f' Reached max pages ({MAX_PAGES}) — stopping.') break if is_next_disabled(pg): print(' Next button disabled — last page.') break print(' Clicking next page...') try: pg.click('a.next.icon') except Exception as e: print(f' Could not click next: {e} — stopping.') break time.sleep(PAGE_LOAD_WAIT_MS / 1000) page_num += 1 return total_inserted, total_skipped def run(): print('=' * 70) print('Evergabe-Online - CPV Code Scraper') print('=' * 70) print(f'CPV codes to search: {len(CPV_CODES)}') print('=' * 70) if not CPV_CODES: print('ERROR: No CPV codes found in managed_keywords table with language_code="cpv"') print('Please check that managed_keywords table contains CPV codes with language_code="cpv"') return conn = get_db_connection() cursor = conn.cursor() grand_inserted = 0 grand_skipped = 0 with sync_playwright() as pw: browser = pw.chromium.launch(headless=True) context = browser.new_context( user_agent=( 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) ' 'AppleWebKit/537.36 (KHTML, like Gecko) ' 'Chrome/124.0.0.0 Safari/537.36' ) ) pg = context.new_page() for i, (cpv_code, cpv_label) in enumerate(CPV_CODES.items(), 1): print(f'\n[{i}/{len(CPV_CODES)}]', end='') if is_keyword_done(conn, cpv_code, SOURCE): print(f' [SKIP] Already ran today: "{cpv_code}"') continue ins, skp = scrape_cpv(pg, conn, cursor, cpv_code, cpv_label) grand_inserted += ins grand_skipped += skp mark_keyword_done(conn, cpv_code, SOURCE) print(f' [SAVED] Marked CPV done: "{cpv_code}"') time.sleep(2) browser.close() cursor.close() conn.close() print('\n' + '=' * 70) print('Scraping Complete!') print('=' * 70) print(f' CPV codes searched : {len(CPV_CODES)}') print(f' Rows inserted : {grand_inserted:,}') print(f' Rows skipped : {grand_skipped:,}') print('=' * 70) if __name__ == '__main__': try: run() except KeyboardInterrupt: print('\n\nInterrupted by user.') except Exception as e: import traceback print(f'\nFatal error: {e}') traceback.print_exc()