#!/usr/bin/env python3 from __future__ import annotations import argparse, fcntl, hashlib, os, sqlite3, sys from contextlib import contextmanager from datetime import datetime, timedelta, timezone from pathlib import Path from urllib.parse import urlparse import requests from extractors.generic import extract_all from matching import validate_product_match from condition import assess_condition from browser_fetch import HybridBrowser from decision_engine import evaluate from dbutil import setting, setting_bool, setting_int, log_event DB=os.environ.get('FIYAT_RADAR_DB','/var/lib/fiyat-radar/fiyat-radar.sqlite') UA=os.environ.get('FIYAT_RADAR_UA','Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36') TIMEOUT=int(os.environ.get('FIYAT_RADAR_TIMEOUT','20')) LOCK_FILE=os.environ.get('FIYAT_RADAR_LOCK_FILE','/var/lib/fiyat-radar/scan.lock') def now(): return datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S') @contextmanager def scan_lock(): Path(LOCK_FILE).parent.mkdir(parents=True,exist_ok=True) with open(LOCK_FILE,'a+',encoding='utf-8') as fh: fcntl.flock(fh.fileno(),fcntl.LOCK_EX) try: yield finally: fcntl.flock(fh.fileno(),fcntl.LOCK_UN) 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 # v0.3.16: Görsel ağ/indirme kodu fiyat scanner'ından tamamen çıkarıldı. # Scanner yalnız fiyat/kaynak doğrular ve gerektiğinde bağımsız image queue'ya iş bırakır. def http_fetch(session,row): domain=(urlparse(row['url']).hostname or '').lower();merchant_domain=(row['domain'] or '').lower() if merchant_domain and not (domain==merchant_domain or domain.endswith('.'+merchant_domain)):raise ValueError(f'URL domaini mağaza domainiyle eşleşmiyor: {domain}') r=session.get(row['url'],timeout=TIMEOUT,headers={'User-Agent':UA,'Accept-Language':'tr-TR,tr;q=0.9,en;q=0.7','Accept':'text/html,application/xhtml+xml'}) r.raise_for_status();html=r.text if 'html' not in r.headers.get('content-type','').lower() and '0,max_pages=_bmp,timeout_ms=_btm,user_agent=UA) log_event(c,'scan','info','Tarama başladı',run_type='scan',run_id=run_id,product_id=product_id,details={'limit':limit,'force':force,'manual':manual,'fetch_mode':fetch_mode},commit=True) q="""SELECT s.*,p.name product_name,p.name,p.brand,p.model,p.variant,p.condition_type,p.capacity_gb,p.color,p.ean,p.image_path,p.image_source_url,p.image_updated_at,p.image_locked,m.name merchant,m.domain,m.trust_score FROM product_sources s JOIN products p ON p.id=s.product_id JOIN merchants m ON m.id=s.merchant_id WHERE s.active=1 AND p.active=1 AND m.active=1""";args=[] if product_id is not None:q+=' AND p.id=?';args.append(int(product_id)) if not force:q+=" AND (s.next_check_at IS NULL OR s.next_check_at <= datetime('now'))" q+=" ORDER BY CASE WHEN COALESCE(s.condition_status,'pending')<>'verified' THEN 0 ELSE 1 END,CASE WHEN s.last_price IS NULL THEN 0 ELSE 1 END,COALESCE(s.next_check_at,'1970-01-01'),s.id LIMIT ?";args.append(limit) rows=c.execute(q,args).fetchall();ok=fail=review=0;session=requests.Session();image_done=set() try: for row in rows: checked=now();next_at=(datetime.now(timezone.utc)+timedelta(minutes=max(15,row['check_interval_minutes']))).strftime('%Y-%m-%d %H:%M:%S') try: offers,html,fetched_via,final=fetch_offers(session,row,browser,run_id,c) evaluated=[];condition_errors=[] for candidate in offers[:40]: try: evidence_text=candidate.match_text or candidate.title validate_product_match(dict(row),evidence_text,html=html,url=final,condition_text=candidate.condition_text or '',enforce_condition=True) ca=assess_condition(dict(row),evidence_text,final,candidate.condition_text or '',require_known=True) if ca.status!='verified':raise ValueError(f'CONDITION_{ca.status.upper()}: {ca.evidence or "durum kanıtı yok"}') candidate.condition_type=ca.detected or ca.expected candidate.condition_confidence=ca.confidence candidate.condition_evidence=ca.evidence if smart:dec=evaluate(c,dict(row),candidate,html,trust_score=row['trust_score'],fetched_via=fetched_via) else: from matching import confidence conf=confidence(dict(row),candidate.title,html,final,candidate.condition_text or '');dec=type('D',(),{'status':'auto' if conf>=.45 else 'reject','score':conf,'match_confidence':conf,'price_sanity':1.0,'summary':f'Klasik eşleşme: {conf:.2f}','reasons':[]})() evaluated.append((candidate,dec)) except Exception as oe: if str(oe).startswith('CONDITION_'):condition_errors.append(str(oe)) log_event(c,'scan.offer','debug',f'Satıcı teklifi elendi: {oe}',run_type='scan',run_id=run_id,product_id=row['product_id'],source_id=row['id'],merchant_id=row['merchant_id']) if not evaluated: if condition_errors: # Mismatch/conflict unknown'dan daha kesin ve daha önceliklidir. condition_errors.sort(key=lambda x: 0 if x.startswith(('CONDITION_MISMATCH','CONDITION_CONFLICT')) else 1) raise ValueError(condition_errors[0]) raise ValueError('Doğrulanabilir satıcı/fiyat teklifi bulunamadı') auto=[x for x in evaluated if x[1].status=='auto' and x[0].in_stock is not False] or [x for x in evaluated if x[1].status=='auto'] known_shipping=[x for x in auto if x[0].shipping_known] pick_pool=known_shipping or auto offer,dec=(min(pick_pool,key=lambda x:(x[0].total_price,x[0].price)) if pick_pool else max(evaluated,key=lambda x:x[1].score)) log_event(c,'scan.decision','info' if dec.status=='auto' else 'warning',dec.summary,run_type='scan',run_id=run_id,product_id=row['product_id'],source_id=row['id'],merchant_id=row['merchant_id'],details={'decision':dec.status,'score':dec.score,'match':dec.match_confidence,'price_sanity':dec.price_sanity,'fetched_via':fetched_via,'title':offer.title,'price':offer.price,'seller':offer.seller_name,'shipping_price':offer.shipping_price,'shipping_known':offer.shipping_known,'total_price':offer.total_price,'offer_count':len(evaluated)}) if dec.status!='auto': c.execute("UPDATE product_sources SET last_status='review',last_error=?,last_checked_at=?,next_check_at=? WHERE id=?",(dec.summary,checked,next_at,row['id']));review+=1;c.commit();continue saved_offers=sync_source_offers(c,row,evaluated,checked) c.execute("""UPDATE product_sources SET last_price=?,last_currency=?,last_stock=?,last_title=?,last_confidence=?,last_status='ok',last_error=NULL,last_checked_at=?,next_check_at=?,consecutive_failures=0,last_seller_name=?,last_shipping_price=?,last_shipping_known=?,last_total_price=?,stock_confidence=?,expected_condition=?,detected_condition=?,condition_confidence=?,condition_evidence=?,condition_status='verified' WHERE id=?""", (offer.price,offer.currency,None if offer.in_stock is None else int(offer.in_stock),offer.title,dec.score,checked,next_at,offer.seller_name or row['merchant'],offer.shipping_price,int(offer.shipping_known),offer.total_price,offer.stock_confidence,row['condition_type'],offer.condition_type,offer.condition_confidence,offer.condition_evidence,row['id'])) c.execute("""INSERT INTO price_observations(product_id,source_id,merchant_id,price,currency,in_stock,confidence,raw_title,seller_name,shipping_price,shipping_known,total_price,stock_confidence,condition_type,condition_confidence,condition_evidence,observed_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (row['product_id'],row['id'],row['merchant_id'],offer.price,offer.currency,None if offer.in_stock is None else int(offer.in_stock),dec.score,offer.title,offer.seller_name or row['merchant'],offer.shipping_price,int(offer.shipping_known),offer.total_price,offer.stock_confidence,offer.condition_type,offer.condition_confidence,offer.condition_evidence,checked)) # v0.3.16: FİYAT TARAYICISI GÖRSEL İNDİRMEZ. Görsel CDN/arama/ # doğrulama sorunu fiyat transaction'ını geciktiremez veya başarısız kılamaz. # Fiyatı kaydet, gerekiyorsa bağımsız image worker'a yalnız iş bırak. if row['product_id'] not in image_done: need_image=image_force or not row['image_path'] or int(row['image_locked'] or 0)!=1 if need_image: prio=220 if image_force else 80 c.execute("""INSERT INTO image_fetch_requests(product_id,status,priority,attempts,requested_at,next_attempt_at,note) VALUES(?,'pending',?,0,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,?) ON CONFLICT(product_id) DO UPDATE SET status=CASE WHEN image_fetch_requests.status='running' THEN 'running' ELSE 'pending' END, priority=MAX(image_fetch_requests.priority,excluded.priority), next_attempt_at=CASE WHEN image_fetch_requests.status='running' THEN image_fetch_requests.next_attempt_at ELSE CURRENT_TIMESTAMP END, note=excluded.note""", (row['product_id'],prio,'Fiyat taraması tamamlandı; görsel bağımsız worker tarafından doğrulanacak')) image_done.add(row['product_id']) log_event(c,'scan.image-queue','info','Fiyat kaydedildi; görsel bağımsız kuyruğa bırakıldı',run_type='scan',run_id=run_id,product_id=row['product_id'],source_id=row['id'],merchant_id=row['merchant_id']) ok+=1;c.commit() if row['product_id'] in image_done: try:Path('/var/lib/fiyat-radar/image.trigger').touch() except OSError:pass print(f"OK {row['merchant']} | {row['product_name']} | {offer.price} {offer.currency} | toplam={offer.total_price} | satıcı={offer.seller_name or row['merchant']} | teklifler={saved_offers} | karar={round(dec.score*100)} | {fetched_via}") except Exception as e: msg=str(e) # Ürün durumu ürün kimliğinin zorunlu parçasıdır. Sıfır ile yenilenmiş/açık kutu/ikinci el karışamaz. if msg.startswith(('CONDITION_MISMATCH:','CONDITION_CONFLICT:')): status='mismatch' if msg.startswith('CONDITION_MISMATCH:') else 'conflict' c.execute("UPDATE product_sources SET active=0,expected_condition=?,condition_status=?,last_price=NULL,last_total_price=NULL,last_status='invalid_condition',last_error=?,last_checked_at=? WHERE id=?",(row['condition_type'],status,msg[:1000],checked,row['id'])) c.execute('UPDATE source_offers SET active=0 WHERE source_id=?',(row['id'],)) c.execute('DELETE FROM price_observations WHERE source_id=?',(row['id'],)) log_event(c,'scan.condition-guard','warning','Yanlış ürün durumuna ait kaynak devre dışı bırakıldı ve kirli fiyat geçmişi temizlendi',run_type='scan',run_id=run_id,product_id=row['product_id'],source_id=row['id'],merchant_id=row['merchant_id'],details={'url':row['url'],'expected':row['condition_type'],'reason':msg});fail+=1;c.commit();print(f"INVALID-CONDITION {row['merchant']} | {row['product_name']} | {e}",file=sys.stderr);continue if msg.startswith('CONDITION_UNKNOWN:'): # Belirsiz durum sıfır sayılmaz. Kaynağı tekrar denenebilir tut fakat fiyatı kesinlikle yayımlama. next_unknown=(datetime.now(timezone.utc)+timedelta(minutes=30)).strftime('%Y-%m-%d %H:%M:%S') c.execute("UPDATE product_sources SET expected_condition=?,detected_condition=NULL,condition_status='pending',condition_confidence=NULL,condition_evidence=NULL,last_price=NULL,last_total_price=NULL,last_status='condition_unknown',last_error=?,last_checked_at=?,next_check_at=? WHERE id=?",(row['condition_type'],msg[:1000],checked,next_unknown,row['id'])) c.execute('UPDATE source_offers SET active=0 WHERE source_id=?',(row['id'],)) log_event(c,'scan.condition-guard','warning','Ürün durumu belirsiz; fiyat yayından çıkarıldı ve tekrar incelemeye bırakıldı',run_type='scan',run_id=run_id,product_id=row['product_id'],source_id=row['id'],merchant_id=row['merchant_id'],details={'url':row['url'],'expected':row['condition_type']});review+=1;c.commit();print(f"CONDITION-UNKNOWN {row['merchant']} | {row['product_name']}",file=sys.stderr);continue # Kesin kardeş-model çatışması yeniden denenebilir ağ hatası değildir. # Yanlış linki anında kapat ve daha önce yazdığı sahte fiyat geçmişini temizle. if msg.startswith('Kardeş model niteleyicisi uyuşmuyor'): c.execute("UPDATE product_sources SET active=0,last_status='invalid_model',last_error=?,last_checked_at=? WHERE id=?",(msg[:1000],checked,row['id'])) c.execute('UPDATE source_offers SET active=0 WHERE source_id=?',(row['id'],)) c.execute('DELETE FROM price_observations WHERE source_id=?',(row['id'],)) log_event(c,'scan.model-guard','warning','Kardeş modele ait kaynak anında devre dışı bırakıldı',run_type='scan',run_id=run_id,product_id=row['product_id'],source_id=row['id'],merchant_id=row['merchant_id'],details={'url':row['url'],'reason':msg});fail+=1;c.commit();print(f"INVALID-MODEL {row['merchant']} | {row['product_name']} | {e}",file=sys.stderr);continue fails=row['consecutive_failures']+1;backoff=min(1440,max(row['check_interval_minutes'],30)*(2 if fails>=3 else 1));next_at=(datetime.now(timezone.utc)+timedelta(minutes=backoff)).strftime('%Y-%m-%d %H:%M:%S') c.execute("UPDATE product_sources SET last_status='error',last_error=?,last_checked_at=?,next_check_at=?,consecutive_failures=? WHERE id=?",(msg[:1000],checked,next_at,fails,row['id']));fail+=1 log_event(c,'scan','error',msg,run_type='scan',run_id=run_id,product_id=row['product_id'],source_id=row['id'],merchant_id=row['merchant_id'],details={'url':row['url']});c.commit();print(f"ERR {row['merchant']} | {row['product_name']} | {e}",file=sys.stderr) c.execute('UPDATE scan_log SET finished_at=?,checked_count=?,success_count=?,failure_count=? WHERE id=?',(now(),len(rows),ok,fail+review,run_id));log_event(c,'scan','info','Tarama tamamlandı',run_type='scan',run_id=run_id,details={'checked':len(rows),'success':ok,'review':review,'failure':fail,'browser_pages':browser.used if browser else 0});c.commit() if ok: try: Path('/var/lib/fiyat-radar/family-sync.trigger').touch() except OSError: pass print(f'Tamamlandı: {len(rows)} kontrol, {ok} başarılı, {review} inceleme, {fail} hata');return len(rows),ok,fail+review finally: if browser:browser.close() c.close() def process_manual_queue(max_requests=10): processed=0 while processed