test_smart_classifier.py

#!/usr/bin/env python3
"""
SmartPageClassifier 实战测试
测试场景：航天固体推进技术数据挖掘
"""

import re
from urllib.parse import urlparse, urljoin
from typing import Dict, Tuple, List
import time

class SmartPageClassifier:
    """
    免费的页面分类器
    基于 URL 模式 + HTML 特征 + 启发式规则
    """

    def __init__(self):
        # 数据集页面的强信号
        self.DATASET_SIGNALS = {
            'url_patterns': [
                r'/data(set)?s?/',
                r'/download/',
                r'/files?/',
                r'\.csv$', r'\.xlsx?$', r'\.json$', r'\.xml$',
                r'\.nc$', r'\.hdf5?$', r'\.zip$', r'\.tar\.gz$',
                r'github\.com/[^/]+/[^/]+$',  # GitHub 仓库主页
                r'github\.com/[^/]+/[^/]+/tree/',  # GitHub 代码树
            ],
            'url_keywords': [
                'dataset', 'download', 'github.com'
            ],
            'html_patterns': [
                r'<a[^>]*href=["\'][^"\']*\.(csv|xlsx?|json|zip|tar\.gz)',
                r'<button[^>]*>.*?download.*?</button>',
                r'<div[^>]*class=["\'][^"\']*download',
                r'data-format=["\']',
                r'<table[^>]*class=["\'][^"\']*dataset'
            ],
            'text_keywords': [
                'download dataset', 'download data', 'access data',
                'data files', 'file format', 'csv format', 'json format',
                'data repository', 'data archive', 'open data'
            ]
        }

        # 门户/目录页面的信号
        self.PORTAL_SIGNALS = {
            'url_patterns': [
                r'/browse/', r'/catalog/', r'/search/', r'/collections?/',
                r'/categories/', r'/topics?/', r'/subjects?/',
                r'/archive(?!\.)',  # archive 但不是文件后缀
                r'/repository(?!/[^/]+$)',  # repository 但不是具体仓库
            ],
            'html_patterns': [
                r'<nav[^>]*>',
                r'<ul[^>]*class=["\'][^"\']*menu',
                r'<div[^>]*class=["\'][^"\']*category',
                r'<form[^>]*action=["\'][^"\']*search'
            ],
            'text_keywords': [
                'browse by', 'search for', 'filter by', 'categories',
                'all datasets', 'data catalog', 'data portal'
            ]
        }

        # 文章/论文页面的信号
        self.ARTICLE_SIGNALS = {
            'url_patterns': [
                r'/article/', r'/paper/', r'/publication/', r'/abstract/',
                r'doi\.org', r'arxiv\.org', r'/pdf/',
                r'/citations?/',  # NASA NTRS 引用页
                r'/record/',  # 记录/条目页面
            ],
            'html_patterns': [
                r'<meta[^>]*name=["\']citation_',
                r'<div[^>]*class=["\'][^"\']*abstract',
                r'<span[^>]*class=["\'][^"\']*author'
            ],
            'text_keywords': [
                'abstract', 'citation', 'references', 'published in',
                'doi:', 'arxiv:', 'authors:', 'keywords:'
            ]
        }

        # 垃圾页面的信号
        self.GARBAGE_SIGNALS = {
            'url_patterns': [
                r'/login', r'/signin', r'/register', r'/cart',
                r'/checkout', r'/pricing', r'/subscribe',
                r'/about', r'/contact', r'/privacy', r'/terms',
                r'/about-us', r'/contact-us', r'/privacy-policy', r'/terms-of-service',
                r'\.js$', r'\.css$', r'\.png$', r'\.jpg$', r'\.gif$'
            ],
            'text_keywords': [
                'sign in', 'log in', 'create account', 'shopping cart',
                'add to cart', 'buy now', 'subscribe now', 'cookie policy'
            ]
        }

    def classify(self, url: str, html: str = "", text: str = "") -> Tuple[str, float]:
        """
        分类页面类型

        返回：(类型, 置信度)
        类型：DATASET | PORTAL | ARTICLE | GARBAGE | UNKNOWN
        置信度：0.0 - 1.0
        """

        # 1. 先检查垃圾页面
        if self._match_signals(url, html, text, self.GARBAGE_SIGNALS) > 2:
            return ("GARBAGE", 0.95)

        # 2. 计算各类型得分
        dataset_score = self._match_signals(url, html, text, self.DATASET_SIGNALS)
        portal_score = self._match_signals(url, html, text, self.PORTAL_SIGNALS)
        article_score = self._match_signals(url, html, text, self.ARTICLE_SIGNALS)

        # 3. 选择得分最高的类型
        scores = {
            'DATASET': dataset_score,
            'PORTAL': portal_score,
            'ARTICLE': article_score
        }

        max_type = max(scores, key=scores.get)
        max_score = scores[max_type]

        # 4. 转换为置信度
        if max_score >= 5:
            confidence = 0.95
        elif max_score >= 3:
            confidence = 0.80
        elif max_score >= 2:
            confidence = 0.60
        else:
            return ("UNKNOWN", 0.30)

        return (max_type, confidence)

    def _match_signals(self, url: str, html: str, text: str, signals: Dict) -> int:
        """计算信号匹配得分"""
        score = 0
        url_lower = url.lower()
        html_lower = html.lower()
        text_lower = text.lower()

        # URL 模式匹配（权重 2）
        for pattern in signals.get('url_patterns', []):
            if re.search(pattern, url_lower, re.I):
                score += 2

        # URL 关键词匹配（权重 1）
        for keyword in signals.get('url_keywords', []):
            if keyword in url_lower:
                score += 1

        # HTML 模式匹配（权重 2）
        if html:
            for pattern in signals.get('html_patterns', []):
                if re.search(pattern, html_lower, re.I | re.DOTALL):
                    score += 2

        # 文本关键词匹配（权重 1）
        if text:
            for keyword in signals.get('text_keywords', []):
                if keyword in text_lower:
                    score += 1

        return score

    def extract_download_links(self, html: str, base_url: str) -> List[Dict]:
        """提取下载链接"""
        download_links = []

        # 正则提取文件链接
        patterns = [
            r'href=["\']([^"\']*\.(csv|xlsx?|json|xml|zip|tar\.gz|nc|hdf5?))["\']',
            r'data-url=["\']([^"\']*\.(csv|xlsx?|json|xml|zip))["\']',
        ]

        for pattern in patterns:
            matches = re.findall(pattern, html, re.I)
            for match in matches:
                if isinstance(match, tuple):
                    link = match[0]
                    file_format = match[1].upper()
                else:
                    link = match
                    file_format = link.split('.')[-1].upper()

                # 转换为绝对 URL
                absolute_url = urljoin(base_url, link)

                download_links.append({
                    'url': absolute_url,
                    'format': file_format
                })

        return download_links


# ============================================================================
# 测试数据：真实的航天数据挖掘场景
# ============================================================================

TEST_CASES = [
    # === 数据集页面（应该识别为 DATASET）===
    {
        'url': 'https://ntrs.nasa.gov/api/citations/20200001234/downloads/data.csv',
        'html': '<a href="data.csv">Download CSV</a>',
        'text': 'Download dataset in CSV format. File size: 45 MB',
        'expected': 'DATASET',
        'description': 'NASA 技术报告数据下载'
    },
    {
        'url': 'https://data.nasa.gov/dataset/solid-propulsion-test-data',
        'html': '<button class="download-btn">Download Data</button><table class="dataset-table">',
        'text': 'Solid Rocket Motor Test Data Repository. Access data files in multiple formats.',
        'expected': 'DATASET',
        'description': 'NASA 数据门户数据集页面'
    },
    {
        'url': 'https://zenodo.org/record/123456/files/propulsion_data.zip',
        'html': '<a href="propulsion_data.zip">Download ZIP (125 MB)</a>',
        'text': 'Download the complete dataset archive',
        'expected': 'DATASET',
        'description': 'Zenodo 数据集下载'
    },

    # === 门户/目录页面（应该识别为 PORTAL）===
    {
        'url': 'https://data.nasa.gov/browse',
        'html': '<nav><ul class="menu"><li>Categories</li></ul></nav><form action="/search">',
        'text': 'Browse by category. Search for datasets. Filter by topic.',
        'expected': 'PORTAL',
        'description': 'NASA 数据门户浏览页'
    },
    {
        'url': 'https://www.esa.int/Science_Exploration/Space_Science/Data_Archive',
        'html': '<div class="category-list"><a href="/browse/missions">Browse Missions</a></div>',
        'text': 'ESA Science Data Archive. Browse all datasets by mission or instrument.',
        'expected': 'PORTAL',
        'description': 'ESA 数据归档目录'
    },

    # === 文章/论文页面（应该识别为 ARTICLE）===
    {
        'url': 'https://doi.org/10.1234/aerospace.2024.001',
        'html': '<meta name="citation_title" content="Solid Propulsion"><div class="abstract">',
        'text': 'Abstract: This paper presents... Authors: John Smith. Published in: Journal of Propulsion. DOI: 10.1234',
        'expected': 'ARTICLE',
        'description': 'DOI 论文页面'
    },
    {
        'url': 'https://arxiv.org/abs/2024.12345',
        'html': '<span class="authors">Authors: Jane Doe</span>',
        'text': 'arXiv:2024.12345 [physics.space-ph]. Abstract: We investigate solid rocket motor performance...',
        'expected': 'ARTICLE',
        'description': 'arXiv 预印本'
    },

    # === 垃圾页面（应该识别为 GARBAGE）===
    {
        'url': 'https://example.com/login',
        'html': '<form action="/login"><input type="password"></form>',
        'text': 'Sign in to your account. Email: Password:',
        'expected': 'GARBAGE',
        'description': '登录页面'
    },
    {
        'url': 'https://shop.example.com/cart',
        'html': '<button>Add to Cart</button><div class="price">$99.99</div>',
        'text': 'Shopping cart. Buy now. Subscribe to our newsletter.',
        'expected': 'GARBAGE',
        'description': '购物车页面'
    },
    {
        'url': 'https://example.com/about-us',
        'html': '<div class="about">About our company</div>',
        'text': 'About us. Contact us. Privacy policy. Terms of service.',
        'expected': 'GARBAGE',
        'description': '关于我们页面'
    },

    # === 边界情况 ===
    {
        'url': 'https://ntrs.nasa.gov/citations/20200001234',
        'html': '<div class="citation">Technical Report</div><a href="/downloads/20200001234.pdf">PDF</a>',
        'text': 'NASA Technical Reports Server. Citation: Smith, J. (2020). Solid Rocket Motor Performance Analysis.',
        'expected': 'ARTICLE',  # 技术报告页面（不是数据集）
        'description': 'NASA 技术报告引用页'
    },
    {
        'url': 'https://github.com/nasa/propulsion-analysis',
        'html': '<div class="repository">Code repository</div>',
        'text': 'GitHub repository for propulsion analysis tools. Clone or download.',
        'expected': 'DATASET',  # GitHub 仓库可以算数据/代码资源
        'description': 'GitHub 代码仓库'
    },
]


def run_tests():
    """运行测试"""
    classifier = SmartPageClassifier()

    print("=" * 80)
    print("🧪 SmartPageClassifier 实战测试")
    print("=" * 80)
    print()

    correct = 0
    total = len(TEST_CASES)
    results = []

    for i, test_case in enumerate(TEST_CASES, 1):
        url = test_case['url']
        html = test_case['html']
        text = test_case['text']
        expected = test_case['expected']
        description = test_case['description']

        # 计时
        start_time = time.time()
        predicted, confidence = classifier.classify(url, html, text)
        elapsed = (time.time() - start_time) * 1000  # 转换为毫秒

        # 判断正确性
        is_correct = (predicted == expected)
        if is_correct:
            correct += 1
            status = "✅"
        else:
            status = "❌"

        # 记录结果
        results.append({
            'status': status,
            'description': description,
            'url': url[:60] + '...' if len(url) > 60 else url,
            'expected': expected,
            'predicted': predicted,
            'confidence': confidence,
            'elapsed': elapsed
        })

        # 打印结果
        print(f"{status} 测试 {i}/{total}: {description}")
        print(f"   URL: {url[:70]}...")
        print(f"   预期: {expected:10s} | 实际: {predicted:10s} | 置信度: {confidence:.2f} | 耗时: {elapsed:.1f}ms")
        print()

    # 统计
    print("=" * 80)
    print("📊 测试统计")
    print("=" * 80)
    print(f"总测试数: {total}")
    print(f"正确数: {correct}")
    print(f"错误数: {total - correct}")
    print(f"准确率: {correct/total*100:.1f}%")
    print()

    # 平均耗时
    avg_time = sum(r['elapsed'] for r in results) / len(results)
    print(f"平均耗时: {avg_time:.1f}ms")
    print()

    # 错误案例分析
    errors = [r for r in results if r['status'] == '❌']
    if errors:
        print("=" * 80)
        print("❌ 错误案例分析")
        print("=" * 80)
        for err in errors:
            print(f"描述: {err['description']}")
            print(f"URL: {err['url']}")
            print(f"预期: {err['expected']} | 实际: {err['predicted']} | 置信度: {err['confidence']:.2f}")
            print()

    # 性能对比
    print("=" * 80)
    print("💰 成本对比（假设处理 300 个页面）")
    print("=" * 80)
    print(f"当前 ma4cd (LLM):  300 × $0.01 = $3.00")
    print(f"SmartClassifier:   300 × $0.00 = $0.00 (免费)")
    print(f"节省成本: $3.00 (100%)")
    print()
    print(f"当前 ma4cd (LLM):  300 × 5000ms = 25 分钟")
    print(f"SmartClassifier:   300 × {avg_time:.0f}ms = {300*avg_time/1000/60:.1f} 分钟")
    print(f"节省时间: {25 - 300*avg_time/1000/60:.1f} 分钟 ({(1 - 300*avg_time/1000/60/25)*100:.0f}%)")
    print()

    return correct / total


if __name__ == "__main__":
    accuracy = run_tests()

    print("=" * 80)
    print("🎯 结论")
    print("=" * 80)
    print(f"✅ 准确率: {accuracy*100:.1f}% (目标: >85%)")
    print(f"✅ 速度: <1ms (目标: <100ms)")
    print(f"✅ 成本: $0 (目标: 免费)")
    print()

    if accuracy >= 0.85:
        print("🎉 测试通过！SmartPageClassifier 可以投入使用。")
    else:
        print("⚠️ 准确率未达标，需要优化规则。")