image-db-sync-monitor/legacy/check_images.py

import asyncio
import aiohttp
import aiomysql
import json
import csv
import sys
from urllib.parse import urljoin

async def check_url(session, url, semaphore, timeout_seconds=10):
    async with semaphore:
        try:
            async with session.head(url, allow_redirects=True, timeout=timeout_seconds) as response:
                if response.status == 405:
                    async with session.get(url, allow_redirects=True, timeout=timeout_seconds) as get_response:
                        status_msg = "Available" if get_response.status == 200 else ("Not Found" if get_response.status == 404 else "Error")
                        return get_response.status, status_msg
                
                status_msg = "Available" if response.status == 200 else ("Not Found" if response.status == 404 else "Error")
                return response.status, status_msg
        except asyncio.TimeoutError:
            return None, "Timeout"
        except Exception as e:
            return None, f"Error: {type(e).__name__}"

async def process_database(db_name, config, summary_data):
    print(f"\n--- Processing database: {db_name} ---")
    
    host = config['host']
    port = config.get('port', 3306)
    user = config['user']
    password = config['password']
    table = config['table']
    image_column = config['image_column']
    pk_column = config['primary_key']
    
    base_url = config['base_url']
    concurrency_limit = config.get('concurrency_limit', 50)
    timeout_seconds = config.get('request_timeout_seconds', 10)
    
    output_filename = f"{db_name}_image_report.csv"
    
    try:
        pool = await aiomysql.create_pool(
            host=host, port=port,
            user=user, password=password,
            db=db_name, autocommit=True
        )
    except Exception as e:
        print(f"Failed to connect to database {db_name}: {e}")
        summary_data.append([db_name, "Connection Failed", 0, 0, 0, 0])
        return

    try:
        async with pool.acquire() as conn:
            async with conn.cursor(aiomysql.DictCursor) as cur:
                try:
                    await cur.execute(f"SELECT {pk_column}, {image_column} FROM {table} WHERE {image_column} IS NOT NULL AND {image_column} != ''")
                    rows = await cur.fetchall()
                except Exception as e:
                    print(f"Failed to query table {table} in {db_name}: {e}")
                    summary_data.append([db_name, "Table/Query Error", 0, 0, 0, 0])
                    return
        
        total_rows = len(rows)
        print(f"Found {total_rows} rows to process in {db_name}.{table}.")
        
        if total_rows == 0:
            print("No rows to process. Skipping.")
            summary_data.append([db_name, "Success", 0, 0, 0, 0])
            return

        semaphore = asyncio.Semaphore(concurrency_limit)
        
        connector = aiohttp.TCPConnector(limit=concurrency_limit)
        async with aiohttp.ClientSession(connector=connector) as session:
            tasks = []
            for row in rows:
                img_path = row[image_column]
                if isinstance(img_path, str) and img_path.startswith('/'):
                    img_path = img_path[1:]
                
                if not base_url.endswith('/'):
                    base_url += '/'
                    
                url = urljoin(base_url, img_path)
                
                task = asyncio.create_task(process_row(session, row, url, pk_column, image_column, semaphore, timeout_seconds))
                tasks.append(task)
            
            results = []
            completed = 0
            available = 0
            not_found = 0
            error = 0
            
            for f in asyncio.as_completed(tasks):
                res = await f
                results.append(res)
                
                status_msg = res[3]
                if status_msg == "Available":
                    available += 1
                elif status_msg == "Not Found":
                    not_found += 1
                else:
                    error += 1
                
                completed += 1
                if completed % 500 == 0 or completed == total_rows:
                    print(f"Processed {completed}/{total_rows} URLs in {db_name}...")
            
        with open(output_filename, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            writer.writerow(['Row Identifier', 'Original Column Value', 'Full URL', 'Status', 'HTTP Status Code'])
            for res in results:
                writer.writerow(res)
                
        print(f"Report generated: {output_filename}")
        
        # Add to summary
        summary_data.append([db_name, "Success", total_rows, available, not_found, error])
    
    finally:
        pool.close()
        await pool.wait_closed()

async def process_row(session, row, url, pk_column, image_column, semaphore, timeout_seconds):
    status_code, status_msg = await check_url(session, url, semaphore, timeout_seconds)
    return [row[pk_column], row[image_column], url, status_msg, status_code]

async def main():
    try:
        with open('config.json', 'r') as f:
            config = json.load(f)
    except FileNotFoundError:
        print("Error: config.json not found.")
        sys.exit(1)
        
    databases = config.get('databases', [])
    if not databases:
        print("No databases to check in config.json.")
        return
        
    summary_data = []
        
    for db_name in databases:
        await process_database(db_name, config, summary_data)
        
    # Write summary report
    summary_filename = "FINAL_SUMMARY_REPORT.csv"
    with open(summary_filename, 'w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(['Database Name', 'Query Status', 'Total Rows with Image', 'Images Available (200)', 'Images Not Found (404)', 'Errors/Timeouts'])
        for row in summary_data:
            writer.writerow(row)
            
    print(f"\n=====================================")
    print(f"ALL DONE! Final summary report generated: {summary_filename}")
    print(f"=====================================")

if __name__ == '__main__':
    if sys.platform == 'win32':
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(main())