"""Context-builder query functions for the SEO tab (Chapter 09). Per Ch09's own Output instruction, bot-related widgets query bot_hits directly (small, indefinitely-retained, indexed — Ch06) rather than a new rollup; only the human-side of the crawled-vs-visited comparison needs the new human_path_stats_daily table. """ from __future__ import annotations from collections import defaultdict from datetime import date from sqlalchemy import case, func from app.extensions import db from app.models.bot_hit import BotHit from app.models.human_path_stats import HumanPathStatsDaily from app.utils.dates import day_bounds # ASSUMPTION (flagged): Ch09 doesn't define "major crawler" for the # crawl-frequency chart's series cap. TOP_N_BOTS_FOR_CHART = 6 def get_bot_summary(from_date: date, to_date: date) -> list[dict]: start, end = day_bounds(from_date, to_date) rows = ( db.session.query( BotHit.bot_name, func.count().label("hits"), func.sum(case((BotHit.verified.is_(True), 1), else_=0)).label("verified_hits"), func.max(BotHit.timestamp).label("last_seen"), ) .filter(BotHit.timestamp >= start, BotHit.timestamp < end) .group_by(BotHit.bot_name) .order_by(func.count().desc()) .all() ) return [ { "bot_name": r.bot_name, "hits": r.hits, "verified_pct": round(r.verified_hits / r.hits * 100, 1) if r.hits else 0.0, "last_seen": r.last_seen.isoformat() if r.last_seen else None, } for r in rows ] def get_crawl_chart_data(from_date: date, to_date: date, bot_name: str | None) -> dict: start, end = day_bounds(from_date, to_date) base_filters = [BotHit.timestamp >= start, BotHit.timestamp < end] if bot_name: allowed_bots = [bot_name] else: top_rows = ( db.session.query(BotHit.bot_name, func.count().label("hits")) .filter(*base_filters) .group_by(BotHit.bot_name) .order_by(func.count().desc()) .limit(TOP_N_BOTS_FOR_CHART) .all() ) allowed_bots = [r.bot_name for r in top_rows] if not allowed_bots: return {"series": []} rows = ( db.session.query(func.date(BotHit.timestamp).label("day"), BotHit.bot_name, func.count().label("count")) .filter(*base_filters, BotHit.bot_name.in_(allowed_bots)) .group_by("day", BotHit.bot_name) .order_by("day") .all() ) points_by_bot: dict[str, list[dict]] = defaultdict(list) for r in rows: points_by_bot[r.bot_name].append({"t": r.day, "count": r.count}) return {"series": [{"bot_name": b, "points": points_by_bot.get(b, [])} for b in allowed_bots]} def get_bot_status_codes(from_date: date, to_date: date) -> dict: start, end = day_bounds(from_date, to_date) bucket = case( (BotHit.status_code < 300, "2xx"), (BotHit.status_code < 400, "3xx"), (BotHit.status_code < 500, "4xx"), else_="5xx", ) rows = ( db.session.query(bucket.label("bucket"), func.count().label("count")) .filter(BotHit.timestamp >= start, BotHit.timestamp < end) .group_by("bucket") .all() ) breakdown = {"2xx": 0, "3xx": 0, "4xx": 0, "5xx": 0} for r in rows: breakdown[r.bucket] = r.count # Ch09 calls out 404 by name specifically, not just the 4xx bucket. not_found_404 = ( db.session.query(func.count()) .filter(BotHit.timestamp >= start, BotHit.timestamp < end, BotHit.status_code == 404) .scalar() ) return {"breakdown": breakdown, "not_found_404": not_found_404 or 0} def get_crawled_vs_visited(from_date: date, to_date: date, page: int, per_page: int) -> tuple[list[list], int]: """Diffed table: one row per path, bot hits vs. human hits.""" start, end = day_bounds(from_date, to_date) bot_rows = ( db.session.query(BotHit.path, func.count().label("hits")) .filter(BotHit.timestamp >= start, BotHit.timestamp < end) .group_by(BotHit.path) .all() ) human_rows = ( db.session.query(HumanPathStatsDaily.path, func.sum(HumanPathStatsDaily.count).label("hits")) .filter(HumanPathStatsDaily.date >= from_date, HumanPathStatsDaily.date <= to_date) .group_by(HumanPathStatsDaily.path) .all() ) bot_counts = {r.path: r.hits for r in bot_rows} human_counts = {r.path: r.hits for r in human_rows} combined = [] for path in set(bot_counts) | set(human_counts): b, h = bot_counts.get(path, 0), human_counts.get(path, 0) total = b + h combined.append([path, b, h, round(b / total * 100, 1) if total else 0.0]) # ASSUMPTION (flagged): sorted by bot hits desc — Ch09 doesn't specify. combined.sort(key=lambda row: row[1], reverse=True) total_count = len(combined) offset = (page - 1) * per_page return combined[offset : offset + per_page], total_count