from __future__ import annotations import json, re from dataclasses import dataclass from urllib.parse import urljoin from bs4 import BeautifulSoup @dataclass class Offer: title: str | None price: float currency: str in_stock: bool | None image_url: str | None = None seller_name: str | None = None seller_url: str | None = None shipping_price: float | None = None shipping_known: bool = False stock_confidence: float = 0.45 match_text: str | None = None condition_text: str | None = None condition_type: str | None = None condition_confidence: float | None = None condition_evidence: str | None = None @property def total_price(self) -> float: return float(self.price) + (float(self.shipping_price) if self.shipping_known and self.shipping_price is not None else 0.0) def _price(v): if v is None: return None if isinstance(v, (int,float)): return float(v) if isinstance(v, dict): for k in ('value','price','amount'): if k in v: p=_price(v.get(k)) if p is not None: return p return None s=str(v).strip().replace('\xa0',' ') s=re.sub(r'[^0-9,\.]','',s) if not s: return None if ',' in s and '.' in s: if s.rfind(',') > s.rfind('.'): s=s.replace('.','').replace(',','.') else: s=s.replace(',','') elif ',' in s: tail=s.split(',')[-1] s=s.replace('.','').replace(',','.' if len(tail)<=2 else '') elif s.count('.')>1: s=s.replace('.','') try: p=float(s) return p if 0.01 <= p <= 100_000_000 else None except ValueError: return None def _stock(v): if v is None: return None s=str(v).lower() if 'outofstock' in s or 'out_of_stock' in s or 'tükendi' in s or 'stokta yok' in s or 'soldout' in s: return False if 'instock' in s or 'in_stock' in s or 'stokta' in s or 'limitedavailability' in s: return True return None def _image(v): if isinstance(v, str): return v.strip() or None if isinstance(v, list): for x in v: got=_image(x) if got: return got if isinstance(v, dict): for k in ('contentUrl','url','thumbnailUrl'): got=_image(v.get(k)) if got: return got return None def _seller(v): if isinstance(v,str): return (v.strip() or None),None if not isinstance(v,dict): return None,None name=v.get('name') or v.get('legalName') url=v.get('url') return (str(name).strip() if name else None),(str(url).strip() if url else None) def _shipping(obj): """JSON-LD shippingDetails içinden bilinen kargo ücretini çıkarır.""" details=obj.get('shippingDetails') if isinstance(obj,dict) else None if details is None: return None,False items=details if isinstance(details,list) else [details] values=[] for item in items: if not isinstance(item,dict): continue rate=item.get('shippingRate') raw_rate=rate if isinstance(rate,dict): raw_rate=rate.get('value') if rate.get('value') is not None else rate.get('price') p=_price(raw_rate) if p is None and raw_rate is not None: try: zero=float(str(raw_rate).replace(',','.').strip()) if zero==0: p=0.0 except (TypeError,ValueError): pass if p is not None: values.append(p) # Açıkça ücretsiz kargo işareti varsa 0 olarak bilinir. txt=' '.join(str(item.get(k,'')) for k in ('name','description')).lower() if 'free shipping' in txt or 'ücretsiz kargo' in txt: values.append(0.0) if values: return min(values),True return None,False def _product_match_text(obj, page_title=None): if not isinstance(obj,dict): return page_title or '' parts=[page_title or '',str(obj.get('name') or ''),str(obj.get('color') or ''),str(obj.get('size') or ''),str(obj.get('sku') or ''),str(obj.get('mpn') or ''),str(obj.get('itemCondition') or '')] props=obj.get('additionalProperty') if isinstance(props,dict): props=[props] if isinstance(props,list): for pr in props: if isinstance(pr,dict): parts.extend([str(pr.get('name') or ''),str(pr.get('value') or '')]) return ' '.join(x for x in parts if x).strip() def _walk_json(obj, page_title=None, inherited_image=None, inherited_condition=None): out=[] if isinstance(obj,list): for x in obj: out += _walk_json(x,page_title,inherited_image,inherited_condition) elif isinstance(obj,dict): typ=obj.get('@type') types={typ} if isinstance(typ,str) else set(typ or []) if isinstance(typ,list) else set() local_image=_image(obj.get('image')) or inherited_image local_condition=str(obj.get('itemCondition') or inherited_condition or '').strip() or None if 'Product' in types: title=obj.get('name') or page_title match_text=_product_match_text(obj,page_title) offers=obj.get('offers') if offers: out += _walk_offer(offers,title,local_image,match_text,local_condition) elif 'Offer' in types or 'AggregateOffer' in types: out += _walk_offer(obj,page_title,local_image,page_title,local_condition) for k,v in obj.items(): if k not in ('offers',): out += _walk_json(v,page_title,local_image,local_condition) return out def _walk_offer(obj,title,image_url=None,match_text=None,inherited_condition=None): if isinstance(obj,list): out=[] for x in obj: out += _walk_offer(x,title,image_url,match_text,inherited_condition) return out if not isinstance(obj,dict): return [] offer_image=_image(obj.get('image')) or image_url local_condition=str(obj.get('itemCondition') or inherited_condition or '').strip() or None # AggregateOffer içindeki gerçek satıcı tekliflerini önce al. nested=[] if obj.get('offers') is not None: nested=_walk_offer(obj.get('offers'),title,offer_image,match_text,local_condition) seller_name,seller_url=_seller(obj.get('seller') or obj.get('offeredBy')) shipping_price,shipping_known=_shipping(obj) stock=_stock(obj.get('availability')) stock_conf=.92 if obj.get('availability') is not None else .45 current=[] for key in ('price','lowPrice','highPrice'): p=_price(obj.get(key)) if p is not None: current.append(Offer(title,p,str(obj.get('priceCurrency') or 'TRY').upper(),stock,offer_image,seller_name,seller_url,shipping_price,shipping_known,stock_conf,match_text or title,condition_text=local_condition)) break # Gerçek satıcı teklifleri varsa satıcısız aggregate lowPrice'ı tekrar göstermeyelim. if nested and not seller_name: return nested return nested+current def _meta_image(soup): for attrs in ({'property':'og:image'}, {'property':'og:image:secure_url'}, {'name':'twitter:image'}, {'itemprop':'image'}): tag=soup.find(['meta','link'],attrs=attrs) if tag: got=tag.get('content') or tag.get('href') if got:return got return None def _meta_condition(soup): for attrs in ({'itemprop':'itemCondition'}, {'property':'product:item_condition'}, {'name':'itemCondition'}, {'name':'condition'}): tag=soup.find(['meta','link','span'],attrs=attrs) if tag: got=tag.get('content') or tag.get('href') or tag.get_text(' ',strip=True) if got:return str(got).strip() # Bütün sayfada kelime aramak yanlış pozitif üretir (önerilen başka ürünler olabilir). # Yalnız açık bir "Ürün Durumu / Condition" etiketi çevresini kanıt kabul et. label_re=re.compile(r'^(?:ürün\s+durumu|urun\s+durumu|condition|ürün\s+kondisyonu|urun\s+kondisyonu)\s*:?$',re.I) for tag in soup.find_all(['dt','th','td','span','div','li','strong'],limit=800): txt=tag.get_text(' ',strip=True) if not txt or len(txt)>80:continue if label_re.match(txt): sib=tag.find_next_sibling() if sib: val=sib.get_text(' ',strip=True) if val and len(val)<=160:return val parent=tag.parent if parent: val=parent.get_text(' ',strip=True) if val and len(val)<=240:return val return None def extract_all(html:str, url:str) -> list[Offer]: soup=BeautifulSoup(html,'lxml') title=(soup.title.get_text(' ',strip=True) if soup.title else None) candidates=[] for tag in soup.find_all('script', attrs={'type':'application/ld+json'}): raw=tag.string or tag.get_text() if not raw: continue try: candidates += _walk_json(json.loads(raw), title) except Exception: continue meta_image=_meta_image(soup) meta_condition=_meta_condition(soup) meta_price=None; currency='TRY'; stock=None; stock_conf=.45 for tagname,attrs,field in [ ('meta', {'property':'product:price:amount'}, 'content'), ('meta', {'property':'og:price:amount'}, 'content'), ('meta', {'itemprop':'price'}, 'content')]: tag=soup.find(tagname,attrs=attrs) if tag: meta_price=_price(tag.get(field)) if meta_price is not None: break cur=soup.find('meta',attrs={'property':'product:price:currency'}) or soup.find('meta',attrs={'itemprop':'priceCurrency'}) if cur and cur.get('content'): currency=cur.get('content').upper() av=soup.find('link',attrs={'itemprop':'availability'}) or soup.find('meta',attrs={'itemprop':'availability'}) if av: stock=_stock(av.get('href') or av.get('content')); stock_conf=.82 if meta_price is not None: candidates.append(Offer(title,meta_price,currency,stock,meta_image,stock_confidence=stock_conf,match_text=title,condition_text=meta_condition)) if not candidates: for pat in [ r'"price"\s*:\s*"?([0-9][0-9\.,]{1,15})', r'"currentPrice"\s*:\s*"?([0-9][0-9\.,]{1,15})', r'([0-9][0-9\.]{1,12},[0-9]{2})\s*(?:₺|TL|TRY)']: m=re.search(pat,html,re.I) if m: p=_price(m.group(1)) if p is not None: candidates.append(Offer(title,p,'TRY',None,meta_image,match_text=title,condition_text=meta_condition)); break if not candidates: raise ValueError('Sayfada güvenilir fiyat alanı bulunamadı') # URL'leri mutlaklaştır, tekrarları temizle. out=[];seen=set() for o in candidates: if not o.image_url and meta_image:o.image_url=meta_image if not o.condition_text and meta_condition:o.condition_text=meta_condition if o.image_url:o.image_url=urljoin(url,o.image_url) if o.seller_url:o.seller_url=urljoin(url,o.seller_url) key=((o.seller_name or '').casefold(),round(float(o.price),2),round(float(o.shipping_price or 0),2),o.currency,o.in_stock) if key in seen:continue seen.add(key);out.append(o) # Satıcılı teklifler varsa satıcısız aynı/aggregate fiyatları geri plana at. out.sort(key=lambda o:(0 if o.in_stock is not False else 1,0 if o.seller_name else 1,o.total_price,o.price)) return out def extract(html:str, url:str) -> Offer: return extract_all(html,url)[0]