mirror of
https://github.com/cclank/news-aggregator-skill.git
synced 2026-09-18 20:08:23 +08:00
7aced7b4a4
- Add --no-save flag to daily_briefing.py and fetch_news.py to skip writing intermediate JSON files to reports directory - Update SKILL.md and templates.md to use --no-save by default - Add .gitignore patterns for stderr logs, debug files, screenshots - Remove dead sources (TLDR, Cognitive Rev, The Neuron, The Rundown) - Fix Interconnects URL (add www. prefix) - Add new utility scripts: rss_parser, test_all_sources, fetch_bensbites - Add briefing instructions for AI daily and GitHub - Clean up briefing_data_final.json
131 lines
4.5 KiB
Python
131 lines
4.5 KiB
Python
|
|
import subprocess
|
|
import os
|
|
import sys
|
|
import shutil
|
|
from datetime import datetime
|
|
|
|
def get_all_sources():
|
|
"""Retrieve all source keys from fetch_news.py --list-sources"""
|
|
cmd = [sys.executable, 'scripts/fetch_news.py', '--list-sources']
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
print(f"Error listing sources: {result.stderr}")
|
|
return []
|
|
|
|
lines = result.stdout.strip().split('\n')
|
|
sources = []
|
|
start_collecting = False
|
|
for line in lines:
|
|
if line.startswith('---'):
|
|
start_collecting = True
|
|
continue
|
|
if start_collecting and line.strip():
|
|
key = line.split('|')[0].strip()
|
|
if key:
|
|
sources.append(key)
|
|
return sources
|
|
|
|
def test_source(source, out_dir):
|
|
"""Run fetch_news.py for a single source"""
|
|
print(f"Testing {source}...", end=' ', flush=True)
|
|
cmd = [
|
|
sys.executable,
|
|
'scripts/fetch_news.py',
|
|
'--source', source,
|
|
'--limit', '1',
|
|
'--save',
|
|
'--outdir', out_dir
|
|
]
|
|
|
|
start_time = datetime.now()
|
|
try:
|
|
# Timeout after 30s to prevent hanging
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=45)
|
|
duration = (datetime.now() - start_time).total_seconds()
|
|
|
|
if result.returncode == 0:
|
|
# Check if file was actually created
|
|
# fetch_news prints "[Saved] Raw Data: ...JSON"
|
|
if "Raw Data:" in result.stderr:
|
|
print(f"✅ PASS ({duration:.1f}s)")
|
|
return True, duration, None
|
|
else:
|
|
# It might have returned empty list [] and not saved anything?
|
|
# fetch_news only saves if data is not empty?
|
|
# Let's check logic: "if args.save or ... md_file = save_report..."
|
|
# save_report writes file even if data is empty?
|
|
# "if not data: f.write('No items found.')" -> Yes.
|
|
# So if it didn't save, something else happened.
|
|
print(f"⚠️ EMPTY/NO-SAVE ({duration:.1f}s)")
|
|
return False, duration, "Script ran but no save message found"
|
|
else:
|
|
print(f"❌ FAIL ({duration:.1f}s)")
|
|
return False, duration, result.stderr.strip()
|
|
|
|
except subprocess.TimeoutExpired:
|
|
print("❌ TIMEOUT (45s)")
|
|
return False, 45, "Timeout"
|
|
except Exception as e:
|
|
print(f"❌ ERROR: {e}")
|
|
return False, 0, str(e)
|
|
|
|
def main():
|
|
# Setup
|
|
test_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'test_results')
|
|
if os.path.exists(test_dir):
|
|
shutil.rmtree(test_dir)
|
|
os.makedirs(test_dir)
|
|
|
|
print(f"Starting systematic source test...")
|
|
print(f"Output directory: {test_dir}\n")
|
|
|
|
sources = get_all_sources()
|
|
print(f"Found {len(sources)} sources to test.\n")
|
|
|
|
results = []
|
|
|
|
for source in sources:
|
|
success, duration, error = test_source(source, test_dir)
|
|
results.append({
|
|
'source': source,
|
|
'success': success,
|
|
'duration': duration,
|
|
'error': error
|
|
})
|
|
|
|
# Generate Summary
|
|
summary_path = os.path.join(test_dir, 'summary.md')
|
|
with open(summary_path, 'w', encoding='utf-8') as f:
|
|
f.write("# 🧪 News Source Test Report\n")
|
|
f.write(f"Date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
|
|
|
|
passed = [r for r in results if r['success']]
|
|
failed = [r for r in results if not r['success']]
|
|
|
|
f.write(f"## Summary\n")
|
|
f.write(f"- **Total**: {len(results)}\n")
|
|
f.write(f"- **Passed**: {len(passed)}\n")
|
|
f.write(f"- **Failed**: {len(failed)}\n\n")
|
|
|
|
f.write("## ❌ Failures\n")
|
|
if not failed:
|
|
f.write("None! 🎉\n")
|
|
else:
|
|
for r in failed:
|
|
f.write(f"- **{r['source']}**: {r['error']}\n")
|
|
|
|
f.write("\n## 📋 Detailed Results\n")
|
|
f.write("| Source | Status | Duration | Note |\n")
|
|
f.write("|---|---|---|---|\n")
|
|
for r in results:
|
|
status = "✅ PASS" if r['success'] else "❌ FAIL"
|
|
error_msg = r['error'] if r['error'] else ""
|
|
f.write(f"| {r['source']} | {status} | {r['duration']:.1f}s | {error_msg} |\n")
|
|
|
|
print(f"\nTest passed: {len(passed)}/{len(results)}")
|
|
print(f"Report saved to: {summary_path}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|