import sys import os import time import re from bs4 import BeautifulSoup 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 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.sync_api import sync_playwright SOURCE = 'www.evergabe-online.de' SITE_ROOT = 'https://www.evergabe-online.de' BATCH_SIZE = 50 def fetch_pending(cursor, limit=BATCH_SIZE): cursor.execute( """SELECT id, url FROM tenders WHERE detail = 0 AND source = %s ORDER BY id ASC LIMIT %s""", (SOURCE, limit) ) return cursor.fetchall() def extract_url_id(url): """Extract numeric id from URL like tenderdetails.html?id=845946""" m = re.search(r'[?&]id=(\d+)', url) return m.group(1) if m else None def scrape_notice(page, url): """Visit detail page, return inner HTML of notice div.""" page.goto(url, wait_until='networkidle', timeout=60000) try: page.wait_for_selector('div.procedure-details.nat-form, div#htmlDocument', timeout=10000) except Exception: pass div = page.query_selector('div.procedure-details.nat-form') if div: return div.inner_html() div = page.query_selector('div#htmlDocument') if div: return div.inner_html() return None def parse_versanddatum(raw): """Parse '18.03.26, 07:45' or '18.03.2026, 07:45' into 'YYYY-MM-DD', else None.""" if not raw or raw.strip() == '-': return None m = re.match(r'(\d{1,2})\.(\d{2})\.(\d{2,4})', raw.strip()) if not m: return None day, month, year = m.group(1), m.group(2), m.group(3) if len(year) == 2: year = '20' + year return f'{year}-{month}-{int(day):02d}' def scrape_documents(page, url_id): """Visit documents page, return list of {document_name, document_url, date_added}.""" docs_url = f'{SITE_ROOT}/tenderdocuments.html?0&id={url_id}' page.goto(docs_url, wait_until='networkidle', timeout=60000) try: page.wait_for_selector('table tbody', timeout=10000) except Exception: pass docs = page.eval_on_selector_all( 'table tbody tr', """trs => trs.map(tr => { const a = tr.querySelector('td.filename a.title'); if (!a) return null; const dlBtn = tr.querySelector('td.text-right a.icon[download]'); const dateEl = tr.querySelector('td span.inquiry-date'); return { document_name : a.getAttribute('title') || a.innerText.trim(), document_url : dlBtn ? dlBtn.href : a.href, date_raw : dateEl ? dateEl.innerText.trim() : null, }; }).filter(r => r !== null)""" ) for doc in docs: doc['date_added'] = parse_versanddatum(doc.pop('date_raw', None)) return docs def parse_cpv_code(soup): """ Parse CPV code from notice soup. Handles two formats: 1. EU format: 'Hauptklassifizierungscode (cpv): 79710000 ...' 2. Chapter format: div.chapter with label 'f)' containing an 8-digit code Returns the first 8-digit CPV code found, or None. """ if soup is None: return None text = soup.get_text(separator=' ') # EU format: Hauptklassifizierungscode (cpv): 79710000 m = re.search(r'Hauptklassifizierungscode\s*\(cpv\)\s*[:\-]?\s*(\d{8})', text, re.IGNORECASE) if m: return m.group(1) # Generic CPV label pattern m = re.search(r'\bCPV\b[^\d]*(\d{8})', text, re.IGNORECASE) if m: return m.group(1) # Chapter f) — look for 8-digit code in content for chapter in soup.select('div.chapter'): num_div = chapter.select_one('div.chapter-number') if not num_div: continue if num_div.get_text(strip=True).strip('.').strip() != 'f)': continue content_div = chapter.select_one('div.chapter-content') if content_div: m = re.search(r'\b(\d{8})\b', content_div.get_text()) if m: return m.group(1) return None def parse_eligibility(soup): """ Parse 'w) Beurteilung der Eignung' from notice soup. Collects all chapter content from w) up to (but not including) x). """ if soup is None: return None chapters = soup.select('div.chapter') collecting = False parts = [] for chapter in chapters: num_div = chapter.select_one('div.chapter-number') if not num_div: continue label = num_div.get_text(strip=True).strip('.').strip() if label == 'w)': collecting = True if collecting: if label == 'x)': break # stop before x) content_div = chapter.select_one('div.chapter-content') if content_div: for br in content_div.find_all('br'): br.replace_with('\n') text = content_div.get_text(separator='\n', strip=True) if text: parts.append(text) return '\n\n'.join(parts) if parts else None def parse_auftraggeber(soup): """ Parse 'a) Öffentlicher Auftraggeber' from notice soup. Returns dict with contracting_authority_name, contracting_org_address, contracting_org_city, contracting_org_postal. """ if soup is None: return {} for chapter in soup.select('div.chapter'): num_div = chapter.select_one('div.chapter-number') if not num_div: continue if num_div.get_text(strip=True).strip('.').strip() != 'a)': continue content_div = chapter.select_one('div.chapter-content') if not content_div: break for br in content_div.find_all('br'): br.replace_with('\n') lines = [l.strip() for l in content_div.get_text().split('\n') if l.strip()] result = {} for line in lines: if ':' in line: key, _, val = line.partition(':') key = key.strip().lower() val = val.strip() if 'name' in key: result['contracting_authority_name'] = val elif 'straße' in key or 'strasse' in key or 'hausnummer' in key: result['contracting_org_address'] = val elif 'postleitzahl' in key or 'plz' in key: result['contracting_org_postal'] = val elif key == 'ort': result['contracting_org_city'] = val elif 'e-mail' in key or 'email' in key: result['contracting_authority_email'] = val return result return {} def parse_summary(soup): """ Parse short description from 'f) Art und Umfang der Leistung'. Falls back to 'd) Art des Auftrags' if f) is not found. Single pass over chapters collecting both labels. """ if soup is None: return None target = {'f)', 'd)'} found = {} for chapter in soup.select('div.chapter'): num_div = chapter.select_one('div.chapter-number') if not num_div: continue label = num_div.get_text(strip=True).strip('.').strip() if label not in target or label in found: continue content_div = chapter.select_one('div.chapter-content') if content_div: for br in content_div.find_all('br'): br.replace_with('\n') text = content_div.get_text(separator='\n', strip=True) if text: found[label] = text if len(found) == len(target): break # both labels found, stop early return found.get('f)') or found.get('d)') # def parse_ort_der_ausfuehrung(soup): # """ # Parse 'e) Ort der Ausführung' from notice soup. # Returns dict with contracting_org_country/address/city/province/postal. # """ # if soup is None: # return {} # for chapter in soup.select('div.chapter'): # num_div = chapter.select_one('div.chapter-number') # if not num_div: # continue # if num_div.get_text(strip=True).strip('.').strip() != 'e)': # continue # content_div = chapter.select_one('div.chapter-content') # if not content_div: # break # # Replace
with newline, then split into non-empty lines # for br in content_div.find_all('br'): # br.replace_with('\n') # lines = [l.strip() for l in content_div.get_text().split('\n') if l.strip()] # postal = city = address = province = None # # country = 'DE' # country = 'Germany' # postal_city_idx = None # for i, line in enumerate(lines): # m = re.match(r'^(\d{5})\s+(.+)$', line) # if m: # postal = m.group(1) # city = m.group(2).strip() # postal_city_idx = i # break # if postal_city_idx is not None and postal_city_idx > 0: # address = ', '.join(lines[:postal_city_idx]) # elif postal_city_idx is None and lines: # address = ', '.join(lines) # return { # 'contracting_org_country': country, # 'contracting_org_address': address, # 'contracting_org_city': city, # 'contracting_org_province': province, # 'contracting_org_postal': postal, # } def parse_ort_der_ausfuehrung(soup): if soup is None: return {'contracting_org_country': 'Germany'} for chapter in soup.select('div.chapter'): num_div = chapter.select_one('div.chapter-number') if not num_div or num_div.get_text(strip=True).strip('.').strip() != 'e)': continue content_div = chapter.select_one('div.chapter-content') if not content_div: break for br in content_div.find_all('br'): br.replace_with('\n') lines = [l.strip() for l in content_div.get_text().split('\n') if l.strip()] postal = city = address = None country = 'Germany' postal_city_idx = None for i, line in enumerate(lines): m = re.match(r'^(\d{5})\s+(.+)$', line) if m: postal = m.group(1) city = m.group(2).strip() postal_city_idx = i break for line in lines: line_lower = line.lower() if 'land:' in line_lower or 'staat:' in line_lower: country = line.split(':')[-1].strip() elif 'österreich' in line_lower: country = 'Austria' elif 'schweiz' in line_lower: country = 'Switzerland' elif 'luxemburg' in line_lower: country = 'Luxembourg' if postal_city_idx is not None and postal_city_idx > 0: address = ', '.join(lines[:postal_city_idx]) elif postal_city_idx is None and lines: address = ', '.join(lines) return { 'contracting_org_country': country, 'contracting_org_address': address, 'contracting_org_city': city, 'contracting_org_province': None, 'contracting_org_postal': postal, } return {'contracting_org_country': 'Germany'} return {} def save_notice(cursor, tender_id, notice_html, loc, authority, eligibility=None, summary=None, cpv_code=None): cursor.execute( """UPDATE tender_details SET notice = %s, contracting_authority_name = %s, contracting_authority_email = %s, contracting_org_address = %s, contracting_org_city = %s, contracting_org_province = %s, contracting_org_postal = %s, contracting_org_country = %s, languages = %s, framework_agreement = %s, trade_agreements = %s, summary = %s, cpv_code = %s WHERE tender_id = %s""", ( notice_html, authority.get('contracting_authority_name'), authority.get('contracting_authority_email'), authority.get('contracting_org_address'), authority.get('contracting_org_city'), loc.get('contracting_org_province'), authority.get('contracting_org_postal'), loc.get('contracting_org_country', 'Germany'), 'German', eligibility, eligibility, summary, cpv_code, tender_id, ) ) def save_documents(cursor, tender_id, docs): for doc in docs: cursor.execute( """INSERT INTO tender_documents (tender_id, document_name, document_url, date_added) VALUES (%s, %s, %s, %s)""", (tender_id, doc['document_name'], doc['document_url'], doc.get('date_added')) ) def mark_done(cursor, tender_id): cursor.execute( "UPDATE tenders SET detail = 1 WHERE id = %s", (tender_id,) ) def run(): print('=' * 70) print('Evergabe-Online - Detail Scraper') print('=' * 70) conn = get_db_connection() cursor = conn.cursor() cursor.execute( "SELECT COUNT(*) FROM tenders WHERE detail=0 AND source=%s", (SOURCE,) ) total_pending = cursor.fetchone()[0] print(f'Pending tenders: {total_pending:,}') print('=' * 70) total_done = 0 total_failed = 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' ) ) page = context.new_page() while True: batch = fetch_pending(cursor) if not batch: print('\nNo more pending tenders.') break for tender_id, url in batch: print(f' [{tender_id}] {url}', end=' ... ', flush=True) try: # 1. Scrape notice notice_html = scrape_notice(page, url) if not notice_html: print('notice div not found', end=' ') # 2. Scrape documents url_id = extract_url_id(url) docs = [] if url_id: docs = scrape_documents(page, url_id) # 3. Parse fields from notice (parse HTML once, reuse soup) soup = BeautifulSoup(notice_html, 'html.parser') if notice_html else None loc = parse_ort_der_ausfuehrung(soup) authority = parse_auftraggeber(soup) eligibility = parse_eligibility(soup) summary = parse_summary(soup) cpv_code = parse_cpv_code(soup) # 4. Save to DB save_notice(cursor, tender_id, notice_html, loc, authority, eligibility, summary, cpv_code) save_documents(cursor, tender_id, docs) mark_done(cursor, tender_id) conn.commit() total_done += 1 city_info = loc.get('contracting_org_city', '-') or '-' print(f'OK (notice={"yes" if notice_html else "no"}, docs={len(docs)}, city={city_info})') except Exception as e: conn.rollback() total_failed += 1 print(f'FAILED: {e}') print(f'\n Progress: {total_done:,} done, {total_failed:,} failed\n') browser.close() cursor.close() conn.close() print('\n' + '=' * 70) print('Detail Scraping Complete!') print('=' * 70) print(f' Done : {total_done:,}') print(f' Failed : {total_failed:,}') 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()