207 lines
8.0 KiB
Python
207 lines
8.0 KiB
Python
import asyncio
|
|
import aiohttp
|
|
import aiomysql
|
|
import json
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from urllib.parse import urljoin
|
|
|
|
DATA_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), 'data')
|
|
CONFIG_PATH = os.path.join(DATA_DIR, 'config.json')
|
|
SCHEMA_RESULTS_PATH = os.path.join(DATA_DIR, 'schema_results.json')
|
|
IMAGE_RESULTS_PATH = os.path.join(DATA_DIR, 'image_results.json')
|
|
|
|
def load_config():
|
|
with open(CONFIG_PATH, 'r') as f:
|
|
return json.load(f)
|
|
|
|
def load_json(path):
|
|
if not os.path.exists(path):
|
|
return {}
|
|
try:
|
|
with open(path, 'r') as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return {}
|
|
|
|
def save_json(path, data):
|
|
with open(path, 'w') as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
SCAN_STATE = {
|
|
"schema_scanning": False,
|
|
"schema_current_db": None,
|
|
"image_scanning": False,
|
|
"image_current_db": None
|
|
}
|
|
|
|
def get_scan_status():
|
|
return SCAN_STATE
|
|
|
|
async def scan_schema_for_db(db_name: str, config: dict):
|
|
host = config['host']
|
|
port = config.get('port', 3306)
|
|
user = config['user']
|
|
password = config['password']
|
|
|
|
tables_data = []
|
|
try:
|
|
pool = await aiomysql.create_pool(
|
|
host=host, port=port, user=user, password=password, db=db_name, autocommit=True, connect_timeout=5
|
|
)
|
|
async with pool.acquire() as conn:
|
|
async with conn.cursor(aiomysql.DictCursor) as cur:
|
|
await cur.execute("SHOW TABLES")
|
|
tables = [list(row.values())[0] for row in await cur.fetchall()]
|
|
|
|
for table in tables:
|
|
try:
|
|
await cur.execute(f"SELECT COUNT(*) as cnt FROM `{table}`")
|
|
row_count = (await cur.fetchone())['cnt']
|
|
|
|
await cur.execute(f"SELECT COUNT(*) as cnt FROM information_schema.columns WHERE table_schema='{db_name}' AND table_name='{table}'")
|
|
col_count = (await cur.fetchone())['cnt']
|
|
|
|
tables_data.append({
|
|
"table_name": table,
|
|
"row_count": row_count,
|
|
"column_count": col_count,
|
|
"status": "Success"
|
|
})
|
|
except Exception as e:
|
|
tables_data.append({
|
|
"table_name": table,
|
|
"row_count": 0,
|
|
"column_count": 0,
|
|
"status": f"Error: {e}"
|
|
})
|
|
pool.close()
|
|
await pool.wait_closed()
|
|
return {"db_name": db_name, "status": "Success", "tables": tables_data}
|
|
except Exception as e:
|
|
return {"db_name": db_name, "status": f"Connection Failed: {e}", "tables": []}
|
|
|
|
async def check_url(session, url, semaphore, timeout_seconds):
|
|
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:
|
|
return "Available" if get_response.status == 200 else "Not Found"
|
|
return "Available" if response.status == 200 else "Not Found"
|
|
except asyncio.TimeoutError:
|
|
return "Timeout"
|
|
except Exception:
|
|
return "Error"
|
|
|
|
async def scan_images_for_db_table(db_name: str, table: str, config: dict):
|
|
host = config['host']
|
|
port = config.get('port', 3306)
|
|
user = config['user']
|
|
password = config['password']
|
|
image_column = config.get('image_column', 'test_image_path')
|
|
base_url = config['base_url']
|
|
if not base_url.endswith('/'): base_url += '/'
|
|
|
|
concurrency_limit = config.get('concurrency_limit', 50)
|
|
timeout_seconds = config.get('request_timeout_seconds', 10)
|
|
|
|
stats = {"table": table, "total_rows_with_image": 0, "available": 0, "not_found": 0, "error": 0, "status": "Success"}
|
|
|
|
try:
|
|
pool = await aiomysql.create_pool(
|
|
host=host, port=port, user=user, password=password, db=db_name, autocommit=True, connect_timeout=5
|
|
)
|
|
async with pool.acquire() as conn:
|
|
async with conn.cursor(aiomysql.DictCursor) as cur:
|
|
try:
|
|
await cur.execute(f"SELECT `{image_column}` FROM `{table}` WHERE `{image_column}` IS NOT NULL AND `{image_column}` != ''")
|
|
rows = await cur.fetchall()
|
|
except Exception as e:
|
|
stats['status'] = f"Table/Query Error: {e}"
|
|
pool.close()
|
|
await pool.wait_closed()
|
|
return stats
|
|
|
|
pool.close()
|
|
await pool.wait_closed()
|
|
|
|
stats['total_rows_with_image'] = len(rows)
|
|
if len(rows) == 0: return stats
|
|
|
|
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:]
|
|
url = urljoin(base_url, img_path)
|
|
tasks.append(asyncio.create_task(check_url(session, url, semaphore, timeout_seconds)))
|
|
|
|
for f in asyncio.as_completed(tasks):
|
|
res = await f
|
|
if res == "Available": stats['available'] += 1
|
|
elif res == "Not Found": stats['not_found'] += 1
|
|
else: stats['error'] += 1
|
|
|
|
return stats
|
|
except Exception as e:
|
|
stats['status'] = f"Connection/Execution Failed: {e}"
|
|
return stats
|
|
|
|
async def run_schema_scan():
|
|
if SCAN_STATE["schema_scanning"]: return
|
|
SCAN_STATE["schema_scanning"] = True
|
|
try:
|
|
config = load_config()
|
|
databases = config.get('databases', [])
|
|
|
|
results = load_json(SCHEMA_RESULTS_PATH)
|
|
|
|
for db_name in databases:
|
|
SCAN_STATE["schema_current_db"] = db_name
|
|
schema_data = await scan_schema_for_db(db_name, config)
|
|
results[db_name] = {
|
|
"schema_status": schema_data['status'],
|
|
"schema_tables": schema_data['tables'],
|
|
"last_updated": datetime.now(timezone.utc).isoformat()
|
|
}
|
|
save_json(SCHEMA_RESULTS_PATH, results) # Instantly available
|
|
|
|
finally:
|
|
SCAN_STATE["schema_scanning"] = False
|
|
SCAN_STATE["schema_current_db"] = None
|
|
|
|
async def run_image_scan():
|
|
if SCAN_STATE["image_scanning"]: return
|
|
SCAN_STATE["image_scanning"] = True
|
|
try:
|
|
config = load_config()
|
|
databases = config.get('databases', [])
|
|
image_tables = config.get('image_tables', [])
|
|
|
|
schema_results = load_json(SCHEMA_RESULTS_PATH)
|
|
image_results = load_json(IMAGE_RESULTS_PATH)
|
|
|
|
for db_name in databases:
|
|
SCAN_STATE["image_current_db"] = db_name
|
|
db_schema = schema_results.get(db_name, {})
|
|
|
|
if db_schema.get('schema_status') == "Success":
|
|
existing_tables = [t['table_name'] for t in db_schema.get('schema_tables', [])]
|
|
stats_list = []
|
|
for itable in image_tables:
|
|
if itable in existing_tables:
|
|
img_data = await scan_images_for_db_table(db_name, itable, config)
|
|
stats_list.append(img_data)
|
|
|
|
image_results[db_name] = {
|
|
"image_stats": stats_list,
|
|
"last_updated": datetime.now(timezone.utc).isoformat()
|
|
}
|
|
save_json(IMAGE_RESULTS_PATH, image_results) # Instantly available
|
|
finally:
|
|
SCAN_STATE["image_scanning"] = False
|
|
SCAN_STATE["image_current_db"] = None
|