如何解决Gmail中供应商与产品信息提取的正则及Spacy失效问题

需要从Gmail邮件中提取采购相关字段（Mail_Date、Mail_Subject、Product_Name、Product_Quantity、Product_Price、Vendor_Name、Vendor_Email、Vendor_Phone、Vendor_Address、Vendor_GST_No、Vendor_Website）并导出到Excel，但现有方案存在两个核心问题：

• 正则表达式仅在数据结构完全一致时有效，非结构化场景下会提取出"aaa""676776"这类随机无效值
• 使用Spacy默认模型的NLP提取逻辑输出错误，无法准确关联产品、数量和价格

一、正则匹配优化：增加语义约束与上下文过滤 #

1. 供应商信息提取优化 #

• 邮箱提取：优先匹配签名区域的邮箱，而非全文任意邮箱
• 手机号提取：增加前后非数字边界约束，避免匹配订单号、物流号等长数字串
• 签名提取：支持多种常见签名前缀（Thanks/Best regards等），过滤纯数字/乱码行
• 新增地址、GST号提取：针对采购场景的格式规则匹配

2. 产品信息提取优化 #

• 产品正则增加关键词前缀（如"Product:""Item:"），减少无关文本匹配
• 表格格式匹配增加表头关键词过滤，确保只匹配采购清单表格
• 无效值过滤：过滤长度过短或纯数字的产品名

二、Spacy NLP增强：规则匹配+上下文关联 #

1. 自定义规则匹配器 #

用Spacy的Matcher替代单纯依赖默认NER，针对采购场景定义产品+数量+价格的组合匹配规则

2. 上下文关联逻辑 #

通过句子分割，确保产品、数量、价格属于同一业务条目，避免跨条目错误关联

3. 实体验证：通过依存关系验证数量与产品的关联性 #

修改后的完整代码 #

import re
import yaml
import imaplib
import spacy
import pandas as pd
from email import message_from_bytes
from bs4 import BeautifulSoup
from dateutil import parser
from spacy.matcher import Matcher

# 加载模型并初始化Matcher
nlp = spacy.load("en_core_web_sm")
matcher = Matcher(nlp.vocab)

# 定义采购领域匹配规则：产品+数量+价格的常见组合
product_rules = [
    # 匹配包含产品关键词的名词
    [{"LOWER": {"IN": ["product", "item", "goods"]}, "OP": "?"}, {"POS": "NOUN", "DEP": "ROOT", "OP": "+"}],
    # 匹配数量相关数字
    [{"LOWER": {"IN": ["qty", "quantity", "units", "nos"]}, "OP": "?"}, {"POS": "NUM", "ENT_TYPE": "CARDINAL"}],
    # 匹配价格相关数字
    [{"LOWER": {"IN": ["price", "cost", "rate"]}, "OP": "?"}, {"POS": "NUM", "ENT_TYPE": "MONEY"}]
]
matcher.add("PRODUCT_QUANTITY_PRICE", [product_rules])

class ProcurementEmailParser:
    def __init__(self, credentials_path):
        self.credentials = self.load_credentials(credentials_path)
        self.mail = None
        self.output_columns = [
            'Mail_Date', 'Mail_Subject', 'Product_Name', 
            'Product_Quantity', 'Product_Price',
            'Vendor_Name', 'Vendor_Email', 'Vendor_Phone', 
            'Vendor_Website', 'Vendor_Address', 'Vendor_GST_No'
        ]
    
    def load_credentials(self, path):
        with open(path) as f:
            return yaml.safe_load(f)
    
    def connect_email(self):
        self.mail = imaplib.IMAP4_SSL('imap.gmail.com')
        self.mail.login(self.credentials['email'], self.credentials['password'])
        self.mail.select('inbox')
    
    def extract_vendor_info(self, text):
        vendor_info = {
            'Vendor_Name': '',
            'Vendor_Email': '',
            'Vendor_Phone': '',
            'Vendor_Website': '',
            'Vendor_Address': '',
            'Vendor_GST_No': ''
        }
        
        # 提取签名块：支持多种常见签名前缀
        signature_prefixes = ['Regards,', 'Thanks,', 'Best regards,', 'Warm regards,', 'Sincerely,']
        signature_block = ''
        for prefix in signature_prefixes:
            if prefix in text:
                signature_block = text.split(prefix)[-1].strip()
                break
        
        # 优先从签名块提取供应商信息
        if signature_block:
            lines = [line.strip() for line in signature_block.split('\n') if line.strip()]
            # 提取供应商名称：过滤纯数字/乱码行，优先取非邮箱/手机号的第一行
            for line in lines:
                if not re.match(r'^\d+$', line) and '@' not in line and not re.match(r'^\+?\d+$', line):
                    vendor_info['Vendor_Name'] = re.sub(r'[^a-zA-Z0-9\s&.,]', '', line).strip()
                    break
            
            # 从签名块提取邮箱
            email_match = re.search(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', signature_block)
            if email_match:
                vendor_info['Vendor_Email'] = email_match.group(0)
            
            # 从签名块提取手机号：增加前后非数字约束，避免匹配长数字串
            phone_match = re.search(r'(?:^|\D)(?:\+?91[\s-]?)?[6-9]\d{9}(?:$|\D)', signature_block)
            if phone_match:
                vendor_info['Vendor_Phone'] = re.sub(r'\D', '', phone_match.group(0))
            
            # 从签名块提取网站
            website_match = re.search(r'(?:https?://)?(?:www\.)?[\w.-]+\.[a-zA-Z]{2,}', signature_block)
            if website_match:
                vendor_info['Vendor_Website'] = website_match.group(0)
            
            # 提取地址：匹配包含城市/邮编的行
            address_lines = []
            for line in lines:
                if re.search(r'(?:city|state|pin|postal|address)', line.lower()) or re.match(r'.*\d{6}.*', line):
                    address_lines.append(line)
            if address_lines:
                vendor_info['Vendor_Address'] = ' '.join(address_lines)
            
            # 提取GST号：匹配印度GST格式
            gst_match = re.search(r'GSTIN:\s*[0-9A-Z]{15}', signature_block, re.IGNORECASE)
            if gst_match:
                vendor_info['Vendor_GST_No'] = gst_match.group(0).replace('GSTIN:', '').strip()
        
        # 若签名块无结果，从全文补充提取（但优先级低）
        if not vendor_info['Vendor_Email']:
            email_match = re.search(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', text)
            if email_match:
                vendor_info['Vendor_Email'] = email_match.group(0)
        
        return vendor_info
    
    def safe_float_conversion(self, value):
        try:
            if not value:
                return None
            # 移除货币符号和逗号
            cleaned = re.sub(r'[₹$]', '', value).replace(',', '')
            return float(cleaned)
        except:
            return None
    
    def extract_products(self, text):
        products = []
        
        # 优化后的表格格式匹配：增加表头关键词过滤
        if re.search(r'(?:item|product)\s+(?:description|name)\s+(?:qty|quantity)\s+(?:price|rate)', text, re.IGNORECASE):
            table_pattern = r'(\d+)\s+(.+?)\s+(\d+)\s+([\d,]+)\s+([\d,]+)'
            for match in re.finditer(table_pattern, text):
                product_name = match.group(2).strip()
                # 过滤无效产品名（长度过短或纯数字）
                if len(product_name) > 2 and not product_name.isdigit():
                    products.append({
                        'Product_Name': product_name,
                        'Product_Quantity': int(match.group(3)),
                        'Product_Price': self.safe_float_conversion(match.group(5))
                    })
        
        # 优化后的行项目匹配：增加关键词前缀
        line_pattern = r'(?:Product:|Item:|Goods:)\s*(.+?)\s+[-–]\s+(\d+)\s*(?:nos|units|qty)\s*[-–]?\s*([₹$]?[\d,]+)?'
        for match in re.finditer(line_pattern, text, re.IGNORECASE):
            product_name = match.group(1).strip()
            if len(product_name) > 2 and not product_name.isdigit():
                products.append({
                    'Product_Name': product_name,
                    'Product_Quantity': int(match.group(2)),
                    'Product_Price': self.safe_float_conversion(match.group(3))
                })
        
        # 增强版NLP提取：规则匹配+上下文关联
        if not products:
            doc = nlp(text)
            # 按句子分割，确保每个条目在同一句子内
            for sent in doc.sents:
                matches = matcher(sent)
                if matches:
                    product_data = {'name': '', 'qty': None, 'price': None}
                    # 提取句子中的实体
                    for ent in sent.ents:
                        if ent.label_ == 'PRODUCT' or (ent.label_ == 'ORG' and len(ent.text) > 2):
                            product_data['name'] = ent.text
                        elif ent.label_ == 'CARDINAL':
                            # 验证数量是否和产品相关（通过依存关系）
                            if any(token.dep_ in ['nummod', 'attr'] for token in ent.root.children):
                                product_data['qty'] = int(ent.text)
                        elif ent.label_ == 'MONEY':
                            product_data['price'] = self.safe_float_conversion(ent.text)
                    
                    # 只有当产品名和数量都存在时才添加
                    if product_data['name'] and product_data['qty'] is not None:
                        products.append({
                            'Product_Name': product_data['name'],
                            'Product_Quantity': product_data['qty'],
                            'Product_Price': product_data['price']
                        })
        
        return products
    
    def process_email(self, email_msg):
        try:
            text_content = ''
            for part in email_msg.walk():
                if part.get_content_type() == 'text/plain':
                    text_content += part.get_payload(decode=True).decode('utf-8', 'ignore')
                elif part.get_content_type() == 'text/html':
                    html_content = part.get_payload(decode=True).decode('utf-8', 'ignore')
                    soup = BeautifulSoup(html_content, 'html.parser')
                    text_content += '\n' + soup.get_text(separator='\n', strip=True)
            
            # 保留换行符，便于签名块和句子分割
            text_content = re.sub(r'\s+', ' ', text_content).strip()
            
            vendor_info = self.extract_vendor_info(text_content)
            products = self.extract_products(text_content)
            
            records = []
            for product in products:
                record = {
                    'Mail_Date': parser.parse(email_msg['Date']).strftime('%Y-%m-%d %H:%M:%S'),
                    'Mail_Subject': email_msg.get('Subject', 'No Subject'),
                    **product,
                    **vendor_info
                }
                records.append(record)
            
            return records
        
        except Exception as e:
            print(f"处理邮件出错: {str(e)}")
            return []
    
    def process_emails(self, limit=50, save_path='output.xlsx'):
        self.connect_email()
        _, msg_ids = self.mail.search(None, 'ALL')
        all_data = []
        
        for msg_id in msg_ids[0].split()[-limit:]:
            try:
                _, msg_data = self.mail.fetch(msg_id, '(RFC822)')
                email_msg = message_from_bytes(msg_data[0][1])
                all_data.extend(self.process_email(email_msg))
            except Exception as e:
                print(f"处理邮件 {msg_id.decode()} 出错: {str(e)}")
                continue
        
        df = pd.DataFrame(all_data, columns=self.output_columns)
        df.to_excel(save_path, index=False)
        return df

if __name__ == "__main__":
    parser = ProcurementEmailParser(r"C:\Users\one\credentials.yml")
    df = parser.process_emails(limit=50, save_path=r"C:\Users\one\vp_details.xlsx")
    print(f"成功处理 {len(df)} 条记录")

关键优化点说明 #

1. 签名信息过滤：优先从签名块提取供应商信息，避免抓取邮件正文内的无关邮箱/手机号
2. 正则语义约束：产品匹配增加"Product:"/"Item:"等关键词前缀，减少误匹配
3. NLP规则增强：使用Spacy Matcher定义采购领域规则，结合句子分割确保产品与数量/价格的关联准确性
4. 无效值过滤：增加产品名长度、数字校验，过滤"aaa""676776"这类无效条目
5. 扩展字段支持：新增Vendor_Address和Vendor_GST_No的提取逻辑

内容的提问来源于stack exchange，提问作者Krishnakumar Jr.AI developer