Spaces:
Running on Zero
Running on Zero
File size: 5,983 Bytes
62be02c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | import logging
import re
import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin
from typing import List, Dict, Optional
from .constants import INDIACODE_BASE_URL, KANOON_BASE_URL
from .utils import polite_fetch, clean_text, get_text_hash, is_valid_section
logger = logging.getLogger(__name__)
class BaseScraper:
def __init__(self, session: requests.Session):
self.session = session
def scrape(self, act_name: str) -> List[Dict]:
raise NotImplementedError("Subclasses must implement scrape()")
class IndiaCodeScraper(BaseScraper):
def find_handle(self, act_name: str) -> Optional[str]:
search_url = f"{INDIACODE_BASE_URL}/search?query={act_name.replace(' ', '+')}"
html = polite_fetch(search_url, self.session)
if not html: return None
soup = BeautifulSoup(html, 'html.parser')
for a in soup.find_all('a', href=True):
if "/handle/123456789/" in a['href']:
return urljoin(INDIACODE_BASE_URL, a['href'])
return None
def scrape(self, act_name: str) -> List[Dict]:
sections = []
handle_url = self.find_handle(act_name)
if not handle_url:
logger.info(f"IndiaCode: No handle found for {act_name}")
return []
html = polite_fetch(handle_url, self.session)
if not html: return []
soup = BeautifulSoup(html, 'html.parser')
sub_links = []
for a in soup.find_all('a', href=True):
href = a['href']
if "/handle/123456789/" in href and href not in handle_url:
sub_links.append(urljoin(INDIACODE_BASE_URL, href))
sub_links = list(dict.fromkeys(sub_links))
logger.info(f"IndiaCode: Found {len(sub_links)} sub-links for {act_name}")
for link in sub_links[:100]: # Safety cap
page_html = polite_fetch(link, self.session)
if not page_html: continue
page_soup = BeautifulSoup(page_html, 'html.parser')
if any(x in page_soup.get_text() for x in ["Browsing Central Act", "Sort by", "Items/Page"]):
continue
content_div = page_soup.find("div", {"id": "content"}) or page_soup.find("div", class_="item-body")
if not content_div: continue
text = clean_text(content_div.get_text())
if not is_valid_section(text): continue
match = re.search(r'(Section|Sec\.|Article)\s+(\d+[A-Z]*)', text, re.I)
sec_num = match.group(2) if match else "0"
title_el = page_soup.find("h1") or page_soup.find("h2") or page_soup.find("title")
title = clean_text(title_el.get_text()) if title_el else f"Section {sec_num}"
sections.append({
"section_number": sec_num,
"title": title,
"text": text,
"hash": get_text_hash(text)
})
return sections
class KanoonScraper(BaseScraper):
def scrape(self, act_name: str) -> List[Dict]:
sections = []
search_url = f"{KANOON_BASE_URL}/search/?formInput={act_name.replace(' ', '+')}+doctypes:acts"
html = polite_fetch(search_url, self.session)
if not html: return []
soup = BeautifulSoup(html, 'html.parser')
results = soup.select('.result_title a')
if not results: return []
act_url = urljoin(KANOON_BASE_URL, results[0]['href'])
html = polite_fetch(act_url, self.session)
if not html: return []
soup = BeautifulSoup(html, 'html.parser')
content = soup.find("div", class_="judgments")
if not content: return []
links = content.find_all('a', href=True)
doc_links = [urljoin(KANOON_BASE_URL, a['href']) for a in links if "/doc/" in a['href'] and not a['href'].endswith("/")]
if len(doc_links) > 2:
logger.info(f"Kanoon: TOC detected for {act_name}. Following {len(doc_links)} links.")
for link in doc_links:
sec_html = polite_fetch(link, self.session)
if not sec_html: continue
sec_soup = BeautifulSoup(sec_html, 'html.parser')
sec_div = sec_soup.find("div", class_="judgments")
if not sec_div: continue
text = clean_text(sec_div.get_text())
if not is_valid_section(text): continue
title_el = sec_soup.find("div", class_="doc_title") or sec_soup.find("h1")
title = clean_text(title_el.get_text()) if title_el else "Section Content"
match = re.search(r'(Section|Sec\.|Article)\s+(\d+[A-Z]*)', title, re.I)
sec_num = match.group(2) if match else str(len(sections)+1)
sections.append({
"section_number": sec_num,
"title": title,
"text": text,
"hash": get_text_hash(text)
})
return sections
class LegalScraperEngine:
def __init__(self):
self.session = requests.Session()
self.indiacode = IndiaCodeScraper(self.session)
self.kanoon = KanoonScraper(self.session)
def scrape_act(self, act_name: str) -> tuple[List[Dict], str]:
"""Try IndiaCode first, then Kanoon as fallback."""
logger.info(f"Starting scrape for: {act_name}")
sections = self.indiacode.scrape(act_name)
if len(sections) >= 3:
return sections, "IndiaCode"
logger.info(f"IndiaCode failed/insufficient for {act_name}. Trying Kanoon fallback.")
sections = self.kanoon.scrape(act_name)
if sections:
return sections, "IndianKanoon"
return [], "None"
|