kussssh/IPO-Analyzer
0
1"""2SEBI IPO Document Scraper — Scrapes DRHP, RHP, and Prospectus listings from SEBI.3Cross-references companies across all 3 document types to build a unified IPO tracker.4Filters out addendum, abridged, and corrigendum documents.5Supports FULL PAGINATION — fetches ALL entries across all pages, sorted latest-first.6"""7 8import re9import time10import datetime11import requests12from bs4 import BeautifulSoup13from typing import List, Dict, Optional, Set, Tuple14from urllib.parse import urlencode, urlparse, parse_qs, urlunparse15from pathlib import Path16 17from backend.config import UPLOAD_DIR18 19SEBI_BASE = "https://www.sebi.gov.in"20 21SEBI_LISTING_URLS = {22 "drhp": "https://www.sebi.gov.in/sebiweb/home/HomeAction.do?doListing=yes&sid=3&ssid=15&smid=10",23 "rhp": "https://www.sebi.gov.in/sebiweb/home/HomeAction.do?doListing=yes&sid=3&ssid=15&smid=11",24 "prospectus": "https://www.sebi.gov.in/sebiweb/home/HomeAction.do?doListing=yes&sid=3&ssid=15&smid=12",25}26 27EXCLUDE_KEYWORDS = ["addendum", "abridged", "corrigendum"]28 29NAME_SUFFIXES = [30 " - drhp",31 " - draft red herring prospectus",32 " -drhp",33 " - rhp",34 " - red herring prospectus",35 " -rhp",36 " - prospectus",37 " -prospectus",38 " - draft abridged prospectus",39 " drhp",40 " rhp",41]42 43HEADERS = {44 "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "45 "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",46 "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",47}48 49MAX_PAGES = 500 # Cap increased for full coverage50 51 52# ─────────────────────────────────────────────53# Helpers54# ─────────────────────────────────────────────55 56 57def normalize_company_name(name: str) -> str:58 n = name.strip().lower()59 for suffix in NAME_SUFFIXES:60 if n.endswith(suffix):61 n = n[: -len(suffix)]62 n = re.sub(r"\s*(limited|ltd\.?)\s*$", "", n)63 n = re.sub(r"[^\w\s]", " ", n)64 n = re.sub(r"\s+", " ", n).strip()65 return n66 67 68def _parse_date(date_str: str) -> str:69 """Return a YYYY-MM-DD sortable string from various SEBI date formats."""70 if not date_str:71 return "0000-00-00"72 for fmt in ("%b %d, %Y", "%d-%b-%Y", "%d/%m/%Y", "%B %d, %Y", "%d %b %Y"):73 try:74 return datetime.datetime.strptime(date_str.strip(), fmt).strftime(75 "%Y-%m-%d"76 )77 except ValueError:78 pass79 return date_str # return raw if unparseable80 81 82def _absolute_url(href: str) -> str:83 if href.startswith("http"):84 return href85 return SEBI_BASE + ("" if href.startswith("/") else "/") + href86 87 88def _update_query_param(url: str, key: str, value: str) -> str:89 parsed = urlparse(url)90 params = parse_qs(parsed.query)91 params[key] = [value]92 return urlunparse(parsed._replace(query=urlencode(params, doseq=True)))93 94 95def _page_signature(entries: List[Dict]) -> Tuple:96 """Fingerprint top-5 URLs to detect duplicate/repeated pages."""97 return tuple(e["detail_url"] for e in entries[:5])98 99 100def _extract_total_records(soup: BeautifulSoup) -> Optional[int]:101 text = soup.get_text(" ", strip=True)102 m = re.search(r"\d+\s+to\s+\d+\s+of\s+([\d,]+)\s+records?", text, re.IGNORECASE)103 if m:104 return int(m.group(1).replace(",", ""))105 m = re.search(r"total\s*:?\s*([\d,]+)", text, re.IGNORECASE)106 if m:107 return int(m.group(1).replace(",", ""))108 return None109 110 111def _parse_table_entries(soup: BeautifulSoup) -> List[Dict]:112 """Parse one page's table into a list of entry dicts."""113 table = soup.find("table", class_="table")114 if not table:115 tables = soup.find_all("table")116 table = tables[0] if tables else None117 if not table:118 return _fallback_link_entries(soup)119 120 entries = []121 for row in table.find_all("tr"):122 cells = row.find_all("td")123 if len(cells) < 2:124 continue125 126 date_str = cells[0].get_text(strip=True)127 title_cell = cells[1]128 links = title_cell.find_all("a", href=True)129 if not links:130 continue131 132 first_link = links[0]133 name = first_link.get_text(strip=True)134 href = first_link["href"]135 136 if not name or len(name) < 3:137 continue138 if any(kw in name.lower() for kw in EXCLUDE_KEYWORDS):139 continue140 141 full_url = _absolute_url(href)142 entries.append(143 {144 "name": name,145 "date": date_str,146 "detail_url": full_url,147 "is_pdf": href.lower().endswith(".pdf"),148 "normalized_name": normalize_company_name(name),149 }150 )151 return entries152 153 154def _fallback_link_entries(soup: BeautifulSoup) -> List[Dict]:155 entries = []156 for link in soup.find_all("a", href=True):157 href = link["href"]158 name = link.get_text(strip=True)159 if not name or len(name) < 3:160 continue161 if "javascript:" in href:162 continue163 if not (164 "/filings/" in href or "/sebi_data/" in href or "HomeAction.do" in href165 ):166 continue167 if any(kw in name.lower() for kw in EXCLUDE_KEYWORDS):168 continue169 entries.append(170 {171 "name": name,172 "date": "",173 "detail_url": _absolute_url(href),174 "is_pdf": href.lower().endswith(".pdf"),175 "normalized_name": normalize_company_name(name),176 }177 )178 return entries179 180 181def _try_fetch_page(182 session: requests.Session, page_no: int, doc_type: str183) -> Optional[str]:184 """185 Fetch page N using the actual SEBI AJAX pagination endpoint.186 page_no is 1-indexed.187 """188 ajax_url = "https://www.sebi.gov.in/sebiweb/ajax/home/getnewslistinfo.jsp"189 smid_map = {"drhp": "10", "rhp": "11", "prospectus": "12"}190 191 payload = {192 "nextValue": str(page_no - 1), # SEBI is 0-indexed internally193 "next": "n",194 "search": "",195 "fromDate": "",196 "toDate": "",197 "deptId": "-1",198 "sid": "3",199 "ssid": "15",200 "smid": smid_map.get(doc_type, "10"),201 }202 203 try:204 resp = session.post(ajax_url, headers=HEADERS, data=payload, timeout=30)205 if resp.ok and len(resp.text) > 500:206 return resp.text207 except Exception as e:208 print(f"Error fetching page {page_no}: {e}")209 pass210 211 return None212 213 214# ─────────────────────────────────────────────215# Main scraper with full pagination216# ─────────────────────────────────────────────217 218 219def scrape_sebi_listing(doc_type: str) -> List[Dict]:220 """221 Scrape ALL pages from a SEBI listing (DRHP / RHP / Prospectus) using AJAX pagination.222 """223 base_url = SEBI_LISTING_URLS.get(doc_type)224 if not base_url:225 print(f"❌ Unknown doc_type: {doc_type}")226 return []227 228 session = requests.Session()229 session.headers.update(HEADERS)230 231 # ── Page 1 ──232 try:233 resp = session.get(base_url, timeout=30)234 resp.raise_for_status()235 except Exception as e:236 print(f"Failed to fetch SEBI {doc_type.upper()} main page: {e}")237 return []238 239 soup1 = BeautifulSoup(resp.text, "html.parser")240 total_records = _extract_total_records(soup1)241 242 # Actually fetch page 1 from AJAX to be consistent243 page1_html = _try_fetch_page(session, 1, doc_type)244 if page1_html:245 page1_entries = _parse_table_entries(BeautifulSoup(page1_html, "html.parser"))246 else:247 page1_entries = _parse_table_entries(soup1)248 249 page_size = max(len(page1_entries), 25)250 251 if total_records and total_records > 0:252 total_pages = min(MAX_PAGES, -(-total_records // page_size)) # ceil division253 else:254 total_pages = MAX_PAGES255 256 print(257 f"[{doc_type.upper()}] Total records detected: {total_records or 'unknown'} | Pages estimated: {total_pages}"258 )259 260 seen_urls: Set[str] = set()261 seen_sigs: Set[Tuple] = set()262 all_entries: List[Dict] = []263 264 # Initial page additions265 sig = _page_signature(page1_entries)266 seen_sigs.add(sig)267 for e in page1_entries:268 if e["detail_url"] not in seen_urls:269 seen_urls.add(e["detail_url"])270 all_entries.append(e)271 272 # ── Pages 2..N ──273 consecutive_empty = 0274 for page_no in range(2, total_pages + 1):275 html = _try_fetch_page(session, page_no, doc_type)276 if not html:277 consecutive_empty += 1278 if consecutive_empty >= 3:279 print(f" Stopping at page {page_no} due to 3 consecutive empty pages")280 break281 continue282 283 soup = BeautifulSoup(html, "html.parser")284 page_entries = _parse_table_entries(soup)285 286 if not page_entries:287 consecutive_empty += 1288 if consecutive_empty >= 3:289 print(f" Stopping at page {page_no} due to 3 consecutive empty pages")290 break291 continue292 293 sig = _page_signature(page_entries)294 if sig in seen_sigs:295 print(f" Page {page_no}: duplicate detected — pagination exhausted")296 break297 298 seen_sigs.add(sig)299 consecutive_empty = 0300 new_count = 0301 for e in page_entries:302 if e["detail_url"] not in seen_urls:303 seen_urls.add(e["detail_url"])304 all_entries.append(e)305 new_count += 1306 307 print(308 f" Page {page_no}: {len(page_entries)} entries | {new_count} new | "309 f"Running total: {len(all_entries)}"310 )311 time.sleep(0.4) # polite scraping delay312 313 # ── Sort latest-first ──314 all_entries.sort(key=lambda e: _parse_date(e.get("date", "")), reverse=True)315 316 print(317 f"[{doc_type.upper()}] {len(all_entries)} total entries scraped (all pages, latest first)"318 )319 return all_entries320 321 322# ── Convenience wrappers ──323 324 325def scrape_drhp_list() -> List[Dict]:326 return scrape_sebi_listing("drhp")327 328 329def scrape_rhp_list() -> List[Dict]:330 return scrape_sebi_listing("rhp")331 332 333def scrape_prospectus_list() -> List[Dict]:334 return scrape_sebi_listing("prospectus")335 336 337# ─────────────────────────────────────────────338# IPO Tracker339# ─────────────────────────────────────────────340 341 342def build_ipo_tracker() -> List[Dict]:343 """344 Scrape all 3 SEBI listing pages with full pagination and cross-reference345 companies to build a unified IPO tracker, sorted latest-date-first.346 """347 print("Building IPO Tracker - scraping ALL pages from SEBI...")348 349 drhp_entries = scrape_drhp_list()350 rhp_entries = scrape_rhp_list()351 prospectus_entries = scrape_prospectus_list()352 353 rhp_map: Dict[str, Dict] = {}354 for e in rhp_entries:355 rhp_map.setdefault(e["normalized_name"], e)356 357 prospectus_map: Dict[str, Dict] = {}358 for e in prospectus_entries:359 prospectus_map.setdefault(e["normalized_name"], e)360 361 tracker: List[Dict] = []362 seen: Set[str] = set()363 364 # DRHP-anchored entries365 for drhp in drhp_entries:366 norm = drhp["normalized_name"]367 if norm in seen:368 continue369 seen.add(norm)370 371 rhp = rhp_map.get(norm)372 prosp = prospectus_map.get(norm)373 stage = "prospectus" if prosp else ("rhp" if rhp else "drhp")374 375 latest_date = (prosp or rhp or drhp).get("date", "")376 377 tracker.append(378 {379 "company_name": _clean_display_name(drhp["name"]),380 "normalized_name": norm,381 "drhp": _entry_summary(drhp),382 "rhp": _entry_summary(rhp),383 "prospectus": _entry_summary(prosp),384 "stage": stage,385 "latest_date": latest_date,386 }387 )388 389 # RHP-only entries390 for entry in rhp_entries:391 norm = entry["normalized_name"]392 if norm in seen:393 continue394 seen.add(norm)395 prosp = prospectus_map.get(norm)396 tracker.append(397 {398 "company_name": _clean_display_name(entry["name"]),399 "normalized_name": norm,400 "drhp": None,401 "rhp": _entry_summary(entry),402 "prospectus": _entry_summary(prosp),403 "stage": "prospectus" if prosp else "rhp",404 "latest_date": (prosp or entry).get("date", ""),405 }406 )407 408 # Prospectus-only entries409 for entry in prospectus_entries:410 norm = entry["normalized_name"]411 if norm in seen:412 continue413 seen.add(norm)414 tracker.append(415 {416 "company_name": _clean_display_name(entry["name"]),417 "normalized_name": norm,418 "drhp": None,419 "rhp": None,420 "prospectus": _entry_summary(entry),421 "stage": "prospectus",422 "latest_date": entry.get("date", ""),423 }424 )425 426 # Sort latest-first427 tracker.sort(key=lambda c: _parse_date(c.get("latest_date", "")), reverse=True)428 429 print(f"✅ IPO Tracker: {len(tracker)} unique companies (sorted by latest date)")430 return tracker431 432 433def _clean_display_name(name: str) -> str:434 cleaned = name.strip()435 cleaned = re.sub(436 r"\s*-\s*(DRHP|RHP|Prospectus|Draft\s+Abridged\s+Prospectus)\s*$",437 "",438 cleaned,439 flags=re.IGNORECASE,440 )441 if cleaned == cleaned.upper() and len(cleaned) > 5:442 cleaned = cleaned.title()443 return cleaned.strip()444 445 446def _entry_summary(entry: Optional[Dict]) -> Optional[Dict]:447 if not entry:448 return None449 return {450 "name": entry["name"],451 "date": entry["date"],452 "detail_url": entry["detail_url"],453 "is_pdf": entry["is_pdf"],454 }455 456 457# ─────────────────────────────────────────────458# PDF Discovery & Download459# ─────────────────────────────────────────────460 461 462def find_pdf_on_detail_page(detail_url: str) -> Optional[str]:463 """Navigate to a SEBI detail page and find the embedded PDF link."""464 try:465 resp = requests.get(detail_url, headers=HEADERS, timeout=30)466 resp.raise_for_status()467 except Exception as e:468 print(f"❌ Failed to fetch detail page: {e}")469 return None470 471 raw_html = resp.text472 473 # Strategy 1: iframe ?file= param474 file_param = re.findall(475 r'[?&]file=(https?://[^"\'&\s]+\.pdf)', raw_html, re.IGNORECASE476 )477 if file_param:478 print(f" ✅ Found PDF via ?file= param")479 return file_param[0]480 481 soup = BeautifulSoup(raw_html, "html.parser")482 483 # Strategy 2: iframe src484 for iframe in soup.find_all("iframe"):485 src = iframe.get("src", "")486 if ".pdf" in src.lower():487 m = re.search(r"[?&]file=(https?://[^&\s]+)", src)488 if m:489 return m.group(1)490 if src.startswith("http"):491 return src492 return SEBI_BASE + src493 494 # Strategy 3: anchor tags495 for a in soup.find_all("a", href=True):496 href = a["href"]497 if ".pdf" in href.lower() and "sebi_data" in href.lower():498 return _absolute_url(href)499 500 # Strategy 4: raw regex in HTML501 pdf_refs = re.findall(502 r'(https?://www\.sebi\.gov\.in/sebi_data/[^"\'&\s]+\.pdf)',503 raw_html,504 re.IGNORECASE,505 )506 if pdf_refs:507 return pdf_refs[0]508 509 print(f" ❌ Could not find PDF for: {detail_url}")510 return None511 512 513def download_pdf(pdf_url: str, filename: str, max_retries: int = 5) -> str:514 """Download a PDF from SEBI with resumable retry logic. Returns local file path."""515 pdf_path = UPLOAD_DIR / filename516 if pdf_path.exists() and pdf_path.stat().st_size > 10000:517 print(f"📦 PDF already downloaded: {filename}")518 return str(pdf_path)519 520 temp_path = pdf_path.with_suffix(".tmp")521 522 for attempt in range(1, max_retries + 1):523 try:524 # Resume from partial download if available — avoids re-downloading on every retry525 resume_from = temp_path.stat().st_size if temp_path.exists() else 0526 req_headers = {**HEADERS}527 if resume_from > 0:528 req_headers["Range"] = f"bytes={resume_from}-"529 print(f"⬇️ Resuming PDF from {resume_from / 1024 / 1024:.1f} MB (attempt {attempt}/{max_retries}): {pdf_url}")530 else:531 print(f"⬇️ Downloading PDF (attempt {attempt}/{max_retries}): {pdf_url}")532 533 resp = requests.get(pdf_url, headers=req_headers, timeout=300, stream=True)534 535 # Server doesn't support Range — restart fresh536 if resume_from > 0 and resp.status_code == 200:537 resume_from = 0538 temp_path.unlink(missing_ok=True)539 # Range not satisfiable — server says file is already fully sent540 elif resp.status_code == 416:541 if temp_path.exists() and temp_path.stat().st_size > 10000:542 if pdf_path.exists():543 pdf_path.unlink()544 temp_path.rename(pdf_path)545 return str(pdf_path)546 temp_path.unlink(missing_ok=True)547 resume_from = 0548 continue549 550 resp.raise_for_status()551 552 total_bytes = resume_from553 write_mode = "ab" if resume_from > 0 else "wb"554 with open(temp_path, write_mode) as f:555 for chunk in resp.iter_content(chunk_size=1024 * 1024): # 1 MB chunks556 if chunk:557 f.write(chunk)558 total_bytes += len(chunk)559 560 if total_bytes < 10000:561 print(f" ⚠️ File too small ({total_bytes} bytes), retrying...")562 temp_path.unlink(missing_ok=True)563 continue564 565 if pdf_path.exists():566 pdf_path.unlink()567 temp_path.rename(pdf_path)568 print(f"✅ Downloaded: {filename} ({total_bytes / 1024 / 1024:.1f} MB)")569 return str(pdf_path)570 571 except (572 requests.exceptions.ChunkedEncodingError,573 requests.exceptions.ConnectionError,574 requests.exceptions.Timeout,575 ) as e:576 print(f" ⚠️ Attempt {attempt} failed: {e}")577 if attempt < max_retries:578 wait = min(attempt * 5, 30)579 print(f" ⏳ Retrying in {wait}s (partial file kept for resume)...")580 time.sleep(wait)581 else:582 temp_path.unlink(missing_ok=True)583 raise Exception(584 f"Failed to download PDF after {max_retries} attempts: {e}"585 )586 587 raise Exception(f"Failed to download PDF after {max_retries} attempts")588