jtricerolph
72f607132c
Live-tested again: 800 tokens still wasn't enough for Sonnet's bullet density (6 dense bullets, cut off mid-sentence again). Capped bullet count to 4-5, added a hard 35-word single-sentence limit per bullet, told it not to add a closing summary, and raised the cap to 1200 for genuine headroom. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
634 lines
23 KiB
Python
634 lines
23 KiB
Python
"""
|
|
AI Daily Insights Generation Job
|
|
|
|
Gathers Pickup-V2 forecast data, booking pace, competitor rates, rate parity,
|
|
UK bank holidays, and the previous insight, then sends a compact prompt to
|
|
Anthropic's Haiku model to generate a daily briefing for hotel revenue staff.
|
|
|
|
Schedule: Daily at 7:15 AM (after all forecasts and accuracy calc complete)
|
|
"""
|
|
import json
|
|
import logging
|
|
from datetime import date, timedelta, datetime, timezone
|
|
from typing import Dict, List, Any, Optional
|
|
|
|
import holidays
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from database import AsyncSessionLocal
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_MODEL = "claude-haiku-4-5-20251001"
|
|
DEFAULT_DAILY_TOKEN_BUDGET = 12000
|
|
MAX_OUTPUT_TOKENS = 1200
|
|
FORECAST_HORIZON_DAYS = 30
|
|
RATE_MOVEMENT_THRESHOLD = 3.0 # GBP — minimum delta to surface as a market move
|
|
|
|
|
|
async def get_config(db: AsyncSession) -> Dict[str, str]:
|
|
"""Get AI insights config from system_config."""
|
|
result = await db.execute(
|
|
text("""
|
|
SELECT config_key, config_value, COALESCE(is_encrypted, false) as is_encrypted
|
|
FROM system_config
|
|
WHERE config_key LIKE 'ai_insights_%'
|
|
""")
|
|
)
|
|
config = {}
|
|
for row in result.fetchall():
|
|
value = row.config_value
|
|
if row.is_encrypted and value:
|
|
import base64
|
|
try:
|
|
value = base64.b64decode(value.encode()).decode()
|
|
except Exception:
|
|
pass
|
|
config[row.config_key] = value
|
|
return config
|
|
|
|
|
|
async def check_daily_budget(db: AsyncSession, budget: int) -> tuple[bool, int]:
|
|
"""Check if we're within the daily token budget. Returns (within_budget, tokens_used_today)."""
|
|
result = await db.execute(
|
|
text("""
|
|
SELECT COALESCE(SUM(input_tokens + output_tokens), 0) as total
|
|
FROM ai_insights
|
|
WHERE generated_at >= CURRENT_DATE
|
|
""")
|
|
)
|
|
row = result.fetchone()
|
|
used = int(row.total) if row else 0
|
|
return used < budget, used
|
|
|
|
|
|
async def get_previous_insight(db: AsyncSession) -> Optional[Dict[str, Any]]:
|
|
"""Fetch the most recent insight so the new one can compare against it."""
|
|
result = await db.execute(
|
|
text("""
|
|
SELECT headline, content, generated_at, data_snapshot
|
|
FROM ai_insights
|
|
ORDER BY generated_at DESC
|
|
LIMIT 1
|
|
""")
|
|
)
|
|
row = result.fetchone()
|
|
if not row:
|
|
return None
|
|
return {
|
|
"headline": row.headline,
|
|
"content": row.content,
|
|
"generated_at": row.generated_at,
|
|
"data_snapshot": row.data_snapshot or {},
|
|
}
|
|
|
|
|
|
async def gather_occupancy_data(db: AsyncSession, days: int = FORECAST_HORIZON_DAYS) -> List[Dict]:
|
|
"""Gather Pickup-V2 occupancy forecast data for the next N days."""
|
|
from services.forecasting.pickup_v2_model import run_pickup_v2_forecast
|
|
|
|
today = date.today()
|
|
end = today + timedelta(days=days - 1)
|
|
|
|
try:
|
|
forecasts = await run_pickup_v2_forecast(
|
|
db, 'hotel_occupancy_pct', today, end, include_details=False
|
|
)
|
|
return forecasts
|
|
except Exception as e:
|
|
logger.warning(f"Failed to gather occupancy data: {e}")
|
|
return []
|
|
|
|
|
|
async def gather_revenue_data(db: AsyncSession, days: int = FORECAST_HORIZON_DAYS) -> List[Dict]:
|
|
"""Gather Pickup-V2 revenue forecast data for the next N days."""
|
|
from services.forecasting.pickup_v2_model import run_pickup_v2_forecast
|
|
|
|
today = date.today()
|
|
end = today + timedelta(days=days - 1)
|
|
|
|
try:
|
|
forecasts = await run_pickup_v2_forecast(
|
|
db, 'net_accom', today, end, include_details=False
|
|
)
|
|
return forecasts
|
|
except Exception as e:
|
|
logger.warning(f"Failed to gather revenue data: {e}")
|
|
return []
|
|
|
|
|
|
async def gather_budget_data(db: AsyncSession, days: int = FORECAST_HORIZON_DAYS) -> Dict[str, float]:
|
|
"""Gather budget values for forecast comparison."""
|
|
today = date.today()
|
|
end = today + timedelta(days=days - 1)
|
|
|
|
result = await db.execute(
|
|
text("""
|
|
SELECT date, budget_type, budget_value
|
|
FROM daily_budgets
|
|
WHERE date BETWEEN :start AND :end
|
|
AND budget_type IN ('net_accom', 'occupancy')
|
|
"""),
|
|
{"start": today, "end": end}
|
|
)
|
|
|
|
budgets = {}
|
|
for row in result.fetchall():
|
|
key = f"{row.date}_{row.budget_type}"
|
|
budgets[key] = float(row.budget_value) if row.budget_value else None
|
|
return budgets
|
|
|
|
|
|
async def gather_competitor_data(db: AsyncSession, days: int = FORECAST_HORIZON_DAYS) -> Dict[str, Any]:
|
|
"""Gather competitor rate data from Booking.com scraper."""
|
|
today = date.today()
|
|
end = today + timedelta(days=days - 1)
|
|
|
|
# Own hotel rate on Booking.com
|
|
own_result = await db.execute(
|
|
text("""
|
|
SELECT DISTINCT ON (r.rate_date)
|
|
r.rate_date,
|
|
r.rate_gross as booking_rate,
|
|
r.availability_status
|
|
FROM booking_com_rates r
|
|
JOIN booking_com_hotels h ON r.hotel_id = h.id
|
|
WHERE h.tier = 'own'
|
|
AND r.rate_date BETWEEN :start AND :end
|
|
ORDER BY r.rate_date, r.scraped_at DESC
|
|
"""),
|
|
{"start": today, "end": end}
|
|
)
|
|
own_rates = {str(row.rate_date): {
|
|
'rate': float(row.booking_rate) if row.booking_rate else None,
|
|
'status': row.availability_status
|
|
} for row in own_result.fetchall()}
|
|
|
|
# Cheapest competitor rate per date
|
|
comp_result = await db.execute(
|
|
text("""
|
|
SELECT DISTINCT ON (r.rate_date)
|
|
r.rate_date,
|
|
r.rate_gross as comp_rate,
|
|
h.name as comp_name
|
|
FROM booking_com_rates r
|
|
JOIN booking_com_hotels h ON r.hotel_id = h.id
|
|
WHERE h.tier IN ('competitor', 'market')
|
|
AND r.rate_date BETWEEN :start AND :end
|
|
AND r.availability_status = 'available'
|
|
AND r.rate_gross IS NOT NULL
|
|
ORDER BY r.rate_date, r.rate_gross ASC
|
|
"""),
|
|
{"start": today, "end": end}
|
|
)
|
|
comp_rates = {str(row.rate_date): {
|
|
'rate': float(row.comp_rate),
|
|
'name': row.comp_name
|
|
} for row in comp_result.fetchall()}
|
|
|
|
# Own rack rate from Newbook
|
|
rack_result = await db.execute(
|
|
text("""
|
|
SELECT DISTINCT ON (rate_date)
|
|
rate_date,
|
|
rate_gross
|
|
FROM newbook_current_rates
|
|
WHERE rate_date BETWEEN :start AND :end
|
|
ORDER BY rate_date, valid_from DESC
|
|
"""),
|
|
{"start": today, "end": end}
|
|
)
|
|
rack_rates = {str(row.rate_date): float(row.rate_gross) if row.rate_gross else None
|
|
for row in rack_result.fetchall()}
|
|
|
|
return {
|
|
'own_booking': own_rates,
|
|
'competitors': comp_rates,
|
|
'rack': rack_rates
|
|
}
|
|
|
|
|
|
def get_uk_bank_holidays(start: date, end: date) -> Dict[str, str]:
|
|
"""UK (England) bank holidays within the given range, keyed by YYYY-MM-DD."""
|
|
years = list(range(start.year, end.year + 1))
|
|
try:
|
|
uk_holidays = holidays.UK(subdiv='England', years=years)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load UK bank holidays: {e}")
|
|
return {}
|
|
|
|
return {
|
|
str(d): name for d, name in uk_holidays.items()
|
|
if start <= d <= end
|
|
}
|
|
|
|
|
|
def _fmt_date(d: str) -> str:
|
|
"""Convert YYYY-MM-DD to DD/MM/YYYY for UK display."""
|
|
try:
|
|
parts = str(d).split('-')
|
|
if len(parts) == 3:
|
|
return f"{parts[2]}/{parts[1]}/{parts[0]}"
|
|
except Exception:
|
|
pass
|
|
return str(d)
|
|
|
|
|
|
def build_market_movement_section(
|
|
current: Dict[str, Any],
|
|
previous: Optional[Dict[str, Any]],
|
|
threshold: float = RATE_MOVEMENT_THRESHOLD
|
|
) -> List[str]:
|
|
"""
|
|
Diff competitor/own rates against the previous insight's snapshot.
|
|
Only same-channel deltas are compared (B.com vs B.com, Rack vs Rack) so
|
|
a direct rack rate is never held up against a competitor's B.com rate.
|
|
"""
|
|
lines = ["## Market Movement Since Last Insight"]
|
|
|
|
if not previous:
|
|
lines.append("No previous insight to compare against (first run).")
|
|
lines.append("")
|
|
return lines
|
|
|
|
prev_competitors = previous.get('competitors', {}) or {}
|
|
prev_own_booking = previous.get('own_booking', {}) or {}
|
|
prev_rack = previous.get('rack', {}) or {}
|
|
|
|
cur_competitors = current.get('competitors', {}) or {}
|
|
cur_own_booking = current.get('own_booking', {}) or {}
|
|
cur_rack = current.get('rack', {}) or {}
|
|
|
|
moves = []
|
|
|
|
for d, cur in cur_competitors.items():
|
|
prev = prev_competitors.get(d)
|
|
if not prev or not cur.get('rate') or not prev.get('rate'):
|
|
continue
|
|
delta = cur['rate'] - prev['rate']
|
|
if abs(delta) >= threshold:
|
|
moves.append(
|
|
f"{_fmt_date(d)} | Cheapest competitor ({cur.get('name', '?')}, B.com): "
|
|
f"£{prev['rate']:.0f} -> £{cur['rate']:.0f} ({delta:+.0f})"
|
|
)
|
|
|
|
for d, cur in cur_own_booking.items():
|
|
prev = prev_own_booking.get(d)
|
|
if not prev or not cur.get('rate') or not prev.get('rate'):
|
|
continue
|
|
delta = cur['rate'] - prev['rate']
|
|
if abs(delta) >= threshold:
|
|
moves.append(
|
|
f"{_fmt_date(d)} | Own rate (B.com): "
|
|
f"£{prev['rate']:.0f} -> £{cur['rate']:.0f} ({delta:+.0f})"
|
|
)
|
|
|
|
for d, cur_rate in cur_rack.items():
|
|
prev_rate = prev_rack.get(d)
|
|
if not prev_rate or not cur_rate:
|
|
continue
|
|
delta = cur_rate - prev_rate
|
|
if abs(delta) >= threshold:
|
|
moves.append(
|
|
f"{_fmt_date(d)} | Own rate (Rack/Newbook): "
|
|
f"£{prev_rate:.0f} -> £{cur_rate:.0f} ({delta:+.0f})"
|
|
)
|
|
|
|
if moves:
|
|
moves.sort()
|
|
lines.extend(moves)
|
|
else:
|
|
lines.append(f"No competitor or own-rate movements of £{threshold:.0f}+ since last insight.")
|
|
|
|
lines.append("")
|
|
return lines
|
|
|
|
|
|
def build_prompt(
|
|
occupancy: List[Dict],
|
|
revenue: List[Dict],
|
|
budgets: Dict[str, float],
|
|
competitor: Dict[str, Any],
|
|
bank_holidays: Dict[str, str],
|
|
previous_insight: Optional[Dict[str, Any]] = None,
|
|
) -> tuple[str, str]:
|
|
"""Build system and user prompts from gathered data. Returns (system_msg, user_msg)."""
|
|
|
|
system_msg = (
|
|
"You are an AI assistant for a hotel revenue manager in the UK. Analyze the data below and provide "
|
|
"a daily briefing. Use UK date format (DD/MM/YYYY) and GBP (£) for all monetary values. "
|
|
"Be specific with numbers and dates. Keep it actionable — no fluff or generic advice.\n\n"
|
|
"Output format: first line must be `HEADLINE: <one sentence, the single most important takeaway>`, "
|
|
"then a blank line, then EXACTLY 4-5 bullet points covering the most important of: occupancy/pace, "
|
|
"pricing opportunities, notable competitor rate movements (only if the Market Movement section has "
|
|
"any), UK bank holidays in the window if they affect pace, and anything unusual requiring attention. "
|
|
"Each bullet must be a SINGLE sentence, hard limit 35 words — pick the 1-2 most important numbers per "
|
|
"bullet rather than listing every date. Do not add a closing summary or wrap-up paragraph after the "
|
|
"bullets — the last bullet ends the response.\n\n"
|
|
"A 'Previous Insight' section may be included below — compare against it explicitly: call out what's "
|
|
"changed, what's resolved, and what's still an open issue. Don't just repeat it verbatim.\n\n"
|
|
"Never compare a direct/rack rate to a competitor's Booking.com rate as if they were the same channel — "
|
|
"only compare rates within the same channel (B.com vs B.com, rack vs rack) when discussing parity or "
|
|
"pricing moves."
|
|
)
|
|
|
|
lines = []
|
|
|
|
if previous_insight:
|
|
gen_at = previous_insight.get('generated_at')
|
|
gen_at_str = gen_at.strftime('%d/%m/%Y %H:%M') if gen_at else 'unknown time'
|
|
lines.append(f"## Previous Insight ({gen_at_str})")
|
|
if previous_insight.get('headline'):
|
|
lines.append(f"Headline: {previous_insight['headline']}")
|
|
lines.append(previous_insight.get('content', ''))
|
|
lines.append("")
|
|
|
|
# Occupancy section
|
|
if occupancy:
|
|
lines.append(f"## Occupancy Forecast - Pickup-V2 (next {FORECAST_HORIZON_DAYS} days)")
|
|
lines.append("Date | DoW | OTB | Forecast | PY Final | Pace vs LY | Budget | Notes")
|
|
for fc in occupancy:
|
|
d = fc.get('date', '')
|
|
dow = fc.get('day_of_week', '')
|
|
otb = fc.get('current_otb')
|
|
forecast = fc.get('forecast')
|
|
py_final = fc.get('prior_year_final')
|
|
pace = fc.get('pace_vs_prior_pct')
|
|
budget_key = f"{d}_occupancy"
|
|
budget_val = budgets.get(budget_key)
|
|
|
|
otb_str = f"{otb:.0f}%" if otb is not None else "-"
|
|
fc_str = f"{forecast:.0f}%" if forecast is not None else "-"
|
|
py_str = f"{py_final:.0f}%" if py_final is not None else "-"
|
|
pace_str = f"{pace:+.0f}%" if pace is not None else "-"
|
|
bud_str = f"{budget_val:.0f}%" if budget_val is not None else "-"
|
|
holiday_note = bank_holidays.get(str(d), "")
|
|
|
|
lines.append(
|
|
f"{_fmt_date(d)} | {dow} | {otb_str} | {fc_str} | {py_str} | {pace_str} | {bud_str} | {holiday_note}"
|
|
)
|
|
lines.append("")
|
|
|
|
# Revenue section — per-day + aggregate
|
|
if revenue:
|
|
lines.append(f"## Revenue Forecast - Pickup-V2 (next {FORECAST_HORIZON_DAYS} days)")
|
|
lines.append("Date | DoW | OTB Rev | Forecast Rev | PY Rev | Pace vs LY | Budget Rev | Opportunity")
|
|
for fc in revenue:
|
|
d = fc.get('date', '')
|
|
dow = fc.get('day_of_week', '')
|
|
otb_rev = fc.get('current_otb_rev')
|
|
forecast_rev = fc.get('forecast')
|
|
py_rev = fc.get('prior_year_final_rev')
|
|
pace = fc.get('pace_vs_prior_pct')
|
|
budget_val = budgets.get(f"{d}_net_accom")
|
|
lost = fc.get('lost_potential') or 0
|
|
|
|
otb_str = f"£{otb_rev:,.0f}" if otb_rev is not None else "-"
|
|
fc_str = f"£{forecast_rev:,.0f}" if forecast_rev is not None else "-"
|
|
py_str = f"£{py_rev:,.0f}" if py_rev is not None else "-"
|
|
pace_str = f"{pace:+.0f}%" if pace is not None else "-"
|
|
bud_str = f"£{budget_val:,.0f}" if budget_val is not None else "-"
|
|
opp_str = f"£{lost:,.0f} left on table" if fc.get('has_pricing_opportunity') else "-"
|
|
|
|
lines.append(
|
|
f"{_fmt_date(d)} | {dow} | {otb_str} | {fc_str} | {py_str} | {pace_str} | {bud_str} | {opp_str}"
|
|
)
|
|
lines.append("")
|
|
|
|
total_forecast = sum(fc.get('forecast', 0) or 0 for fc in revenue)
|
|
total_otb = sum(fc.get('current_otb_rev', 0) or 0 for fc in revenue)
|
|
total_py = sum(fc.get('prior_year_final_rev', 0) or 0 for fc in revenue)
|
|
opportunity_days = sum(1 for fc in revenue if fc.get('has_pricing_opportunity'))
|
|
total_lost = sum(fc.get('lost_potential', 0) or 0 for fc in revenue)
|
|
|
|
lines.append("## Revenue Signals (aggregate)")
|
|
lines.append(
|
|
f"{FORECAST_HORIZON_DAYS}-day forecast: £{total_forecast:,.0f} | "
|
|
f"OTB: £{total_otb:,.0f} | PY: £{total_py:,.0f}"
|
|
)
|
|
if opportunity_days > 0:
|
|
lines.append(f"Pricing opportunity days: {opportunity_days} | Total lost potential: £{total_lost:,.0f}")
|
|
lines.append("")
|
|
|
|
# Competitor rates section
|
|
own_booking = competitor.get('own_booking', {})
|
|
comp_rates = competitor.get('competitors', {})
|
|
rack_rates = competitor.get('rack', {})
|
|
|
|
if own_booking or comp_rates:
|
|
lines.append(f"## Competitor Rates (next {FORECAST_HORIZON_DAYS} days)")
|
|
lines.append("Date | Own Rack | Own B.com | Cheapest Competitor (B.com) | Competitor Name")
|
|
|
|
all_dates = sorted(set(list(own_booking.keys()) + list(comp_rates.keys()) + list(rack_rates.keys())))
|
|
for d in all_dates:
|
|
rack = rack_rates.get(d)
|
|
own = own_booking.get(d, {})
|
|
comp = comp_rates.get(d, {})
|
|
|
|
rack_str = f"£{rack:.0f}" if rack else "-"
|
|
own_str = f"£{own['rate']:.0f}" if own.get('rate') else "-"
|
|
comp_str = f"£{comp['rate']:.0f}" if comp.get('rate') else "-"
|
|
comp_name = comp.get('name', '-')
|
|
|
|
note = ""
|
|
if own.get('rate') and comp.get('rate') and own['rate'] < comp['rate']:
|
|
note = " <- cheapest on B.com"
|
|
|
|
lines.append(f"{_fmt_date(d)} | {rack_str} | {own_str} | {comp_str} | {comp_name}{note}")
|
|
lines.append("")
|
|
|
|
# Market movement since last insight (same-channel deltas only)
|
|
lines.extend(build_market_movement_section(competitor, (previous_insight or {}).get('data_snapshot', {}).get('competitor_rates')))
|
|
|
|
user_msg = "\n".join(lines)
|
|
return system_msg, user_msg
|
|
|
|
|
|
async def call_llm(api_key: str, system_msg: str, user_msg: str, model: str) -> Dict[str, Any]:
|
|
"""Call Anthropic API and return response with token usage."""
|
|
import anthropic
|
|
|
|
client = anthropic.AsyncAnthropic(api_key=api_key)
|
|
|
|
try:
|
|
response = await client.messages.create(
|
|
model=model,
|
|
max_tokens=MAX_OUTPUT_TOKENS,
|
|
temperature=0.2,
|
|
system=system_msg,
|
|
messages=[{"role": "user", "content": user_msg}]
|
|
)
|
|
|
|
content = response.content[0].text if response.content else ""
|
|
return {
|
|
"content": content,
|
|
"input_tokens": response.usage.input_tokens,
|
|
"output_tokens": response.usage.output_tokens,
|
|
"model": model,
|
|
}
|
|
finally:
|
|
await client.close()
|
|
|
|
|
|
def parse_headline(raw_content: str) -> tuple[Optional[str], str]:
|
|
"""Split a `HEADLINE: ...` first line off the model's response. Returns (headline, remaining_content)."""
|
|
lines = raw_content.split("\n")
|
|
if lines and lines[0].strip().upper().startswith("HEADLINE:"):
|
|
headline = lines[0].split(":", 1)[1].strip()
|
|
remaining = "\n".join(lines[1:]).strip()
|
|
return headline, remaining
|
|
return None, raw_content
|
|
|
|
|
|
async def save_insight(
|
|
db: AsyncSession,
|
|
content: str,
|
|
headline: Optional[str],
|
|
model: str,
|
|
input_tokens: int,
|
|
output_tokens: int,
|
|
data_snapshot: Dict,
|
|
triggered_by: str = "scheduler"
|
|
):
|
|
"""Save generated insight to database."""
|
|
await db.execute(
|
|
text("""
|
|
INSERT INTO ai_insights
|
|
(content, headline, model, input_tokens, output_tokens, data_snapshot, triggered_by)
|
|
VALUES (:content, :headline, :model, :input_tokens, :output_tokens,
|
|
CAST(:data_snapshot AS jsonb), :triggered_by)
|
|
"""),
|
|
{
|
|
"content": content,
|
|
"headline": headline,
|
|
"model": model,
|
|
"input_tokens": input_tokens,
|
|
"output_tokens": output_tokens,
|
|
"data_snapshot": json.dumps(data_snapshot),
|
|
"triggered_by": triggered_by,
|
|
}
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def cleanup_old_insights(db: AsyncSession, keep_days: int = 90):
|
|
"""Remove insights older than keep_days."""
|
|
cutoff = datetime.now(timezone.utc) - timedelta(days=keep_days)
|
|
await db.execute(
|
|
text("DELETE FROM ai_insights WHERE generated_at < :cutoff"),
|
|
{"cutoff": cutoff}
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def generate_insight(db: AsyncSession, triggered_by: str = "scheduler") -> Dict[str, Any]:
|
|
"""
|
|
Core insight generation logic. Used by both scheduler and manual trigger.
|
|
Returns result dict with success/error status.
|
|
"""
|
|
config = await get_config(db)
|
|
|
|
# Check enabled
|
|
if config.get('ai_insights_enabled', 'false').lower() not in ('true', '1', 'yes'):
|
|
return {"success": False, "error": "AI insights disabled"}
|
|
|
|
# Check API key
|
|
api_key = config.get('ai_insights_api_key')
|
|
if not api_key:
|
|
return {"success": False, "error": "No API key configured"}
|
|
|
|
model = config.get('ai_insights_model', DEFAULT_MODEL)
|
|
budget = int(config.get('ai_insights_daily_token_budget', str(DEFAULT_DAILY_TOKEN_BUDGET)))
|
|
|
|
# Check daily budget
|
|
within_budget, used_today = await check_daily_budget(db, budget)
|
|
if not within_budget:
|
|
return {
|
|
"success": False,
|
|
"error": f"Daily token budget exceeded ({used_today}/{budget} tokens used today)"
|
|
}
|
|
|
|
# Gather data
|
|
logger.info("Gathering data for AI insight...")
|
|
previous_insight = await get_previous_insight(db)
|
|
occupancy = await gather_occupancy_data(db)
|
|
revenue = await gather_revenue_data(db)
|
|
budgets = await gather_budget_data(db)
|
|
competitor = await gather_competitor_data(db)
|
|
|
|
today = date.today()
|
|
end = today + timedelta(days=FORECAST_HORIZON_DAYS - 1)
|
|
bank_holidays = get_uk_bank_holidays(today, end)
|
|
|
|
if not occupancy and not revenue:
|
|
return {"success": False, "error": "No forecast data available"}
|
|
|
|
# Build prompt
|
|
system_msg, user_msg = build_prompt(
|
|
occupancy, revenue, budgets, competitor, bank_holidays, previous_insight
|
|
)
|
|
|
|
# Store data snapshot — full competitor rates so the NEXT insight can diff against it
|
|
data_snapshot = {
|
|
"occupancy_days": len(occupancy),
|
|
"revenue_days": len(revenue),
|
|
"competitor_dates": len(competitor.get('own_booking', {})),
|
|
"competitor_rates": competitor,
|
|
"prompt_preview": user_msg[:500],
|
|
}
|
|
|
|
# Call LLM
|
|
logger.info(f"Calling {model} for AI insight...")
|
|
try:
|
|
result = await call_llm(api_key, system_msg, user_msg, model)
|
|
except Exception as e:
|
|
logger.error(f"LLM call failed: {e}")
|
|
return {"success": False, "error": f"LLM call failed: {str(e)}"}
|
|
|
|
headline, content = parse_headline(result["content"])
|
|
|
|
# Save
|
|
await save_insight(
|
|
db,
|
|
content=content,
|
|
headline=headline,
|
|
model=result["model"],
|
|
input_tokens=result["input_tokens"],
|
|
output_tokens=result["output_tokens"],
|
|
data_snapshot=data_snapshot,
|
|
triggered_by=triggered_by,
|
|
)
|
|
|
|
logger.info(
|
|
f"AI insight generated: {result['input_tokens']} input, "
|
|
f"{result['output_tokens']} output tokens ({triggered_by})"
|
|
)
|
|
|
|
# Cleanup old insights
|
|
try:
|
|
await cleanup_old_insights(db)
|
|
except Exception as e:
|
|
logger.warning(f"Old insight cleanup failed: {e}")
|
|
|
|
return {
|
|
"success": True,
|
|
"content": content,
|
|
"headline": headline,
|
|
"input_tokens": result["input_tokens"],
|
|
"output_tokens": result["output_tokens"],
|
|
"model": result["model"],
|
|
}
|
|
|
|
|
|
async def run_ai_insights_generation():
|
|
"""Scheduled job entry point."""
|
|
async with AsyncSessionLocal() as db:
|
|
try:
|
|
result = await generate_insight(db, triggered_by="scheduler")
|
|
if result.get("success"):
|
|
logger.info("Scheduled AI insight generation completed")
|
|
else:
|
|
logger.info(f"Scheduled AI insight skipped: {result.get('error')}")
|
|
except Exception as e:
|
|
logger.error(f"Scheduled AI insight generation failed: {e}", exc_info=True)
|