AI大模型训练数据来源：如何用代理IP规范获取多语种语料?

"""
多语种语料采集框架
集成网帆代理动态住宅IP + 语言识别 + 质量评分
"""
import requests
import time
import hashlib
from collections import defaultdict

class CorpusCollector:
    def __init__(self, api_key):
        self.api_key = api_key
        self.api_url = "https://api.fanproxy.com/api/proxies"
        self.collected = []
        self.url_seen = set()
        self.content_hashes = set()

    def get_residential_ip(self, country, session_duration=30):
        """获取指定国家的动态住宅IP"""
        params = {
            "api_key": self.api_key,
            "country": country,
            "count": 1,
            "format": "json",
            "protocol": "http"
        }
        try:
            resp = requests.get(self.api_url, params=params, timeout=10)
            data = resp.json().get("data", [])
            return data[0] if data else None
        except Exception:
            return None

    def collect_page(self, url, country):
        """采集单个页面"""
        # URL去重
        if url in self.url_seen:
            return None
        self.url_seen.add(url)

        # 获取目标国家住宅IP
        proxy_info = self.get_residential_ip(country)
        if not proxy_info:
            return None

        proxy_url = f"http://{proxy_info['ip']}:{proxy_info['port']}"
        proxies = {"http": proxy_url, "https": proxy_url}

        try:
            # 模拟真实用户请求
            session = requests.Session()
            session.proxies = proxies
            session.headers.update({
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                              "AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36",
                "Accept-Language": "en-US,en;q=0.9"
            })

            response = session.get(url, timeout=15)
            if response.status_code != 200:
                return None

            content = response.text

            # 内容指纹去重
            content_hash = hashlib.md5(content.encode()).hexdigest()
            if content_hash in self.content_hashes:
                return None
            self.content_hashes.add(content_hash)

            # 质量评分
            quality = self.score_quality(content)

            result = {
                "url": url,
                "country": country,
                "ip": proxy_info["ip"],
                "status": response.status_code,
                "content_length": len(content),
                "quality_score": quality,
                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
            }

            self.collected.append(result)
            return result

        except Exception:
            return None

    def score_quality(self, text):
        """基础质量评分（0-100）"""
        score = 0
        # 文本长度评分
        if len(text) > 500:
            score += 30
        elif len(text) > 200:
            score += 15
        # 信息密度（简单评估）
        if len(text) > 0:
            alpha_ratio = sum(c.isalpha() for c in text) / len(text)
            if alpha_ratio > 0.5:
                score += 30
        # 编码完整性
        try:
            text.encode('utf-8')
            score += 20
        except Exception:
            pass
        # 结构完整性（含HTML标签）
        if '<p>' in text or '<article>' in text:
            score += 20
        return score

    def collect_corpus(self, url_list, country):
        """批量采集某国家的语料"""
        results = []
        for url in url_list:
            result = self.collect_page(url, country)
            if result and result["quality_score"] >= 60:
                results.append(result)
            # 请求间隔（模拟真实用户）
            time.sleep(3 + hash(url) % 5)  # 3-7秒随机间隔
        return results

# 使用示例
collector = CorpusCollector("your_api_key")

# 采集肯尼亚的斯瓦希里语语料
swahili_urls = [
    "https://example-kenyan-news.co.ke/article/1",
    "https://example-kenyan-news.co.ke/article/2",
]
results = collector.collect_corpus(swahili_urls, country="KE")
for r in results:
    print(f"URL: {r['url']} | 质量: {r['quality_score']} | IP: {r['ip']}")