Files
xianyan/scripts/verify_data_flow.py
Developer 41a60b0288 feat: 新增多项核心服务与功能增强
refactor(theme): 扩展AppTheme支持卡片样式和圆角风格动态配置
feat(services): 新增HapticService触觉反馈服务
feat(services): 实现ScreenWakeService屏幕常亮管理
feat(services): 添加SoundService音效播放服务
feat(services): 集成AppLockService应用锁功能
feat(services): 实现BatteryOptimizationService电池优化
feat(services): 新增NetworkProxyService网络代理
feat(services): 完善DataExportService数据导出
feat(services): 增强PermissionService权限管理
feat(tools): 工具中心新增拼音转换等多项功能
fix(localization): 修复时区初始化错误
docs: 更新工具中心开发清单和设置重构文档
chore: 更新依赖版本和CI配置
2026-05-07 09:05:35 +08:00

220 lines
7.6 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# ============================================================
# 闲言APP — 句子卡片循环重复根因深度验证
# 创建时间: 2026-05-01
# 作用: 精确模拟客户端 refreshDailySentences 的完整数据流
# ============================================================
import requests
import json
import time
import hashlib
BASE = "https://tools.wktyl.com"
def fetch_mix(mode="random", limit=5, channels=None, sort=None):
params = {"mode": mode, "limit": limit}
if channels:
params["channels"] = ",".join(channels)
if sort:
params["sort"] = sort
try:
r = requests.get(f"{BASE}/api/feed/mix", params=params, timeout=10)
data = r.json()
if data.get("code") == 1:
items = data.get("data", {}).get("list", [])
return items
except Exception as e:
print(f" ❌ 请求失败: {e}")
return []
def make_id(item):
return f"{item.get('feed_type','?')}_{item.get('id')}"
def simulate_client_flow(rounds=5, limit=5):
"""
精确模拟客户端 refreshDailySentences 逻辑:
1. existingIds = state.dailySentences.map((s) => s.id).toSet()
2. fetchMix(config)
3. dailyList = result.list.map(HomeSentence.fromFeedItem).toList()
4. unique = dailyList.where((s) => !existingIds.contains(s.id)).toList()
5. if unique.isEmpty && existingIds.isNotEmpty: retry with 3x limit
6. finalList = unique.isNotEmpty ? unique : dailyList
7. state = state.copyWith(dailySentences: finalList)
"""
print("=" * 70)
print("模拟客户端 refreshDailySentences 完整数据流")
print("=" * 70)
existing_ids = set()
all_seen_ids = set()
daily_sentences_ids = []
for i in range(rounds):
print(f"\n--- 第 {i+1} 轮 ---")
print(f" existingIds = {existing_ids}")
# Step 1: fetchMix
items = fetch_mix(mode="random", limit=limit)
if not items:
print(" ❌ API 返回空数据!")
continue
# Step 2: 构建 dailyList
daily_list = [make_id(item) for item in items]
print(f" API 返回: {daily_list}")
# Step 3: 计算 unique
unique = [sid for sid in daily_list if sid not in existing_ids]
print(f" unique (不在existingIds中): {unique}")
# Step 4: 如果 unique 为空,重试
if not unique and existing_ids:
print(f" ⚠️ unique为空! 重试 (limit={limit*3})")
retry_items = fetch_mix(mode="random", limit=limit*3)
if retry_items:
retry_list = [make_id(item) for item in retry_items]
print(f" 重试API返回: {retry_list}")
unique = [sid for sid in retry_list if sid not in existing_ids]
print(f" 重试unique: {unique}")
if unique:
daily_list = unique
# Step 5: finalList
final_list = unique if unique else daily_list
print(f" finalList: {final_list}")
# Step 6: 更新 state
daily_sentences_ids = final_list
existing_ids = set(final_list)
all_seen_ids.update(final_list)
print(f" 更新后 dailySentences = {daily_sentences_ids}")
print(f" 更新后 existingIds = {existing_ids}")
print(f"\n{'='*70}")
print(f"总结: {rounds}轮后共看到 {len(all_seen_ids)} 个不同句子")
print(f"所有ID: {all_seen_ids}")
def simulate_with_dedup_issue(rounds=5, limit=5):
"""
模拟潜在问题: existingIds 只包含当前 dailySentences 的ID
而不是所有历史看过的ID导致旧句子可能重新出现
"""
print("\n" + "=" * 70)
print("模拟问题场景: existingIds 只包含当前5条的ID")
print("=" * 70)
existing_ids = set()
history_ids = set()
for i in range(rounds):
items = fetch_mix(mode="random", limit=limit)
if not items:
continue
daily_list = [make_id(item) for item in items]
unique = [sid for sid in daily_list if sid not in existing_ids]
final_list = unique if unique else daily_list
# 关键: existingIds 只包含当前 dailySentences
# 之前看过的句子如果不在当前 dailySentences 中,会被视为"新"的
old_ids_reappeared = [sid for sid in final_list if sid in history_ids and sid not in existing_ids]
existing_ids = set(final_list)
history_ids.update(final_list)
print(f"{i+1}轮: finalList={final_list}")
if old_ids_reappeared:
print(f" ⚠️ 旧句子重新出现: {old_ids_reappeared}")
def test_specific_mode_with_hottest():
"""
测试 mode=specific + sort=hottest 是否导致数据固定
"""
print("\n" + "=" * 70)
print("测试: mode=specific + sort=hottest 数据是否固定")
print("=" * 70)
all_ids = []
for i in range(3):
items = fetch_mix(mode="specific", limit=5,
channels=["poetry", "wisdom", "story"],
sort="hottest")
ids = [make_id(item) for item in items]
all_ids.extend(ids)
print(f"{i+1}次: {ids}")
time.sleep(0.3)
unique = len(set(all_ids))
total = len(all_ids)
print(f" 📊 总ID={total}, 去重={unique}, 重复率={1-unique/total:.1%}")
def test_specific_mode_with_newest():
"""
测试 mode=specific + sort=newest 数据是否变化
"""
print("\n" + "=" * 70)
print("测试: mode=specific + sort=newest 数据是否变化")
print("=" * 70)
all_ids = []
for i in range(3):
items = fetch_mix(mode="specific", limit=5,
channels=["poetry", "wisdom", "story"],
sort="newest")
ids = [make_id(item) for item in items]
all_ids.extend(ids)
print(f"{i+1}次: {ids}")
time.sleep(0.3)
unique = len(set(all_ids))
total = len(all_ids)
print(f" 📊 总ID={total}, 去重={unique}, 重复率={1-unique/total:.1%}")
def test_default_config_no_sort():
"""
测试: 默认配置 (mode=random, limit=5, 无sort参数)
这正是客户端当前发送的请求
"""
print("\n" + "=" * 70)
print("测试: 客户端当前请求参数 (mode=random, limit=5, 无sort)")
print("=" * 70)
all_ids = []
for i in range(5):
items = fetch_mix(mode="random", limit=5)
ids = [make_id(item) for item in items]
all_ids.extend(ids)
print(f"{i+1}次: {ids}")
time.sleep(0.3)
unique = len(set(all_ids))
total = len(all_ids)
print(f" 📊 总ID={total}, 去重={unique}, 重复率={1-unique/total:.1%}")
if __name__ == "__main__":
simulate_client_flow(rounds=5, limit=5)
simulate_with_dedup_issue(rounds=5, limit=5)
test_specific_mode_with_hottest()
test_specific_mode_with_newest()
test_default_config_no_sort()
print("\n" + "=" * 70)
print("🔍 根因分析总结")
print("=" * 70)
print()
print("问题: 句子卡片5个句子循环重复永远不会出现新的")
print()
print("可能根因:")
print(" 1. API层面: mode=specific+sort=hottest 返回固定热门数据")
print(" 2. 客户端层面: existingIds去重逻辑导致数据子集化")
print(" 3. 状态层面: didUpdateWidget未正确检测新数据")
print(" 4. 并发层面: 多次refreshDailySentences竞态条件")
print()
print("修复方案:")
print(" A. FeedMixConfig添加sort字段默认'newest'")
print(" B. refreshDailySentences添加防重入锁")
print(" C. 清空existingIds后再请求确保获取全新数据")
print(" D. 添加详细日志追踪数据流")