fix(analytics): link report data to selected period

services/analysis_period_service.py

@@ -0,0 +1,89 @@
+"""Canonical period helpers shared by analytics report pages."""
+
+from __future__ import annotations
+
+from calendar import monthrange
+from datetime import date, datetime
+
+
+def parse_iso_date(value) -> date | None:
+    if not value:
+        return None
+    if isinstance(value, datetime):
+        return value.date()
+    if isinstance(value, date):
+        return value
+    try:
+        return datetime.strptime(str(value)[:10], "%Y-%m-%d").date()
+    except (TypeError, ValueError):
+        return None
+
+
+def parse_month(value) -> date | None:
+    if not value:
+        return None
+    if isinstance(value, datetime):
+        return value.date().replace(day=1)
+    if isinstance(value, date):
+        return value.replace(day=1)
+    try:
+        return datetime.strptime(str(value)[:7], "%Y-%m").date()
+    except (TypeError, ValueError):
+        return None
+
+
+def month_end(value) -> date | None:
+    month_start = parse_month(value)
+    if not month_start:
+        return None
+    return month_start.replace(day=monthrange(month_start.year, month_start.month)[1])
+
+
+def make_analysis_period(start, end=None, *, mode="range", label=None) -> dict:
+    start_date = parse_iso_date(start)
+    end_date = parse_iso_date(end) if end else start_date
+    if not start_date and end_date:
+        start_date = end_date
+    if not start_date or not end_date:
+        return {
+            "mode": mode,
+            "label": label or "全部資料",
+            "start_date": "",
+            "end_date": "",
+            "start_month": "",
+            "end_month": "",
+            "is_single_day": False,
+            "is_single_month": False,
+        }
+    if start_date > end_date:
+        start_date, end_date = end_date, start_date
+
+    start_month = start_date.strftime("%Y-%m")
+    end_month_value = end_date.strftime("%Y-%m")
+    is_single_day = start_date == end_date
+    is_single_month = start_month == end_month_value
+    if not label:
+        if is_single_day:
+            label = start_date.strftime("%Y-%m-%d")
+        elif is_single_month and start_date.day == 1 and end_date == month_end(start_date):
+            label = start_date.strftime("%Y 年 %m 月")
+        else:
+            label = f"{start_date:%Y-%m-%d} 至 {end_date:%Y-%m-%d}"
+
+    return {
+        "mode": mode,
+        "label": label,
+        "start_date": start_date.isoformat(),
+        "end_date": end_date.isoformat(),
+        "start_month": start_month,
+        "end_month": end_month_value,
+        "is_single_day": is_single_day,
+        "is_single_month": is_single_month,
+    }
+
+
+def make_month_period(value, *, mode="month") -> dict:
+    start = parse_month(value)
+    if not start:
+        return make_analysis_period(None, mode=mode)
+    return make_analysis_period(start, month_end(start), mode=mode)

services/competitor_intel_repository.py

@@ -1533,20 +1533,49 @@ def _fetch_manual_review_summary(engine) -> dict[str, Any]:
     return summary
 
 
-def fetch_competitor_gap_trend(engine, days: int = 30) -> dict:
+def fetch_competitor_gap_trend(
+    engine,
+    days: int = 30,
+    *,
+    start_date=None,
+    end_date=None,
+) -> dict:
     days = max(7, min(int(days or 30), 120))
+    start_label = _date_label(start_date) if start_date else ""
+    end_label = _date_label(end_date) if end_date else ""
     return _cached_payload(
-        f"gap_trend:v2:days={days}:floor={PCHOME_MATCH_SCORE_FLOOR}",
-        lambda: _fetch_competitor_gap_trend_uncached(engine, days=days),
+        f"gap_trend:v3:days={days}:start={start_label}:end={end_label}:floor={PCHOME_MATCH_SCORE_FLOOR}",
+        lambda: _fetch_competitor_gap_trend_uncached(
+            engine,
+            days=days,
+            start_date=start_label or None,
+            end_date=end_label or None,
+        ),
     )
 
 
-def _fetch_competitor_gap_trend_uncached(engine, days: int = 30) -> dict:
-    """近 N 天 PChome 價差壓力趨勢。"""
+def _fetch_competitor_gap_trend_uncached(
+    engine,
+    days: int = 30,
+    *,
+    start_date=None,
+    end_date=None,
+) -> dict:
+    """PChome gap trend for an explicit period or a trailing N-day window."""
     if not inspect(engine).has_table("competitor_price_history"):
         return {"labels": [], "avg_gap_pct": [], "risk_count": [], "momo_advantage_count": [], "match_count": []}
 
     days = max(7, min(int(days or 30), 120))
+    if start_date or end_date:
+        period_clause = """
+              AND (:start_date IS NULL OR cph.crawled_at >= CAST(:start_date AS date))
+              AND (:end_date IS NULL OR cph.crawled_at < CAST(:end_date AS date) + INTERVAL '1 day')
+        """
+        params = {"start_date": start_date, "end_date": end_date}
+    else:
+        period_clause = "AND cph.crawled_at >= CURRENT_DATE - (:days * INTERVAL '1 day')"
+        params = {"days": days}
+
     sql = text(f"""
         WITH latest_history AS (
             SELECT
@@ -1560,7 +1589,7 @@ def _fetch_competitor_gap_trend_uncached(engine, days: int = 30) -> dict:
                 ) AS rn
             FROM competitor_price_history cph
             WHERE cph.source = 'pchome'
-              AND cph.crawled_at >= CURRENT_DATE - (:days * INTERVAL '1 day')
+              {period_clause}
               AND cph.momo_price IS NOT NULL
               AND cph.momo_price > 0
               AND cph.price IS NOT NULL
@@ -1580,7 +1609,7 @@ def _fetch_competitor_gap_trend_uncached(engine, days: int = 30) -> dict:
         ORDER BY bucket_date
     """)
     with engine.connect() as conn:
-        rows = conn.execute(sql, {"days": days}).mappings().all()
+        rows = conn.execute(sql, params).mappings().all()
 
     return {
         "labels": [_date_label(row.get("bucket_date")) for row in rows],
@@ -2370,14 +2399,31 @@ def fetch_competitor_comparison_results(
     return results
 
 
-def build_competitor_intel_payload(engine, days: int = 30) -> dict:
+def build_competitor_intel_payload(
+    engine,
+    days: int = 30,
+    *,
+    start_date=None,
+    end_date=None,
+    include_current_snapshot: bool = True,
+) -> dict:
     """頁面、AI、PPT 可共用的摘要 payload。"""
-    review_queue = fetch_competitor_review_queue(engine, limit=12)
+    review_queue = fetch_competitor_review_queue(engine, limit=12) if include_current_snapshot else []
     return {
-        "coverage": fetch_competitor_coverage(engine),
-        "trend": fetch_competitor_gap_trend(engine, days=days),
-        "top_risks": fetch_top_competitor_risks(engine, limit=10),
+        "coverage": fetch_competitor_coverage(engine) if include_current_snapshot else {},
+        "trend": fetch_competitor_gap_trend(
+            engine,
+            days=days,
+            start_date=start_date,
+            end_date=end_date,
+        ),
+        "top_risks": fetch_top_competitor_risks(engine, limit=10) if include_current_snapshot else [],
         "review_queue": review_queue,
         "review_decision_brief": summarize_review_decision_envelopes(review_queue, limit=5),
         "match_score_floor": PCHOME_MATCH_SCORE_FLOOR,
+        "snapshot_in_period": include_current_snapshot,
+        "period": {
+            "start_date": _date_label(start_date) if start_date else "",
+            "end_date": _date_label(end_date) if end_date else "",
+        },
     }