import re import sys import json import requests import mysql.connector from datetime import datetime, date from html.parser import HTMLParser # 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})\)$") MONTH_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" ] # LOGGING def log_info(msg): print(f"[INFO] {msg}") def log_warn(msg): print(f"[WARN] {msg}") def log_error(msg): print(f"[ERROR] {msg}") def log_debug(msg): if DEBUG: print(f"[DEBUG] {msg}") now = datetime.now() if now.year < 2026: log_error(f"System year {now.year} < 2026. Aborting.") sys.exit(1) # DB CONNECTION try: with open("../db-config.json") 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) last_cursor_by_type = {} for sym_type in ("Equity-linked", "Commodity-linked"): cursor.execute(""" SELECT list_dt, sym_ticker FROM fin_security WHERE sym_type = %s ORDER BY idfin_security DESC LIMIT 1 """, (sym_type,)) row = cursor.fetchone() last_cursor_by_type[sym_type] = { "last_list_dt": row[0] if row else None, "last_ticker": row[1] if row else None } log_info( f"Last {sym_type} cursor : " f"date={last_cursor_by_type[sym_type]['last_list_dt']}, " f"ticker={last_cursor_by_type[sym_type]['last_ticker']}" ) # FETCH PAGE try: log_info("Fetching ML prospectus page") html = requests.get(URL, timeout=30).text except Exception as e: log_error(f"Failed to fetch page: {e}") sys.exit(1) # HTML PARSER (NO BS4) class MLParser(HTMLParser): def __init__(self): super().__init__() self.current_month = None self.current_year = None self.current_sym_type = None self.records = [] self.capture_text = False self.text_buffer = "" def handle_starttag(self, tag, attrs): if tag == "a": self.capture_text = True self.text_buffer = "" def handle_endtag(self, tag): if tag == "a" and self.capture_text: text = self.text_buffer.strip() self.capture_text = False m = TICKER_RE.search(text) if m and self.current_month and self.current_sym_type: self.records.append({ "year": self.current_year, "month": self.current_month, "sym_type": self.current_sym_type, "sym_ticker": m.group(1), "sym_details": text }) def handle_data(self, data): data = data.strip() if not data: return m = MONTH_RE.match(data) if m: self.current_month = MONTH_NAMES.index(m.group(1)) + 1 self.current_year = int(m.group(2)) self.current_sym_type = None log_debug(f"Detected month header: {data}") return if "Equity-linked" in data: self.current_sym_type = "Equity-linked" return if "Commodity-linked" in data: self.current_sym_type = "Commodity-linked" return if self.capture_text: self.text_buffer += data # PARSE HTML parser = MLParser() parser.feed(html) records = parser.records if not records: log_warn("No records extracted from page") sys.exit(0) # Determine earliest cursor date across sym_types cursor_dates = [ v["last_list_dt"] for v in last_cursor_by_type.values() if v["last_list_dt"] is not None ] if cursor_dates: earliest_dt = min(cursor_dates) start_year = earliest_dt.year start_month = earliest_dt.month else: start_year = records[0]["year"] start_month = records[0]["month"] log_info(f"Starting processing from {start_year}-{start_month:02d}") # insert_rows = [] # found_cursor_by_type = { # "Equity-linked": last_cursor_by_type["Equity-linked"]["last_ticker"] is None, # "Commodity-linked": last_cursor_by_type["Commodity-linked"]["last_ticker"] is None # } # # for rec in records: # sym_type = rec["sym_type"] # if (rec["year"], rec["month"]) < (start_year, start_month): # continue # # if not found_cursor_by_type[sym_type]: # if rec["sym_ticker"] == last_cursor_by_type[sym_type]["last_ticker"]: # found_cursor_by_type[sym_type] = True # log_info( # f"Found last {sym_type} ticker: " # f"{last_cursor_by_type[sym_type]['last_ticker']}" # ) # continue # # insert_rows.append(( # date(rec["year"], rec["month"], 1), # sym_type, # rec["sym_ticker"], # rec["sym_details"], # LIST_DT # )) insert_rows = [] found_cursor_by_type = { "Equity-linked": last_cursor_by_type["Equity-linked"]["last_ticker"] is None, "Commodity-linked": last_cursor_by_type["Commodity-linked"]["last_ticker"] is None } for rec in records: sym_type = rec["sym_type"] if sym_type not in ("Equity-linked", "Commodity-linked"): continue if (rec["year"], rec["month"]) < (start_year, start_month): continue cursor_dt = last_cursor_by_type[sym_type]["last_list_dt"] if cursor_dt: cursor_month = (cursor_dt.year, cursor_dt.month) record_month = (rec["year"], rec["month"]) if record_month > cursor_month: insert_rows.append(( date(rec["year"], rec["month"], 1), sym_type, rec["sym_ticker"], rec["sym_details"], LIST_DT )) continue if not found_cursor_by_type[sym_type]: if rec["sym_ticker"] == last_cursor_by_type[sym_type]["last_ticker"]: found_cursor_by_type[sym_type] = True log_info( f"Found last {sym_type} ticker: " f"{rec['sym_ticker']}" ) continue insert_rows.append(( date(rec["year"], rec["month"], 1), sym_type, rec["sym_ticker"], rec["sym_details"], LIST_DT )) # INSERT if insert_rows: cursor.executemany(""" INSERT INTO fin_security (mon_dt, sym_type, sym_ticker, sym_details, list_dt) VALUES (%s, %s, %s, %s, %s) """, insert_rows) conn.commit() log_info(f"Inserted {len(insert_rows)} new records") else: log_info("No new records to insert") cursor.close() conn.close() log_info("Scraping completed successfully")