ghstedpixel/app.py
0
1import json2import re3import httpx4import uvicorn5import asyncio6import os7import urllib.parse8from urllib.parse import urlparse, parse_qs, quote_plus9from fastapi import FastAPI, Request, Form10from fastapi.responses import HTMLResponse11from fastapi.templating import Jinja2Templates12 13from config import HF_TOKEN14from security import scrub_conversational_bloat, check_url_safe15from database import (16 get_cached_deals, set_cached_deals,17 get_cached_coupons, set_cached_coupons,18 get_base_price, set_base_price,19 redis_client20)21from schema import extract_universal_sizing, is_size_mismatched, parse_price_locally22from network import extract_page_data23from coupons import call_free_llm, fetch_deals_standalone24from shopping import scan_prices25 26# Clean import from your data separation module file27from popular import get_trending_deals28 29app = FastAPI()30 31BASE_DIR = os.path.dirname(os.path.abspath(__file__))32templates = Jinja2Templates(directory=BASE_DIR)33 34def unwrap_direct_product_url(url_str: str) -> str:35 """Recursively peels back tracking redirect strings to isolate raw retailer tokens"""36 if not url_str or not isinstance(url_str, str):37 return ""38 current_url = url_str.strip()39 for _ in range(5):40 if "google.com" not in current_url and "googleadservices.com" not in current_url:41 break42 parsed = urlparse(current_url)43 queries = parse_qs(parsed.query)44 found = False45 for key in ["adurl", "url", "q", "u", "link", "destination", "ad_url", "gclid"]:46 if queries.get(key):47 candidate = queries[key][0].strip()48 if "http" in candidate:49 if not candidate.startswith("http"):50 candidate = candidate[candidate.find("http"):]51 current_url = urllib.parse.unquote(candidate)52 found = True53 break54 if not found:55 match = re.search(r'(?:adurl|url|q|link)=(https?%3A%2F%2F[^&]+|https?://[^&]+)', current_url, re.IGNORECASE)56 if match:57 current_url = urllib.parse.unquote(match.group(1))58 found = True59 else:60 break61 return "" if "google.com" in current_url or "googleadservices.com" in current_url else current_url62 63def clean_retailer_url(url_str: str) -> str:64 """Strips click-tracking bloat parameters to present short paths safe from ad blockers"""65 if not url_str:66 return ""67 try:68 parsed = urlparse(url_str)69 queries = parse_qs(parsed.query)70 clean_queries = {}71 drop_params = {72 'gclid', 'gclsrc', 'utm_source', 'utm_medium', 'utm_campaign', 73 'utm_content', 'utm_term', 'fbclid', '_ga', '_gl', 'srsltid'74 }75 for k, v in queries.items():76 if k.lower() not in drop_params:77 clean_queries[k] = v78 new_query = urllib.parse.urlencode(clean_queries, doseq=True)79 return urllib.parse.urlunparse((80 parsed.scheme, parsed.netloc, parsed.path, 81 parsed.params, new_query, parsed.fragment82 ))83 except:84 return url_str85 86def get_status_context():87 """Generates reactive state tags for UI elements mapping cache statuses"""88 status_text = "Google Pricing Intelligence & Historical Tracking Active" if redis_client else "Security Protection Active | Core Memory Tracking"89 status_color = "var(--accent-green)" if redis_client else "#f59e0b"90 return status_text, status_color91 92@app.get("/", response_class=HTMLResponse)93def dashboard(request: Request):94 status_text, status_color = get_status_context()95 return templates.TemplateResponse(96 request=request,97 name="dashboard.html", 98 context={99 "status_text": status_text,100 "status_color": status_color,101 "prev_name": "",102 "prev_size": "",103 "prev_mode": "balanced",104 "prev_gender": "unisex",105 "show_results": False,106 "error_msg": None,107 "popular_deals": get_trending_deals(),108 "active_tab": "scan"109 }110 )111 112@app.post("/search", response_class=HTMLResponse)113async def handle_search(114 request: Request,115 product_name: str = Form(...), 116 product_size: str = Form(None), 117 search_mode: str = Form("balanced"), 118 product_gender: str = Form("unisex")119):120 status_text, status_color = get_status_context()121 base_ui_context = {122 "status_text": status_text,123 "status_color": status_color,124 "prev_name": product_name,125 "prev_size": product_size or "",126 "prev_mode": search_mode,127 "prev_gender": product_gender,128 "show_results": False,129 "error_msg": None,130 "popular_deals": get_trending_deals(),131 "active_tab": "scan"132 }133 134 if not HF_TOKEN:135 base_ui_context["error_msg"] = "Please add a valid HF_TOKEN into Space Secrets to activate the free AI models."136 return templates.TemplateResponse(request=request, name="dashboard.html", context=base_ui_context)137 138 original_name_input = product_name139 original_size_input = product_size or ""140 raw_input = product_name.strip()141 142 force_refresh = raw_input.endswith("*")143 if force_refresh:144 raw_input = raw_input.rstrip("*").strip()145 print(f"⚡ [Force Refresh] Evicting active cache entries for query mapping context: '{raw_input}'")146 147 extracted_size_from_name = extract_universal_sizing(raw_input)148 if original_size_input:149 requested_size = extract_universal_sizing(original_size_input) or original_size_input.strip()150 else:151 requested_size = extracted_size_from_name152 153 if not original_size_input and extracted_size_from_name:154 patterns_to_strip = [155 r'(\d+(?:\.\d+)?)\s*(?:FL\.?\s*)?(OZ|ML|GB|TB)\b',156 r'\b(EXTRA\s*LARGE|X-LARGE|LARGE|MEDIUM|SMALL)\b',157 r'\b(XS|S|M|L|XL|2XL|XXL|3XL|4XL)\b',158 r'\b(?:SIZE|SZ|US|UK|EU)?\s*(\d+(?:\.5)?)\b'159 ]160 target_item = raw_input161 for pattern in patterns_to_strip:162 target_item = re.sub(pattern, "", target_item, flags=re.IGNORECASE).strip()163 target_item = " ".join(target_item.split())164 else:165 target_item = raw_input166 167 target_item = scrub_conversational_bloat(target_item) if not target_item.startswith(("http", "https")) else target_item168 gender_prefix = ""169 if product_gender == "mens": gender_prefix = "men's "170 elif product_gender == "womens": gender_prefix = "women's "171 elif product_gender == "kids": gender_prefix = "kids' "172 173 if gender_prefix and not target_item.startswith(("http://", "https://")):174 if gender_prefix.strip().lower() not in target_item.lower(): target_item = f"{gender_prefix}{target_item}"175 176 deals = []177 coupons = []178 original_site_deal = None179 180 if target_item.startswith(("http://", "https://")):181 input_url = target_item182 print(f"🔗 [Input URL] Pasted hyperlink reference intercepted: {input_url}")183 184 async with httpx.AsyncClient(trust_env=False) as safety_client:185 if not await check_url_safe(input_url, safety_client):186 base_ui_context["error_msg"] = "The provided link target matches active security alert threat signatures and was isolated."187 return templates.TemplateResponse(request=request, name="dashboard.html", context=base_ui_context)188 189 raw_text, img_url = await extract_page_data(input_url, "product")190 if not raw_text:191 base_ui_context["error_msg"] = "Tarz could not parse or connect to the provided URL."192 return templates.TemplateResponse(request=request, name="dashboard.html", context=base_ui_context)193 194 try:195 url_prompt = f"""196 Identify the exact product name, find its listed retail numerical price, and sizing text from the block:197 198 {raw_text}199 200 Return ONLY raw JSON. No text wraps or explanations.201 Format structure: {{"product_name": "exact brand name", "price": 49.99, "size": "3.4 oz", "found": true}}202 """203 try:204 ai_res = call_free_llm(url_prompt, max_tokens=150)205 url_data = json.loads(ai_res)206 if url_data.get('found') and url_data.get('price') is not None:207 target_item = url_data['product_name']208 if not requested_size: requested_size = extract_universal_sizing(url_data.get('size'))209 210 original_site_deal = {211 "title": target_item,212 "url": clean_retailer_url(input_url), 213 "price": round(float(url_data['price']), 2), 214 "image": img_url,215 "source": urlparse(input_url).netloc.replace("www.", ""),216 "size": requested_size or "Standard"217 }218 except Exception as api_err:219 print(f"⚠️ [Link Solver Fallback] API unavailable. Parsing pasted link via heuristics...")220 local_report = parse_price_locally(raw_text, "product")221 if local_report["found"]:222 target_item = "Extracted Product Deal"223 original_site_deal = {224 "title": "Extracted Product Deal",225 "url": clean_retailer_url(input_url), 226 "price": local_report["price"], 227 "image": img_url,228 "source": urlparse(input_url).netloc.replace("www.", ""),229 "size": requested_size or "Standard"230 }231 else:232 base_ui_context["error_msg"] = "Programmatic local parser could not identify currency text variables."233 return templates.TemplateResponse(request=request, name="dashboard.html", context=base_ui_context)234 235 if not original_site_deal:236 base_ui_context["error_msg"] = "Pasted link parsed, but price extraction could not resolve valid currency markers."237 return templates.TemplateResponse(request=request, name="dashboard.html", context=base_ui_context)238 except Exception as e:239 print(f"❌ [Link Analytics Error] Failure processing link data parsing: {e}")240 base_ui_context["error_msg"] = "Error decoding reference link metadata structure."241 return templates.TemplateResponse(request=request, name="dashboard.html", context=base_ui_context)242 243 cached_deals = None if force_refresh else get_cached_deals(target_item, requested_size, search_mode)244 cached_coupons = None if force_refresh else get_cached_coupons(target_item)245 246 if cached_deals is not None:247 print(f"⚡ [Cache Hit] Found unexpired retail pricing matrix for '{target_item}' | Mode: '{search_mode}'. Skipping live queries.")248 deals = cached_deals249 coupons = cached_coupons or []250 else:251 print(f"❄️ [Cache Miss] Initiating parallel engine verification sequence for pricing & deals...")252 253 markdown_deals, product_offers = await asyncio.gather(254 fetch_deals_standalone(target_item),255 scan_prices(target_item, requested_size, search_mode)256 )257 258 if original_site_deal:259 if not any(d['url'] == original_site_deal['url'] for d in product_offers): 260 product_offers.append(original_site_deal)261 262 if product_offers:263 product_offers.sort(key=lambda x: x.get('price', float('inf')))264 market_max_price = max([x.get('price', 0) for x in product_offers if x.get('price', 0) > 0]) or 1.0265 processed_percentage_badges = []266 267 for item in product_offers:268 cur_p = item.get('price', 0)269 if not cur_p:270 continue271 272 item_source = item.get('source', 'Retailer').strip()273 item_title = item.get('title', 'Product Listing')274 275 old_p = item.get('original_price') or item.get('old_price') or item.get('regular_price')276 market_pct = int(round((1 - cur_p / market_max_price) * 100))277 278 if old_p and float(old_p) > float(cur_p) and float(old_p) > 0:279 calculated_pct = int(round((1 - float(cur_p) / float(old_p)) * 100))280 badge_string = f"{calculated_pct}% OFF"281 elif market_pct > 10:282 badge_string = f"{market_pct}% OFF"283 item['original_price'] = round(market_max_price, 2)284 else:285 badge_string = "BEST PRICE"286 item['original_price'] = round(cur_p, 2)287 288 sanitized_title = re.sub(r'[^a-zA-Z0-9]', '_', item_title).lower()[:50]289 redis_history_key = f"hist_low:prod:{sanitized_title}"290 291 historical_low = None292 if redis_client:293 try:294 cached_low = redis_client.get(redis_history_key)295 if cached_low:296 historical_low = float(cached_low)297 except Exception as redis_err:298 print(f"⚠️ [Redis Error] Failed reading baseline matrix: {redis_err}")299 300 if historical_low is not None:301 if cur_p < historical_low and redis_client:302 try: redis_client.set(redis_history_key, str(cur_p))303 except: pass304 else:305 if redis_client:306 try: redis_client.set(redis_history_key, str(cur_p))307 except: pass308 309 item['historical_low_event'] = False310 if historical_low:311 item['all_time_low_price'] = historical_low312 313 item['discount_percent'] = badge_string314 item['discount'] = badge_string315 item['percentage'] = badge_string316 item['badge'] = badge_string317 318 if not any(c['description'] == badge_string and c['source'].lower() == item_source.lower() for c in processed_percentage_badges):319 processed_percentage_badges.append({320 "description": badge_string,321 "source": item_source322 })323 324 current_lowest = product_offers[0]['price']325 base_price = get_base_price(target_item, requested_size, search_mode)326 if base_price is not None and base_price != current_lowest:327 delta = current_lowest - base_price328 product_offers[0]['price_delta'] = round(delta, 2)329 set_base_price(target_item, requested_size, search_mode, current_lowest)330 elif base_price == current_lowest:331 product_offers[0]['price_delta'] = 0.0332 else:333 set_base_price(target_item, requested_size, search_mode, current_lowest)334 product_offers[0]['price_delta'] = 0.0335 336 deals = product_offers337 set_cached_deals(target_item, requested_size, search_mode, deals)338 else:339 deals = []340 set_cached_deals(target_item, requested_size, search_mode, deals)341 342 coupons = []343 set_cached_coupons(target_item, coupons)344 345 size_header = f" ({requested_size})" if requested_size else ""346 347 validated_deals = []348 for d in deals:349 if "url" not in d and "link" in d:350 d["url"] = d["link"]351 elif "link" not in d and "url" in d:352 d["link"] = d["url"]353 354 unwrapped = unwrap_direct_product_url(d.get("url", ""))355 if unwrapped:356 cleaned_url = clean_retailer_url(unwrapped)357 d["url"] = cleaned_url358 d["link"] = cleaned_url359 try:360 d["source_domain"] = d.get("source") or urlparse(cleaned_url).netloc.replace("www.", "")361 except:362 d["source_domain"] = "retailer.link"363 validated_deals.append(d)364 365 base_ui_context.update({366 "show_results": True,367 "target_item": target_item,368 "size_header": size_header,369 "deals": validated_deals,370 "coupons": coupons,371 "is_ref_url": original_site_deal["url"] if original_site_deal else None372 })373 374 return templates.TemplateResponse(request=request, name="dashboard.html", context=base_ui_context)375 376if __name__ == "__main__":377 uvicorn.run("main:app", host="0.0.0.0", port=7860, reload=True)