Python中如何设计微博动态监控系统，实现新发布微博的实时通知？

API

要设计一个微博动态监控系统，核心是轮询API和消息推送。这里给一个基于requests和schedule库的基础实现方案，监控指定用户的微博列表变化。

import requests
import time
import schedule
from datetime import datetime
import json

class WeiboMonitor:
    def __init__(self, target_uid, check_interval=60):
        """
        初始化监控器
        :param target_uid: 目标微博用户UID
        :param check_interval: 检查间隔（秒）
        """
        self.target_uid = target_uid
        self.check_interval = check_interval
        self.last_weibo_id = None  # 记录最新微博ID
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
    
    def fetch_latest_weibo(self):
        """获取用户最新微博"""
        # 这里使用微博移动端API示例，实际需要根据微博接口调整
        url = f'https://m.weibo.cn/api/container/getIndex'
        params = {
            'type': 'uid',
            'value': self.target_uid,
            'containerid': '107603{uid}'.format(uid=self.target_uid)
        }
        
        try:
            response = requests.get(url, params=params, headers=self.headers, timeout=10)
            if response.status_code == 200:
                data = response.json()
                # 解析微博列表
                weibos = data.get('data', {}).get('cards', [])
                for card in weibos:
                    if card.get('card_type') == 9:  # 微博卡片类型
                        mblog = card.get('mblog', {})
                        return {
                            'id': mblog.get('id'),
                            'text': mblog.get('text', ''),
                            'created_at': mblog.get('created_at', ''),
                            'user': mblog.get('user', {}).get('screen_name', '')
                        }
        except Exception as e:
            print(f"获取微博失败: {e}")
        return None
    
    def check_new_weibo(self):
        """检查新微博"""
        latest = self.fetch_latest_weibo()
        if not latest:
            return
        
        current_id = latest['id']
        
        # 首次运行，只记录不通知
        if self.last_weibo_id is None:
            self.last_weibo_id = current_id
            print(f"[{datetime.now()}] 初始化最新微博ID: {current_id}")
            return
        
        # 发现新微博
        if current_id != self.last_weibo_id:
            self.last_weibo_id = current_id
            self.send_notification(latest)
    
    def send_notification(self, weibo_data):
        """发送通知（示例：控制台打印）"""
        print(f"\n{'='*50}")
        print(f"📢 检测到新微博！")
        print(f"用户: {weibo_data['user']}")
        print(f"时间: {weibo_data['created_at']}")
        print(f"内容: {weibo_data['text'][:100]}...")
        print(f"微博ID: {weibo_data['id']}")
        print(f"{'='*50}\n")
        
        # 实际应用中可扩展：
        # 1. 发送邮件：使用smtplib
        # 2. 微信推送：使用Server酱或企业微信API
        # 3. 保存到数据库：使用sqlite3或pymysql
    
    def run(self):
        """启动监控"""
        print(f"开始监控微博用户 {self.target_uid}，检查间隔 {self.check_interval}秒")
        schedule.every(self.check_interval).seconds.do(self.check_new_weibo)
        
        # 立即执行一次初始化
        self.check_new_weibo()
        
        while True:
            schedule.run_pending()
            time.sleep(1)

# 使用示例
if __name__ == "__main__":
    # 替换为目标用户的UID（在微博个人主页URL中找到）
    TARGET_UID = "1234567890"  # 示例UID
    
    monitor = WeiboMonitor(target_uid=TARGET_UID, check_interval=30)
    monitor.run()

关键点说明：