#!/usr/bin/env python3 """ FormDS Web Scraper - Filtered Version Scrapes FormDS.com with company type exclusion filtering. Author: HemantaBhusal Version: 3.0 (Filtered) """ import os import sys import urllib.request import urllib.error from html.parser import HTMLParser import mysql.connector import time import logging import concurrent.futures from datetime import datetime sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from global_config import db_config logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s') class TableParser(HTMLParser): """Custom HTML parser to extract table data with company links""" def __init__(self): super().__init__() self.in_table = False self.in_row = False self.in_cell = False self.current_row = [] self.current_cell = "" self.rows = [] self.cell_count = 0 self.current_link = "" self.in_link = False def handle_starttag(self, tag, attrs): if tag == 'table': self.in_table = True elif tag == 'tr' and self.in_table: self.in_row = True self.current_row = [] self.cell_count = 0 elif tag == 'td' and self.in_row: self.in_cell = True self.current_cell = "" self.current_link = "" self.cell_count += 1 elif tag == 'a' and self.in_cell: self.in_link = True for attr_name, attr_value in attrs: if attr_name == 'href': self.current_link = attr_value break def handle_endtag(self, tag): if tag == 'table': self.in_table = False elif tag == 'tr' and self.in_row: self.rows.append(self.current_row) self.in_row = False elif tag == 'td' and self.in_cell: if self.cell_count == 1 and self.current_link: combined = f"{self.current_cell.strip()}|{self.current_link}" self.current_row.append(combined) else: self.current_row.append(self.current_cell.strip()) self.in_cell = False elif tag == 'a': self.in_link = False def handle_data(self, data): if self.in_cell: self.current_cell += data class CompanyDetailParser(HTMLParser): """Custom HTML parser to extract company details""" def __init__(self): super().__init__() self.in_address = False self.in_address_tag = False self.in_table = False self.in_thead = False self.in_tbody = False self.in_row = False self.in_cell = False self.address_lines = [] self.current_text = "" self.table_headers = [] self.current_row = [] self.table_data = [] self.cell_index = 0 def handle_starttag(self, tag, attrs): if tag == 'div': for attr_name, attr_value in attrs: if attr_name == 'id' and attr_value == 'address-div': self.in_address = True break elif tag == 'address' and self.in_address: self.in_address_tag = True elif tag == 'table': self.in_table = True self.table_headers = [] self.table_data = [] elif tag == 'thead' and self.in_table: self.in_thead = True elif tag == 'tbody' and self.in_table: self.in_tbody = True elif tag == 'tr': self.in_row = True self.current_row = [] self.cell_index = 0 elif tag == 'th' and self.in_row: self.in_cell = True self.current_text = "" elif tag == 'td' and self.in_row: self.in_cell = True self.current_text = "" def handle_endtag(self, tag): if tag == 'div' and self.in_address: self.in_address = False elif tag == 'address' and self.in_address_tag: self.in_address_tag = False elif tag == 'table': self.in_table = False elif tag == 'thead': self.in_thead = False elif tag == 'tbody': self.in_tbody = False elif tag == 'tr' and self.in_row: if self.in_thead and self.current_row: self.table_headers = self.current_row elif not self.in_thead and self.current_row: self.table_data.append(self.current_row) self.in_row = False elif tag in ['th', 'td'] and self.in_cell: self.current_row.append(self.current_text.strip()) self.in_cell = False self.cell_index += 1 def handle_data(self, data): if self.in_address_tag: clean_data = data.strip() if clean_data: self.address_lines.append(clean_data) elif self.in_cell: self.current_text += data def get_address(self): result = ' '.join(self.address_lines) return result def get_contacts(self): contacts = [] if self.table_headers: headers_lower = [h.lower() for h in self.table_headers] if 'name' in headers_lower and 'role' in headers_lower: for row in self.table_data: if len(row) >= 2: name = row[0].strip() role = row[1].strip().replace(', ', '/').replace(',', '/') if name and role: contacts.append(f"{name} : {role}") result = ", ".join(contacts) return result class FormDSScraper: def __init__(self, max_workers=8, delay=0.8): self.base_url = "https://www.formds.com/filings/newest" os.environ['TZ'] = 'America/New_York' time.tzset() # apply the timezone # Store current date (New York) self.current_date = datetime.now().strftime('%Y-%m-%d') #self.current_date = '2025-09-16' #Hardcoded Date self.max_workers = max_workers self.delay = delay # Load exclusion list self.excluded_types = self.load_exclusions() # Get validation data from database self.latest_company_name, self.max_order_num = self.get_latest_validation_data() logging.info(f"Filtered scraper initialized for {self.current_date}") if self.latest_company_name: logging.info(f"Latest company: {self.latest_company_name[:50]}... (order: {self.max_order_num})") else: logging.info("First run - no existing data") def load_exclusions(self): """Load excluded company types from exclude_comp.txt""" try: # Get the directory containing this script script_dir = os.path.dirname(os.path.abspath(__file__)) # Try multiple possible paths possible_paths = [ # One directory back from script (your intended structure) os.path.join(os.path.dirname(script_dir), 'exclude_comp.txt'), # Same directory as script (fallback) os.path.join(script_dir, 'exclude_comp.txt'), # Current working directory (fallback) 'exclude_comp.txt', # Absolute path from current working directory parent os.path.join(os.path.dirname(os.getcwd()), 'exclude_comp.txt') ] exclude_file_path = None for path in possible_paths: if os.path.exists(path): exclude_file_path = path break else: logging.info(f"File not found at: {path}") if not exclude_file_path: logging.warning("exclude_comp.txt not found in any expected location") return [] with open(exclude_file_path, 'r', encoding='utf-8') as f: content = f.read().strip() # Split by | and clean whitespace excluded_types = [comp_type.strip() for comp_type in content.split('|')] excluded_types = [comp_type for comp_type in excluded_types if comp_type] # Remove empty logging.info(f"Loaded {len(excluded_types)} excluded company types: {excluded_types}") return excluded_types except FileNotFoundError: logging.warning("exclude_comp.txt not found - no exclusions will be applied") return [] except Exception as e: logging.error(f"Error reading exclude_comp.txt: {e}") return [] def extract_company_type(self, company_name): """Extract company type from company name like 'Company Name (Type)'""" company_name = company_name.strip() # Look for content in parentheses at the end if company_name.endswith(')'): start_idx = company_name.rfind('(') if start_idx != -1: company_type = company_name[start_idx+1:-1].strip() return company_type return "" def is_excluded(self, company_name): """Check if company should be excluded based on its type""" if not self.excluded_types: return False company_type = self.extract_company_type(company_name) return company_type in self.excluded_types def get_latest_validation_data(self): """Get latest company name and max order number from highest order_num globally""" try: connection = mysql.connector.connect(**db_config) cursor = connection.cursor() cursor.execute(""" SELECT company, order_num FROM filings3 ORDER BY order_num DESC LIMIT 1 """) result = cursor.fetchone() if result: return result[0], result[1] # (company_name, order_num) else: return None, 0 except mysql.connector.Error as error: logging.error(f"Database error: {error}") return None, 0 finally: if connection and connection.is_connected(): cursor.close() connection.close() def make_request(self, url): """Make HTTP request using urllib""" try: req = urllib.request.Request( url, headers={ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.5', 'Connection': 'keep-alive' } ) with urllib.request.urlopen(req, timeout=10) as response: return response.read().decode('utf-8') except urllib.error.URLError as e: logging.error(f"Failed to fetch {url}: {e}") return None except Exception as e: logging.error(f"Unexpected error fetching {url}: {e}") return None def get_page(self, page_num=1): """Fetch page from FormDS and parse HTML""" try: url = f"{self.base_url}?page={page_num}" html_content = self.make_request(url) if not html_content: return None parser = TableParser() parser.feed(html_content) return parser.rows except Exception as e: logging.error(f"Failed to parse page {page_num}: {e}") return None def get_company_details(self, company_url): """Fetch single company detail page and extract address and contacts""" try: if not company_url: return "", "" full_url = f"https://www.formds.com{company_url}" html_content = self.make_request(full_url) if not html_content: return "", "" parser = CompanyDetailParser() parser.feed(html_content) address = parser.get_address() contacts = parser.get_contacts() return address, contacts except Exception as e: logging.warning(f"Failed to get details for {company_url}: {e}") return "", "" def get_company_details_batch(self, company_urls_data): """Fetch multiple company details concurrently""" if not company_urls_data: return [] results = [None] * len(company_urls_data) def fetch_single_company(index_url_tuple): index, (company_name, url) = index_url_tuple if not url: return index, ("", "") try: address, contacts = self.get_company_details(url) return index, (address, contacts) except Exception as e: logging.warning(f"Error fetching details for {company_name}: {e}") return index, ("", "") # Use ThreadPoolExecutor for concurrent requests with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor: # Submit all tasks futures = [] for i, (company_name, url) in enumerate(company_urls_data): future = executor.submit(fetch_single_company, (i, (company_name, url))) futures.append(future) if i > 0: # Add small delay between submissions time.sleep(self.delay / self.max_workers) # Collect results for future in concurrent.futures.as_completed(futures): try: index, details = future.result() results[index] = details except Exception as e: logging.error(f"Thread execution error: {e}") return results def process_company_details(self, page_data, company_urls_data): """Process company details for a batch of companies""" if not page_data or not company_urls_data: return logging.info(f"Fetching details for {len(company_urls_data)} companies concurrently...") company_details = self.get_company_details_batch(company_urls_data) # Assign details to filing data for i, (address, contacts) in enumerate(company_details): if i < len(page_data): page_data[i]['company_address'] = address page_data[i]['company_all_contact'] = contacts def scrape_pages(self, max_pages=20): """Main scraping with filtering and optimized batch processing""" page_num = 1 scraped_data = [] is_first_run = self.latest_company_name is None excluded_count = 0 mode = "first" if is_first_run else "incremental" logging.info(f"Starting {mode} scraping with filtering...") while page_num <= max_pages: rows = self.get_page(page_num) if not rows: break page_data = [] company_urls_data = [] # For batch processing for row_idx, row in enumerate(rows[1:], 1): # Skip header if len(row) >= 6: try: # Extract data with company URL handling company_data = row[0] # May contain company|link if '|' in company_data: company, company_url = company_data.split('|', 1) else: company = company_data company_url = "" # Clean company name - remove newlines and extra whitespace company = ' '.join(company.split()) filing_date = row[4] # CONDITION: Different current date → Process current page data first, then return if filing_date != self.current_date: if page_data: # Process company details for current page before returning self.process_company_details(page_data, company_urls_data) scraped_data.extend(page_data) logging.info(f"Date changed to {filing_date} - returning {len(scraped_data)} filings") logging.info(f"Total excluded {excluded_count} companies based on type filtering") return scraped_data # CONDITION: Same company (second run only) → Process current page data first, then return if not is_first_run and company == self.latest_company_name: if page_data: # Process company details for current page before returning self.process_company_details(page_data, company_urls_data) scraped_data.extend(page_data) logging.info(f"Found existing company - returning {len(scraped_data)} filings") logging.info(f"Excluded {excluded_count} companies based on type filtering") return scraped_data # Check if company type is excluded if self.is_excluded(company): excluded_count += 1 continue # Skip this company # Collect basic filing data filing_data = { 'company': company, 'reported_funding': row[2], 'incremental_cash': row[3], 'filing_date': filing_date, 'new_or_amended': row[5], 'company_address': "", # Will be filled by batch processing 'company_all_contact': "" # Will be filled by batch processing } page_data.append(filing_data) company_urls_data.append((company, company_url)) except Exception as e: logging.warning(f"Error parsing row {row_idx}: {e}") continue # Batch process company details for the entire page if page_data: self.process_company_details(page_data, company_urls_data) scraped_data.extend(page_data) # Stop if no current date data found on this page if not page_data and excluded_count == 0: logging.info(f"No current date data - returning {len(scraped_data)} filings") return scraped_data page_num += 1 time.sleep(self.delay) # Delay between pages logging.info(f"Max pages reached - returning {len(scraped_data)} filings") logging.info(f"Total excluded: {excluded_count} companies based on type filtering") return scraped_data def save_to_mysql(self, data): """Save data with proper order numbering using batch operations""" if not data: print("No data to save") return connection = None try: connection = mysql.connector.connect(**db_config) cursor = connection.cursor() insert_query = """ INSERT INTO filings3 (company, reported_funding, incremental_cash, filing_date, new_or_amended, order_num, company_address, company_all_contact) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) """ # Prepare batch data with order numbers batch_data = [] for i, filing in enumerate(data): order_num = self.max_order_num + len(data) - i batch_data.append(( filing['company'], filing['reported_funding'], filing['incremental_cash'], filing['filing_date'], filing['new_or_amended'], order_num, filing['company_address'], filing['company_all_contact'] )) # Execute batch insert cursor.executemany(insert_query, batch_data) connection.commit() if self.latest_company_name is None: print(f"Successfully saved {len(data)} records (first run)") else: print(f"Successfully saved {len(data)} new records") except mysql.connector.Error as error: print(f"MySQL Error: {error}") logging.error(f"MySQL database error during batch save: {error}") if connection: connection.rollback() except Exception as e: print(f"Save error: {e}") logging.error(f"Unexpected error during batch save: {e}") if connection: connection.rollback() finally: if connection and connection.is_connected(): cursor.close() connection.close() def main(): """Main execution""" try: # Run filtered scraper with default optimized settings scraper = FormDSScraper() scraped_data = scraper.scrape_pages(max_pages=25) if scraped_data and len(scraped_data) > 0: scraper.save_to_mysql(scraped_data) else: print("No new filings found") except Exception as e: print(f"Scraper failed: {e}") logging.error(f"Fatal error: {e}") if __name__ == "__main__": main()