105 lines
4.1 KiB
Python
105 lines
4.1 KiB
Python
import logging
|
|
from repositories.product_repository import product_repository
|
|
from repositories.competitor_repository import competitor_repository
|
|
from repositories.battlecard_repository import battlecard_repository
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
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}
|
|
Price range: {price_low} - {price_high} USD
|
|
Avg duration: {avg_duration} days
|
|
Positioning: {positioning_summary}
|
|
Recent changes: {recent_changes}
|
|
|
|
Write:
|
|
1. One line positioning summary (max 15 words)
|
|
2. Three strengths (bullet points, max 8 words each)
|
|
3. Three weaknesses (bullet points, max 8 words each)
|
|
4. One pricing observation (max 20 words)
|
|
|
|
Return ONLY valid JSON:
|
|
{{
|
|
"positioning": "...",
|
|
"strengths": ["...", "...", "..."],
|
|
"weaknesses": ["...", "...", "..."],
|
|
"pricing_observation": "..."
|
|
}}'''
|
|
|
|
|
|
def _summarise_products(products):
|
|
"""Reduces a list of product records into the plain values BATTLECARD_PROMPT needs."""
|
|
top_trips = ', '.join(p.get('trip_name', '') for p in products[:5] 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',
|
|
'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',
|
|
}
|
|
|
|
|
|
def generate_battlecard(competitor_id, tenant_id):
|
|
"""
|
|
Pulls latest product data for a competitor from PocketBase, passes
|
|
structured data to the OpenRouter extraction model, and saves the
|
|
generated battlecard text/JSON back to PocketBase.
|
|
|
|
Data flow:
|
|
competitor_id → PocketBase products (latest 10) →
|
|
structured prompt → OpenRouter (claude-haiku 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=10)
|
|
summary = _summarise_products(products)
|
|
|
|
prompt = BATTLECARD_PROMPT.format(
|
|
name=competitor.get('name', 'Unknown') if competitor else 'Unknown',
|
|
top_trips=summary['top_trips'],
|
|
price_low=summary['price_low'],
|
|
price_high=summary['price_high'],
|
|
avg_duration=summary['avg_duration'],
|
|
positioning_summary='Derived from current product catalogue.',
|
|
recent_changes='See price_history for this competitor.',
|
|
)
|
|
|
|
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', '')}"
|
|
)
|