from __future__ import annotations

import json
import os
import shlex
import subprocess
from datetime import date
from typing import Any


class PostgresRepository:
    """Read-only repository for the Liangzhu ecommerce PostgreSQL database.

    Default mode uses SSH to run psql on `wwyl-cloud` so no DB password is stored in
    this project. When deployed on the cloud host later, set
    LIANGZHU_PSQL_MODE=local to call local psql directly.
    """

    def __init__(self, ssh_host: str | None = None, db_name: str = "liangzhu_qianniu"):
        self.ssh_host = ssh_host or os.getenv("LIANGZHU_DB_SSH_HOST", "wwyl-cloud")
        self.db_name = db_name
        self.mode = os.getenv("LIANGZHU_PSQL_MODE", "ssh")

    def _psql_json(self, sql: str) -> list[dict[str, Any]]:
        wrapped = "SELECT COALESCE(jsonb_agg(to_jsonb(t))::text, '[]') FROM (" + sql + ") t;"
        if self.mode == "local":
            if os.getenv("LIANGZHU_PSQL_USE_SUDO") == "1":
                cmd = ["sudo", "-u", "postgres", "psql", "-d", self.db_name, "-At", "-c", wrapped]
            else:
                cmd = ["psql", "-d", self.db_name, "-At", "-c", wrapped]
        else:
            remote = "sudo -u postgres psql -d " + shlex.quote(self.db_name) + " -At -c " + shlex.quote(wrapped)
            cmd = ["ssh", "-o", "BatchMode=yes", self.ssh_host, remote]
        proc = subprocess.run(cmd, text=True, capture_output=True, timeout=180)
        if proc.returncode:
            raise RuntimeError((proc.stderr or proc.stdout or "psql failed")[-2000:])
        return json.loads((proc.stdout or "[]").strip() or "[]")

    def latest_date(self) -> str | None:
        rows = self._psql_json("SELECT max(stat_date)::text AS latest_date FROM mart.omnichannel_daily_store_summary")
        return rows[0].get("latest_date") if rows else None

    def available_dates(self) -> list[str]:
        rows = self._psql_json("""
SELECT stat_date::text AS stat_date
FROM mart.omnichannel_daily_store_summary
GROUP BY stat_date
ORDER BY stat_date
""")
        return [r["stat_date"] for r in rows if r.get("stat_date")]

    def store_summary(self, start_date: date, end_date: date) -> list[dict[str, Any]]:
        return self._psql_json(f"""
SELECT source_platform, store_name,
       SUM(visitor_count) AS visitor_count,
       SUM(page_view_count) AS page_view_count,
       SUM(payment_amount) AS payment_amount,
       SUM(refund_amount) AS refund_amount,
       SUM(net_payment_amount) AS net_payment_amount,
       SUM(payment_buyer_count) AS payment_buyer_count,
       SUM(payment_item_count) AS payment_item_count
FROM mart.omnichannel_daily_store_summary
WHERE stat_date BETWEEN '{start_date}'::date AND '{end_date}'::date
GROUP BY source_platform, store_name
ORDER BY source_platform, store_name
""")

    def trend(self, start_date: date, end_date: date) -> list[dict[str, Any]]:
        return self._psql_json(f"""
SELECT stat_date::text AS stat_date,
       SUM(visitor_count) AS visitor_count,
       SUM(page_view_count) AS page_view_count,
       SUM(net_payment_amount) AS net_payment_amount,
       SUM(payment_amount) AS payment_amount,
       SUM(payment_buyer_count) AS payment_buyer_count
FROM mart.omnichannel_daily_store_summary
WHERE stat_date BETWEEN '{start_date}'::date AND '{end_date}'::date
GROUP BY stat_date
ORDER BY stat_date
""")

    def product_rank(self, start_date: date, end_date: date, limit: int = 20) -> list[dict[str, Any]]:
        return self._psql_json(f"""
SELECT store_name, product_id, left(product_name, 100) AS product_name,
       SUM(product_visitor_count) AS product_visitor_count,
       SUM(payment_amount) AS payment_amount,
       SUM(refund_amount) AS refund_amount,
       SUM(net_payment_amount) AS net_payment_amount,
       SUM(payment_item_count) AS payment_item_count,
       SUM(payment_buyer_count) AS payment_buyer_count
FROM fact.product_daily_core
WHERE stat_date BETWEEN '{start_date}'::date AND '{end_date}'::date
  AND COALESCE(net_payment_amount, payment_amount, 0) > 0
GROUP BY store_name, product_id, product_name
ORDER BY SUM(COALESCE(net_payment_amount, payment_amount, 0)) DESC NULLS LAST
LIMIT {int(limit)}
""")

    def traffic_rank(self, start_date: date, end_date: date, limit: int = 20) -> list[dict[str, Any]]:
        return self._psql_json(f"""
SELECT store_name,
       COALESCE(parent_source_name, '') AS parent_source_name,
       COALESCE(source_name, '未标明') AS source_name,
       SUM(visitor_count) AS visitor_count,
       SUM(page_view_count) AS page_view_count,
       SUM(payment_amount) AS payment_amount,
       SUM(payment_buyer_count) AS payment_buyer_count
FROM fact.traffic_daily_store
WHERE stat_date BETWEEN '{start_date}'::date AND '{end_date}'::date
GROUP BY store_name, parent_source_name, source_name
ORDER BY SUM(visitor_count) DESC NULLS LAST
LIMIT {int(limit)}
""")

    def promotion_store_summary(self, start_date: date, end_date: date) -> list[dict[str, Any]]:
        """Promotion effect by store, using scene report as the non-duplicated spend/GMV口径."""
        return self._psql_json(f"""
SELECT p.store_code,
       p.store_name,
       SUM(COALESCE(p.spend_amount, 0)) AS promotion_spend_amount,
       SUM(COALESCE(p.promoted_transaction_amount, 0)) AS promoted_transaction_amount,
       SUM(COALESCE(p.transaction_count, 0)) AS promoted_transaction_count,
       SUM(COALESCE(p.click_count, 0)) AS promotion_click_count,
       SUM(COALESCE(p.impression_count, 0)) AS promotion_impression_count,
       CASE WHEN SUM(COALESCE(p.impression_count, 0)) = 0 THEN NULL ELSE SUM(COALESCE(p.click_count, 0)) / SUM(COALESCE(p.impression_count, 0)) END AS promotion_ctr,
       CASE WHEN SUM(COALESCE(p.click_count, 0)) = 0 THEN NULL ELSE SUM(COALESCE(p.spend_amount, 0)) / SUM(COALESCE(p.click_count, 0)) END AS promotion_cpc,
       CASE WHEN SUM(COALESCE(p.spend_amount, 0)) = 0 THEN NULL ELSE SUM(COALESCE(p.promoted_transaction_amount, 0)) / SUM(COALESCE(p.spend_amount, 0)) END AS promotion_roi,
       SUM(COALESCE(p.net_payment_amount, 0)) AS net_payment_amount_for_rate,
       CASE WHEN SUM(COALESCE(p.net_payment_amount, 0)) = 0 THEN NULL ELSE SUM(COALESCE(p.spend_amount, 0)) / SUM(COALESCE(p.net_payment_amount, 0)) END AS promotion_spend_rate,
       CASE WHEN SUM(COALESCE(p.net_payment_amount, 0)) = 0 THEN NULL ELSE SUM(COALESCE(p.promoted_transaction_amount, 0)) / SUM(COALESCE(p.net_payment_amount, 0)) END AS promotion_transaction_share
FROM mart.promotion_daily_effect_summary p
WHERE p.stat_date BETWEEN '{start_date}'::date AND '{end_date}'::date
GROUP BY p.store_code, p.store_name
ORDER BY SUM(COALESCE(p.spend_amount, 0)) DESC NULLS LAST
""")

    def promotion_trend(self, start_date: date, end_date: date) -> list[dict[str, Any]]:
        return self._psql_json(f"""
SELECT p.stat_date::text AS stat_date,
       SUM(COALESCE(p.spend_amount, 0)) AS promotion_spend_amount,
       SUM(COALESCE(p.promoted_transaction_amount, 0)) AS promoted_transaction_amount,
       SUM(COALESCE(p.transaction_count, 0)) AS promoted_transaction_count,
       SUM(COALESCE(p.click_count, 0)) AS promotion_click_count,
       SUM(COALESCE(p.impression_count, 0)) AS promotion_impression_count,
       CASE WHEN SUM(COALESCE(p.spend_amount, 0)) = 0 THEN NULL ELSE SUM(COALESCE(p.promoted_transaction_amount, 0)) / SUM(COALESCE(p.spend_amount, 0)) END AS promotion_roi,
       CASE WHEN SUM(COALESCE(p.net_payment_amount, 0)) = 0 THEN NULL ELSE SUM(COALESCE(p.spend_amount, 0)) / SUM(COALESCE(p.net_payment_amount, 0)) END AS promotion_spend_rate
FROM mart.promotion_daily_effect_summary p
WHERE p.stat_date BETWEEN '{start_date}'::date AND '{end_date}'::date
GROUP BY p.stat_date
ORDER BY p.stat_date
""")

    def promotion_scene_rank(self, start_date: date, end_date: date, limit: int = 20) -> list[dict[str, Any]]:
        return self._psql_json(f"""
SELECT store_name,
       scene_name,
       SUM(COALESCE(spend_amount, 0)) AS promotion_spend_amount,
       SUM(COALESCE(promoted_transaction_amount, 0)) AS promoted_transaction_amount,
       SUM(COALESCE(transaction_count, 0)) AS promoted_transaction_count,
       SUM(COALESCE(click_count, 0)) AS promotion_click_count,
       SUM(COALESCE(impression_count, 0)) AS promotion_impression_count,
       CASE WHEN SUM(COALESCE(impression_count, 0)) = 0 THEN NULL ELSE SUM(COALESCE(click_count, 0)) / SUM(COALESCE(impression_count, 0)) END AS promotion_ctr,
       CASE WHEN SUM(COALESCE(click_count, 0)) = 0 THEN NULL ELSE SUM(COALESCE(spend_amount, 0)) / SUM(COALESCE(click_count, 0)) END AS promotion_cpc,
       CASE WHEN SUM(COALESCE(spend_amount, 0)) = 0 THEN NULL ELSE SUM(COALESCE(promoted_transaction_amount, 0)) / SUM(COALESCE(spend_amount, 0)) END AS promotion_roi
FROM mart.promotion_daily_effect_summary
WHERE stat_date BETWEEN '{start_date}'::date AND '{end_date}'::date
GROUP BY store_name, scene_name
ORDER BY SUM(COALESCE(spend_amount, 0)) DESC NULLS LAST
LIMIT {int(limit)}
""")

    def promotion_plan_rank(self, start_date: date, end_date: date, limit: int = 20) -> list[dict[str, Any]]:
        return self._psql_json(f"""
SELECT store_name,
       scene_name,
       plan_name,
       SUM(COALESCE(spend_amount, 0)) AS promotion_spend_amount,
       SUM(COALESCE(promoted_transaction_amount, 0)) AS promoted_transaction_amount,
       SUM(COALESCE(transaction_count, 0)) AS promoted_transaction_count,
       SUM(COALESCE(click_count, 0)) AS promotion_click_count,
       SUM(COALESCE(impression_count, 0)) AS promotion_impression_count,
       CASE WHEN SUM(COALESCE(impression_count, 0)) = 0 THEN NULL ELSE SUM(COALESCE(click_count, 0)) / SUM(COALESCE(impression_count, 0)) END AS promotion_ctr,
       CASE WHEN SUM(COALESCE(click_count, 0)) = 0 THEN NULL ELSE SUM(COALESCE(spend_amount, 0)) / SUM(COALESCE(click_count, 0)) END AS promotion_cpc,
       CASE WHEN SUM(COALESCE(spend_amount, 0)) = 0 THEN NULL ELSE SUM(COALESCE(promoted_transaction_amount, 0)) / SUM(COALESCE(spend_amount, 0)) END AS promotion_roi
FROM mart.promotion_daily_plan_summary
WHERE stat_date BETWEEN '{start_date}'::date AND '{end_date}'::date
GROUP BY store_name, scene_name, plan_name
ORDER BY SUM(COALESCE(spend_amount, 0)) DESC NULLS LAST
LIMIT {int(limit)}
""")