app.py

import snscrape.modules.twitter as sntwitter
import gradio as gr
import threading
import time

# 用于存储已展示的推文 ID
seen_ids = set()

# 控制循环的开关
running = False

def search_tweets(keyword):
    """生成器：按关键词抓取推文"""
    for tweet in sntwitter.TwitterSearchScraper(keyword).get_items():
        if tweet.id not in seen_ids:
            seen_ids.add(tweet.id)

            media_urls = []
            if tweet.media:
                for m in tweet.media:
                    if hasattr(m, "fullUrl"):
                        media_urls.append(m.fullUrl)

            yield {
                "text": tweet.content,
                "url": tweet.url,
                "media": media_urls,
            }

def start_stream(keyword):
    """后台线程：不断抓取推文并返回"""
    global running
    running = True
    tweets = search_tweets(keyword)
    for tweet in tweets:
        if not running:
            break
        media = tweet["media"][0] if tweet["media"] else None
        # 判断是图片还是视频
        if media and (media.endswith(".mp4") or "video" in media):
            yield tweet["text"], tweet["url"], None, media
        else:
            yield tweet["text"], tweet["url"], media, None
        time.sleep(3)  # 控制间隔

def stop_stream():
    """停止按钮调用"""
    global running
    running = False
    return "已停止", "", None, None


with gr.Blocks() as demo:
    gr.Markdown("## 🐦 推文展示工具（自动播放）")

    keyword = gr.Textbox(label="搜索关键词", value="AI")
    start_btn = gr.Button("开始")
    stop_btn = gr.Button("停止")

    output_text = gr.Textbox(label="推文内容")
    output_url = gr.Textbox(label="推文链接")
    output_img = gr.Image(label="推文图片")
    output_video = gr.Video(label="推文视频")

    # 开始按钮：流式展示
    start_btn.click(
        fn=start_stream,
        inputs=[keyword],
        outputs=[output_text, output_url, output_img, output_video],
        api_name="start",
        show_progress=True,
    )

    # 停止按钮
    stop_btn.click(
        fn=stop_stream,
        inputs=[],
        outputs=[output_text, output_url, output_img, output_video],
    )

demo.launch()