import requests from bs4 import BeautifulSoup import re from datetime import datetime, date import time import sys import os import threading from concurrent.futures import ThreadPoolExecutor, as_completed # Set UTF-8 encoding for console output 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'), ) SOURCE = 'canadabuys' # Base URL BASE_URL = "https://canadabuys.canada.ca" HEADERS = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.9', } def parse_date(date_str): """Parse date string to DATE format""" try: if date_str and date_str.strip(): date_match = re.search(r'(\d{4}/\d{2}/\d{2})', date_str) if date_match: return datetime.strptime(date_match.group(1), '%Y/%m/%d').date() except: pass return None def parse_closing_time(closing_time_str): """Parse closing time string like '2:00 pm EST' or '14:00 EDT'. Returns (time_str, timezone_str) or (None, None).""" if not closing_time_str or not closing_time_str.strip(): return None, None # Match time (12h or 24h) and optional timezone abbreviation match = re.match( r'(\d{1,2}:\d{2}(?:\s*[apAP][mM])?)\s*([A-Z]{2,5})?', closing_time_str.strip() ) if not match: return None, None time_part = match.group(1).strip() timezone_part = match.group(2) or None return time_part, timezone_part def build_deadline_datetime(closing_date, closing_time_str): """Combine closing_date (date) and closing_time string into a datetime. Returns (deadline_datetime, timezone_str).""" if not closing_date: return None, None time_part, timezone_part = parse_closing_time(closing_time_str) if not time_part: return None, timezone_part try: # Try 12h format first (e.g. "2:00 pm") for fmt in ('%I:%M %p', '%I:%M%p', '%H:%M'): try: t = datetime.strptime(time_part.upper(), fmt.upper()).time() break except ValueError: continue else: return None, timezone_part deadline_dt = datetime.combine(closing_date, t) return deadline_dt, timezone_part except Exception: return None, timezone_part def get_text_safe(element, selector=None, attribute=None): """Safely get text from element""" try: if selector: elem = element.select_one(selector) if elem: if attribute: return elem.get(attribute, '').strip() return elem.get_text(strip=True) elif element: if attribute: return element.get(attribute, '').strip() return element.get_text(strip=True) except: pass return None def extract_node_id(soup): """Extract node ID from page""" try: elem = soup.select_one('[data-history-node-id]') if elem: return elem.get('data-history-node-id') except: pass return None def extract_detail_data(html): """Extract all detail data from tender page HTML""" soup = BeautifulSoup(html, 'html.parser') data = { 'node_id': None, 'status': None, 'status_text': None, 'solicitation_number': None, 'closing_time': None, 'amendment_date': None, 'description': None, 'contract_duration': None, 'contract_duration_full': None, 'contract_start_date': None, 'notice_type': None, 'procurement_method': None, 'unspsc_code': None, 'unspsc_description': None, 'trade_agreements': None, 'limited_tendering_reason': None, 'organization': None, 'contracting_org_address': None, 'contracting_org_city': None, 'contracting_org_province': None, 'contracting_org_postal': None, 'contracting_org_country': None, 'contracting_authority_name': None, 'contracting_authority_phone': None, 'contracting_authority_email': None, 'regions_of_delivery': None, 'languages': None, 'region_of_opportunity': None, 'selection_criteria': None, 'contract_authority_address': None, 'documents': [] } # Node ID data['node_id'] = extract_node_id(soup) # Status — last non-empty line of the element text (skips label like "Status") status_elem = soup.select_one('#tender-status-label.field-content') if status_elem: lines = [l.strip() for l in status_elem.get_text().splitlines() if l.strip()] status_text = lines[-1] if lines else None data['status_text'] = status_text data['status'] = status_text.lower().replace(' ', '_') if status_text else None # Solicitation Number sol_num = soup.select_one('.field--name-field-tender-solicitation-number .field--item') if sol_num: data['solicitation_number'] = sol_num.get_text(strip=True) # Closing Time time_span = soup.select_one('.closing-date-field .timeclass') if time_span: data['closing_time'] = time_span.get_text(strip=True) # Amendment Date amend_date = soup.select_one('.field--name-field-tender-amendment-date time') if amend_date: data['amendment_date'] = parse_date(amend_date.get('datetime', '')) # Description desc = soup.select_one('.field--name-body.tender-detail-description') if desc: data['description'] = desc.get_text(strip=True) # Contract Duration - from description section duration = soup.select_one('.field--name-field-tender-contract-date') if duration: # Full text with proper spacing full_text = duration.get_text(separator=' ', strip=True) data['contract_duration_full'] = full_text # Extract start date start_date_match = re.search(r'(\d{4}/\d{2}/\d{2})', full_text) if start_date_match: data['contract_start_date'] = parse_date(start_date_match.group(1)) # Contract Duration - from summary section (more reliable) duration_summary = soup.select_one('.views-field-field-tender-contract-duration .field-content') if duration_summary: duration_text = duration_summary.get_text(strip=True) data['contract_duration'] = duration_text # Also update full if not already set if not data['contract_duration_full']: data['contract_duration_full'] = duration_text # Notice Type notice_type = soup.select_one('.views-field-field-tender-notice-type .field-content .field--item') if notice_type: data['notice_type'] = notice_type.get_text(strip=True) # Procurement Method proc_method = soup.select_one('.views-field-field-tender-procurement-method .field-content .field--item') if proc_method: data['procurement_method'] = proc_method.get_text(strip=True) # UNSPSC unspsc_link = soup.select_one('.views-field-field-tender-unspsc .unspsc_link_clr') if unspsc_link: full_text = unspsc_link.get_text(strip=True) # Extract code and description match = re.match(r'(\d+)\s+(.+)', full_text) if match: data['unspsc_code'] = match.group(1) data['unspsc_description'] = match.group(2) # Trade Agreements trade_agreements = [] trade_items = soup.select('.field--name-field-tender-trade-agreements .field--item .field--item') for item in trade_items: text = item.get_text(strip=True) if text: trade_agreements.append(text) if trade_agreements: data['trade_agreements'] = ' | '.join(trade_agreements) # Limited Tendering Reason ltr_items = soup.select('.field--name-field-tender-ltr .field--item .field--item') ltr_reasons = [] for item in ltr_items: text = item.get_text(strip=True) if text: ltr_reasons.append(text) if ltr_reasons: data['limited_tendering_reason'] = ' | '.join(ltr_reasons) # Contracting Organization org_name = soup.select_one('.field--name-field-tender-contact-orgname') if org_name: data['organization'] = org_name.get_text(strip=True) # Address address_line = soup.select_one('.field--name-field-tender-contact-a-line') if address_line: data['contracting_org_address'] = address_line.get_text(strip=True) # City, Province, Postal Code city_elem = soup.select_one('.field--name-field-tender-contact-a-city') if city_elem: city_text = city_elem.get_text(strip=True) # Parse "Ottawa, Ontario, K1A 0H5" format parts = [p.strip() for p in city_text.split(',')] if len(parts) >= 1: data['contracting_org_city'] = parts[0] if len(parts) >= 2: data['contracting_org_province'] = parts[1] if len(parts) >= 3: data['contracting_org_postal'] = parts[2] # Country country = soup.select_one('.field--name-field-tender-contact-a-country') if country: data['contracting_org_country'] = country.get_text(strip=True) # Second address block (parent: dl.tender-contracting) → contract_authority_address contracting_block = soup.select_one('dl.tender-contracting') if contracting_block: parts = [] addr2 = contracting_block.select_one('.field--name-field-tender-contact-a-line') if addr2: parts.append(addr2.get_text(strip=True)) city2 = contracting_block.select_one('.field--name-field-tender-contact-a-city') if city2: parts.append(city2.get_text(strip=True)) country2 = contracting_block.select_one('.field--name-field-tender-contact-a-country') if country2: parts.append(country2.get_text(strip=True)) if parts: data['contract_authority_address'] = ', '.join(parts) # Contracting Authority auth_name = soup.select_one('.field--name-field-tender-contact-contactname .field--item') if auth_name: data['contracting_authority_name'] = auth_name.get_text(strip=True) auth_phone = soup.select_one('.field--name-field-tender-contact-phone .phone-number-format') if auth_phone: data['contracting_authority_phone'] = auth_phone.get_text(strip=True) auth_email = soup.select_one('.field--name-field-tender-contact-email .field--item') if auth_email: data['contracting_authority_email'] = auth_email.get_text(strip=True) # Region(s) of delivery regions_elem = soup.select_one('.views-field-field-tender-delivery-regions .field-content') if regions_elem: # Try to get from nested field--item region_item = regions_elem.select_one('.field--item') if region_item: data['regions_of_delivery'] = region_item.get_text(strip=True) else: data['regions_of_delivery'] = regions_elem.get_text(strip=True) # Language(s) lang_elem = soup.select_one('.views-field-field-tender-notice-languages .field-content') if lang_elem: data['languages'] = lang_elem.get_text(strip=True) # Region of opportunity opp_region_elem = soup.select_one('.views-field-field-tender-opportunity-regions .field-content') if opp_region_elem: data['region_of_opportunity'] = opp_region_elem.get_text(strip=True) # Selection criteria selection_elem = soup.select_one('.views-field-field-tender-selection-criteria .field-content') if selection_elem: # Try to get from nested field--item selection_item = selection_elem.select_one('.field--item') if selection_item: data['selection_criteria'] = selection_item.get_text(strip=True) else: data['selection_criteria'] = selection_elem.get_text(strip=True) # Documents doc_rows = soup.select('.tender-documents-table tbody tr') for row in doc_rows: try: doc = {} # Document title and URL link = row.select_one('.field-document_link a') if link: doc['title'] = link.get_text(strip=True) doc['url'] = link.get('href', '') if doc['url'] and not doc['url'].startswith('http'): doc['url'] = BASE_URL + doc['url'] # Amendment number amend = row.select_one('.field-amendment_number') if amend: doc['amendment_number'] = amend.get_text(strip=True) # Language lang_cell = row.select_one('.field-language') if lang_cell: langs = [] for link in lang_cell.select('a'): langs.append(link.get_text(strip=True)) doc['language'] = ' and '.join(langs) if langs else lang_cell.get_text(strip=True) # Download count dl_count = row.select_one('.field-download_count') if dl_count: try: doc['download_count'] = int(dl_count.get_text(strip=True)) except: doc['download_count'] = 0 # Date added date_elem = row.select_one('.field-date_added') if date_elem: doc['date_added'] = parse_date(date_elem.get_text(strip=True)) if doc.get('title'): data['documents'].append(doc) except Exception as e: print(f" Warning: Error parsing document row: {e}") continue return data def fetch_tender_detail(url): """Fetch tender detail page""" try: full_url = BASE_URL + url if not url.startswith('http') else url response = requests.get(full_url, headers=HEADERS, timeout=30) if response.status_code == 200: return response.text else: print(f" ✗ HTTP {response.status_code}") return None except Exception as e: print(f" ✗ Request error: {e}") return None def save_tender_details(tender_id, detail_data, closing_date, conn, db_lock): """Save detail data to database using shared connection with lock""" try: deadline_dt, deadline_tz = build_deadline_datetime(closing_date, detail_data['closing_time']) fields = ( detail_data['solicitation_number'], detail_data['closing_time'], detail_data['amendment_date'], detail_data['contract_duration'], detail_data['contract_duration_full'], detail_data['contract_start_date'], detail_data['notice_type'], detail_data['procurement_method'], detail_data['unspsc_code'], detail_data['unspsc_description'], detail_data['trade_agreements'], detail_data['limited_tendering_reason'], detail_data['contracting_org_address'], detail_data['contracting_org_city'], detail_data['contracting_org_province'], detail_data['contracting_org_postal'], detail_data['contracting_org_country'], detail_data['contracting_authority_name'], detail_data['contracting_authority_phone'], detail_data['contracting_authority_email'], detail_data['regions_of_delivery'], detail_data['languages'], detail_data['region_of_opportunity'], detail_data['selection_criteria'], detail_data['node_id'], detail_data['contract_authority_address'], ) with db_lock: cursor = conn.cursor() cursor.execute( """UPDATE tenders SET description=%s, organization=%s, detail=1, status=%s, status_text=%s, updated_at=NOW() WHERE id=%s""", (detail_data['description'], detail_data['organization'], detail_data['status'], detail_data['status_text'], tender_id) ) cursor.execute( "SELECT id FROM tender_details WHERE tender_id=%s LIMIT 1", (tender_id,) ) det_row = cursor.fetchone() if det_row: cursor.execute( """UPDATE tender_details SET solicitation_number=%s, closing_time=%s, amendment_date=%s, contract_duration=%s, contract_duration_full=%s, contract_start_date=%s, notice_type=%s, procurement_method=%s, unspsc_code=%s, unspsc_description=%s, trade_agreements=%s, limited_tendering_reason=%s, contracting_org_address=%s, contracting_org_city=%s, contracting_org_province=%s, contracting_org_postal=%s, contracting_org_country=%s, contracting_authority_name=%s, contact_phone=%s, contracting_authority_email=%s, regions_of_delivery=%s, languages=%s, region_of_opportunity=%s, selection_criteria=%s, node_id=%s, contract_authority_address=%s, deadline=%s, deadline_timezone=%s, publication_date=open_date, updated_at=NOW() WHERE tender_id=%s""", fields + (deadline_dt, deadline_tz, tender_id,) ) else: cursor.execute( """INSERT INTO tender_details (tender_id, solicitation_number, closing_time, amendment_date, contract_duration, contract_duration_full, contract_start_date, notice_type, procurement_method, unspsc_code, unspsc_description, trade_agreements, limited_tendering_reason, contracting_org_address, contracting_org_city, contracting_org_province, contracting_org_postal, contracting_org_country, contracting_authority_name, contact_phone, contracting_authority_email, regions_of_delivery, languages, region_of_opportunity, selection_criteria, node_id, contract_authority_address, deadline, deadline_timezone, created_at, updated_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,NOW(),NOW())""", (tender_id,) + fields + (deadline_dt, deadline_tz,) ) doc_count = 0 for doc in detail_data['documents']: cursor.execute( "SELECT id FROM tender_documents WHERE tender_id=%s AND document_url=%s LIMIT 1", (tender_id, doc.get('url')) ) if cursor.fetchone(): continue cursor.execute( """INSERT INTO tender_documents (tender_id, document_name, document_url, amendment_number, language, download_count, date_added) VALUES (%s,%s,%s,%s,%s,%s,%s)""", (tender_id, doc.get('title'), doc.get('url'), doc.get('amendment_number'), doc.get('language'), doc.get('download_count', 0), doc.get('date_added')) ) doc_count += 1 conn.commit() cursor.close() return True, doc_count except Exception as e: print(f" ✗ Database error: {e}") return False, 0 def process_record(record, conn, db_lock, print_lock, counters, total): """Fetch, parse and save a single tender record (runs in thread)""" tender_id = record['id'] url = record['url'] title = record['title'] closing_date = record.get('closing_date') html = fetch_tender_detail(url) if not html: with print_lock: counters['errors'] += 1 counters['done'] += 1 print(f" [{counters['done']}/{total}] ✗ {tender_id} — fetch failed") return try: detail_data = extract_detail_data(html) saved, doc_count = save_tender_details(tender_id, detail_data, closing_date, conn, db_lock) with print_lock: counters['done'] += 1 if saved: counters['success'] += 1 counters['docs'] += doc_count org = detail_data.get('organization') or '' print(f" [{counters['done']}/{total}] ✓ {tender_id} | {title[:50]} | {org} | docs={doc_count}") else: counters['errors'] += 1 print(f" [{counters['done']}/{total}] ✗ {tender_id} — save failed") if counters['done'] % 50 == 0: print(f"\n{'─' * 80}") print(f"Progress: {counters['done']}/{total} | Success: {counters['success']} | Errors: {counters['errors']} | Docs: {counters['docs']}") print(f"{'─' * 80}\n") except Exception as e: with print_lock: counters['errors'] += 1 counters['done'] += 1 print(f" [{counters['done']}/{total}] ✗ {tender_id} — parse error: {e}") def scrape_tender_details(batch_size=None, start_from=0, workers=5): """Main function to scrape tender details""" print("=" * 80) print("Canada Buys - Tender Detail Scraper") print("=" * 80) conn = get_db_connection() cursor = conn.cursor(dictionary=True) cursor.execute( "SELECT COUNT(*) as total FROM tenders WHERE source=%s AND detail=0 AND url LIKE '%%/en/tender-opportunities/%%'", (SOURCE,) ) total = cursor.fetchone()['total'] print(f"Total records to process: {total:,}") query = "SELECT id, source_id AS tender_id, url, title, closing_date FROM tenders WHERE source=%s AND detail=0 AND url LIKE '%/en/tender-opportunities/%'" params = [SOURCE] if start_from > 0: query += " AND id > %s" params.append(start_from) query += " ORDER BY id" if batch_size: query += " LIMIT %s" params.append(batch_size) cursor.execute(query, params) records = cursor.fetchall() cursor.close() total_records = len(records) print(f"Processing {total_records} records with {workers} workers...") print("=" * 80) db_lock = threading.Lock() print_lock = threading.Lock() counters = {'done': 0, 'success': 0, 'errors': 0, 'docs': 0} with ThreadPoolExecutor(max_workers=workers) as executor: futures = [ executor.submit(process_record, record, conn, db_lock, print_lock, counters, total_records) for record in records ] for future in as_completed(futures): try: future.result() except Exception as e: print(f" ✗ Unexpected thread error: {e}") conn.close() print("\n" + "=" * 80) print("Scraping Complete!") print("=" * 80) print(f" Total processed: {total_records}") print(f" Successful: {counters['success']}") print(f" Errors: {counters['errors']}") print(f" Documents saved: {counters['docs']}") print("=" * 80) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description='Scrape Canada Buys tender details') parser.add_argument('--batch-size', type=int, default=None, help='Number of records to process') parser.add_argument('--start-from', type=int, default=0, help='Start from record ID') parser.add_argument('--workers', type=int, default=5, help='Number of concurrent workers (default: 5)') args = parser.parse_args() try: scrape_tender_details(batch_size=args.batch_size, start_from=args.start_from, workers=args.workers) except KeyboardInterrupt: print("\n\n✗ Interrupted by user") except Exception as e: print(f"\n✗ Fatal Error: {e}") import traceback traceback.print_exc()