1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
|
#!/usr/bin/env python3
"""
测试无限制历史模式
验证系统是否能处理大规模历史数据获取，
测试不同的配置参数和性能表现。
"""
import os
import sys
import time
import logging
# 设置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)
# Add the parent directory to the path so we can import the main module
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from scripts.fetch_papers import ArxivPaperFetcher
def test_configuration_options():
"""测试不同的配置选项"""
print("🔍 测试无限制历史模式配置")
print("=" * 60)
# 测试不同的配置场景
test_scenarios = [
{
"name": "小规模测试",
"MAX_HISTORICAL_PAPERS": "1000",
"MAX_PAPERS_PER_CATEGORY": "200",
"MAX_CONCURRENT": "10",
"description": "适合快速测试和开发"
},
{
"name": "中规模测试",
"MAX_HISTORICAL_PAPERS": "5000",
"MAX_PAPERS_PER_CATEGORY": "1000",
"MAX_CONCURRENT": "25",
"description": "适合日常使用"
},
{
"name": "大规模测试",
"MAX_HISTORICAL_PAPERS": "50000",
"MAX_PAPERS_PER_CATEGORY": "10000",
"MAX_CONCURRENT": "50",
"description": "适合完整历史数据获取"
},
{
"name": "超大规模测试",
"MAX_HISTORICAL_PAPERS": "100000",
"MAX_PAPERS_PER_CATEGORY": "20000",
"MAX_CONCURRENT": "100",
"description": "适合研究级别的数据挖掘"
}
]
print("📊 支持的配置场景:")
for i, scenario in enumerate(test_scenarios, 1):
print(f"\n{i}. {scenario['name']}:")
print(f" - 最大论文数: {int(scenario['MAX_HISTORICAL_PAPERS']):,}")
print(f" - 每类别限制: {int(scenario['MAX_PAPERS_PER_CATEGORY']):,}")
print(f" - 并发数: {scenario['MAX_CONCURRENT']}")
print(f" - 描述: {scenario['description']}")
# 计算理论性能
print(f"\n⚡ 理论性能估算:")
print(f" 基于以下假设:")
print(f" - 每篇论文GPT处理时间: 1-2秒")
print(f" - 并行处理加速比: 10-20x")
print(f" - 网络延迟和API限制: 考虑在内")
for scenario in test_scenarios:
max_papers = int(scenario['MAX_HISTORICAL_PAPERS'])
concurrent = int(scenario['MAX_CONCURRENT'])
# 串行时间估算
serial_time = max_papers * 1.5 # 1.5秒每篇
# 并行时间估算
parallel_time = max_papers / concurrent * 1.5 + 60 # 额外60秒开销
print(f"\n {scenario['name']}:")
print(f" - 串行处理时间: {serial_time/3600:.1f} 小时")
print(f" - 并行处理时间: {parallel_time/3600:.1f} 小时")
print(f" - 加速比: {serial_time/parallel_time:.1f}x")
def test_memory_requirements():
"""测试内存需求"""
print(f"\n" + "="*60)
print("💾 内存需求分析")
print("="*60)
# 估算每篇论文的内存占用
avg_title_length = 100 # 平均标题长度
avg_abstract_length = 1500 # 平均摘要长度
avg_authors = 4 # 平均作者数
avg_categories = 2 # 平均类别数
# 每篇论文大约的内存占用（字符数）
chars_per_paper = (
avg_title_length +
avg_abstract_length +
avg_authors * 30 + # 每个作者约30字符
avg_categories * 10 + # 每个类别约10字符
200 # 其他字段
)
bytes_per_paper = chars_per_paper * 2 # 假设每字符2字节（UTF-8）
print(f"📊 每篇论文内存占用估算:")
print(f" - 标题: ~{avg_title_length} 字符")
print(f" - 摘要: ~{avg_abstract_length} 字符")
print(f" - 作者: ~{avg_authors * 30} 字符")
print(f" - 类别: ~{avg_categories * 10} 字符")
print(f" - 其他: ~200 字符")
print(f" - 总计: ~{chars_per_paper} 字符 (~{bytes_per_paper/1024:.1f} KB)")
# 不同规模的内存需求
paper_counts = [1000, 5000, 20000, 50000, 100000]
print(f"\n📈 不同规模的内存需求:")
for count in paper_counts:
total_mb = count * bytes_per_paper / 1024 / 1024
print(f" - {count:,} 篇论文: ~{total_mb:.1f} MB")
print(f"\n💡 建议:")
print(f" - 16GB内存: 支持最多 ~100,000 篇论文")
print(f" - 8GB内存: 支持最多 ~50,000 篇论文")
print(f" - 4GB内存: 支持最多 ~20,000 篇论文")
print(f" - 如果内存不足，可以降低MAX_HISTORICAL_PAPERS")
def test_api_cost_estimation():
"""测试API成本估算"""
print(f"\n" + "="*60)
print("💰 API成本估算")
print("="*60)
# OpenAI GPT-4o 价格 (2024年价格)
# Input: $2.50 per 1M tokens
# Output: $10.00 per 1M tokens
input_price_per_1m = 2.50
output_price_per_1m = 10.00
# 估算每篇论文的token消耗
avg_input_tokens = 400 # 标题+摘要+系统prompt
avg_output_tokens = 1 # 只返回"0"或"1"
cost_per_paper = (
(avg_input_tokens / 1000000) * input_price_per_1m +
(avg_output_tokens / 1000000) * output_price_per_1m
)
print(f"📊 每篇论文API成本估算:")
print(f" - 输入tokens: ~{avg_input_tokens}")
print(f" - 输出tokens: ~{avg_output_tokens}")
print(f" - 每篇成本: ~${cost_per_paper:.4f}")
# 不同规模的成本
paper_counts = [1000, 5000, 20000, 50000, 100000]
print(f"\n💸 不同规模的API成本:")
for count in paper_counts:
total_cost = count * cost_per_paper
print(f" - {count:,} 篇论文: ~${total_cost:.2f}")
print(f"\n🎯 成本优化建议:")
print(f" - 先用小规模测试验证效果")
print(f" - 使用MAX_HISTORICAL_PAPERS控制规模")
print(f" - 考虑分批处理大规模数据")
print(f" - 监控API使用量避免超支")
def demonstrate_configuration():
"""演示配置使用方法"""
print(f"\n" + "="*60)
print("🛠️ 配置使用方法")
print("="*60)
print(f"🔧 环境变量配置:")
print(f"""
# 基础配置 (推荐用于测试)
export MAX_HISTORICAL_PAPERS=1000
export MAX_PAPERS_PER_CATEGORY=200
export MAX_CONCURRENT=10
# 中等规模配置 (推荐用于日常使用)
export MAX_HISTORICAL_PAPERS=5000
export MAX_PAPERS_PER_CATEGORY=1000
export MAX_CONCURRENT=25
# 大规模配置 (推荐用于研究)
export MAX_HISTORICAL_PAPERS=50000
export MAX_PAPERS_PER_CATEGORY=10000
export MAX_CONCURRENT=50
# 无限制配置 (谨慎使用)
export MAX_HISTORICAL_PAPERS=1000000
export MAX_PAPERS_PER_CATEGORY=100000
export MAX_CONCURRENT=100
""")
print(f"🚀 运行命令:")
print(f"""
# 使用默认配置运行历史模式
FETCH_MODE=historical python scripts/fetch_papers.py
# 使用自定义配置运行
MAX_HISTORICAL_PAPERS=10000 \\
MAX_PAPERS_PER_CATEGORY=2000 \\
MAX_CONCURRENT=30 \\
FETCH_MODE=historical \\
python scripts/fetch_papers.py
""")
print(f"⚠️ 注意事项:")
print(f" - 首次运行建议使用小规模配置")
print(f" - 监控内存使用情况")
print(f" - 注意API成本控制")
print(f" - 考虑网络稳定性")
print(f" - 大规模运行可能需要数小时")
if __name__ == "__main__":
print("🎯 ArXiv无限制历史模式测试")
print("=" * 60)
test_configuration_options()
test_memory_requirements()
test_api_cost_estimation()
demonstrate_configuration()
print(f"\n✅ 测试完成！")
print(f"💡 现在可以根据需求配置合适的参数来运行历史模式")
print(f"🚀 建议先从小规模开始测试，确保一切正常后再扩大规模")
|
