75 lines
3.0 KiB
Python
75 lines
3.0 KiB
Python
#!/usr/bin/env python
|
|
"""
|
|
Test script to download images for existing games that don't have local images yet.
|
|
"""
|
|
import asyncio
|
|
import aiohttp
|
|
from pathlib import Path
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from src.libs.config import Config
|
|
from src.libs.database import Game_table, Metadata_table
|
|
from src.libs.functions import download_image, get_image_filename
|
|
|
|
async def test_image_downloads():
|
|
config = Config()
|
|
url = f"sqlite+pysqlite:///{config.database_path}"
|
|
engine = create_engine(url, future=True)
|
|
|
|
with Session(engine) as session:
|
|
# Get first 3 games that have remote images but no local images
|
|
stmt = (
|
|
select(Game_table)
|
|
.join(Metadata_table)
|
|
.where(
|
|
(Metadata_table.cover_image.is_not(None)) &
|
|
(Metadata_table.cover_image_path.is_(None))
|
|
)
|
|
.limit(3)
|
|
)
|
|
games = session.scalars(stmt).all()
|
|
|
|
print(f"Found {len(games)} games to test image downloads for")
|
|
|
|
async with aiohttp.ClientSession() as http_session:
|
|
for game in games:
|
|
metadata = game.metadata_obj
|
|
print(f"\nTesting: {game.title}")
|
|
|
|
# Download cover image
|
|
if metadata.cover_image:
|
|
cover_filename = get_image_filename(metadata.cover_image, game.title, 'cover')
|
|
cover_path = config.images_path / cover_filename
|
|
|
|
print(f" Downloading cover: {metadata.cover_image}")
|
|
success = await download_image(metadata.cover_image, cover_path, http_session)
|
|
|
|
if success:
|
|
print(f" ✓ Cover saved to: {cover_path}")
|
|
# Update database with local path
|
|
metadata.cover_image_path = cover_path
|
|
else:
|
|
print(f" ✗ Failed to download cover")
|
|
|
|
# Download screenshot
|
|
if metadata.screenshot:
|
|
screenshot_filename = get_image_filename(metadata.screenshot, game.title, 'screenshot')
|
|
screenshot_path = config.images_path / screenshot_filename
|
|
|
|
print(f" Downloading screenshot: {metadata.screenshot}")
|
|
success = await download_image(metadata.screenshot, screenshot_path, http_session)
|
|
|
|
if success:
|
|
print(f" ✓ Screenshot saved to: {screenshot_path}")
|
|
# Update database with local path
|
|
metadata.screenshot_path = screenshot_path
|
|
else:
|
|
print(f" ✗ Failed to download screenshot")
|
|
|
|
# Commit the updates
|
|
session.commit()
|
|
print(f"\n✓ Database updated with local image paths")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test_image_downloads()) |