amz_review_analyse/voc_业务_2/data_loader.py

# -*- coding: utf-8 -*-
"""
数据加载模块：从 SQLite 数据库和 CSV 中提取结构化数据。
"""
from __future__ import annotations

import csv
import json
import logging
import math
import sqlite3
from collections import Counter, defaultdict
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple

logger = logging.getLogger("voc.data_loader")

STRUCTURED_DB_NAME = "voc_structured.sqlite"
CLUSTER_DB_NAME = "voc_clustering.sqlite"
EMBED_DB_NAME = "voc_embeddings.sqlite"
CLEANED_CSV_NAME = "merged_reviews_cleaned.csv"
WORD_FREQ_CSV_NAME = "output/word_freq.csv"
TERMS_JSON_NAME = "output/voc_terms.json"

# 与 report_utils._PERSONA_PHRASE_STOPWORDS 保持一致（避免循环 import）
_PERSONA_PHRASE_STOPWORDS = frozenset({
    "self", "unknown", "user", "customer", "buyer", "myself", "the user",
    "a user", "consumer", "reviewer", "amazon", "product", "item",
})


def _phrase_segments_local(phrases: List[str]) -> List[str]:
    """将聚类短语拆成可匹配的英文片段。"""
    segs: List[str] = []
    for p in phrases:
        p = (p or "").lower().strip()
        if not p:
            continue
        segs.append(p)
        for part in p.split(","):
            part = part.strip()
            if len(part) >= 3:
                segs.append(part)
    return list(dict.fromkeys(segs))


def _cluster_semantic_label(
    cluster: "ClusterData",
    phrase_doc_freq: Dict[str, int],
    n_clusters_in_stage: int,
) -> str:
    """从 top_phrases 生成簇短语义标签（频次 × IDF，同类 stage 内去重）。"""
    scored: List[Tuple[float, str]] = []
    for rank, phrase in enumerate(cluster.top_phrases):
        tf = 1.0 / (1 + rank)
        for seg in _phrase_segments_local([phrase]):
            if seg in _PERSONA_PHRASE_STOPWORDS or len(seg) < 4:
                continue
            df = phrase_doc_freq.get(seg, 1)
            idf = math.log((n_clusters_in_stage + 1) / (df + 0.5))
            scored.append((tf * idf, seg))
    if scored:
        scored.sort(key=lambda x: (-x[0], -len(x[1])))
        return scored[0][1][:48]
    for phrase in cluster.top_phrases:
        p = (phrase or "").strip()
        if len(p) >= 4:
            return p[:48]
    return f"cluster_{cluster.cluster_label}"


@dataclass
class ReviewRecord:
    asin: str
    rating: float
    title: str
    content: str
    verified: bool
    vine: bool
    review_date: str
    source_row: int


@dataclass
class ASINStats:
    asin: str
    total: int
    avg_rating: float
    pos_rate: float
    neg_rate: float
    star_dist: Dict[int, int] = field(default_factory=dict)


@dataclass
class ClusterData:
    stage: str
    cluster_label: int
    review_count: int
    phrase_count: int
    top_phrases: List[str] = field(default_factory=list)
    source_rows: List[int] = field(default_factory=list)
    source_asins: List[str] = field(default_factory=list)
    sentiment: str = ""
    audience_label: int = -1


@dataclass
class MarketStats:
    total_reviews: int
    weighted_avg_rating: float
    pos_rate: float
    neg_rate: float
    neg_review_count: int
    pos_review_count: int
    asins: List[ASINStats] = field(default_factory=list)


# Persona A/B/C 维度信号词（用于运行时判定簇的 suggested_dimension）
_A_DIM_SIGNALS = frozenset({
    # 生理/物理特征 — 跨品类通用
    "sensitive skin", "sensitive", "allergy", "allergic",
    "elderly", "senior", "older",
    "kid", "child", "children", "toddler", "infant", "baby", "newborn",
    "pregnant", "pregnancy", "nursing", "breastfeeding",
    "diabetic", "diabetes",
    "my dog", "my cat", "my pet", "puppy", "kitten", "pet owner",
    "professional", "beginner", "first time user",
    "large breed", "small breed", "small dog", "large dog",
    "oily skin", "dry skin", "acne", "eczema", "psoriasis",
    "curly hair", "fine hair", "color treated",
})
_B_DIM_SIGNALS = frozenset({
    # 行为/使用场景 — 跨品类通用
    "travel", "on the go", "portable", "lightweight", "compact",
    "daily", "everyday", "first time", "first-time", "beginner",
    "outdoor", "indoor", "home", "office", "kitchen", "car",
    "quick", "easy to use", "simple", "convenient",
    "maintenance", "cleaning", "storage", "organization",
    "camping", "hiking", "gym", "workout",
    "cooking", "baking", "cleaning house",
})
_C_DIM_SIGNALS = frozenset({
    # 购买动机/背景 — 跨品类通用
    "gift", "present", "birthday", "christmas", "holiday",
    "for my wife", "for my husband", "for my mom", "for my daughter",
    "replacement", "replace", "upgrade", "switched from",
    "price", "cheap", "expensive", "affordable", "worth the money",
    "waste of money", "not worth", "overpriced", "good value",
    "recommend", "recommended", "saw on", "social media",
    "amazon", "online", "review", "reviews", "rating",
    "bought", "purchased", "ordered", "arrived",
})


def _suggest_dimension(stage: str, top_phrases: List[str]) -> str:
    """根据簇的 top_phrases 内容判定建议的 Persona 维度（A/B/C）。"""
    blob = " ".join(top_phrases).lower()
    a_hits = sum(1 for w in _A_DIM_SIGNALS if w in blob)
    b_hits = sum(1 for w in _B_DIM_SIGNALS if w in blob)
    c_hits = sum(1 for w in _C_DIM_SIGNALS if w in blob)
    if a_hits >= 2:
        return "A"
    if a_hits >= 1 and (b_hits + c_hits) == 0:
        return "A"
    if b_hits > c_hits and b_hits >= 2:
        return "B"
    if c_hits > b_hits and c_hits >= 2:
        return "C"
    # 默认按 stage 推理
    if stage.startswith("3a"):
        return "C"
    return "B"


class DataLoader:
    def __init__(self, project_root: Path, product_name: str = "", industry: str = ""):
        self.project_root = Path(project_root).resolve()
        self.product_name = product_name
        self.industry = industry
        self.structured_db = self.project_root / STRUCTURED_DB_NAME
        self.cluster_db = self.project_root / CLUSTER_DB_NAME
        self.embed_db = self.project_root / EMBED_DB_NAME
        self.cleaned_csv = self.project_root / CLEANED_CSV_NAME
        self.word_freq_csv = self.project_root / WORD_FREQ_CSV_NAME
        self.terms_json = self.project_root / TERMS_JSON_NAME

    @staticmethod
    def load_raw_review_samples(input_dir: Path, max_samples: int = 50):
        """从原始 CSV 目录中加载样本评论内容（仅 content 字段），供 LLM 识别产品/行业。
        Returns: (samples: list[str], dir_name: str)"""
        import csv as _csv
        samples = []
        dir_path = Path(input_dir)
        dir_name = dir_path.name if dir_path.is_dir() else ""
        if not dir_path.is_dir():
            logger.warning("原始数据目录不存在: %s", dir_path)
            return samples, dir_name
        for fname in sorted(dir_path.iterdir()):
            if not fname.suffix.lower() == ".csv":
                continue
            try:
                with fname.open(encoding="utf-8-sig", newline="") as f:
                    reader = _csv.DictReader(f)
                    if not reader.fieldnames or "content" not in reader.fieldnames:
                        continue
                    for row in reader:
                        text = (row.get("content") or "").strip()
                        if text and len(text) >= 20:
                            samples.append(text)
                            if len(samples) >= max_samples:
                                break
                if len(samples) >= max_samples:
                    break
            except Exception as e:
                logger.debug("跳过文件 %s: %s", fname.name, e)
        return samples, dir_name

    def load_reviews(self) -> List[ReviewRecord]:
        if not self.cleaned_csv.is_file():
            raise FileNotFoundError(f"清洗后评论文件不存在: {self.cleaned_csv}")
        reviews: List[ReviewRecord] = []
        with self.cleaned_csv.open(encoding="utf-8-sig", newline="") as f:
            reader = csv.DictReader(f)
            for i, row in enumerate(reader, start=1):
                text = (row.get("content") or "").strip()
                if not text:
                    continue
                reviews.append(ReviewRecord(
                    asin=(row.get("asin") or "").strip(),
                    rating=float(row.get("rating") or 0),
                    title=(row.get("title") or "").strip(),
                    content=text,
                    verified=(row.get("verified") or "").strip().lower() == "true",
                    vine=(row.get("vine") or "").strip().lower() == "true",
                    review_date=(row.get("review_date") or "").strip(),
                    source_row=i,
                ))
        logger.info("已加载 %s 条有效评论", len(reviews))
        return reviews

    def get_review_by_source_row(self, source_row: int) -> Optional[ReviewRecord]:
        reviews = self.load_reviews()
        for r in reviews:
            if r.source_row == source_row:
                return r
        return None

    def load_comment_extractions(
        self,
        job_id: Optional[int] = None,
    ) -> Dict[int, Dict[str, Any]]:
        """加载 source_row → extraction_json 映射（最新 job 或指定 job_id）。"""
        if not self.structured_db.is_file():
            logger.warning("未找到结构化库 %s，引用匹配将仅使用原文关键词", self.structured_db)
            return {}
        conn = sqlite3.connect(self.structured_db)
        try:
            if job_id is None:
                row = conn.execute(
                    "SELECT id FROM analysis_jobs ORDER BY id DESC LIMIT 1"
                ).fetchone()
                if not row:
                    logger.warning("voc_structured.sqlite 中无结构化任务")
                    return {}
                job_id = int(row[0])
            cur = conn.execute(
                """
                SELECT source_row, extraction_json
                FROM comment_extractions
                WHERE job_id = ?
                ORDER BY source_row
                """,
                (job_id,),
            )
            out: Dict[int, Dict[str, Any]] = {}
            for sr, js in cur.fetchall():
                try:
                    out[int(sr)] = json.loads(js)
                except json.JSONDecodeError as e:
                    logger.debug("跳过无效 extraction source_row=%s: %s", sr, e)
            logger.info("已加载 %s 条结构化提取 (job_id=%s)", len(out), job_id)
            return out
        finally:
            conn.close()

    def load_basic_stats(self) -> MarketStats:
        reviews = self.load_reviews()
        if not reviews:
            raise ValueError("无有效评论")
        asin_groups: Dict[str, List[ReviewRecord]] = defaultdict(list)
        for r in reviews:
            asin_groups[r.asin].append(r)
        asin_stats_list: List[ASINStats] = []
        weighted_sum = 0.0
        total_count = 0
        for asin in sorted(asin_groups.keys()):
            grp = asin_groups[asin]
            ratings = [r.rating for r in grp if r.rating > 0]
            n = len(ratings)
            if n == 0:
                continue
            avg = sum(ratings) / n
            pos = sum(1 for r in ratings if r >= 4) / n
            neg = sum(1 for r in ratings if r <= 2) / n
            dist = Counter(int(r) for r in ratings)
            asin_stats_list.append(ASINStats(
                asin=asin, total=n, avg_rating=round(avg, 2),
                pos_rate=round(pos, 3), neg_rate=round(neg, 3),
                star_dist={i: dist.get(i, 0) for i in range(1, 6)},
            ))
            weighted_sum += avg * n
            total_count += n
        all_ratings = [r.rating for r in reviews if r.rating > 0]
        wavg = round(weighted_sum / total_count, 2) if total_count else 0.0
        pos_total = sum(1 for r in all_ratings if r >= 4)
        neg_total = sum(1 for r in all_ratings if r <= 2)
        return MarketStats(
            total_reviews=total_count,
            weighted_avg_rating=wavg,
            pos_rate=round(pos_total / total_count, 3) if total_count else 0,
            neg_rate=round(neg_total / total_count, 3) if total_count else 0,
            neg_review_count=neg_total,
            pos_review_count=pos_total,
            asins=asin_stats_list,
        )

    def _latest_cluster_run_id(self) -> int:
        if not self.cluster_db.is_file():
            raise FileNotFoundError(f"聚类库不存在: {self.cluster_db}")
        conn = sqlite3.connect(self.cluster_db)
        try:
            row = conn.execute("SELECT id FROM cluster_runs ORDER BY id DESC LIMIT 1").fetchone()
            if not row:
                raise RuntimeError("voc_clustering.sqlite 中无聚类记录")
            return int(row[0])
        finally:
            conn.close()

    def load_cluster_data(self) -> Dict[str, List[ClusterData]]:
        run_id = self._latest_cluster_run_id()
        conn = sqlite3.connect(self.cluster_db)
        try:
            cur = conn.execute(
                """SELECT stage, cluster_label, source_row, embed_text,
                   entity_type, audience, sentiment, content
                   FROM cluster_assignments WHERE run_id = ?
                   ORDER BY stage, cluster_label, source_row""",
                (run_id,),
            )
            rows = cur.fetchall()
        finally:
            conn.close()
        groups: Dict[Tuple[str, int], List[Tuple[int, str, str, str, str]]] = defaultdict(list)
        for stage, label, src_row, embed_text, etype, audience, sentiment, content in rows:
            groups[(stage, int(label))].append((
                int(src_row), embed_text, etype or "", audience or "", sentiment or "",
            ))
        reviews = self.load_reviews()
        row_to_asin: Dict[int, str] = {r.source_row: r.asin for r in reviews}
        result: Dict[str, List[ClusterData]] = defaultdict(list)
        for (stage, label), items in sorted(groups.items()):
            unique_src_rows = sorted(set(sr for sr, _, _, _, _ in items))
            asins = sorted(set(row_to_asin.get(sr, "?") for sr in unique_src_rows))
            phrases = [et for _, et, _, _, _ in items if et.strip()]
            phrase_counter = Counter(phrases)
            top_phrases = [p for p, _ in phrase_counter.most_common(15)]
            sentiment = items[0][3] if items else ""
            audience_str = items[0][2] if items else ""
            aud_label = -1
            if audience_str:
                try:
                    aud_label = int(audience_str)
                except ValueError:
                    aud_label = -1
            cd = ClusterData(
                stage=stage, cluster_label=label,
                review_count=len(unique_src_rows), phrase_count=len(phrases),
                top_phrases=top_phrases, source_rows=unique_src_rows,
                source_asins=asins, sentiment=sentiment, audience_label=aud_label,
            )
            result[stage].append(cd)
        logger.info("已加载聚类: %s stages, %s 簇", len(result), sum(len(v) for v in result.values()))
        return dict(result)



    def build_persona_cluster_catalog(
        self,
        sample_reviews_per_cluster: int = 5,
        max_review_chars: int = 320,
    ) -> List[Dict[str, Any]]:
        """供 Persona LLM 绑定的聚类簇目录（含每簇代表性评论和运行时维度建议）。"""
        all_c = self.load_cluster_data()
        catalog: List[Dict[str, Any]] = []
        for stage in (
            "3a_pain_global",
            "3b_aspect_opinion_negative",
            "3b_aspect_opinion_positive",
        ):
            stage_clusters = sorted(all_c.get(stage, []), key=lambda x: -x.review_count)
            phrase_doc_freq: Dict[str, int] = defaultdict(int)
            for c in stage_clusters:
                segs = set(_phrase_segments_local(c.top_phrases))
                for seg in segs:
                    if seg not in _PERSONA_PHRASE_STOPWORDS and len(seg) >= 4:
                        phrase_doc_freq[seg] += 1
            n_stage = max(len(stage_clusters), 1)
            used_labels: Set[str] = set()
            for c in stage_clusters:
                label = _cluster_semantic_label(c, phrase_doc_freq, n_stage)
                if label in used_labels:
                    for rank, phrase in enumerate(c.top_phrases[1:], start=1):
                        alt = (phrase or "").strip().lower()[:48]
                        if alt and len(alt) >= 4 and alt not in used_labels:
                            label = alt
                            break
                used_labels.add(label)
                entry: Dict[str, Any] = {
                    "stage": stage,
                    "label": c.cluster_label,
                    "review_count": c.review_count,
                    "semantic_label": label,
                    "top_phrases": c.top_phrases[:12],
                    "suggested_dimension": _suggest_dimension(stage, c.top_phrases),
                }
                if sample_reviews_per_cluster > 0:
                    quotes = self.get_representative_quotes(
                        c.source_rows, max_quotes=sample_reviews_per_cluster,
                    )
                    entry["sample_reviews"] = [
                        {
                            "rating": int(q.get("rating") or 0),
                            "content": (q.get("content") or "")[:max_review_chars],
                        }
                        for q in quotes
                    ]
                catalog.append(entry)
        return catalog

    def load_audience_clusters(self) -> List[ClusterData]:
        return self.load_cluster_data().get("1_audience", [])

    def load_global_pain_clusters(self) -> List[ClusterData]:
        return self.load_cluster_data().get("3a_pain_global", [])

    def load_global_feedback_clusters(self) -> Dict[str, List[ClusterData]]:
        all_c = self.load_cluster_data()
        return {
            "negative": all_c.get("3b_aspect_opinion_negative", []),
            "positive": all_c.get("3b_aspect_opinion_positive", []),
            "neutral": all_c.get("3b_aspect_opinion_neutral", []),
        }

    def load_per_audience_clusters(self) -> Dict[int, Dict[str, List[ClusterData]]]:
        all_c = self.load_cluster_data()
        aud_c: Dict[int, Dict[str, List[ClusterData]]] = defaultdict(
            lambda: {"pain": [], "negative": [], "positive": [], "neutral": []}
        )
        for stage, clusters in all_c.items():
            if stage.startswith("2a_pain_audience_c"):
                al = int(stage.split("_c")[-1])
                aud_c[al]["pain"].extend(clusters)
            elif stage.startswith("2b_aspect_opinion_"):
                parts = stage.replace("2b_aspect_opinion_", "").split("_audience_c")
                sentiment = parts[0]
                al = int(parts[1]) if len(parts) > 1 else -1
                if sentiment in ("positive", "negative", "neutral"):
                    aud_c[al][sentiment].extend(clusters)
        return dict(aud_c)

    def build_cluster_prompt_data(
        self,
        persona_sample_reviews: int = 5,
    ) -> Dict[str, Any]:
        global_pain = self.load_global_pain_clusters()
        global_fb = self.load_global_feedback_clusters()
        stats = self.load_basic_stats()

        def _sum(clusters: List[ClusterData]) -> List[Dict[str, Any]]:
            return [
                {"label": c.cluster_label, "phrase_count": c.phrase_count,
                 "review_count": c.review_count, "asins": c.source_asins,
                 "top_phrases": c.top_phrases}
                for c in sorted(clusters, key=lambda x: -x.phrase_count)
            ]
        return {
            "audience_clusters": [],
            "global_pains": _sum(global_pain),
            "global_negative": _sum(global_fb["negative"]),
            "global_positive": _sum(global_fb["positive"]),
            "persona_cluster_catalog": self.build_persona_cluster_catalog(
                sample_reviews_per_cluster=persona_sample_reviews,
            ),
            "per_audience": {},
            "total_reviews": stats.total_reviews,
            "neg_review_count": stats.neg_review_count,
            "pos_review_count": stats.pos_review_count,
            "asins": [a.asin for a in stats.asins],
        }

    def load_word_freq(self, top_n: int = 200) -> List[Tuple[str, int]]:
        if not self.word_freq_csv.is_file():
            logger.warning("词频文件不存在: %s", self.word_freq_csv)
            return []
        rows: List[Tuple[str, int]] = []
        with self.word_freq_csv.open(encoding="utf-8", newline="") as f:
            reader = csv.DictReader(f)
            for row in reader:
                w = (row.get("word") or "").strip()
                if not w:
                    continue
                try:
                    c = int(row.get("count") or 0)
                except ValueError:
                    c = 0
                rows.append((w, c))
        rows.sort(key=lambda x: x[1], reverse=True)
        return rows[:top_n]

    def get_representative_quotes(self, source_rows: List[int], max_quotes: int = 5) -> List[Dict[str, str]]:
        reviews = self.load_reviews()
        row_map = {r.source_row: r for r in reviews}
        candidates = []
        for sr in source_rows:
            r = row_map.get(sr)
            if r and len(r.content) >= 30:
                candidates.append(r)
        candidates.sort(key=lambda r: abs(len(r.content) - 200))
        selected = candidates[:max_quotes]
        return [
            {"content": r.content, "asin": r.asin, "rating": str(int(r.rating)),
             "amazon_url": f"https://www.amazon.com/dp/{r.asin}"}
            for r in selected
        ]

    def get_quotes_for_cluster(self, cluster: ClusterData, max_quotes: int = 5) -> List[Dict[str, str]]:
        return self.get_representative_quotes(cluster.source_rows, max_quotes)


def build_amazon_url(asin: str) -> str:
    return f"https://www.amazon.com/dp/{asin}"


def asin_link_html(asin: str) -> str:
    url = build_amazon_url(asin)
    return f'<a href="{url}" target="_blank" rel="noopener">{asin}</a>'