员工邮箱生成与验证工具开发技术问题咨询

项目概述 #

开发一款Python工具，用于生成目标企业员工邮箱并验证真实性。核心流程：基于员工姓名生成多种候选邮箱组合，通过SMTP协议验证邮箱有效性，结合谷歌搜索确认邮箱网络存在情况，最终为每个邮箱给出1-5分的评分（5分为最优）。可用数据包括：员工名、中间名、姓氏、企业域名、LinkedIn主页（如有）。

待解决技术问题 #

1. 数据库中员工姓名包含无用中间名，需提取仅名和姓的部分（例：将“Graham F. Sutherland”处理为“Graham Sutherland”）；
2. 需要生成名首字母+姓氏、名+姓氏首字母的邮箱组合（例：A.bonnet@xyz.com 或 Alex.b@xyz.com）；
3. SMTP验证时所有邮箱评分均为3分，无法区分高低分。

现有代码 #

import mysql.connector
import re
import requests
from googlesearch import search
import smtplib
import dns.resolver


def connect_db():
    try:
        conn = mysql.connector.connect(
            host="localhost",
            user="root",
            password="1234",
            database="usa"
        )
        return conn
    except mysql.connector.Error as err:
        print("Error:", err)
        exit()


def fetch_data():
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute("""
        SELECT k.ID, k.company_name, k.Person_Name, k.Designation, u.ID AS Company_ID, u.Company_Website 
        FROM Key_people_info k
        JOIN usa_companies_final u ON k.company_name = u.NAME
    """)
    data = cursor.fetchall()
    cursor.close()
    conn.close()
    return data


def extract_domain(website_url):
    match = re.search(r"https?://(?:www\.)?([^/]+)", website_url)
    if match:
        return match.group(1)
    return None


def generate_possible_emails(name, domain):
    first_name, last_name = name.lower().split(' ', 1)
    emails = [
        f"{first_name}.{last_name}@{domain}",
        f"{first_name}{last_name}@{domain}",
        f"{first_name}@{domain}",
        f"{first_name[0]}{last_name}@{domain}",
        f"{first_name[0]}.{last_name}@{domain}"
    ]
    return emails


def check_email_exists(email):
    domain = email.split('@')[-1]
    try:
        mx_records = dns.resolver.resolve(domain, 'MX')
        mx_record = str(mx_records[0].exchange)

        server = smtplib.SMTP()
        server.set_debuglevel(0)
        server.connect(mx_record)
        server.helo(server.local_hostname)
        server.mail('me@domain.com')
        code, message = server.rcpt(email)
        server.quit()

        return code == 250
    except Exception as e:
        return False


def search_google_for_info(name, company, designation):
    query = f"{name} {designation} {company} site:linkedin.com"
    search_results = list(search(query, num_results=5))

    linkedin_profile = None
    linkedin_name = None

    for url in search_results:
        if 'linkedin.com/in/' in url:
            linkedin_profile = url
            linkedin_name = url.split('/')[-1].split('?')[0]
            break

    email_query = f"{name} {company} email"
    search_results = list(search(email_query, num_results=10))

    emails_found = set()
    for url in search_results:
        try:
            response = requests.get(url, timeout=5)
            if response.status_code == 200:
                found_emails = re.findall(r"[\w\.-]+@[\w\.-]+", response.text)
                emails_found.update(found_emails)
        except:
            continue

    return linkedin_profile, linkedin_name, emails_found


def rate_email(email, smtp_status, found_in_google):
    if smtp_status:
        return 5
    elif email in found_in_google:
        return 4
    else:
        return 3


def process_and_store():
    data = fetch_data()
    results = []

    for row in data:
        person_id, company_name, person_name, designation, company_id, website = row
        print(f"\nProcessing: {person_name} ({designation}) at {company_name}")

        domain = extract_domain(website)
        if not domain:
            print("Could not extract domain.")
            continue

        linkedin_profile, linkedin_name, emails_found_in_google = search_google_for_info(person_name, company_name, designation)

        possible_emails = generate_possible_emails(person_name, domain)
        best_email = None
        best_rating = 0

        for email in possible_emails:
            smtp_valid = check_email_exists(email)
            rating = rate_email(email, smtp_valid, emails_found_in_google)
            print(f"Checked {email}: SMTP={smtp_valid}, In Google={email in emails_found_in_google}, Rating={rating}")
            if rating > best_rating:
                best_email = email
                best_rating = rating

        if best_email:
            results.append((company_id, company_name, person_name, designation, best_email, best_rating, linkedin_name, linkedin_profile))

    insert_results(results)


def insert_results(results):
    conn = connect_db()
    cursor = conn.cursor()
    for row in results:
        cursor.execute("""
            INSERT INTO usa_top_companies_key_people_email
            (Company_ID, Company_Name, Person_Name, Designation, Email_Id, Rating, LinkedIn_Name, LinkedIn_Profile)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
            ON DUPLICATE KEY UPDATE
                Email_Id = VALUES(Email_Id),
                Rating = VALUES(Rating),
                LinkedIn_Name = VALUES(LinkedIn_Name),
                LinkedIn_Profile = VALUES(LinkedIn_Profile)
        """, row)
    conn.commit()
    cursor.close()
    conn.close()


if __name__ == "__main__":
    process_and_store()

问题解决方案 #

1. 清理姓名，去除中间名 #

新增clean_full_name函数，过滤掉中间名（包括带点的缩写），仅保留名和姓：

def clean_full_name(full_name):
    name_parts = re.split(r'\s+', full_name.strip())
    if len(name_parts) >= 2:
        # 过滤掉单字母带点的中间名缩写
        filtered_parts = [part for part in name_parts if not re.match(r'^\w\.$', part)]
        if len(filtered_parts) >= 2:
            return f"{filtered_parts[0]} {filtered_parts[-1]}"
        return f"{name_parts[0]} {name_parts[-1]}"
    return full_name

在process_and_store中调用该函数处理原始姓名：

cleaned_name = clean_full_name(person_name)
possible_emails = generate_possible_emails(cleaned_name, domain)

2. 新增邮箱组合格式 #

修改generate_possible_emails函数，添加两种新的邮箱格式：

def generate_possible_emails(name, domain):
    first_name, last_name = name.lower().split(' ', 1)
    emails = [
        f"{first_name}.{last_name}@{domain}",
        f"{first_name}{last_name}@{domain}",
        f"{first_name}@{domain}",
        f"{first_name[0]}{last_name}@{domain}",
        f"{first_name[0]}.{last_name}@{domain}",
        # 新增名+姓氏首字母、名首字母+姓氏的格式
        f"{first_name}.{last_name[0]}@{domain}",
        f"{first_name}{last_name[0]}@{domain}"
    ]
    return emails

3. 修复SMTP验证与评分问题 #

问题分析

当前SMTP验证仅返回布尔值，无法区分明确无效邮箱和连接失败的情况；评分逻辑档位过少，导致无法区分不同可信度的邮箱。

解决方案

1. 修改check_email_exists，返回具体SMTP状态码，便于精准判断：

def check_email_exists(email):
    domain = email.split('@')[-1]
    try:
        mx_records = dns.resolver.resolve(domain, 'MX')
        mx_record = str(mx_records[0].exchange)

        server = smtplib.SMTP()
        server.set_debuglevel(0)
        server.connect(mx_record)
        server.helo(server.local_hostname)
        server.mail('me@domain.com')
        code, message = server.rcpt(email)
        server.quit()

        return code  # 返回具体状态码
    except dns.resolver.NXDOMAIN:
        return 400  # 域名不存在
    except smtplib.SMTPConnectError:
        return 500  # 无法连接SMTP服务器
    except Exception as e:
        return 0  # 其他未知错误

2. 优化rate_email函数，细化评分规则，覆盖更多场景：

def rate_email(email, smtp_code, found_in_google):
    # SMTP返回250：邮箱明确有效
    if smtp_code == 250:
        return 5
    # SMTP返回550：邮箱明确不存在
    elif smtp_code == 550:
        return 2
    # 谷歌搜索中精确匹配到该邮箱
    elif email in found_in_google:
        return 4
    # 域名无效或无法连接SMTP服务器，且未在谷歌找到
    elif smtp_code in (400, 500, 0):
        return 1
    # SMTP返回未知状态，未在谷歌找到
    else:
        return 3

3. 在process_and_store中调整调用逻辑，适配新的返回值：

smtp_code = check_email_exists(email)
smtp_valid = smtp_code == 250
rating = rate_email(email, smtp_code, emails_found_in_google)
print(f"Checked {email}: SMTP Code={smtp_code}, In Google={email in emails_found_in_google}, Rating={rating}")

内容的提问来源于stack exchange，提问作者Akash Saxena