Russell Ballestrini
99c836e2c8
- filevault.py: Vault (sync) + AsyncVault (async) in one file - Delete storage.py (redundant async wrapper) - Both classes share identical API: store, get, exists, delete, stats - Add content_hash() and hash_to_path() utility functions - Update all imports: neopig.py, serp.py, test files - Update README with unified documentation - 72 lines saved, cleaner architecture
116 lines
3.3 KiB
Python
116 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Functional test for neopig crawler.
|
|
|
|
Tests crawling upload.unturf.com for images.
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Add current dir to path
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from neopig import NeoPig
|
|
from async_web_fetcher import CrawlMode
|
|
|
|
|
|
async def test_crawl_upload_unturf():
|
|
"""Test crawling upload.unturf.com for images."""
|
|
print("=" * 60)
|
|
print("neopig functional test")
|
|
print("Target: https://upload.unturf.com")
|
|
print("=" * 60)
|
|
|
|
# Use test database and vault
|
|
pig = NeoPig(
|
|
db_path="test_neopig.db",
|
|
vault_path="test_vault"
|
|
)
|
|
await pig.init()
|
|
|
|
print("\n[1/3] Starting crawl...")
|
|
stats = await pig.crawl(
|
|
target_uri="https://upload.unturf.com",
|
|
keywords=["unturf", "upload"],
|
|
mode=CrawlMode.IMAGES,
|
|
depth=-1, # Unlimited depth
|
|
max_pages=100, # More pages
|
|
download_media=True,
|
|
)
|
|
|
|
print("\n[2/3] Crawl complete!")
|
|
print(f" Pages crawled: {stats['pages_crawled']}")
|
|
print(f" Media found: {stats['media_found']}")
|
|
print(f" Media downloaded: {stats['media_downloaded']}")
|
|
print(f" Duplicates skipped: {stats['duplicates_skipped']}")
|
|
print(f" Errors: {stats['errors']}")
|
|
|
|
# Check vault (filevault)
|
|
print("\n[3/3] Checking filevault...")
|
|
vault_stats = await pig.vault.stats()
|
|
print(f" Vault path: {vault_stats['path']}")
|
|
print(f" Vault depth: {vault_stats['depth']}")
|
|
print(f" Files stored: {vault_stats['count']}")
|
|
if 'total_size' in vault_stats:
|
|
size_mb = vault_stats['total_size'] / (1024 * 1024)
|
|
print(f" Total size: {size_mb:.2f} MB")
|
|
|
|
# Check database
|
|
from database import Database
|
|
db = Database("test_neopig.db")
|
|
await db.init()
|
|
db_stats = await db.get_stats()
|
|
print(f"\n Database stats:")
|
|
print(f" Total media: {db_stats['total_media']}")
|
|
print(f" By type: {db_stats['by_type']}")
|
|
print(f" Total sources: {db_stats['total_sources']}")
|
|
|
|
print("\n" + "=" * 60)
|
|
if stats['media_downloaded'] > 0:
|
|
print("TEST PASSED - Successfully crawled and stored images")
|
|
elif stats['media_found'] > 0:
|
|
print("TEST PARTIAL - Found images but none downloaded (may be dupes)")
|
|
else:
|
|
print("TEST WARNING - No images found on target")
|
|
print("=" * 60)
|
|
|
|
return stats
|
|
|
|
|
|
async def test_crawl_modes():
|
|
"""Test different crawl modes."""
|
|
print("\n" + "=" * 60)
|
|
print("Testing crawl modes...")
|
|
print("=" * 60)
|
|
|
|
pig = NeoPig(
|
|
db_path="test_neopig.db",
|
|
vault_path="test_vault"
|
|
)
|
|
await pig.init()
|
|
|
|
# Test media mode (images + videos + audio)
|
|
print("\n[MEDIA mode] Crawling for all media...")
|
|
stats = await pig.crawl(
|
|
target_uri="https://upload.unturf.com",
|
|
keywords=["test"],
|
|
mode=CrawlMode.MEDIA,
|
|
depth=1,
|
|
max_pages=5,
|
|
download_media=False, # Just index, don't download
|
|
)
|
|
print(f" Found: {stats['media_found']} media items (not downloaded)")
|
|
|
|
return stats
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("neopig functional test suite\n")
|
|
|
|
# Run tests
|
|
asyncio.run(test_crawl_upload_unturf())
|
|
asyncio.run(test_crawl_modes())
|
|
|
|
print("\nDone! Check test_vault/ and test_neopig.db for results.")
|