207 lines
8.8 KiB
Python
207 lines
8.8 KiB
Python
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
from repositories.product_repository import product_repository
|
|
from repositories.competitor_repository import competitor_repository
|
|
from repositories.battlecard_repository import battlecard_repository
|
|
from repositories.price_history_repository import price_history_repository
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# How many of a competitor's most-recently-scraped active products feed
|
|
# a battlecard. Previously 10 — skewed the summary toward whatever
|
|
# happened to be scraped most recently (e.g. this week's watchlist
|
|
# picks) instead of a reasonably complete view of the tracked catalogue.
|
|
# Adjustable here only, same convention as core/scraper.py's
|
|
# CACHE_STALENESS_HOURS.
|
|
BATTLECARD_PRODUCT_LIMIT = 50
|
|
|
|
# How far back price_history is considered "recent" for the battlecard's
|
|
# recent-changes line.
|
|
RECENT_CHANGES_WINDOW_DAYS = 30
|
|
|
|
BATTLECARD_PROMPT = '''You are a competitive intelligence analyst for an adventure travel operator.
|
|
Based on this competitor data, write a concise battlecard.
|
|
|
|
Competitor: {name}
|
|
Top trips: {top_trips}
|
|
Number of tracked trips: {trip_count}
|
|
Price range: {price_low} - {price_high} USD
|
|
Avg duration: {avg_duration} days
|
|
Service levels offered: {service_levels}
|
|
Typical target audience: {target_audiences}
|
|
Common activities/themes: {activities_summary}
|
|
Recent price changes (last 30 days): {recent_changes}
|
|
|
|
Write — ground every point in the specific data above (real prices,
|
|
service levels, activities, audience, recent price moves) rather than
|
|
generic statements that could apply to any competitor:
|
|
1. One line positioning summary (max 20 words)
|
|
2. Three strengths (bullet points, max 14 words each)
|
|
3. Three weaknesses (bullet points, max 14 words each)
|
|
4. One pricing observation (max 30 words) — reference the actual price
|
|
range and/or a real recent price change when one is available
|
|
|
|
Return ONLY valid JSON:
|
|
{{
|
|
"positioning": "...",
|
|
"strengths": ["...", "...", "..."],
|
|
"weaknesses": ["...", "...", "..."],
|
|
"pricing_observation": "..."
|
|
}}'''
|
|
|
|
|
|
def _top_values(values, limit=3):
|
|
"""
|
|
Dedupes a list of possibly-null strings, preserving first-seen
|
|
order, and joins the top `limit` into one comma-separated string.
|
|
Used to summarise service_level/target_audience/activities across a
|
|
competitor's tracked products without just picking the single most
|
|
recent one.
|
|
"""
|
|
seen = []
|
|
for value in values:
|
|
if value and value not in seen:
|
|
seen.append(value)
|
|
return ', '.join(seen[:limit])
|
|
|
|
|
|
def _summarise_products(products):
|
|
"""
|
|
Reduces a competitor's tracked products into the plain values
|
|
BATTLECARD_PROMPT needs. Previously only derived 3 numbers
|
|
(price range, avg duration) plus a joined trip-name list — ignored
|
|
service_level/target_audience/activities even though every products
|
|
row already has them (CLAUDE.md §12).
|
|
"""
|
|
top_trips = ', '.join(p.get('trip_name', '') for p in products[:8] if p.get('trip_name'))
|
|
prices = [p['majority_price_usd'] for p in products if p.get('majority_price_usd')]
|
|
durations = [p['duration_days'] for p in products if p.get('duration_days')]
|
|
return {
|
|
'top_trips': top_trips or 'No products on record',
|
|
'trip_count': len(products),
|
|
'price_low': min(prices) if prices else 'unknown',
|
|
'price_high': max(prices) if prices else 'unknown',
|
|
'avg_duration': round(sum(durations) / len(durations)) if durations else 'unknown',
|
|
'service_levels': _top_values([p.get('service_level') for p in products]) or 'unknown',
|
|
'target_audiences': _top_values([p.get('target_audience') for p in products]) or 'unknown',
|
|
'activities_summary': _top_values([p.get('activities') for p in products], limit=5) or 'unknown',
|
|
}
|
|
|
|
|
|
def _summarise_recent_price_changes(products, days=RECENT_CHANGES_WINDOW_DAYS):
|
|
"""
|
|
Replaces the old hardcoded 'See price_history for this competitor.'
|
|
placeholder with a real summary of that competitor's actual recent
|
|
price moves — the model was previously being told to reference data
|
|
it was never actually given.
|
|
|
|
Every price_history row already represents a genuine change
|
|
(analysis_service._save_scrape_result() only writes one when
|
|
old_price != new_price), so no extra magnitude filter is needed here
|
|
— just a recency window. price_history has no competitor_id column
|
|
(only product_id/tenant_id per CLAUDE.md §12), so this reuses the
|
|
existing price_history_repository.list_for_product() per already-
|
|
fetched product rather than adding a new repository method.
|
|
|
|
Data flow:
|
|
products → per product: price_history_repository.list_for_product(limit=3) →
|
|
entries within `days` with a real change_percent →
|
|
sorted by |change_percent| descending (most notable first) →
|
|
top 5 joined into one line, or a "no changes" fallback if none qualify
|
|
"""
|
|
cutoff = datetime.now(timezone.utc) - timedelta(days=days)
|
|
changes = []
|
|
for product in products:
|
|
entries = price_history_repository.list_for_product(product['id'], limit=3)
|
|
for entry in entries:
|
|
scraped_at = entry.get('scraped_at')
|
|
change_percent = entry.get('change_percent')
|
|
if not scraped_at or change_percent is None:
|
|
continue
|
|
try:
|
|
scraped_dt = datetime.fromisoformat(scraped_at.replace('Z', '+00:00'))
|
|
except (ValueError, AttributeError):
|
|
continue
|
|
if scraped_dt < cutoff:
|
|
continue
|
|
direction = 'dropped' if change_percent < 0 else 'rose'
|
|
price = entry.get('majority_price_usd')
|
|
price_part = f' to ${price}' if price is not None else ''
|
|
changes.append((
|
|
abs(change_percent),
|
|
f"{product.get('trip_name') or 'a trip'} {direction} {abs(change_percent)}%{price_part}",
|
|
))
|
|
|
|
if not changes:
|
|
return f'No significant price changes in the last {days} days.'
|
|
|
|
changes.sort(key=lambda c: c[0], reverse=True)
|
|
return '; '.join(text for _, text in changes[:5])
|
|
|
|
|
|
def generate_battlecard(competitor_id, tenant_id):
|
|
"""
|
|
Pulls a competitor's tracked products (and their real recent price
|
|
history) from PocketBase, passes a real structured summary to the
|
|
OpenRouter extraction model, and saves the generated battlecard
|
|
text/JSON back to PocketBase.
|
|
|
|
Data flow:
|
|
competitor_id → PocketBase products (latest BATTLECARD_PRODUCT_LIMIT) →
|
|
_summarise_products() + _summarise_recent_price_changes() →
|
|
structured prompt → OpenRouter (via core.extractor) →
|
|
battlecard text + JSON → PocketBase battlecards table →
|
|
battlecard dict returned for immediate use
|
|
"""
|
|
# core.extractor is a Phase 2 module — imported lazily so this
|
|
# service is importable (and unit-testable against mocks) before
|
|
# the scraping engine refactor lands.
|
|
from core.extractor import extract_with_openrouter
|
|
|
|
competitor = competitor_repository.get_by_id(competitor_id)
|
|
products = product_repository.get_latest_for_competitor(competitor_id, limit=BATTLECARD_PRODUCT_LIMIT)
|
|
summary = _summarise_products(products)
|
|
recent_changes = _summarise_recent_price_changes(products)
|
|
|
|
prompt = BATTLECARD_PROMPT.format(
|
|
name=competitor.get('name', 'Unknown') if competitor else 'Unknown',
|
|
top_trips=summary['top_trips'],
|
|
trip_count=summary['trip_count'],
|
|
price_low=summary['price_low'],
|
|
price_high=summary['price_high'],
|
|
avg_duration=summary['avg_duration'],
|
|
service_levels=summary['service_levels'],
|
|
target_audiences=summary['target_audiences'],
|
|
activities_summary=summary['activities_summary'],
|
|
recent_changes=recent_changes,
|
|
)
|
|
|
|
try:
|
|
extracted = extract_with_openrouter(prompt, content='')
|
|
except Exception as e:
|
|
# A failed battlecard regeneration must not break the scrape
|
|
# pipeline that triggered it — log and leave the prior
|
|
# battlecard (if any) untouched.
|
|
logger.error(f'Battlecard generation failed for {competitor_id}: {str(e)}')
|
|
return None
|
|
|
|
battlecard = battlecard_repository.upsert(competitor_id, {
|
|
'tenant_id': tenant_id,
|
|
'content': _render_text(extracted),
|
|
'content_json': extracted,
|
|
})
|
|
return battlecard
|
|
|
|
|
|
def _render_text(extracted):
|
|
"""Flattens the structured battlecard JSON into a plain-text summary for the `content` field."""
|
|
if not isinstance(extracted, dict):
|
|
return str(extracted)
|
|
strengths = '\n'.join(f'+ {s}' for s in extracted.get('strengths', []))
|
|
weaknesses = '\n'.join(f'- {w}' for w in extracted.get('weaknesses', []))
|
|
return (
|
|
f"{extracted.get('positioning', '')}\n\n"
|
|
f"Strengths:\n{strengths}\n\nWeaknesses:\n{weaknesses}\n\n"
|
|
f"{extracted.get('pricing_observation', '')}"
|
|
)
|