import logging from datetime import datetime, timedelta, timezone from repositories.competitor_repository import competitor_repository from repositories.product_repository import product_repository from repositories.price_history_repository import price_history_repository from repositories.scrape_repository import scrape_repository from repositories.ngs_meta_repository import ngs_meta_repository from services import alert_service from services.url_utils import is_generic_homepage logger = logging.getLogger(__name__) # A price move smaller than this is normal noise, not alert-worthy — # matches the >5% threshold in CLAUDE.md's n8n price-change-alert spec. PRICE_CHANGE_ALERT_THRESHOLD_PERCENT = 5 def needs_refresh(competitor, confirmed_url=None): """ Decides whether a competitor's specific confirmed page needs a live re-scrape (REFRESH path) or can be served from PocketBase (FAST path). When confirmed_url is given (trip-intent matching, CLAUDE.md §8), freshness is judged against the cached PRODUCT for that exact URL, not the competitor's global change_detected/last_scraped flags. Those flags reflect whichever page was scraped MOST RECENTLY for this competitor — which may be a completely different trip than the one being asked about now. Gating on them directly made the fast path unreachable (or worse, risked serving the wrong cached trip) whenever a PM researched more than one product for the same competitor across different runs — confirmed directly against real scrape_runs history, where the same competitor's recorded hash changed on nearly every run despite re-scraping one specific URL twice in a row producing an identical hash both times. Falls back to the original competitor-wide check when no confirmed_url is available — the daily cron path (§15), which scrapes one canonical page per competitor and has no trip-intent context at all, is unaffected by this fix. Data flow (confirmed_url given): confirmed_url → product_repository.get_by_competitor_and_url() → no cached product for this exact page → True (refresh) → cached product's last_seen missing/stale (> CACHE_STALENESS_HOURS) → True (refresh) → otherwise → False (fast path is safe) Data flow (no confirmed_url — cron path, unchanged from before): competitor record → change_detected flag set? → True (refresh) → last_scraped missing or older than CACHE_STALENESS_HOURS? → True (refresh) → otherwise → False (fast path is safe) """ # CACHE_STALENESS_HOURS' canonical, adjustable home is core/scraper.py # per CLAUDE.md §7 ("in core/scraper.py — adjust here only") — imported # lazily here for the same reason every other core.* reference in this # file is lazy: this module must stay importable independent of Phase 2. from core.scraper import CACHE_STALENESS_HOURS if confirmed_url: product = product_repository.get_by_competitor_and_url(competitor['id'], confirmed_url) if not product: return True last_seen = product.get('last_seen') if not last_seen: return True try: seen_at = datetime.fromisoformat(last_seen.replace('Z', '+00:00')) except (ValueError, AttributeError): return True return (datetime.now(timezone.utc) - seen_at) > timedelta(hours=CACHE_STALENESS_HOURS) if competitor.get('change_detected'): return True last_scraped = competitor.get('last_scraped') if not last_scraped: return True try: scraped_at = datetime.fromisoformat(last_scraped.replace('Z', '+00:00')) except (ValueError, AttributeError): return True age = datetime.now(timezone.utc) - scraped_at return age > timedelta(hours=CACHE_STALENESS_HOURS) def analyse_from_cache(tenant_id, competitor_id, context): """ FAST path analysis — uses cached PocketBase data instead of a live scrape. The LLM's only job is to write a fresh comparative narrative against the client's *current* context and reassess relevancy — it does NOT re-extract trip/pricing fields from raw page content, since those already exist on the cached `products` record. The structured fields (tripName, price, duration, etc.) returned here come directly from that cached record, not from the LLM call. An earlier version returned the LLM's raw output as the entire result — since the prompt only ever asked for a narrative, with no schema, the model would invent its own ad-hoc JSON shape (bare `{"analysis": "..."}`, or with extra invented keys) that never matched what ResultsTable.vue actually renders (item.result.tripName, .majorityPrice/.gbpPrice, .duration, .comments) — the dashboard showed a "complete" job with an empty-looking results table because none of those fields were ever present. Now the LLM output is merged into (not used as) the returned dict. Data flow: competitor_id → PocketBase products (latest) → structured fields mapped from the cached record (trip_name → tripName, majority_price_usd → majorityPrice, etc.) → client trip context (destination/duration/style) → LLM prompt constrained to {comments, relevancy} only (via core.extractor) → cached structured fields + fresh comments/relevancy merged and returned context['confirmed_url'], when present, selects the cached product for that EXACT trip page (product_repository.get_by_competitor_and_url()) rather than whichever product happens to be most recently scraped for this competitor overall — the same competitor can have cached data for several different trips, and "most recent" is not necessarily the one this run is actually about. Falls back to "latest for competitor" when no confirmed_url is given (cron path) or no product matches it yet. """ confirmed_url = context.get('confirmed_url') latest = product_repository.get_by_competitor_and_url(competitor_id, confirmed_url) if confirmed_url else None if not latest: products = product_repository.get_latest_for_competitor(competitor_id, limit=1) if not products: raise ValueError('No cached product data available for fast path') latest = products[0] # core.extractor is a Phase 2 module — imported lazily so this # service is importable before the scraping engine refactor lands. from core.extractor import extract_with_openrouter # Ground the comparison in the tenant's own REAL extracted trip # (context['own_trip'], from embedding_service.scrape_own_trip()) # whenever it's available — only falling back to the PM's typed # destination/duration/style/product name when no own-trip data was # returned for this run (no website confirmed, or the scrape/ # extraction failed). Previously this dumped the raw `context` dict # (Python repr) into the prompt with no priority framing at all, so # own_trip's presence was easy for the model to under-weight against # the typed fields sitting right next to it in the same blob. own_trip = context.get('own_trip') company_name = context.get('company_name') or 'the client' if own_trip: own_trip_name = own_trip.get('tripName') or context.get('product_name') client_context_desc = ( f"{company_name}'s own {own_trip_name} — ACTUAL extracted data from " f"their own site, not a typed description: " f"{own_trip.get('majorityPrice', 'price not extracted')} USD, " f"{own_trip.get('duration') or context.get('duration')} days, " f"activities: {own_trip.get('activities') or 'not extracted'}" ) else: client_context_desc = ( f"{company_name}'s typed product intent (no own-site data confirmed " f"this run): product={context.get('product_name')}, " f"destination={context.get('destination')}, " f"duration={context.get('duration')} days, style={context.get('travel_style')}" ) prompt = ( 'You are a travel industry analyst. Given this already-extracted ' 'competitor trip data, assess it against ' f"{client_context_desc}.\n\nCompetitor data:\n{latest}\n\n" 'Return ONLY a valid JSON object:\n' '{\n' ' "comments": "short comparative analysis: key differences, strengths, weaknesses, pricing position",\n' ' "relevancy": relevancy score 1-5 as integer\n' '}' ) narrative = extract_with_openrouter(prompt, content='') result = { 'link': latest.get('url'), 'tripName': latest.get('trip_name'), 'tripCode': latest.get('trip_code'), 'serviceLevel': latest.get('service_level'), 'groupSize': latest.get('group_size'), 'duration': latest.get('duration_days'), 'meals': latest.get('meals'), 'startLocation': latest.get('start_location'), 'endLocation': latest.get('end_location'), 'departures': latest.get('departures'), 'seasonality': latest.get('seasonality'), 'startDays': latest.get('start_days'), 'targetAudience': latest.get('target_audience'), 'hotels': latest.get('hotels'), 'activities': latest.get('activities'), 'majorityPrice': latest.get('majority_price_usd'), 'priceLow': latest.get('price_low_usd'), 'priceHigh': latest.get('price_high_usd'), 'dateUsed': latest.get('date_used'), 'audPrice': latest.get('aud_price'), 'cadPrice': latest.get('cad_price'), 'eurPrice': latest.get('eur_price'), 'gbpPrice': latest.get('gbp_price'), 'comments': narrative.get('comments', ''), 'relevancy': narrative.get('relevancy', latest.get('relevancy')), } # NGS fields live on a separate products_ngs_meta record (see # _save_ngs_meta()) — only look it up when this run is actually NGS, # to avoid an extra PocketBase round trip on the common Standard-tab # fast path. if context.get('tab_type') == 'NGS': ngs_meta = ngs_meta_repository.get_by_product_id(latest['id']) if ngs_meta: result['exclusiveAccess'] = ngs_meta.get('exclusive_access') result['groupLeader'] = ngs_meta.get('group_leader') result['sustainability'] = ngs_meta.get('sustainability') return result def detect_change(competitor_id, url, new_hash): """ Compares new_hash against the last content_hash recorded for this exact competitor+url. On a diff: sets competitor.change_detected = true and change_detected_at = now. Replaces Firecrawl Monitor's webhook role (CLAUDE.md §4/§10) — the daily/on-demand scrape itself is now the only place a change is ever detected. Scoped by url, not just competitor_id — see scrape_repository.get_last_hash_for_url()'s docstring for why comparing against whatever page was scraped most recently for this competitor (regardless of which page that was) made change_detected effectively never settle back to False once a competitor had more than one trip page scraped over time. Data flow: competitor_id + url + new_hash → scrape_repository.get_last_hash_for_url() → no prior hash for this url: True (first scrape of this page — always process) → hash differs: competitor_repository.update() sets change flags → True → hash matches: no update → False """ last_hash = scrape_repository.get_last_hash_for_url(competitor_id, url) if last_hash is None: return True if last_hash != new_hash: competitor_repository.update(competitor_id, { 'change_detected': True, 'change_detected_at': datetime.now(timezone.utc).isoformat(), }) return True return False def run_competitor_analysis(tenant_id, competitor, context): """ Decides the fast vs refresh path for one competitor and executes it. Called per-competitor by jobs.run_research_job — failures here are caught by the caller so one competitor's failure never aborts the whole batch (CLAUDE.md §18 resilience rule). context['force_refresh'] (PM-initiated, via TripIntentForm.vue's collapsed "Advanced options" toggle) bypasses BOTH cache-shortcut points below — the initial needs_refresh() fast-path gate, and the post-scrape "content unchanged, reuse cached narrative" shortcut — so a forced run always ends up calling extract_with_openrouter() for real, never analyse_from_cache(). The content-hash bookkeeping itself (scrape_runs row, change_detected clearing) still runs normally even when forced, so future non-forced runs keep an accurate hash history. Data flow: competitor + context → target_url resolved and cleaned → needs_refresh() decision, scoped to target_url (or force_refresh) → FAST: analyse_from_cache() (seconds) → REFRESH: core.scraper.scrape() → compute_content_hash() → detect_change() against the last recorded hash FOR THIS URL → unchanged (and not force_refresh): change_detected cleared (confirmed stable since last look), skip LLM extraction, reuse cached narrative, bump last_scraped only (no wasted extraction cost) → changed (or force_refresh): core.extractor.extract_with_openrouter() → PocketBase products/price_history updated → last_scraped bumped, change_detected left True (stays visible until the next scrape confirms no further change) → result dict returned """ force_refresh = context.get('force_refresh', False) # Phase 2 module — imported lazily, same reasoning as the other # core.* imports further down in this function. from core.scraper import clean_url # context['confirmed_url'] is the page confirmed at UrlConfirmation.vue # (a Firecrawl /map match or a manual swap) — falls back to the # competitor's stored root domain only when nothing was confirmed for # this run (e.g. the daily cron path, which has no trip-intent context # at all). Regression fix: this used to always scrape # competitor['website'] unconditionally, silently discarding whatever # trip-specific URL was actually matched/confirmed for this run. # # Cleaned up front (matching what scrape() does internally) so every # url-scoped lookup below (needs_refresh, detect_change, # analyse_from_cache) compares against the same normalised form # _save_scrape_result() actually persists as products.url — a # trailing-slash or tracking-param difference would otherwise # silently defeat the url match and force an unnecessary refresh. # Cleaned regardless of source (confirmed_url or the website # fallback) so scrape_runs.scraped_url/products.url are always # consistent with each other either way. raw_confirmed_url = context.get('confirmed_url') target_url = clean_url(raw_confirmed_url or competitor['website']) # Only url-scope the fast-path/cache lookups below when this run # actually had an explicit confirmed_url (trip-intent matching) — # the cron/website-fallback path keeps using the competitor-wide # change_detected/last_scraped flags, unchanged from before this fix. confirmed_url = target_url if raw_confirmed_url else None if not needs_refresh(competitor, confirmed_url=confirmed_url) and not force_refresh: logger.info(f'{competitor["name"]} — fast path (cache fresh)') fast_result = analyse_from_cache( tenant_id, competitor['id'], {**context, 'confirmed_url': confirmed_url} ) return { 'path': 'fast', 'result': fast_result, 'is_generic_page': is_generic_homepage(fast_result.get('link')), } logger.info(f'{competitor["name"]} — refresh path (force_refresh)' if force_refresh else f'{competitor["name"]} — refresh path (stale or changed)') # Phase 2 modules — imported lazily, same reasoning as above. from core.scraper import scrape, compute_content_hash from core.extractor import extract_with_openrouter from industries.adventure_travel.prompt import build_prompt # Computed once here and reused by both possible refresh-path returns # below — a meta-fact about which page was actually scraped this run, # not an extracted trip field, so it's a top-level sibling of `result`/ # `path`/`product` in the returned dict rather than nested inside `result`. is_generic = is_generic_homepage(target_url) page_data = scrape(target_url, country=competitor.get('primary_market')) if not page_data.get('success'): raise ValueError(page_data.get('text', 'Scrape failed')) content_hash = compute_content_hash(page_data['text']) changed = detect_change(competitor['id'], target_url, content_hash) # Every refresh scrape gets its own scrape_runs record purely to # persist content_hash (and the url it was hashed from) for the next # comparison — the outer batch job's scrape_runs row (created before # run_research_job) tracks overall job status, not per-competitor # hash history. scrape_repository.create({ 'tenant_id': tenant_id, 'competitor_id': competitor['id'], 'triggered_by': 'cron', 'status': 'complete', 'content_hash': content_hash, 'content_length': len(page_data['text']), 'hash_algorithm': 'sha256', 'scraped_url': target_url, 'competitors_total': 1, 'competitors_done': 1, 'results': {}, 'completed_at': datetime.now(timezone.utc).isoformat(), }) if not changed and not force_refresh: # Confirmed no further change since the last scrape — this is # the moment change_detected gets cleared. If it was never true # to begin with, this is a harmless no-op. logger.info(f'{competitor["name"]} — content unchanged, skipping LLM extraction') competitor_repository.update(competitor['id'], { 'change_detected': False, 'last_scraped': datetime.now(timezone.utc).isoformat(), }) return { 'path': 'refresh', 'result': analyse_from_cache(tenant_id, competitor['id'], {**context, 'confirmed_url': confirmed_url}), 'unchanged': True, 'is_generic_page': is_generic, } if not changed: # force_refresh bypasses the "reuse cache" shortcut, but the hash # comparison itself is still accurate — content really is # unchanged — so change_detected is still cleared here. Only the # decision to skip real extraction is bypassed; falls through to # a real extraction below instead of returning early. logger.info(f'{competitor["name"]} — unchanged but force_refresh requested, running real extraction anyway') competitor_repository.update(competitor['id'], {'change_detected': False}) prompt = build_prompt( competitor['name'], page_data, context.get('product_name'), context.get('destination'), context.get('duration'), context.get('travel_style'), company_name=context.get('company_name'), is_ngs=context.get('tab_type') == 'NGS', own_trip=context.get('own_trip'), ) extracted = extract_with_openrouter(prompt, content=page_data['text']) product = _save_scrape_result(tenant_id, competitor, extracted, page_data) # change_detected is deliberately left as detect_change() set it # (True) rather than reset here — detection and extraction now # happen in the same scrape, so there's no separate "handled" event # to reset it on. It stays visible (CompetitorsPage's "⚠ Changed" # badge, Apps Script sidebar status, tomorrow's n8n cron filter) # until the *next* scrape confirms no further change and clears it # in the branch above — at worst one extra hash-only confirmation # scrape per genuine change, no LLM cost since it lands on the # unchanged branch. competitor_repository.update(competitor['id'], { 'last_scraped': datetime.now(timezone.utc).isoformat(), }) return {'path': 'refresh', 'result': extracted, 'product': product, 'is_generic_page': is_generic} def _save_scrape_result(tenant_id, competitor, extracted, page_data): """ Persists one competitor's freshly-extracted trip data, deciding between "this is a new product" and "this is an update to a product we already track" by matching on (competitor_id, trip_name) — the same dedup key CLAUDE.md's is_new detection and price-history diffing rely on (§15/§19 Phase 5). Data flow: extracted fields → product_repository.get_by_competitor_and_name() → no match: create product with is_new=True → alert_service.create_alert('new_product') → match found: compare majority_price_usd → changed: price_history record written, product updated, alert_service.create_alert('price_change') if the move exceeds PRICE_CHANGE_ALERT_THRESHOLD_PERCENT → unchanged: product's last_seen/fields refreshed, no alert → product record returned either way """ now = datetime.now(timezone.utc).isoformat() trip_name = extracted.get('tripName') new_price = extracted.get('majorityPrice') common_fields = { 'trip_name': trip_name, 'url': page_data['url'], 'destination': extracted.get('destination'), 'duration_days': extracted.get('duration'), 'majority_price_usd': new_price, 'price_low_usd': extracted.get('priceLow'), 'price_high_usd': extracted.get('priceHigh'), 'aud_price': extracted.get('audPrice'), 'cad_price': extracted.get('cadPrice'), 'eur_price': extracted.get('eurPrice'), 'gbp_price': extracted.get('gbpPrice'), 'date_used': extracted.get('dateUsed'), 'comments': extracted.get('comments'), 'relevancy': extracted.get('relevancy'), # These all already have PocketBase columns and were previously # extracted by the LLM (industries/adventure_travel/prompt.py's # build_prompt() JSON schema) but silently dropped on write — the # dashboard's transposed results grid (ResultsTable.vue / # resultFields.js) needs every one of these to mirror the legacy # Sheet's row layout. 'trip_code': extracted.get('tripCode'), 'service_level': extracted.get('serviceLevel'), 'group_size': extracted.get('groupSize'), 'meals': extracted.get('meals'), 'start_location': extracted.get('startLocation'), 'end_location': extracted.get('endLocation'), 'departures': extracted.get('departures'), 'seasonality': extracted.get('seasonality'), 'start_days': extracted.get('startDays'), 'target_audience': extracted.get('targetAudience'), 'hotels': extracted.get('hotels'), 'activities': extracted.get('activities'), 'last_seen': now, 'is_active': True, } existing = product_repository.get_by_competitor_and_name(competitor['id'], trip_name) if not existing: product = product_repository.create({ 'tenant_id': tenant_id, 'competitor_id': competitor['id'], 'is_new': True, 'first_seen': now, **common_fields, }) alert_service.create_alert( tenant_id, competitor['id'], 'new_product', f"{competitor['name']} added a new product: {trip_name or 'Unnamed trip'}", product_id=product['id'], ) _save_ngs_meta(product['id'], extracted) return product product = product_repository.update(existing['id'], {**common_fields, 'is_new': False}) old_price = existing.get('majority_price_usd') if old_price is not None and new_price is not None and old_price != new_price: change_amount = new_price - old_price change_percent = round((change_amount / old_price) * 100, 1) if old_price else None price_history_repository.create({ 'tenant_id': tenant_id, 'product_id': product['id'], 'majority_price_usd': new_price, 'price_low_usd': extracted.get('priceLow'), 'price_high_usd': extracted.get('priceHigh'), 'aud_price': extracted.get('audPrice'), 'cad_price': extracted.get('cadPrice'), 'eur_price': extracted.get('eurPrice'), 'gbp_price': extracted.get('gbpPrice'), 'date_used': extracted.get('dateUsed'), 'change_field': 'majorityPrice', 'change_amount': change_amount, 'change_percent': change_percent, }) if change_percent is not None and abs(change_percent) > PRICE_CHANGE_ALERT_THRESHOLD_PERCENT: direction = 'dropped' if change_percent < 0 else 'raised' alert_service.create_alert( tenant_id, competitor['id'], 'price_change', f"{competitor['name']} {direction} {trip_name or 'a product'} {change_percent}%", product_id=product['id'], change_amount=change_amount, change_percent=change_percent, ) _save_ngs_meta(product['id'], extracted) return product def _save_ngs_meta(product_id, extracted): """ Upserts NGS-only extraction fields (exclusiveAccess/groupLeader/ sustainability) into products_ngs_meta, keyed on product_id. Only fires when build_prompt() actually asked for these fields (is_ngs=True in run_competitor_analysis()) — inferred here from their presence in `extracted` rather than threading a tab_type/context param through _save_scrape_result()'s already-long signature, since a Standard-tab extraction never has these keys at all. Silently no-ops for a Standard-tab run. """ if not any(extracted.get(k) is not None for k in ('exclusiveAccess', 'groupLeader', 'sustainability')): return fields = { 'product_id': product_id, 'exclusive_access': extracted.get('exclusiveAccess'), 'group_leader': extracted.get('groupLeader'), 'sustainability': extracted.get('sustainability'), } existing = ngs_meta_repository.get_by_product_id(product_id) if existing: ngs_meta_repository.update(existing['id'], fields) else: ngs_meta_repository.create(fields)