98 lines
3.4 KiB
Plaintext
Executable File
98 lines
3.4 KiB
Plaintext
Executable File
|
|
import os, json, base64, subprocess, glob, time
|
|
from datetime import datetime
|
|
|
|
SS_DIR = '/var/www/html/l99-screenshots'
|
|
ANALYSIS = '/var/www/html/api/l99-analysis.json'
|
|
LOG = '/var/log/wevia-visual-batch.log'
|
|
|
|
def lg(m):
|
|
ts = datetime.now().strftime('%H:%M')
|
|
with open(LOG,'a') as f: f.write(f'{ts} {m}\n')
|
|
print(f'{ts} {m}')
|
|
|
|
# Load existing analyses
|
|
try:
|
|
data = json.load(open(ANALYSIS))
|
|
except:
|
|
data = {'scans':{},'stats':{'total':0,'success':0,'warn':0,'fail':0,'partial':0},'engine':'moondream-sovereign','total':0}
|
|
|
|
scans = data.get('scans',{})
|
|
already = set(scans.keys())
|
|
|
|
# Find screenshots NOT yet analyzed (latest first)
|
|
all_ss = sorted(glob.glob(f'{SS_DIR}/*.png'), key=os.path.getmtime, reverse=True)
|
|
todo = [s for s in all_ss if os.path.basename(s).replace('.png','') not in already]
|
|
lg(f'Total: {len(all_ss)} | Already: {len(already)} | Todo: {len(todo)}')
|
|
|
|
# Analyze up to 15 per run (15 * 60s = 15 min max)
|
|
batch = todo[:15]
|
|
new_ok = new_fail = 0
|
|
|
|
for path in batch:
|
|
name = os.path.basename(path).replace('.png','')
|
|
size = os.path.getsize(path)
|
|
if size > 500000:
|
|
lg(f' Skip {name}: {size} bytes too large')
|
|
continue
|
|
|
|
lg(f' Analyzing {name} ({size} bytes)...')
|
|
b64 = base64.b64encode(open(path,'rb').read()).decode()
|
|
|
|
body = json.dumps({
|
|
'model': 'moondream',
|
|
'prompt': f'Analyze this web dashboard screenshot "{name}". Describe what you see: layout, elements, colors, any errors or issues. Score UX quality 0-100.',
|
|
'images': [b64],
|
|
'stream': False
|
|
})
|
|
open('/tmp/va_req.json','w').write(body)
|
|
|
|
try:
|
|
t0 = time.time()
|
|
r = subprocess.run(['curl','-sS','-X','POST','http://127.0.0.1:11435/api/generate',
|
|
'-H','Content-Type: application/json','-d','@/tmp/va_req.json',
|
|
'--max-time','90'], capture_output=True, text=True, timeout=95)
|
|
dur = time.time() - t0
|
|
|
|
d = json.loads(r.stdout)
|
|
resp = d.get('response','')
|
|
|
|
if len(resp) > 10:
|
|
scans[name] = {
|
|
'status': 'success',
|
|
'analysis': resp[:500],
|
|
'duration': round(dur,1),
|
|
'engine': 'moondream-sovereign',
|
|
'timestamp': datetime.now().isoformat(),
|
|
'score': 0 # Will be parsed from response
|
|
}
|
|
# Try to extract score
|
|
import re
|
|
m = re.search(r'(\d{1,3})/100|score.*?(\d{1,3})', resp.lower())
|
|
if m:
|
|
scans[name]['score'] = int(m.group(1) or m.group(2))
|
|
new_ok += 1
|
|
lg(f' OK {dur:.0f}s [{len(resp)}c]')
|
|
else:
|
|
scans[name] = {'status': 'fail', 'analysis': 'Empty response', 'duration': round(dur,1), 'engine': 'moondream-sovereign'}
|
|
new_fail += 1
|
|
lg(f' FAIL {dur:.0f}s empty')
|
|
except Exception as e:
|
|
scans[name] = {'status': 'fail', 'analysis': str(e)[:80], 'duration': 0, 'engine': 'moondream-sovereign'}
|
|
new_fail += 1
|
|
lg(f' ERROR: {e}')
|
|
|
|
# Save
|
|
data['scans'] = scans
|
|
data['stats'] = {
|
|
'total': len(scans),
|
|
'success': sum(1 for s in scans.values() if s.get('status')=='success'),
|
|
'fail': sum(1 for s in scans.values() if s.get('status')=='fail'),
|
|
'warn': 0, 'partial': 0
|
|
}
|
|
data['engine'] = 'moondream-sovereign'
|
|
data['total'] = len(scans)
|
|
json.dump(data, open(ANALYSIS,'w'), indent=2)
|
|
|
|
lg(f'DONE: +{new_ok} OK +{new_fail} FAIL | Total: {len(scans)} analyzed')
|