Email Scraping với Python: Hướng dẫn đầy đủ kèm ví dụ

Bình luận: 0

Để tiếp cận trực tiếp hoạt động hiệu quả, bạn cần một nền tảng vững chắc – cơ sở dữ liệu chứa các địa chỉ email thật và được cập nhật. Đó là lúc email scraping bằng Python trở nên hữu ích: một cách để tự động thu thập địa chỉ từ các trang web.

Trong hướng dẫn này, chúng ta sẽ xem cách xây dựng công cụ email scraping bằng Python từ đầu, cách xử lý các trang động, cách lọc và xác minh các địa chỉ thu thập được, và cách sử dụng dữ liệu kết quả trong quy trình marketing hoặc kinh doanh thực tế.

Tài liệu này rất hữu ích nếu bạn cần:

  • Tự tìm hiểu cách thu thập địa chỉ email từ trang web bằng Python mà không cần dịch vụ có sẵn;
  • Tự động hóa việc tạo danh sách gửi thư cho bản tin, CRM hoặc nghiên cứu;
  • Kết nối mã nguồn với các tình huống thực tế – từ trích xuất đến tích hợp.

Tiếp theo, chúng ta sẽ xem cách biến các trang công khai thành kênh giao tiếp trực tiếp với những người có thể trở thành khách hàng của bạn – bằng cách sử dụng Python.

Email Scraping là gì và nó giúp ích như thế nào

Về bản chất, loại scraping này là quá trình quét tự động các trang HTML hoặc trang động và tìm trong nội dung hoặc thuộc tính các mẫu khớp với định dạng địa chỉ (ví dụ: username@domain.tld). Sau đó, bạn lọc, xác minh và lưu kết quả.

Các tác vụ sử dụng Python Email Scraper

Nó được sử dụng rộng rãi trong kinh doanh, marketing, nghiên cứu và tự động hóa các quy trình lặp lại. Đặc biệt hữu ích khi bạn cần thu thập và cấu trúc lượng lớn thông tin công khai từ nhiều nguồn khác nhau.

Các ví dụ cụ thể về nhiệm vụ áp dụng email scraping bằng Python:

  • Xây dựng cơ sở dữ liệu liên hệ cho các chiến dịch email;
  • Marketing và tạo khách hàng tiềm năng;
  • Nghiên cứu và phân tích các liên hệ công khai;
  • Điền và cập nhật hệ thống CRM;
  • Theo dõi hoạt động của đối thủ cạnh tranh;
  • Kiểm tra và xác minh dữ liệu liên hệ của chính bạn.

Nếu bạn quan tâm đến việc thu thập dữ liệu liên hệ cho các dự án thương mại điện tử, hãy xem hướng dẫn của chúng tôi về Ecommerce data scraping.

Các kiến thức cơ bản: Công cụ và Chuẩn bị

Để việc scraping đạt hiệu quả, bạn cần chuẩn bị môi trường và chọn công cụ phù hợp. Chúng giúp bạn lấy dữ liệu nhanh hơn, xử lý các trang phức tạp hoặc động, và tổ chức các dự án lớn.

Chọn thư viện để thu thập địa chỉ email

Các công cụ Python phổ biến cho việc scraping:

Công cụ Mục đích sử dụng
requests / httpx Lấy các trang tĩnh
BeautifulSoup Phân tích HTML / tìm kiếm phần tử
re (biểu thức chính quy) Trích xuất mẫu
lxml Phân tích nhanh hơn
Selenium / Playwright Xử lý các trang chạy bằng JavaScript
Scrapy Khung toàn diện cho các tác vụ thu thập dữ liệu lớn

Chuẩn bị môi trường làm việc

  1. Tạo môi trường ảo (venv hoặc virtualenv).
  2. Cài đặt các gói phụ thuộc:
    pip install requests beautifulsoup4 lxml
    pip install selenium  # nếu cần hiển thị động
  3. (Nếu cần) thiết lập trình điều khiển trình duyệt (ChromeDriver, GeckoDriver).
  4. Chuẩn bị danh sách các URL hoặc tên miền bắt đầu.
  5. Quyết định chiến lược thu thập — đệ quy hoặc giới hạn.

Để xem các phương pháp tương tự được áp dụng cho nền tảng khác như thế nào, hãy xem hướng dẫn chi tiết của chúng tôi về scrape Reddit using Python.

Ví dụ: Email Scraping với Python — Logic cốt lõi (Pseudocode)

# 1. Create an HTTP session with timeouts and retries
session = make_session()
# 2. Load the page
html = session.get(url)
# 3. Look for email addresses:
#    - via regex across the entire text
#    - via mailto: links in HTML
emails = extract_emails_from_text(html)
emails.update(find_mailto_links(html))
# 4. Return a unique list of addresses
return emails

Tại sao lại theo cách này?

  • Session + retries — để tránh lỗi ngẫu nhiên và thực hiện lại yêu cầu khi có lỗi.
  • Regex + mailto: — hai cách đơn giản và hiệu quả ngay lập tức.
  • lxml trong BeautifulSoup — bộ phân tích HTML nhanh hơn và chính xác hơn.
  • Chuẩn hóa mailto: — loại bỏ mọi phần thừa (?subject=...), chỉ giữ lại địa chỉ.

Biến thể mở rộng: Trình thu thập dữ liệu đa cấp (Multi-Level Crawler)

"""
Iterate over internal links within one domain and collect email addresses.
Highlights:
- Page limit (max_pages) to stop safely
- Verifying that a link belongs to the base domain
- Avoiding re-visits
- Optional respect for robots.txt
"""

from __future__ import annotations
from collections import deque
from typing import Set
from urllib.parse import urljoin, urlparse, urlsplit, urlunsplit
import time
import requests
from bs4 import BeautifulSoup
import lxml # Import lxml to ensure it's available for BeautifulSoup
from urllib import robotparser  # standard robots.txt parser
# We use functions from the previous block:
# - make_session()
# - scrape_emails_from_url()
import re

# General regular expression for email addresses
EMAIL_RE = re.compile(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9.-]+")

def scrape_emails_from_url(url: str, session: requests.Session) -> Set[str]:
   """Collect email addresses from the given URL page."""
   emails: Set[str] = set()
   try:
       resp = session.get(url, timeout=getattr(session, "_default_timeout", 10.0))
       resp.raise_for_status()
       # Regular expression for email addresses
       # Note: this regex isn't perfect, but it's sufficient for typical cases
       email_pattern = re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}")
       emails.update(email_pattern.findall(resp.text))
   except requests.RequestException:
       pass
   return emails

def make_session() -> requests.Session:
   """Create and return a requests session with basic settings."""
   session = requests.Session()
   session.headers.update({
       "User-Agent": "EmailScraper/1.0",
       "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
       "Accept-Language": "en-US,en;q=0.9",
       # Don't force Accept-Encoding to avoid br issues without brotli
       "Connection": "keep-alive",
   })
   return session
def same_host(url: str, base_netloc: str) -> bool:
   """True if the link belongs to the same host (domain/subdomain)."""
   return urlparse(url).netloc == base_netloc
def load_robots(start_url: str, user_agent: str = "EmailScraper") -> robotparser.RobotFileParser:
   """Read robots.txt and return a parser for permission checks."""
   base = urlparse(start_url)
   robots_url = f"{base.scheme}://{base.netloc}/robots.txt"
   rp = robotparser.RobotFileParser()
   rp.set_url(robots_url)
   try:
       rp.read()
   except Exception:
       pass
   rp.useragent = user_agent
   return rp

def normalize_url(url: str, base: str | None = None) -> str | None:
   try:
       abs_url = urljoin(base, url) if base else url
       parts = urlsplit(abs_url)
       if parts.scheme not in ("http", "https"):
           return None
       host = parts.hostname
       if not host:
           return None
       host = host.lower()
       netloc = host
       if parts.port:
           netloc = f"{host}:{parts.port}"
       parts = parts._replace(fragment="")
       return urlunsplit((parts.scheme.lower(), netloc, parts.path or "/", parts.query, ""))
   except Exception:
       return None

def in_scope(url: str, base_host: str, include_subdomains: bool) -> bool:
   try:
       host = urlsplit(url).hostname
       if not host:
           return False
       host = host.lower()
       base_host = (base_host or "").lower()
       if include_subdomains:
           return host == base_host or host.endswith("." + base_host)
       else:
           return host == base_host
   except Exception:
       return False
def collect_emails_from_site(
   start_url: str,
   max_pages: int = 100,
   delay_sec: float = 0.5,
   respect_robots: bool = True,
   include_subdomains: bool = True,
) -> Set[str]:
   """
   Traverse pages within a domain and return unique email addresses.
   - max_pages: hard limit on visited pages.
   - delay_sec: polite pause between requests.
   - respect_robots: if True — checks access rules.
   - include_subdomains: if True — allows subdomains (www, etc.).
   """
   session = make_session()
   base_host = (urlparse(start_url).netloc or "").lower()
   visited: Set[str] = set()
   queue: deque[str] = deque()
   enqueued: Set[str] = set()
   all_emails: Set[str] = set()

   start_norm = normalize_url(start_url)
   if start_norm:
       queue.append(start_norm)
       enqueued.add(start_norm)

   rp = load_robots(start_url, user_agent="EmailScraper/1.0") if respect_robots else None

   while queue and len(visited) < max_pages:
       url = queue.popleft()
       if url in visited:
           continue

       # robots.txt check
       if respect_robots and rp is not None:
           try:
               if not rp.can_fetch("EmailScraper/1.0", url):
                   continue
           except Exception:
               pass

       # One request: used both for emails and links
       try:
           resp = session.get(url, timeout=10)
           resp.raise_for_status()
           html_text = resp.text or ""
       except requests.RequestException:
           continue

       visited.add(url)

       # Skip non-HTML pages
       ctype = resp.headers.get("Content-Type", "")
       if ctype and "text/html" not in ctype:
           continue

       # Collect emails
       for m in EMAIL_RE.findall(html_text):
           all_emails.add(m.lower())

       # Parse links
       soup = BeautifulSoup(html_text, "lxml")

       # Emails from mailto:
       for a in soup.find_all("a", href=True):
           href = a["href"].strip()
           if href.lower().startswith("mailto:"):
               addr_part = href[7:].split("?", 1)[0]
               for piece in addr_part.split(","):
                   email = piece.strip()
                   if EMAIL_RE.fullmatch(email):
                       all_emails.add(email.lower())

       for a in soup.find_all("a", href=True):
           href = a["href"].strip()
           if not href or href.startswith(("javascript:", "mailto:", "tel:", "data:")):
               continue
           next_url = normalize_url(href, base=url)
           if not next_url:
               continue
           if not in_scope(next_url, base_host, include_subdomains):
               continue
           if next_url not in visited and next_url not in enqueued:
               queue.append(next_url)
               enqueued.add(next_url)

       if delay_sec > 0:
           time.sleep(delay_sec)

   try:
       session.close()
   except Exception:
       pass
   return all_emails
if __name__ == "__main__":
   import argparse

parser = argparse.ArgumentParser(
   description="An email scraper that traverses pages within a site and prints discovered addresses."
)

parser.add_argument(
   "start_url",
   help="Starting URL, for example: https://example.com"
)

parser.add_argument(
   "--max-pages",
   type=int,
   default=100,
   dest="max_pages",
   help="Maximum number of pages to traverse (default: 100)"
)

parser.add_argument(
   "--delay",
   type=float,
   default=0.5,
   help="Delay between requests in seconds (default: 0.5)"
)

parser.add_argument(
   "--no-robots",
   action="store_true",
   help="Ignore robots.txt (use carefully)"
)

scope = parser.add_mutually_exclusive_group()

scope.add_argument(
   "--include-subdomains",
   dest="include_subdomains",
   action="store_true",
   default=True,
   help="Include subdomains (default)"
)

scope.add_argument(
   "--exact-host",
   dest="include_subdomains",
   action="store_false",
   help="Restrict traversal to the exact host (no subdomains)"
)

parser.add_argument(
   "--output",
   type=str,
   default=None,
   help="Optional: path to a file to save found email addresses (one per line)"

   args = parser.parse_args()

   emails = collect_emails_from_site(
       args.start_url,
       max_pages=args.max_pages,
       delay_sec=args.delay,
       respect_robots=not args.no_robots,
       include_subdomains=args.include_subdomains,
   )

   for e in sorted(emails):
       print(e)

   print(f"Found {len(emails)} unique emails.")

   if args.output:
       try:
           with open(args.output, "w", encoding="utf-8") as f:
               for e in sorted(emails):
                   f.write(e + "\n")
       except Exception as ex:
           print(f"Could not write the output file: {ex}")

Cách chạy và cấu hình script mở rộng

main.py https://example.com
Tham số của script
  • start_url – URL bắt đầu nơi quá trình thu thập dữ liệu bắt đầu (ví dụ: https://example.com).
  • --max-pages – số lượng trang tối đa cần thu thập. Mặc định: 100.
  • --delay – độ trễ giữa các yêu cầu (tính bằng giây) để giảm tải cho máy chủ. Mặc định: 0.5.
  • --no-robots – bỏ qua các quy tắc trong robots.txt. Hãy sử dụng cẩn thận vì một số trang có thể không cho phép thu thập tự động.
  • --include-subdomains – bao gồm các tên miền phụ trong quá trình thu thập. Được bật theo mặc định.
  • --exact-host – giới hạn thu thập trong đúng tên miền chính (không bao gồm tên miền phụ).
  • --output – đường dẫn tới tệp để lưu các địa chỉ tìm được (mỗi dòng một địa chỉ). Nếu không cung cấp, các địa chỉ sẽ được hiển thị trong bảng điều khiển.

Xử lý mã hóa và nội dung động

Khi bạn chạy một script, mọi thứ không phải lúc nào cũng đơn giản: nhiều trang web cố tình ẩn địa chỉ email hoặc chỉ hiển thị chúng sau khi JavaScript được tải xong. Dưới đây là những yếu tố có thể gây trở ngại — và cách xử lý chúng.

Các vấn đề tiềm ẩn

1. Mã hóa (Obfuscation)

Các trang web thường sử dụng kỹ thuật để ẩn địa chỉ khỏi bot:

  • JavaScript ghép địa chỉ từ các phần nhỏ (ví dụ: user + “@” + domain.com);
  • Chuỗi được mã hóa hoặc mã hoá (ví dụ: Base64, HTML entities);
  • Chú thích HTML hoặc các phần chèn trong đó một phần của địa chỉ bị ẩn;
  • Email dưới dạng hình ảnh (ảnh chứa chữ), khi đó script sẽ không thấy gì;
  • Thay thế ký tự: user [at] example [dot] com và các dạng “dễ đọc với con người” khác (address munging).

2. Trang động

Các trang web hiện đại thường tải nội dung qua JavaScript (ví dụ: fetch, AJAX). Một lệnh requests.get() thông thường có thể trả về một khung HTML “rỗng” không chứa nội dung email.

Cách vượt qua những trở ngại này

Các phương pháp thực tế khi gặp phải những trang như vậy:

  1. Selenium hoặc Playwright:

    Khởi chạy trình duyệt, cho trang “tải” hoàn tất, chờ các phần tử cần thiết, sau đó lấy toàn bộ HTML. Cách này hiệu quả khi email được thêm vào bằng JS sau khi tải xong.

  2. Gọi API:

    Thường thì trang thực sự lấy dữ liệu từ API. Kiểm tra các yêu cầu mạng (DevTools → Network) để xem có yêu cầu nào trả về email hoặc thông tin liên hệ ở dạng JSON hay không. Nếu có, tốt hơn nên truy cập trực tiếp API đó.

  3. Phân tích JS / script nội tuyến:

    Đôi khi địa chỉ được “nhúng” trong JavaScript (ví dụ: chuỗi Base64 hoặc chia nhỏ thành các phần). Bạn có thể đọc đoạn JS đó, trích xuất chuỗi và giải mã địa chỉ.

  4. Nếu email ở trong hình ảnh:

    Tải hình ảnh về và sử dụng OCR (Optical Character Recognition), ví dụ với Tesseract. Phương pháp này tiêu tốn tài nguyên hơn nhưng đôi khi là cần thiết.

  5. Độ trễ và thời gian:

    Một số phần tử chỉ xuất hiện sau vài giây hoặc sau các sự kiện cụ thể (cuộn, nhấp chuột). Hãy áp dụng các cách sau:

    • Sử dụng sleep() hoặc chờ một bộ chọn;
    • Thử nhiều lần;
    • Áp dụng chiến lược “thử lại nếu không tìm thấy”.

Kết luận

Bằng cách áp dụng các kỹ thuật được trình bày trong bài viết này về email scraping bằng Python, bạn có thể giúp script của mình hoạt động ổn định trong điều kiện thực tế. Hãy nhớ rằng chất lượng dữ liệu ảnh hưởng trực tiếp đến hiệu quả của các chiến dịch sau này, vì vậy bạn nên triển khai việc lọc, xác minh và lưu dữ liệu ở định dạng thuận tiện ngay từ đầu.

Bình luận:

0 Bình luận