Skip to content

Instantly share code, notes, and snippets.

@rohithreddy

rohithreddy/normalize_data.py

Created June 20, 2026 08:04
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save rohithreddy/325cdab8161d0aed705c52dac891e43d to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save rohithreddy/325cdab8161d0aed705c52dac891e43d to your computer and use it in GitHub Desktop.
Download ZIP
Nettlio sample work
Raw
normalize_data.py
#!/usr/bin/env python3
import json
import os
import re
def clean_drug_name(name):
if not name:
return ""
# Strip common professional and consumer page title suffixes
name = re.sub(r':\s*(Package Insert|Prescribing Info|Prescribing Information|FDA prescribing|Professional|Consumer|Side Effects|Uses|Dosage|Warnings).*$', '', name, flags=re.IGNORECASE)
# Strip parenthetical route/form suffixes
name = re.sub(r'\s*\((Oral|Topical|Otic|Ophthalmic|Inhalation|Injection|Rectal|Sublingual|Vaginal|Nasal|Systemic|Intravenous|Intramuscular)\)$', '', name, flags=re.IGNORECASE)
# General fallback for any remaining colon with metadata keywords
if ":" in name:
parts = name.split(":", 1)
after = parts[1].lower()
if any(w in after for w in ["info", "insert", "prescribing", "professional", "monograph", "uses", "guidelines", "side effect", "dosage"]):
name = parts[0].strip()
return name.strip()
def main():
input_file = "drugs_data.json"
if not os.path.exists(input_file):
print(f"Error: {input_file} not found.")
return
with open(input_file, "r", encoding="utf-8") as f:
try:
data = json.load(f)
except Exception as e:
print(f"Error reading JSON: {e}")
return
print(f"Loaded {len(data)} records from {input_file}.")
normalized_data = {}
for record in data:
url = record.get("url")
if not url:
continue
# Normalize the drug name
orig_name = record.get("drug_name", "")
new_name = clean_drug_name(orig_name)
record["drug_name"] = new_name
# Dedup: if URL is seen multiple times, keep the latest one or merge/override
normalized_data[url] = record
# Sort alphabetically by drug_name (case-insensitive) and then by url
sorted_records = sorted(
normalized_data.values(),
key=lambda x: (x.get("drug_name", "").lower(), x.get("url", ""))
)
# Save to a temporary file first
temp_file = input_file + ".tmp"
with open(temp_file, "w", encoding="utf-8") as f:
json.dump(sorted_records, f, indent=2, ensure_ascii=False)
os.replace(temp_file, input_file)
print(f"Normalized and deduplicated. Saved {len(sorted_records)} records to {input_file}.")
if __name__ == "__main__":
main()
Raw
scraper.py
#!/usr/bin/env python3
"""
Drugs.com Scraper Script
Scrapes drug information from drugs.com for drugs starting with letters K to U.
Saves data in drugs_data.json.
"""
import os
import sys
import re
import json
import time
import random
import logging
import argparse
from urllib.parse import urljoin, urlparse
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from bs4 import BeautifulSoup
# Configure Logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger("drugs_scraper")
def init_driver():
"""Initializes and returns a headless Selenium Chrome WebDriver."""
chrome_options = Options()
chrome_options.add_argument("--headless=new")
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--disable-dev-shm-usage")
chrome_options.add_argument("--disable-gpu")
chrome_options.add_argument("--window-size=1920,1080")
# Set a realistic user agent
chrome_options.add_argument(
"user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
)
# Exclude automation switches
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
chrome_options.add_experimental_option("useAutomationExtension", False)
driver = webdriver.Chrome(options=chrome_options)
# Execute CDP script to prevent webdriver detection
driver.execute_cdp_cmd(
"Page.addScriptToEvaluateOnNewDocument",
{
"source": """
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined
})
"""
}
)
return driver
def load_existing_data(output_file):
"""Loads existing drugs from the output JSON file if it exists."""
if os.path.exists(output_file):
try:
with open(output_file, "r", encoding="utf-8") as f:
data = json.load(f)
if isinstance(data, list):
logger.info(f"Loaded {len(data)} existing records from {output_file}.")
return data
except Exception as e:
logger.warning(f"Could not load existing data from {output_file}: {e}. Starting fresh.")
return []
def save_data(data, output_file):
"""Saves the crawled drug records to the JSON output file in sorted order."""
try:
# Sort data alphabetically by drug_name, then by url
sorted_data = sorted(data, key=lambda x: (x.get("drug_name", "").lower(), x.get("url", "")))
# Save to a temporary file first to avoid corrupting data on interrupt
temp_file = output_file + ".tmp"
with open(temp_file, "w", encoding="utf-8") as f:
json.dump(sorted_data, f, indent=2, ensure_ascii=False)
os.replace(temp_file, output_file)
except Exception as e:
logger.error(f"Failed to save data to {output_file}: {e}")
def extract_field_text(b_tag):
"""Extracts text following a bold label tag in a paragraph until <br> or <b>."""
if not b_tag:
return ""
text_parts = []
curr = b_tag.next_sibling
while curr:
if curr.name in ["br", "b", "strong"]:
break
part = curr.get_text() if curr.name else str(curr)
text_parts.append(part)
curr = curr.next_sibling
return "".join(text_parts).strip()
def clean_generic_name(val):
"""Cleans phonetic pronunciations and format metadata from generic name."""
# Remove text inside [...] (phonetic pronunciation)
val = re.sub(r'\[.*?\]', '', val)
# Remove text inside (...) (qualifiers like oral/injection)
val = re.sub(r'\(.*?\)', '', val)
# Clean up multiple spaces
val = re.sub(r'\s+', ' ', val)
return val.strip()
def clean_brand_name(brand):
"""Cleans brand name prefix and dynamic toggle texts."""
# Remove "of <generic> include(s):" prefix if it exists
brand = re.sub(r'^of\s+.*?\s+include[s]?:\s*', '', brand, flags=re.IGNORECASE)
# Remove "... show all X brands" text
brand = re.sub(r'\.{0,3}\s*show all \d+\s*brands', '', brand, flags=re.IGNORECASE)
# Clean leading/trailing punctuation and spaces
brand = brand.strip(" ,\n\r\t.")
return brand
def clean_drug_name(name):
"""Normalizes the drug_name to the drug's name only."""
if not name:
return ""
# Strip common professional and consumer page title suffixes
name = re.sub(r':\s*(Package Insert|Prescribing Info|Prescribing Information|FDA prescribing|Professional|Consumer|Side Effects|Uses|Dosage|Warnings).*$', '', name, flags=re.IGNORECASE)
# Strip parenthetical route/form suffixes
name = re.sub(r'\s*\((Oral|Topical|Otic|Ophthalmic|Inhalation|Injection|Rectal|Sublingual|Vaginal|Nasal|Systemic|Intravenous|Intramuscular)\)$', '', name, flags=re.IGNORECASE)
# General fallback for any remaining colon with metadata keywords
if ":" in name:
parts = name.split(":", 1)
after = parts[1].lower()
if any(w in after for w in ["info", "insert", "prescribing", "professional", "monograph", "uses", "guidelines", "side effect", "dosage"]):
name = parts[0].strip()
return name.strip()
def parse_drug_html(html, url):
"""Parses drug page HTML and extracts relevant structured fields."""
soup = BeautifulSoup(html, "html.parser")
# 1. Drug Name (H1)
h1 = soup.find("h1")
raw_name = h1.get_text(strip=True) if h1 else ""
drug_name = clean_drug_name(raw_name)
# Locate subtitles elements inside the metadata paragraph
b_generic = soup.find(lambda tag: tag.name in ["b", "strong"] and "generic name" in tag.text.lower())
b_brands = soup.find(lambda tag: tag.name in ["b", "strong"] and any(term in tag.text.lower() for term in ["brand name", "brand names"]))
b_class = soup.find(lambda tag: tag.name in ["b", "strong"] and "drug class" in tag.text.lower())
# 2. Generic Name
generic_raw = extract_field_text(b_generic)
generic_name = clean_generic_name(generic_raw) if generic_raw else ""
# 3. Brand Names (array)
brands_raw = extract_field_text(b_brands)
brand_names = []
if brands_raw:
for b in brands_raw.split(","):
cleaned = clean_brand_name(b)
if cleaned:
brand_names.append(cleaned)
# 4. Drug Class (comma-separated string)
drug_class = extract_field_text(b_class)
# Clean multiple spaces and formatting
drug_class = re.sub(r'\s+', ' ', drug_class).strip()
# 5. Related Conditions (array)
related_conditions = []
# Method A: Search under "Related treatment guides" section
h_guides = soup.find(
lambda tag: tag.name in ["h2", "h3", "h4"]
and any(term in tag.text.lower() for term in ["related treatment guides", "related conditions", "related treatment"])
)
if h_guides:
curr = h_guides.find_next_sibling()
# Scan next few siblings to find the <ul> list
for _ in range(4):
if not curr:
break
if curr.name == "ul":
for li in curr.find_all("li"):
cond_text = li.get_text(strip=True)
if cond_text and cond_text not in related_conditions:
related_conditions.append(cond_text)
break
curr = curr.find_next_sibling()
# Method B: Fallback to searching data-type="drug-condition"
if not related_conditions:
for a in soup.find_all("a", attrs={"data-type": "drug-condition"}):
cond_text = a.get_text(strip=True)
if cond_text and cond_text not in related_conditions:
related_conditions.append(cond_text)
# Method C: Fallback to searching lists of conditions in specific sections (strictly matching 'conditions')
if not related_conditions:
# Search inside ul class containing 'conditions'
cond_ul = soup.find("ul", class_=re.compile(r"conditions"))
if cond_ul:
for li in cond_ul.find_all("li"):
cond_text = li.get_text(strip=True)
if cond_text and cond_text not in related_conditions:
related_conditions.append(cond_text)
return {
"drug_name": drug_name,
"url": url,
"drug_class": drug_class,
"generic_name": generic_name,
"brand_names": brand_names,
"related_conditions": related_conditions
}
def crawl_alphabet_page(driver, url):
"""Crawls an alphabet letter index page to extract drug links and subpage pagination links."""
logger.info(f"Crawling alphabet index page: {url}")
driver.get(url)
# Give it a moment to render
time.sleep(1)
html = driver.page_source
soup = BeautifulSoup(html, "html.parser")
# Locate main content container to avoid layout leakage
content_div = soup.find("div", class_="ddc-main-content")
if not content_div:
content_div = soup
# 1. Find drug links
drug_links = set()
ul_lists = content_div.find_all("ul", class_=re.compile("ddc-list-column|ddc-list-unstyled"))
for ul in ul_lists:
for li in ul.find_all("li"):
a = li.find("a")
if a and a.get("href"):
href = a.get("href").strip()
abs_url = urljoin(url, href)
parsed = urlparse(abs_url)
if parsed.netloc == "www.drugs.com":
path = parsed.path.lower()
if path.endswith(".html"):
exclude_patterns = [
"/alpha/", "/support/", "/answers/", "/medical-answers/",
"/newsletters/", "/podcasts/", "sitemap.html", "privacy.html",
"terms.html", "about.html", "contact.html", "feedback.html"
]
if not any(pat in path for pat in exclude_patterns):
drug_links.add(abs_url)
# 2. Find subpage links (pagination)
subpage_links = set()
paging_anchors = soup.find_all("a", class_="ddc-paging-item")
for anchor in paging_anchors:
href = anchor.get("href")
if href:
abs_url = urljoin(url, href)
subpage_links.add(abs_url)
return list(drug_links), list(subpage_links)
def main():
parser = argparse.ArgumentParser(description="Drugs.com Selenium Scraper (Letters K-U)")
parser.add_argument("--start-letter", default="k", help="Letter to start scraping from (e.g. k)")
parser.add_argument("--end-letter", default="u", help="Letter to end scraping at (e.g. u)")
parser.add_argument("--output", default="drugs_data.json", help="Path to output JSON file")
parser.add_argument("--delay", type=float, default=1.5, help="Average delay between page requests in seconds")
parser.add_argument("--test", action="store_true", help="Run in test mode (scrapes very few drugs to verify correctness)")
args = parser.parse_args()
start_char = args.start_letter.lower()
end_char = args.end_letter.lower()
output_file = args.output
base_delay = args.delay
is_test = args.test
# Define alphabet letters to crawl
all_letters = [chr(i) for i in range(ord('a'), ord('z') + 1)]
try:
start_idx = all_letters.index(start_char)
end_idx = all_letters.index(end_char)
target_letters = all_letters[start_idx : end_idx + 1]
except ValueError:
logger.error(f"Invalid start or end letter. Using defaults (k to u).")
target_letters = [chr(i) for i in range(ord('k'), ord('u') + 1)]
logger.info(f"Target letters for crawl: {target_letters}")
if is_test:
logger.info("TEST MODE ENABLED. Crawl will be restricted to a small subset.")
target_letters = [target_letters[0]] # Only first letter
# Load existing data (checkpoint)
crawled_data = load_existing_data(output_file)
crawled_urls = {item["url"] for item in crawled_data}
# Initialize driver
driver = init_driver()
try:
# Phase 1: Gather or load all drug links
discovered_file = "discovered_links.json"
all_drug_urls = set()
# In test mode, we bypass caching to allow dynamic subset runs
if not is_test and os.path.exists(discovered_file):
try:
with open(discovered_file, "r", encoding="utf-8") as f:
links_list = json.load(f)
if isinstance(links_list, list):
all_drug_urls = set(links_list)
logger.info(f"Loaded {len(all_drug_urls)} discovered drug links from {discovered_file}. Skipping index crawling.")
except Exception as e:
logger.warning(f"Could not load discovered links from {discovered_file}: {e}. Crawling index.")
if not all_drug_urls:
logger.info("--- PHASE 1: GATHERING DRUG LINKS ---")
for letter in target_letters:
letter_url = f"https://www.drugs.com/alpha/{letter}.html"
try:
drug_links, subpage_links = crawl_alphabet_page(driver, letter_url)
logger.info(f"Letter '{letter}' main page: Found {len(drug_links)} drug links and {len(subpage_links)} subpages.")
# Add drug links from main page
all_drug_urls.update(drug_links)
# If test mode, limit subpages
if is_test:
subpage_links = subpage_links[:1]
logger.info(f"Test mode: Limited to {len(subpage_links)} subpages.")
# Crawl each subpage
for subpage_url in subpage_links:
logger.info(f"Crawling subpage: {subpage_url}")
sub_drug_links, _ = crawl_alphabet_page(driver, subpage_url)
logger.info(f" Found {len(sub_drug_links)} drug links on subpage.")
all_drug_urls.update(sub_drug_links)
# Polite delay between subpage requests
time.sleep(base_delay + random.uniform(0.5, 1.5))
if is_test:
break
except Exception as e:
logger.error(f"Error crawling letter index for '{letter}': {e}")
# Save discovered links cache (if not in test mode)
if not is_test and all_drug_urls:
try:
with open(discovered_file, "w", encoding="utf-8") as f:
json.dump(list(all_drug_urls), f, indent=2, ensure_ascii=False)
logger.info(f"Saved {len(all_drug_urls)} discovered drug links to {discovered_file}.")
except Exception as e:
logger.error(f"Failed to save discovered links to {discovered_file}: {e}")
# Filter out already crawled URLs
urls_to_crawl = [url for url in all_drug_urls if url not in crawled_urls]
logger.info(f"Total unique drug links discovered: {len(all_drug_urls)}")
logger.info(f"Already crawled: {len(crawled_urls)}")
logger.info(f"Remaining to crawl: {len(urls_to_crawl)}")
if is_test:
urls_to_crawl = urls_to_crawl[:3]
logger.info(f"Test mode: Restricting remaining links to crawl to {len(urls_to_crawl)} items.")
# Phase 2: Crawl each drug page and extract details
logger.info("--- PHASE 2: CRAWLING DRUG DETAILS ---")
success_count = 0
failed_urls = []
# Helper function to crawl a list of URLs
def crawl_urls_list(urls, attempt=1):
nonlocal success_count
next_failed = []
for idx, drug_url in enumerate(urls, 1):
logger.info(f"[Attempt {attempt} - {idx}/{len(urls)}] Scraping drug: {drug_url}")
try:
driver.get(drug_url)
# Polite delay for dynamic elements to load
time.sleep(base_delay + random.uniform(0.5, 1.5))
html = driver.page_source
record = parse_drug_html(html, drug_url)
if record["drug_name"]:
# Ensure we update crawled_data in-place
crawled_data_dict = {item["url"]: item for item in crawled_data}
crawled_data_dict[drug_url] = record
crawled_data[:] = list(crawled_data_dict.values())
crawled_urls.add(drug_url)
success_count += 1
logger.info(f" Successfully parsed: {record['drug_name']}")
# Save progress increment (every 5 records, or always in test mode)
if success_count % 5 == 0 or is_test:
save_data(crawled_data, output_file)
logger.info(" Saved checkpoint data to disk.")
else:
logger.warning(f" Could not extract drug name from {drug_url}. Adding to failed list.")
next_failed.append(drug_url)
except Exception as e:
logger.error(f" Failed to scrape {drug_url}: {e}")
next_failed.append(drug_url)
# Random sleep delay
time.sleep(base_delay + random.uniform(0.5, 1.5))
return next_failed
# First pass
failed_urls = crawl_urls_list(urls_to_crawl, attempt=1)
# Retry passes
max_attempts = 2
for attempt in range(2, max_attempts + 2):
if not failed_urls or is_test:
break
logger.info(f"--- RETRY PASS {attempt-1} FOR {len(failed_urls)} FAILED URLS ---")
# Extra wait before retry pass
time.sleep(5)
failed_urls = crawl_urls_list(failed_urls, attempt=attempt)
# Final save
save_data(crawled_data, output_file)
# Verification Summary
logger.info("--- SCRAPING VERIFICATION SUMMARY ---")
final_crawled_urls = {item["url"] for item in crawled_data}
# Discovered URLs could include crawled ones already, so let's combine sets
missing_urls = all_drug_urls - final_crawled_urls
logger.info(f"Total Unique Drugs Discovered in Index: {len(all_drug_urls)}")
logger.info(f"Total Drugs Successfully Scraped & Saved: {len(crawled_data)}")
logger.info(f"Total Drugs Missing/Failed: {len(missing_urls)}")
if len(missing_urls) == 0:
logger.info("VERIFICATION SUCCESS: All discovered drugs have been successfully scraped and saved!")
else:
logger.warning(f"VERIFICATION PARTIAL: Scraped {len(crawled_data)} of {len(all_drug_urls)} drugs. {len(missing_urls)} failed permanently.")
logger.warning("Failed URLs:")
for url in sorted(missing_urls):
logger.warning(f" - {url}")
finally:
driver.quit()
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment