#!/usr/bin/env python3 from __future__ import annotations import argparse, fcntl, json, os, sqlite3, subprocess, sys from datetime import datetime, timezone from pathlib import Path from matching import model_identity_conflict, validate_product_match, norm from dbutil import log_event from condition import assess_condition DB=os.environ.get('FIYAT_RADAR_DB','/var/lib/fiyat-radar/fiyat-radar.sqlite') APP=Path(os.environ.get('FIYAT_RADAR_APP','/opt/fiyat-radar-app')) VENV=Path(os.environ.get('FIYAT_RADAR_VENV','/opt/fiyat-radar-venv')) LOCK=Path('/var/lib/fiyat-radar/quality-review.lock') def now(): return datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S') def connect(): c=sqlite3.connect(DB,timeout=30);c.row_factory=sqlite3.Row c.execute('PRAGMA foreign_keys=ON');c.execute('PRAGMA busy_timeout=30000');c.execute('PRAGMA synchronous=NORMAL') return c def claim_next(): c=connect() try: row=c.execute("""UPDATE quality_review_requests SET status='running',started_at=CURRENT_TIMESTAMP,attempts=attempts+1 WHERE id=(SELECT id FROM quality_review_requests WHERE status='pending' ORDER BY priority DESC,id ASC LIMIT 1) RETURNING *""").fetchone();c.commit();return dict(row) if row else None finally:c.close() def run_cmd(args,timeout): env=os.environ.copy();env['FIYAT_RADAR_DB']=DB;env['PYTHONDONTWRITEBYTECODE']='1' p=subprocess.run(args,cwd=str(APP/'worker'),env=env,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,text=True,timeout=timeout) return p.returncode,(p.stdout or '')[-6000:] def legacy_ids(c,variant_id): return [int(x[0]) for x in c.execute("SELECT lvm.legacy_product_id FROM legacy_variant_map lvm JOIN products p ON p.id=lvm.legacy_product_id WHERE lvm.variant_id=? AND p.active=1 ORDER BY p.id",(variant_id,)).fetchall()] def variants_for_request(c,req): if req.get('variant_id'): r=c.execute("SELECT * FROM product_variants WHERE id=? AND family_id=? AND active=1",(int(req['variant_id']),int(req['family_id']))).fetchone();return [dict(r)] if r else [] return [dict(x) for x in c.execute("SELECT * FROM product_variants WHERE family_id=? AND active=1 ORDER BY capacity_gb,color,id",(int(req['family_id']),)).fetchall()] def source_guard(c,pid): p=c.execute("SELECT id,name,brand,model,variant,condition_type,capacity_gb,color FROM products WHERE id=?",(pid,)).fetchone() if not p:return {'disabled':0,'candidates':0,'indirect':0,'condition_pending':0,'condition_bad':0} product=dict(p);disabled=rejected=indirect=pending=condition_bad=0 for r in c.execute("SELECT id,url,last_title FROM product_sources WHERE product_id=? AND active=1",(pid,)).fetchall(): evidence=' '.join(x for x in (r['last_title'] or '',r['url'] or '') if x);reason=model_identity_conflict(product,evidence) if reason: c.execute("UPDATE product_sources SET active=0,last_price=NULL,last_total_price=NULL,last_status='invalid_product',last_error=? WHERE id=?",(reason[:1000],r['id']));c.execute("UPDATE source_offers SET active=0 WHERE source_id=?",(r['id'],));c.execute('DELETE FROM price_observations WHERE source_id=?',(r['id'],));disabled+=1;continue ca=assess_condition(product,evidence,r['url'] or '','',require_known=True) if ca.status in ('mismatch','conflict'): c.execute("UPDATE product_sources SET active=0,last_price=NULL,last_total_price=NULL,last_status='invalid_condition',last_error=?,expected_condition=?,detected_condition=?,condition_confidence=?,condition_evidence=?,condition_status=? WHERE id=?",(f'Durum {ca.status}: {ca.evidence}'[:1000],ca.expected,ca.detected,ca.confidence,ca.evidence,ca.status,r['id']));c.execute('UPDATE source_offers SET active=0 WHERE source_id=?',(r['id'],));c.execute('DELETE FROM price_observations WHERE source_id=?',(r['id'],));condition_bad+=1 elif ca.status=='verified': c.execute("UPDATE product_sources SET expected_condition=?,detected_condition=?,condition_confidence=?,condition_evidence=?,condition_status='verified' WHERE id=?",(ca.expected,ca.detected,ca.confidence,ca.evidence,r['id'])) else: c.execute("UPDATE product_sources SET expected_condition=?,condition_status='pending',last_price=NULL,last_total_price=NULL,last_status='condition_pending' WHERE id=?",(ca.expected,r['id']));c.execute('UPDATE source_offers SET active=0 WHERE source_id=?',(r['id'],));pending+=1 for r in c.execute("SELECT id,url,title FROM discovery_candidates WHERE product_id=? AND status IN ('pending','approved','auto_approved')",(pid,)).fetchall(): text=' '.join(x for x in (r['title'] or '',r['url'] or '') if x);reason=model_identity_conflict(product,text) ca=assess_condition(product,text,r['url'] or '','',require_known=True) if reason or ca.status in ('mismatch','conflict'): c.execute("UPDATE discovery_candidates SET status='rejected',condition_status=?,reviewed_at=CURRENT_TIMESTAMP WHERE id=?",(ca.status if ca.status!='unknown' else 'pending',r['id']));rejected+=1 elif ca.status=='verified': c.execute("UPDATE discovery_candidates SET detected_condition=?,condition_confidence=?,condition_evidence=?,condition_status='verified' WHERE id=?",(ca.detected,ca.confidence,ca.evidence,r['id'])) else:c.execute("UPDATE discovery_candidates SET condition_status='pending' WHERE id=?",(r['id'],)) for r in c.execute("SELECT id,url,title,snippet FROM indirect_offers WHERE product_id=? AND active=1",(pid,)).fetchall(): text=' '.join(x for x in (r['title'] or '',r['snippet'] or '',r['url'] or '') if x);reason=model_identity_conflict(product,text);ca=assess_condition(product,text,r['url'] or '','',require_known=True) if reason or ca.status!='verified':c.execute("UPDATE indirect_offers SET active=0,condition_status=? WHERE id=?",(ca.status if ca.status!='unknown' else 'pending',r['id']));indirect+=1 else:c.execute("UPDATE indirect_offers SET detected_condition=?,condition_confidence=?,condition_evidence=?,condition_status='verified' WHERE id=?",(ca.detected,ca.confidence,ca.evidence,r['id'])) return {'disabled':disabled,'candidates':rejected,'indirect':indirect,'condition_pending':pending,'condition_bad':condition_bad} def fresh_price_ready(c,pids,started): if not pids:return False ph=','.join('?'*len(pids));args=[*pids,started] q=f"SELECT 1 FROM source_offers so JOIN product_sources ps ON ps.id=so.source_id JOIN products p ON p.id=so.product_id WHERE so.active=1 AND ps.active=1 AND ps.condition_status='verified' AND so.condition_type=p.condition_type AND so.product_id IN ({ph}) AND so.last_seen_at>=? AND (so.in_stock IS NULL OR so.in_stock=1) LIMIT 1" if c.execute(q,args).fetchone():return True q=f"SELECT 1 FROM product_sources ps JOIN products p ON p.id=ps.product_id WHERE ps.active=1 AND ps.condition_status='verified' AND ps.detected_condition=p.condition_type AND ps.last_price IS NOT NULL AND ps.product_id IN ({ph}) AND COALESCE(ps.last_checked_at,'')>=? AND ps.last_status='ok' LIMIT 1" return bool(c.execute(q,args).fetchone()) def image_ready(c,pids,started): if not pids:return False ph=','.join('?'*len(pids));q=f"SELECT 1 FROM products WHERE id IN ({ph}) AND image_path IS NOT NULL AND image_path<>'' AND image_locked=1 AND COALESCE(image_verified_at,image_updated_at,image_checked_at,'')>=? LIMIT 1" return bool(c.execute(q,[*pids,started]).fetchone()) def process(req): rid=int(req['id']);fid=int(req['family_id']);started=now();c=connect();vars=variants_for_request(c,req) if not vars: c.execute("UPDATE quality_review_requests SET status='done',finished_at=CURRENT_TIMESTAMP,result_status='missing',note='İncelenecek aktif varyant bulunamadı' WHERE id=?",(rid,));c.commit();c.close();return log_event(c,'quality.review','warning','Derin kalite incelemesi başladı',run_type='quality',run_id=rid,details={'family_id':fid,'variant_id':req.get('variant_id'),'variant_count':len(vars)});c.commit();c.close() diagnostics=[];passed=[];checked_colors=set();work_vars=vars[:40] for v in work_vars: vid=int(v['id']);c=connect();pids=legacy_ids(c,vid);guard={'disabled':0,'candidates':0,'indirect':0,'condition_pending':0,'condition_bad':0} for pid in pids: g=source_guard(c,pid) for k in guard:guard[k]+=g[k] c.commit();c.close() rep=pids[0] if pids else None;steps=[] if rep: cmds=[ ([str(VENV/'bin/python'),str(APP/'worker/variant_engine.py'),'--product-id',str(rep)],50,'variant-ai'), ([str(VENV/'bin/python'),str(APP/'worker/discover.py'),'--product-id',str(rep),'--target-sources','7','--browser-max-pages','8','--web-limit','12'],100,'discovery'), ([str(VENV/'bin/python'),str(APP/'worker/scanner.py'),'--product-id',str(rep),'--force','--limit','80'],100,'scan'), ] for cmd,to,label in cmds: try:rc,out=run_cmd(cmd,to);steps.append({'step':label,'rc':rc,'tail':out[-1200:]}) except subprocess.TimeoutExpired:steps.append({'step':label,'rc':124,'tail':'timeout'}) color_key=norm(str(v.get('color') or '')) if color_key not in checked_colors: checked_colors.add(color_key) try:rc,out=run_cmd([str(VENV/'bin/python'),str(APP/'worker/image_fetcher.py'),'--product-id',str(rep),'--force'],120);steps.append({'step':'image-deep','rc':rc,'tail':out[-1200:]}) except subprocess.TimeoutExpired:steps.append({'step':'image-deep','rc':124,'tail':'timeout'}) c=connect();pids=legacy_ids(c,vid);price_ok=fresh_price_ready(c,pids,started);img_ok=image_ready(c,pids,started) # Aynı renkte daha önce bu inceleme turunda görsel doğrulandıysa propagation sonrası timestamp yeterli olur. ok=bool(price_ok and img_ok) c.execute("UPDATE product_variants SET visibility_status=?,quality_status=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",('visible' if ok else 'hidden','verified' if ok else 'needs_review',vid)) c.commit();c.close();diagnostics.append({'variant_id':vid,'legacy_ids':pids,'guard':guard,'price_fresh':price_ok,'image_fresh':img_ok,'passed':ok,'steps':steps}); if ok:passed.append(vid) c=connect() family_report=(req.get('entity_type')=='family' or not req.get('variant_id')) # Güvenlik: olağanüstü büyük bir ailede servis bütçesi nedeniyle bu turda işlenemeyen # varyantları otomatik yayına döndürme; gizli/needs_review bırak. if family_report and len(vars)>len(work_vars): rest=[int(x['id']) for x in vars[len(work_vars):]] ph=','.join('?'*len(rest)) c.execute(f"UPDATE product_variants SET visibility_status='hidden',quality_status='needs_review',updated_at=CURRENT_TIMESTAMP WHERE id IN ({ph})",rest) diagnostics.append({'unprocessed_variant_ids':rest,'reason':'Derin inceleme turu güvenlik limiti; yeniden manuel inceleme gerekli'}) if family_report: if passed: c.execute("UPDATE product_families SET visibility_status='visible',quality_status='verified',updated_at=CURRENT_TIMESTAMP WHERE id=?",(fid,));result='passed';note=f'Derin inceleme tamamlandı: {len(passed)}/{len(vars)} varyant yeniden yayına uygun.' else: c.execute("UPDATE product_families SET visibility_status='hidden',quality_status='needs_review',updated_at=CURRENT_TIMESTAMP WHERE id=?",(fid,));result='failed';note='Derin inceleme güvenli yayın koşullarını sağlayamadı; ürün gizli tutuluyor.' else: # Tek varyant bildirimi diğer sağlıklı varyantların ürün ailesini yayından düşürmez. result='passed' if passed else 'failed' note=('Bildirilen varyant derin incelemeyi geçti ve yeniden yayına alındı.' if passed else 'Bildirilen varyant doğrulanamadı; yalnız bu varyant gizli tutuluyor.') c.execute("UPDATE product_families SET quality_status=CASE WHEN visibility_status='under_review' THEN quality_status ELSE 'ok' END,updated_at=CURRENT_TIMESTAMP WHERE id=?",(fid,)) c.execute("UPDATE quality_review_requests SET status='done',finished_at=CURRENT_TIMESTAMP,result_status=?,note=?,details_json=? WHERE id=?",(result,note,json.dumps(diagnostics,ensure_ascii=False),rid)) log_event(c,'quality.review','info' if passed else 'warning',note,run_type='quality',run_id=rid,details={'family_id':fid,'passed_variants':passed,'checks':diagnostics});c.commit();c.close() try:run_cmd(['/usr/bin/php',str(APP/'bin/rebuild-family.php'),'--id',str(fid)],30) except Exception:pass def fail(req,e): c=connect();c.execute("UPDATE quality_review_requests SET status='pending',started_at=NULL,note=? WHERE id=?",(f'Tekrar denenecek: {e}'[:1500],int(req['id'])));log_event(c,'quality.review','error',f'Derin inceleme hatası: {e}',run_type='quality',run_id=int(req['id']),details={'family_id':req.get('family_id'),'variant_id':req.get('variant_id')});c.commit();c.close() def run(limit=4): LOCK.parent.mkdir(parents=True,exist_ok=True) with LOCK.open('a+') as fh: try:fcntl.flock(fh.fileno(),fcntl.LOCK_EX|fcntl.LOCK_NB) except BlockingIOError:return 0 n=0 while n