dataapp_debug.py

import os
import time
import urllib.parse
from io import BytesIO

import pandas as pd
import streamlit as st
from playwright.sync_api import sync_playwright

st.set_page_config(page_title="小红书筛选调试器", page_icon="🧪", layout="wide")

USER_DATA_DIR = os.path.join(os.getcwd(), "xhs_browser_profile")
os.makedirs(USER_DATA_DIR, exist_ok=True)

LAUNCH_ARGS = ["--disable-blink-features=AutomationControlled"]


def apply_stealth(page):
    page.add_init_script("""
        Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
    """)
    return page


def open_robot_browser(url="https://www.xiaohongshu.com/explore"):
    with sync_playwright() as p:
        browser = p.chromium.launch_persistent_context(
            user_data_dir=USER_DATA_DIR,
            headless=False,
            viewport={"width": 1600, "height": 900},
            args=LAUNCH_ARGS
        )
        page = browser.pages[0] if browser.pages else browser.new_page()
        apply_stealth(page)
        try:
            page.goto(url, wait_until="domcontentloaded", timeout=20000)
            page.wait_for_event("close", timeout=0)
        finally:
            try:
                browser.close()
            except:
                pass


def get_top_titles(page, n=3):
    titles = []
    items = page.locator("section.note-item")
    count = min(n, items.count())
    for i in range(count):
        card = items.nth(i)
        try:
            title = card.locator(".title").first.inner_text(timeout=1000).strip()
        except:
            try:
                title = card.inner_text(timeout=1000).strip().replace("\n", " ")[:40]
            except:
                title = "读取失败"
        titles.append(title)
    return titles


def get_visible_filter_texts(page):
    texts = []
    loc = page.locator("body")
    try:
        body_text = loc.inner_text(timeout=2000)
        for line in body_text.splitlines():
            line = line.strip()
            if line and any(k in line for k in ["筛选", "排序依据", "最多点赞", "最多评论", "最多收藏", "最新", "图文", "视频", "一天内", "一周内", "半年内", "已筛选"]):
                texts.append(line)
    except:
        pass
    return texts[:80]


def open_filter_panel(page):
    logs = []
    candidates = page.locator("text=/筛选|已筛选/")
    count = candidates.count()
    logs.append(f"找到疑似筛选入口数量: {count}")

    if count == 0:
        raise Exception("页面上没有找到‘筛选/已筛选’入口")

    for i in range(count - 1, -1, -1):
        loc = candidates.nth(i)
        try:
            box = loc.bounding_box()
            logs.append(f"尝试点击第 {i} 个筛选入口，坐标: {box}")
            if box and box["x"] > 1000 and box["y"] < 300:
                loc.click(force=True, timeout=2000)
                time.sleep(1.5)
                if page.locator("text=排序依据").count() > 0:
                    logs.append("成功打开筛选面板")
                    return logs
        except Exception as e:
            logs.append(f"点击第 {i} 个入口失败: {e}")

    raise Exception("点击所有疑似筛选入口后，仍未看到“排序依据”")


def click_option_in_panel(page, text):
    logs = []
    candidates = page.locator(f"text='{text}'")
    count = candidates.count()
    logs.append(f"筛选项 [{text}] 候选数量: {count}")

    if count == 0:
        raise Exception(f"页面上没有找到筛选项: {text}")

    for i in range(count):
        loc = candidates.nth(i)
        try:
            box = loc.bounding_box()
            logs.append(f"尝试点击筛选项 [{text}] 第 {i} 个，坐标: {box}")
            if box and box["x"] > 800 and 80 < box["y"] < 750:
                loc.click(force=True, timeout=2000)
                time.sleep(1.0)
                logs.append(f"点击 [{text}] 成功")
                return logs
        except Exception as e:
            logs.append(f"点击 [{text}] 第 {i} 个失败: {e}")

    raise Exception(f"未能在面板区域点击到筛选项: {text}")


def debug_filter_flow(keyword, sort_by, note_type, time_frame):
    debug_logs = []
    debug_info = {
        "search_url": "",
        "before_titles": [],
        "after_titles": [],
        "visible_filter_texts_before": [],
        "visible_filter_texts_after": [],
        "page_url_after": "",
    }

    p = sync_playwright().start()
    browser = p.chromium.launch_persistent_context(
        user_data_dir=USER_DATA_DIR,
        headless=False,
        viewport={"width": 1600, "height": 900},
        args=LAUNCH_ARGS
    )
    page = browser.pages[0] if browser.pages else browser.new_page()
    apply_stealth(page)

    try:
        encoded_keyword = urllib.parse.quote(keyword)
        target_url = f"https://www.xiaohongshu.com/search_result?keyword={encoded_keyword}&source=web_explore_feed"
        debug_info["search_url"] = target_url
        debug_logs.append(f"进入搜索页: {target_url}")

        page.goto(target_url, wait_until="domcontentloaded", timeout=20000)
        page.wait_for_selector("section.note-item", timeout=15000)
        page.wait_for_timeout(3000)

        debug_info["before_titles"] = get_top_titles(page, 3)
        debug_info["visible_filter_texts_before"] = get_visible_filter_texts(page)
        debug_logs.append(f"筛选前前3条标题: {debug_info['before_titles']}")

        if sort_by != "综合" or note_type != "不限" or time_frame != "不限":
            debug_logs.extend(open_filter_panel(page))

            if sort_by != "综合":
                debug_logs.extend(click_option_in_panel(page, sort_by))

            if note_type != "不限":
                debug_logs.extend(click_option_in_panel(page, note_type))

            if time_frame != "不限":
                debug_logs.extend(click_option_in_panel(page, time_frame))

            # 关闭筛选面板
            page.mouse.click(150, 80)
            debug_logs.append("已点击页面空白处关闭筛选面板")
            page.wait_for_timeout(6000)

        debug_info["page_url_after"] = page.url
        debug_info["after_titles"] = get_top_titles(page, 3)
        debug_info["visible_filter_texts_after"] = get_visible_filter_texts(page)

        debug_logs.append(f"筛选后前3条标题: {debug_info['after_titles']}")
        debug_logs.append(f"筛选后页面URL: {debug_info['page_url_after']}")

        changed = debug_info["before_titles"] != debug_info["after_titles"]
        debug_logs.append(f"筛选前后标题是否发生变化: {changed}")

        return True, debug_logs, debug_info

    except Exception as e:
        import traceback
        traceback.print_exc()
        debug_logs.append(f"发生异常: {e}")
        return False, debug_logs, debug_info

    finally:
        try:
            browser.close()
            p.stop()
        except:
            pass


st.title("🧪 小红书筛选调试器")

with st.sidebar:
    st.header("1. 登录检查")
    if st.button("打开浏览器检查登录状态"):
        with st.spinner("请确认登录后关闭浏览器"):
            open_robot_browser()
            st.success("登录状态检查完成")

    st.markdown("---")
    st.header("2. 调试参数")
    keyword = st.text_input("关键词", value="卡他症状")
    sort_by = st.selectbox("排序依据", ["综合", "最新", "最多点赞", "最多评论", "最多收藏"])
    note_type = st.selectbox("笔记类型", ["不限", "图文", "视频"])
    time_frame = st.selectbox("发布时间", ["不限", "一天内", "一周内", "半年内"])

    start_btn = st.button("开始调试筛选链路", type="primary", use_container_width=True)

if start_btn:
    with st.spinner("正在执行调试，请观察弹出的浏览器动作..."):
        ok, logs, info = debug_filter_flow(keyword, sort_by, note_type, time_frame)

    if ok:
        st.success("调试流程执行完成")
    else:
        st.error("调试流程执行失败，请看下面日志")

    st.subheader("调试日志")
    st.code("\n".join(logs), language="text")

    st.subheader("搜索URL")
    st.write(info["search_url"])

    col1, col2 = st.columns(2)

    with col1:
        st.markdown("### 筛选前前3条标题")
        st.write(info["before_titles"])
        st.markdown("### 筛选前页面可见筛选相关文本")
        st.write(info["visible_filter_texts_before"])

    with col2:
        st.markdown("### 筛选后前3条标题")
        st.write(info["after_titles"])
        st.markdown("### 筛选后页面可见筛选相关文本")
        st.write(info["visible_filter_texts_after"])

    st.markdown("### 筛选后页面URL")
    st.write(info["page_url_after"])