import re import sys import json import requests import mysql.connector from datetime import datetime, date from html.parser import HTMLParser from config_loader import get_db_connection, load_scraper_config from blue_region_extractor import scrape_url, filter_bullets_by_keywords # CONFIG URL = "https://www.ml.com/publish/mkt/prospectus/prospectus.htm" LIST_DT = date.today() DEBUG = False TICKER_RE = re.compile(r"\(\s*([A-Z0-9]{3,8})\s*\)$") 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 (using config_loader module) conn = get_db_connection("../db-config.json") cursor = conn.cursor() log_info("Connected to database") config = load_scraper_config(cursor) excl_prefixes = ', '.join([f'"{p}"' for p in config['exclude_prefixes']]) if config['exclude_prefixes'] else 'None' ml_keywords = ', '.join([f'"{k}"' for k in config['keywords']]) if config['keywords'] else 'None' ml_excl_keys = ', '.join([f'"{k}"' for k in config['exclude_keywords']]) if config['exclude_keywords'] else 'None' log_info(f"Main Page Exclude: {excl_prefixes}") log_info(f"ML Include: {ml_keywords}") log_info(f"ML Exclude: {ml_excl_keys}") last_cursor_by_type = {} for sym_type in ("Equity-linked", "Commodity-linked"): cursor.execute(""" SELECT mon_dt, sym_ticker FROM fin_security_2 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: resp = requests.get(URL, timeout=30) resp.raise_for_status() html = resp.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 = "" self.capture_mode = None def handle_starttag(self, tag, attrs): if tag == "a": self.capture_text = True self.text_buffer = "" # Capture href for detail page URL self.current_href = None for attr_name, attr_value in attrs: if attr_name == "href": self.current_href = attr_value break def handle_endtag(self, tag): if tag == "a" and self.capture_text: text = self.text_buffer.strip() self.capture_text = False if not text: return m = TICKER_RE.search(text) if self.current_sym_type in ("Equity-linked", "Commodity-linked"): if m: # Only capture records from 2026 onwards if self.current_year and self.current_year >= 2026: detail_url = None if self.current_href and "sec.gov" in self.current_href: detail_url = self.current_href 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, "detail_url": detail_url }) return if self.capture_mode == "EQUITY_NEWS": self.records.append({ "year": None, "month": None, "sym_type": "Equity", "sym_ticker": None, "sym_details": text }) return if self.capture_mode == "FIXED_INCOME": self.records.append({ "year": None, "month": None, "sym_type": "Fixed Income", "sym_ticker": None, "sym_details": text }) return 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 data == "Equity New Issues": self.capture_mode = "EQUITY_NEWS" return if data == "Fixed Income": self.capture_mode = "FIXED_INCOME" return if data == "Medium-Term Notes": self.capture_mode = None self.current_sym_type = None 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) def should_exclude(record, exclude_prefixes): """Check if record should be excluded based on prefixes""" ticker = (record.get("sym_ticker") or "").upper() details = (record.get("sym_details") or "").upper() return any(ticker.startswith(p) or details.startswith(p) for p in exclude_prefixes) if config["exclude_prefixes"]: original_count = len(records) records = [r for r in records if not should_exclude(r, config["exclude_prefixes"])] filtered_count = original_count - len(records) if filtered_count > 0: log_info(f"Filtered out {filtered_count} records based on exclude prefixes") # CRITICAL: Group by month and reverse MONTH order, but keep records # within each month in their original order (top-to-bottom from website). # Website shows: Feb (top) -> Jan (bottom) # Parser creates: [Feb rec A, Feb rec B, ..., Jan rec X, Jan rec Y, ...] # After reordering: [Jan rec X, Jan rec Y, ..., Feb rec A, Feb rec B, ...] # Insertion order: Jan gets IDs 1-N, Feb gets IDs N+1-M # Within each month: top website record = lower ID, bottom = higher ID # Cursor query (ORDER BY idfin_security DESC) will get the last record # of the newest month (bottom record of Feb in this example) from collections import OrderedDict month_groups = OrderedDict() for rec in records: if rec["year"] is not None and rec["month"] is not None: key = (rec["year"], rec["month"]) else: key = (None, None) if key not in month_groups: month_groups[key] = [] month_groups[key].append(rec) dated_groups = [(k, v) for k, v in month_groups.items() if k[0] is not None] non_dated_records = month_groups.get((None, None), []) dated_groups.sort(key=lambda x: x[0]) records = [] for _, group in dated_groups: records.extend(group) records.extend(non_dated_records) 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: mli_records = [ r for r in records if r["year"] is not None and r["month"] is not None and r["year"] >= 2026 ] if not mli_records: log_error("No valid MLI month records found on page for year >= 2026") sys.exit(1) # Find the oldest month (minimum year, then minimum month) oldest = min(mli_records, key=lambda r: (r["year"], r["month"])) start_year = oldest["year"] start_month = oldest["month"] log_info(f"Starting processing from {start_year}-{start_month:02d}") # Build insert list (batch insertion) 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: if rec["sym_type"] not in ("Equity-linked", "Commodity-linked"): continue sym_type = rec["sym_type"] 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, None, None # sym_link not available before cursor )) 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 # Scrape detail page for insights (using blue_region_extractor module) insight = None if rec.get("detail_url") and config["keywords"]: log_debug(f"Scraping detail page: {rec['detail_url']}") scrape_result = scrape_url(rec["detail_url"], debug=DEBUG) if scrape_result['status'] == 'success': matching_bullets = filter_bullets_by_keywords( scrape_result['bullets'], config['keywords'], config.get('exclude_keywords', []) ) if matching_bullets: insight = " | ".join(matching_bullets) else: log_debug(f"Scraping failed: {scrape_result.get('error', 'Unknown error')}") insert_rows.append(( date(rec["year"], rec["month"], 1), sym_type, rec["sym_ticker"], rec["sym_details"], LIST_DT, insight, rec.get("detail_url") )) if insert_rows: cursor.executemany(""" INSERT IGNORE INTO fin_security_2 (mon_dt, sym_type, sym_ticker, sym_details, list_dt, sym_insight, sym_link) VALUES (%s, %s, %s, %s, %s, %s, %s) """, insert_rows) conn.commit() inserted = cursor.rowcount skipped = len(insert_rows) - inserted if inserted > 0: log_info(f"Inserted {inserted} new records") if skipped > 0: log_warn(f"Skipped {skipped} duplicate records") else: log_info("No new records to insert") log_info("Processing Equity New Issues and Fixed Income") SNAP_TYPES = ("Equity", "Fixed Income") cursor.execute(""" SELECT sym_type, sym_details, MIN(list_dt) FROM fin_security_2 WHERE sym_type IN ('Equity', 'Fixed Income') GROUP BY sym_type, sym_details """) existing = {} for sym_type, sym_details, first_seen in cursor.fetchall(): existing[(sym_type, sym_details)] = first_seen snapshot_records = [ r for r in records if r["sym_type"] in SNAP_TYPES ] snapshot_inserts = [] today = date.today() for r in snapshot_records: key = (r["sym_type"], r["sym_details"]) if key not in existing: snapshot_inserts.append(( today, r["sym_type"], None, r["sym_details"], today )) continue first_seen = existing[key] days_gap = (today - first_seen).days if days_gap > 90: log_warn(f"Reappeared after {days_gap} days to reinserting: {r['sym_details']}") snapshot_inserts.append(( today, r["sym_type"], None, r["sym_details"], today )) else: log_debug(f"Skip duplicate within 90 days: {r['sym_details']}") if snapshot_inserts: cursor.executemany(""" INSERT INTO fin_security_2 (mon_dt, sym_type, sym_ticker, sym_details, list_dt) VALUES (%s, %s, %s, %s, %s) """, snapshot_inserts) conn.commit() log_info(f"Inserted {len(snapshot_inserts)} new snapshot records") else: log_info("No new Equity/Fixed Income snapshot records to insert") cursor.close() conn.close() log_info("Scraping completed successfully")