Create crawler.py

crawler.py

@@ -0,0 +1,215 @@
+"""
+Habib Bank Limited (HBL) - Enterprise Knowledge RAG Pipeline
+-----------------------------------------------------------
+This production script runs completely offline on an HBL enterprise laptop.
+It deeply crawls the internal domain (including sublinks), downloads PDFs, 
+processes text/tables, and applies Semantic Topic Chunking to create a 
+local FAISS vector store.
+
+Requirements:
+pip install requests beautifulsoup4 pdfplumber langchain langchain-community langchain-experimental sentence-transformers faiss-cpu torch
+"""
+
+import os
+import re
+import time
+from collections import deque
+from urllib.parse import urljoin, urlparse
+
+# Enforce strict offline execution to bypass HBL corporate proxy/firewall checks
+os.environ["TRANSFORMERS_OFFLINE"] = "1"
+os.environ["HF_DATASETS_OFFLINE"] = "1"
+os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "1"
+
+import requests
+import torch
+import pdfplumber
+from bs4 import BeautifulSoup
+from langchain.schema import Document
+from langchain_experimental.text_splitter import SemanticChunker
+from langchain_community.embeddings import HuggingFaceEmbeddings
+from langchain_community.vectorstores import FAISS
+
+# =================================---------
+# CONFIGURATION PORTAL (EDIT AS NEEDED)
+# =================================---------
+# 1. Target links and network boundaries
+TARGET_URL = "http://iamhbl.hbl.com/"  
+ALLOWED_DOMAIN = "iamhbl.hbl.com"
+
+# 2. Crawler limits to prevent infinite loops
+MAX_PAGES = 500       # Maximum HTML pages to scan
+MAX_DEPTH = 3         # Sub-folder depth limit (e.g., Home -> HR -> Circulars)
+
+# 3. Local Model Path - Absolute path to your downloaded bge-m3 weights directory
+# e.g., "C:/Users/Ezan/Models/bge-m3" or "/home/ezan/models/bge-m3"
+LOCAL_MODEL_PATH = "C:/path/to/your/local/folder/bge-m3"  
+
+# 4. Output directories
+DOWNLOAD_DIR = "./iamhbl_downloaded_pdfs"
+INDEX_DIR = "./iamhbl_faiss_index"
+USER_AGENT = "HBL-Internal-Semantic-RAG-Pipeline/1.0"
+# =================================---------
+
+
+def get_pdf_links(seed_url, max_pages=MAX_PAGES, max_depth=MAX_DEPTH):
+    """Recursively crawls the domain and sublinks to discover all PDFs."""
+    print(f"[+] Starting deep-crawl on {seed_url} (Max Depth: {max_depth})")
+    
+    queue = deque([(seed_url, 0)])
+    visited_pages = set()
+    pdf_links = set()
+    headers = {"User-Agent": USER_AGENT}
+    
+    while queue and len(visited_pages) < max_pages:
+        url, depth = queue.popleft()
+        
+        if url in visited_pages or depth > max_depth:
+            continue
+            
+        visited_pages.add(url)
+        
+        try:
+            response = requests.get(url, headers=headers, timeout=10)
+            if response.status_code != 200:
+                continue
+                
+            content_type = response.headers.get("Content-Type", "")
+            if "application/pdf" in content_type or url.lower().endswith(".pdf"):
+                pdf_links.add(url)
+                continue
+            
+            if "text/html" not in content_type:
+                continue
+                
+            print(f"[*] Crawling [Depth {depth}]: {url}")
+            
+            soup = BeautifulSoup(response.text, "html.parser")
+            for link in soup.find_all("a", href=True):
+                next_url = urljoin(url, link["href"]).split("#")[0]
+                
+                if next_url.lower().endswith(".pdf"):
+                    pdf_links.add(next_url)
+                else:
+                    parsed_url = urlparse(next_url)
+                    if ALLOWED_DOMAIN in parsed_url.netloc and next_url not in visited_pages:
+                        queue.append((next_url, depth + 1))
+                        
+        except Exception as e:
+            print(f"[!] Warning: Failed to crawl {url}: {e}")
+
+    print(f"\n[+] Deep-Crawl Complete. Scanned {len(visited_pages)} pages.")
+    print(f"[+] Found {len(pdf_links)} unique PDF targets.")
+    
+    return list(pdf_links)
+
+
+def download_pdfs(pdf_urls, download_dir):
+    """Downloads discovered assets locally without cloud storage dependencies."""
+    os.makedirs(download_dir, exist_ok=True)
+    downloaded_paths = []
+    headers = {"User-Agent": USER_AGENT}
+    
+    for i, url in enumerate(pdf_urls):
+        filename = os.path.join(download_dir, f"hbl_asset_{i+1}.pdf")
+        try:
+            print(f"[->] Fetching [{i+1}/{len(pdf_urls)}]: {url}")
+            response = requests.get(url, headers=headers, stream=True, timeout=15)
+            response.raise_for_status()
+            with open(filename, 'wb') as f:
+                for chunk in response.iter_content(chunk_size=8192):
+                    f.write(chunk)
+            downloaded_paths.append(filename)
+        except Exception as e:
+            print(f"[!] Warning: Failed to download asset from {url}: {e}")
+            
+    return downloaded_paths
+
+
+def extract_content_from_pdfs(pdf_paths):
+    """Parses text content and structured tabular data side-by-side using pdfplumber."""
+    documents = []
+    for path in pdf_paths:
+        print(f"[+] Parsing document extraction layout: {path}")
+        text_content = []
+        try:
+            with pdfplumber.open(path) as pdf:
+                for i, page in enumerate(pdf.pages):
+                    page_text = page.extract_text()
+                    if page_text:
+                        text_content.append(page_text)
+                    
+                    tables = page.extract_tables()
+                    for table in tables:
+                        table_str = "\n".join([
+                            " | ".join(map(lambda x: str(x).replace('\n', ' ') if x else "", row)) 
+                            for row in table
+                        ])
+                        text_content.append(
+                            f"\n--- Structured Matrix Table (Page {i+1}) ---\n"
+                            f"{table_str}\n"
+                            f"--------------------------------------------"
+                        )
+            
+            full_text = "\n\n".join(text_content).strip()
+            if full_text and len(full_text.split()) >= 30:
+                doc = Document(page_content=full_text, metadata={"source": os.path.basename(path)})
+                documents.append(doc)
+        except Exception as e:
+            print(f"[!] Layout Error: Could not read structural data from {path}: {e}")
+            
+    return documents
+
+
+def build_semantic_faiss_index(documents, output_dir, model_path):
+    """Vectorizes data using semantic breakpoints matching subject topics."""
+    if not documents:
+        print("[-] Vectorization Pipeline Aborted: No valid textual documents loaded.")
+        return
+        
+    if not os.path.isdir(model_path):
+        raise FileNotFoundError(f"[-] Local directory for bge-m3 weights not found at: {model_path}")
+        
+    print("[+] Initializing local Embedding Engine...")
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    print(f"[*] Compute Target Acceleration: {device.upper()}")
+    
+    embeddings = HuggingFaceEmbeddings(
+        model_name=model_path,
+        model_kwargs={"device": device},
+        encode_kwargs={"normalize_embeddings": True}
+    )
+    
+    print("[+] Analyzing text layouts for semantic split processing...")
+    # Semantic Chunking evaluates the distance between sentences to group them by topic.
+    text_splitter = SemanticChunker(
+        embeddings,
+        breakpoint_threshold_type="percentile"
+    )
+    
+    chunks = text_splitter.split_documents(documents)
+    print(f"[*] Generated {len(chunks)} contextual topic-based document chunks.")
+    
+    print("[+] Computing vector embeddings & building FAISS Index...")
+    t0 = time.time()
+    vectorstore = FAISS.from_documents(chunks, embeddings)
+    vectorstore.save_local(output_dir)
+    print(f"[+] Success! FAISS indexing layer compiled in {time.time() - t0:.2f}s.")
+    print(f"    Assets stored securely at: {os.path.abspath(output_dir)}")
+    print("    Files ready for distribution: 'index.faiss' and 'index.pkl'")
+
+
+if __name__ == "__main__":
+    print("=================================================================")
+    print("             HBL INTERNAL ASSISTANT - INGESTION ENGINE           ")
+    print("=================================================================")
+    
+    pdf_urls = get_pdf_links(TARGET_URL, max_pages=MAX_PAGES, max_depth=MAX_DEPTH)
+    
+    if pdf_urls:
+        downloaded_files = download_pdfs(pdf_urls, DOWNLOAD_DIR)
+        docs = extract_content_from_pdfs(downloaded_files)
+        build_semantic_faiss_index(docs, output_dir=INDEX_DIR, model_path=LOCAL_MODEL_PATH)
+        print("\n[+] System Execution Sequence Completed Successfully.")
+    else:
+        print("\n[-] Core Pipeline Stopped: Ensure network boundaries or target paths are correct.")