13国语言社交媒体数据采集系统

class BaseScraper(abc.ABC):
    """Abstract base class for all platform scrapers."""

    def __init__(self, platform_name, proxy_manager=None, logger=None):
        self.platform_name = platform_name
        self.proxy_manager = proxy_manager or ProxyManager()
        self._consecutive_errors = 0
        self._max_consecutive_errors = 5

    @abc.abstractmethod
    def authenticate(self) -> bool:
        pass

    @abc.abstractmethod
    def search(self, query, **kwargs) -> Generator[Dict, None, None]:
        pass

    def rate_limit(self):
        """Apply rate limiting based on platform constraints."""
        min_interval = 1.0 / max(self._max_requests_per_second(), 0.1)
        elapsed = time.time() - self._last_request_time
        if elapsed < min_interval:
            time.sleep(min_interval - elapsed + random.uniform(0.1, 0.5))
        self._last_request_time = time.time()

    def make_request(self, method, url, headers=None, params=None, data=None,
                         use_proxy=True, max_retries=3):
        """HTTP request with proxy rotation, retry, and rate limiting."""
        self.rate_limit()
        for attempt in range(max_retries):
            try:
                proxy = self.get_proxy() if use_proxy else None
                response = requests.request(method, url, headers=headers,
                    params=params, json=data,
                    proxies=... if proxy else None, timeout=30)
                if response.status_code == 429:  # rate limited
                    time.sleep(int(response.headers.get('Retry-After', 30)))
                    continue
                response.raise_for_status()
                return response.json()
            except requests.exceptions.ProxyError as e:
                self.proxy_manager.mark_failed(proxy.id)
                time.sleep(random.uniform(1, 3))
        return None

class LanguageDetector:
    """Detect language and extract gender/age/relationship keywords from text."""

    def detect_language(self, text: str) -> Tuple[str, float]:
        """Detect the language of text. Returns (code, confidence)."""
        scores = {}
        # Check character-based patterns for non-Latin scripts
        for lang_code, pattern in LANG_PATTERNS.items():
            matches = pattern.findall(text)
            if matches:
                scores[lang_code] = len(matches) / max(len(text), 1)

        # Keyword-based detection for Latin-script languages
        for lang_code in latin_langs:
            matches = sum(1 for w in words if w in kw_set)
            if matches > 0:
                scores[lang_code] = scores.get(lang_code, 0) + matches / max(len(words), 1)

        best_lang = max(scores, key=scores.get) if scores else "unknown"
        confidence = scores[best_lang] / sum(scores.values()) if scores else 0
        return (best_lang, min(confidence, 1.0))

    def extract_keywords(self, text, lang=None) -> Dict:
        """Extract target keywords, exclude keywords, age/gender hints."""
        for lang_code in check_langs:
            kw_data = self.lang_data[lang_code]
            for kw in kw_data.get("keywords", []):
                if kw.lower() in text_lower:
                    result["matched_keywords"].append({kw, lang_code})
            # Check exclude keywords (married, etc.)
            for kw in kw_data.get("exclude_keywords", []):
                if kw.lower() in text_lower:
                    result["is_excluded"] = True
        return result

class MongoDBStorage:
    """MongoDB storage handler for social media scraper."""

    def __init__(self, logger=None):
        self.config = MONGO_CONFIG
        self.client = None
        self.collections = {}

    def connect(self) -> bool:
        """Connect to MongoDB and init collections."""
        self.client = MongoClient(self.config["uri"],
            maxPoolSize=50, serverSelectionTimeoutMS=5000)
        self.client.admin.command("ping")
        self.db = self.client[self.config["database"]]
        for name, col in self.config["collections"].items():
            self.collections[name] = self.db[col]
        return True

    def insert_raw_post(self, post) -> Optional[str]:
        """Insert a raw scraped post with dedup."""
        post["_stored_at"] = datetime.utcnow()
        result = self.collections["raw_posts"].insert_one(post)
        return str(result.inserted_id)

    def save_filtered_user(self, user_data) -> Optional[str]:
        """Save a filtered target user record."""
        user_data["_stored_at"] = datetime.utcnow()
        result = self.collections["filtered_users"].insert_one(user_data)
        return str(result.inserted_id)

def run_scrape(platforms, languages, demo_mode=True, max_posts=100):
    """Run the complete scraping pipeline."""
    scrapers = create_scrapers(proxy_manager, demo_mode)
    detector = LanguageDetector()
    filter_engine = UserFilter()
    cleaner = DataCleaner()

    for platform_name in platforms:
        scraper = scrapers[platform_name]
        scraper.authenticate()

        for lang in languages:
            posts = scraper.run_search_query(query=query, lang=lang,
                max_results=max_posts // len(languages))
            platform_posts.extend(posts)

        cleaned_posts = cleaner.clean_batch(platform_posts)

        for post in cleaned_posts:
            analysis = detector.process_text(post.get("text", ""), platform_name)
            is_target, _ = filter_engine.filter_post(post)
            if is_target:
                storage.save_filtered_user(post)

    return {"total_posts": len(all), "targets": len(targets)}


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="13-Language Social Media Scraper")
    parser.add_argument("--platforms", nargs="+", choices=["twitter", "reddit", "weibo", "vk"])
    parser.add_argument("--demo", action="store_true", default=True)
    parser.add_argument("--schedule", type=int, default=0, help="Run on schedule (minutes)")
    args = parser.parse_args()
    main()