import json import re import sys import requests import mysql.connector from html.parser import HTMLParser from datetime import date, datetime from typing import List, Dict, Optional # CONFIG URL = "https://www.ml.com/publish/mkt/prospectus/prospectus.htm" LIST_DT = date.today() DEBUG = False TICKER_RE = re.compile(r"\(([A-Z0-9]{3,8})\)\s*$") MONTH_HEADER_RE = re.compile( r"^(January|February|March|April|May|June|July|August|September|October|November|December)\s+\d{4}$" ) MONTH_NAMES = [ "January", "February", "March", "April", "May", "June", "July", "August", "September", "October", "November", "December" ] def log_info(msg: str): print(f"[INFO] {msg}") def log_debug(msg: str): if DEBUG: print(f"[DEBUG] {msg}") def log_error(msg: str): print(f"[ERROR] {msg}") # CUSTOM HTML PARSER class MLProspectusParser(HTMLParser): def __init__(self): super().__init__() self.current_tag = None self.current_attrs = {} self.text_buffer = "" self.elements = [] # List of (tag, attrs, text) def handle_starttag(self, tag, attrs): if self.current_tag: self.elements.append(( self.current_tag, self.current_attrs, self.text_buffer.strip() )) self.current_tag = tag self.current_attrs = dict(attrs) self.text_buffer = "" def handle_endtag(self, tag): if self.current_tag: self.elements.append(( self.current_tag, self.current_attrs, self.text_buffer.strip() )) self.current_tag = None self.current_attrs = {} self.text_buffer = "" def handle_data(self, data): self.text_buffer += data def get_elements(self): if self.current_tag: self.elements.append(( self.current_tag, self.current_attrs, self.text_buffer.strip() )) return self.elements def extract_month_records(html_content: str, month_label: str) -> List[Dict]: parser = MLProspectusParser() parser.feed(html_content) elements = parser.get_elements() header_idx = None for idx, (tag, attrs, text) in enumerate(elements): if tag in ["strong", "h1", "h2", "h3", "p"] and text == month_label: header_idx = idx break if header_idx is None: log_error(f"Month header not found: {month_label}") return [] log_debug(f"Found header for {month_label}") records = [] current_sym_type = None for idx in range(header_idx + 1, len(elements)): tag, attrs, text = elements[idx] if tag in ["strong", "h1", "h2", "h3", "p"]: if MONTH_HEADER_RE.match(text) and text != month_label: break if "Equity-linked" in text: current_sym_type = "Equity-linked" continue if "Commodity-linked" in text: current_sym_type = "Commodity-linked" continue if current_sym_type is None: continue if tag == "a": m = TICKER_RE.search(text) if m: records.append({ "sym_ticker": m.group(1), "sym_details": text, "sym_type": current_sym_type }) return records now = datetime.now() CURRENT_YEAR = now.year CURRENT_MONTH = now.month if CURRENT_YEAR < 2026: log_error(f"System year is {CURRENT_YEAR}. This scraper is restricted to 2026+ only. Aborting.") sys.exit(1) # DB CONNECTION try: with open("../db-config.json", "r") as f: db_cfg = json.load(f) conn = mysql.connector.connect(**db_cfg) cursor = conn.cursor() log_info("Connected to database") except Exception as e: log_error(f"Database connection failed: {e}") sys.exit(1) # FETCH PAGE try: log_info("Fetching ML prospectus page") resp = requests.get(URL, timeout=30) resp.raise_for_status() html_content = resp.text except Exception as e: log_error(f"Failed to fetch page: {e}") sys.exit(1) months_to_process = [] for month_num in range(1, CURRENT_MONTH + 1): label = f"{MONTH_NAMES[month_num - 1]} {CURRENT_YEAR}" mon_dt = date(CURRENT_YEAR, month_num, 1) source_name = f"MLI_{CURRENT_YEAR}_{month_num:02d}" months_to_process.append({ "label": label, "mon_dt": mon_dt, "source": source_name, "month_num": month_num }) log_info(f"Derived months: {', '.join(m['label'] for m in months_to_process)}") # MAIN LOOP for m in months_to_process: is_current_month = (m["month_num"] == CURRENT_MONTH) cursor.execute( "SELECT last_sym_ticker, month_closed FROM fin_scrape_state WHERE source_name = %s", (m["source"],) ) row = cursor.fetchone() last_seen = row[0] if row else None month_closed = row[1] if row else 0 log_debug(f"State → last_seen={last_seen}, month_closed={month_closed}") if not is_current_month and month_closed == 1: log_info(f"Skipping finalized month: {m['label']}") continue log_info(f"Processing {'open' if is_current_month else 'new'} month: {m['label']}") records = extract_month_records(html_content, m["label"]) log_info(f"Extracted {len(records)} records") if not records: continue rows_to_insert = [] for rec in reversed(records): if last_seen and rec["sym_ticker"] == last_seen: log_info("Reached sentinel; stopping delta scan") break rows_to_insert.append(( m["mon_dt"], rec["sym_type"], rec["sym_ticker"], rec["sym_details"], LIST_DT )) # Batch insert if rows_to_insert: try: cursor.executemany(""" INSERT INTO fin_security (mon_dt, sym_type, sym_ticker, sym_details, list_dt) VALUES (%s, %s, %s, %s, %s) """, rows_to_insert) log_info(f"Inserted {len(rows_to_insert)} new records") except Exception as e: conn.rollback() log_error(f"Insert failed for {m['label']}: {e}") continue else: log_info("No new records to insert") # Update sentinel latest_site_ticker = records[-1]["sym_ticker"] should_close_month = (not is_current_month) cursor.execute(""" INSERT INTO fin_scrape_state (source_name, last_sym_ticker, month_closed, updated_at) VALUES (%s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE last_sym_ticker = VALUES(last_sym_ticker), month_closed = VALUES(month_closed), updated_at = NOW() """, (m["source"], latest_site_ticker, 1 if should_close_month else 0)) conn.commit() log_debug(f"Updated sentinel → {latest_site_ticker}") cursor.close() conn.close()