Amazon Review Checker

#!/usr/bin/env python3
"""
Amazon Review Checker - Core Analyzer
AmazonReviewAuthenticityDetection - CoreAnalyzeEngine

Features:
- ProgressiveAnalyze ( has  many  few Data，Give conclusion)
-  many dimensionDegreeDetection (Time/Content/Rating/Account/VP)
- Friendly guidance (HintCanDeepenDirection)

Version: 1.0.0
"""

import json
import re
import math
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from datetime import datetime, timedelta
from collections import Counter
from enum import Enum
import sys


class RiskLevel(Enum):
    LOW = "low"           # 0-30
    MEDIUM = "medium"     # 31-60
    HIGH = "high"         # 61-80
    CRITICAL = "critical" # 81-100


class AnalysisLevel(Enum):
    L1_BASIC = "L1"      # onlyContent
    L2_TIMED = "L2"      # + Time
    L3_SCORED = "L3"     # + Rating
    L4_FULL = "L4"       #  all Field


# ============================================================
# Data Structures
# ============================================================

@dataclass
class Review:
    """Single itemReview"""
    content: str
    rating: Optional[int] = None          # 1-5 star
    date: Optional[str] = None            # DayPeriod
    reviewer_name: Optional[str] = None   # Reviewperson
    verified_purchase: Optional[bool] = None  # VP
    helpful_votes: Optional[int] = None   #  has Help
    reviewer_reviews_count: Optional[int] = None  # ReviewTotalReview Count
    
    @property
    def has_rating(self) -> bool:
        return self.rating is not None
    
    @property
    def has_date(self) -> bool:
        return self.date is not None
    
    @property
    def has_vp(self) -> bool:
        return self.verified_purchase is not None


@dataclass
class DimensionResult:
    """Single dimensionDegreeDetectionResult"""
    name: str
    name_zh: str
    score: float          # 0-100, exceedHighexceedSuspicious
    status: str           # ✅ ⚠️ 🔴
    detail: str
    detail_zh: str
    weight: float = 0.0


@dataclass
class SuspiciousReview:
    """SuspiciousReview"""
    content: str
    risk_score: float
    reasons: List[str]
    reasons_zh: List[str]


@dataclass
class AnalysisResult:
    """AnalyzeResult"""
    asin: str
    total_reviews: int
    analysis_level: AnalysisLevel
    authenticity_score: int       # 0-100, exceedHighexceedReal
    risk_level: RiskLevel
    dimensions: List[DimensionResult]
    suspicious_reviews: List[SuspiciousReview]
    available_fields: List[str]
    missing_fields: List[str]
    deepening_hints: List[str]
    deepening_hints_zh: List[str]
    summary: str
    summary_zh: str


# ============================================================
# DetectionAlgorithm
# ============================================================

def detect_content_similarity(reviews: List[Review]) -> DimensionResult:
    """DetectionSimilar contentDegree"""
    if len(reviews) < 2:
        return DimensionResult(
            name="Content Similarity",
            name_zh="Similar contentDegree",
            score=0,
            status="✅",
            detail="Not enough reviews to compare",
            detail_zh="Review CountInsufficient，CannotComparison",
            weight=0.20
        )
    
    # Simple similarityDegreeDetection：CheckRepeated phrases
    contents = [r.content.lower() for r in reviews]
    
    # Extract3-gram
    def get_ngrams(text, n=3):
        words = re.findall(r'\w+', text)
        return [' '.join(words[i:i+n]) for i in range(len(words)-n+1)]
    
    all_ngrams = []
    for content in contents:
        all_ngrams.extend(get_ngrams(content))
    
    # Statistics heavy complex
    ngram_counts = Counter(all_ngrams)
    repeated = sum(1 for count in ngram_counts.values() if count > 1)
    total = len(ngram_counts) if ngram_counts else 1
    
    similarity_ratio = repeated / total if total > 0 else 0
    score = min(100, similarity_ratio * 500)  # put big 
    
    # CheckHighDegreeSimilarReview for 
    similar_pairs = 0
    for i, c1 in enumerate(contents):
        for c2 in contents[i+1:]:
            if len(c1) > 20 and len(c2) > 20:
                # Simple Jaccard SimilarDegree
                set1, set2 = set(c1.split()), set(c2.split())
                if set1 and set2:
                    jaccard = len(set1 & set2) / len(set1 | set2)
                    if jaccard > 0.5:
                        similar_pairs += 1
    
    if similar_pairs > 0:
        score = max(score, 50 + similar_pairs * 10)
    
    score = min(100, score)
    
    if score < 30:
        status = "✅"
    elif score < 60:
        status = "⚠️"
    else:
        status = "🔴"
    
    return DimensionResult(
        name="Content Similarity",
        name_zh="Similar contentDegree",
        score=round(score, 1),
        status=status,
        detail=f"Found {similar_pairs} similar review pairs",
        detail_zh=f"Found {similar_pairs} groupHighDegreeSimilarReview",
        weight=0.20
    )


def detect_time_clustering(reviews: List[Review]) -> Optional[DimensionResult]:
    """DetectionTimeAggregate"""
    dated_reviews = [r for r in reviews if r.has_date]
    
    if len(dated_reviews) < 5:
        return None  # DataInsufficient
    
    # ParseDayPeriod
    dates = []
    for r in dated_reviews:
        try:
            # TryMultipleFormat
            for fmt in ['%Y-%m-%d', '%B %d, %Y', '%d %B %Y', '%m/%d/%Y']:
                try:
                    dates.append(datetime.strptime(r.date, fmt))
                    break
                except:
                    continue
        except:
            pass
    
    if len(dates) < 5:
        return None
    
    dates.sort()
    
    # CalculateAdjacentReviewTimeseparate
    intervals = [(dates[i+1] - dates[i]).days for i in range(len(dates)-1)]
    
    if not intervals:
        return None
    
    # DetectionAggregate： short Time inside Large amountReview
    short_intervals = sum(1 for i in intervals if i <= 1)  # 1days inside 
    clustering_ratio = short_intervals / len(intervals)
    
    # Detection 48h Outbreak
    burst_count = 0
    window_size = 2  # 2daysWindow
    for i in range(len(dates) - 1):
        count_in_window = sum(1 for d in dates if 0 <= (d - dates[i]).days <= window_size)
        burst_count = max(burst_count, count_in_window)
    
    score = 0
    if clustering_ratio > 0.5:
        score += 40
    if clustering_ratio > 0.7:
        score += 20
    if burst_count > len(dates) * 0.3:
        score += 30
    
    score = min(100, score)
    
    if score < 30:
        status = "✅"
    elif score < 60:
        status = "⚠️"
    else:
        status = "🔴"
    
    return DimensionResult(
        name="Time Clustering",
        name_zh="TimeAggregate",
        score=round(score, 1),
        status=status,
        detail=f"Max {burst_count} reviews in 48h window, {clustering_ratio*100:.0f}% within 1 day",
        detail_zh=f"48h inside  most  many  {burst_count} itemReview，{clustering_ratio*100:.0f}%  in 1days inside ",
        weight=0.25
    )


def detect_rating_distribution(reviews: List[Review]) -> Optional[DimensionResult]:
    """DetectionRating Distribution"""
    rated_reviews = [r for r in reviews if r.has_rating]
    
    if len(rated_reviews) < 5:
        return None
    
    ratings = [r.rating for r in rated_reviews]
    rating_counts = Counter(ratings)
    total = len(ratings)
    
    # CalculateDistribution
    five_star_ratio = rating_counts.get(5, 0) / total
    one_star_ratio = rating_counts.get(1, 0) / total
    extreme_ratio = five_star_ratio + one_star_ratio
    
    # Natural distribution usually is：5star ~50-60%, 4star ~20%, 3star ~10%, 2star ~5%, 1star ~10%
    # Abnormal case： all 5star、Polarized
    
    score = 0
    anomaly_detail = []
    
    if five_star_ratio > 0.85:
        score += 50
        anomaly_detail.append(f"{five_star_ratio*100:.0f}% 5-star (abnormal)")
    elif five_star_ratio > 0.75:
        score += 30
        anomaly_detail.append(f"{five_star_ratio*100:.0f}% 5-star (suspicious)")
    
    # Polarized
    if extreme_ratio > 0.9 and one_star_ratio > 0.1:
        score += 30
        anomaly_detail.append("Polarized distribution")
    
    # MissingMiddleRating
    mid_ratings = sum(rating_counts.get(r, 0) for r in [2, 3, 4])
    if mid_ratings / total < 0.1 and total > 10:
        score += 20
        anomaly_detail.append("Missing mid-range ratings")
    
    score = min(100, score)
    
    if score < 30:
        status = "✅"
    elif score < 60:
        status = "⚠️"
    else:
        status = "🔴"
    
    dist_str = ", ".join([f"{r}★:{rating_counts.get(r,0)}" for r in [5,4,3,2,1]])
    
    return DimensionResult(
        name="Rating Distribution",
        name_zh="Rating Distribution",
        score=round(score, 1),
        status=status,
        detail=f"Distribution: {dist_str}. {'; '.join(anomaly_detail) if anomaly_detail else 'Normal'}",
        detail_zh=f"Distribution: {dist_str}。{'; '.join(anomaly_detail) if anomaly_detail else 'Normal'}",
        weight=0.20
    )


def detect_vp_ratio(reviews: List[Review]) -> Optional[DimensionResult]:
    """Detection Verified Purchase Ratio"""
    vp_reviews = [r for r in reviews if r.has_vp]
    
    if len(vp_reviews) < 5:
        return None
    
    vp_count = sum(1 for r in vp_reviews if r.verified_purchase)
    vp_ratio = vp_count / len(vp_reviews)
    
    # Normal VP Ratio should  should  in  60-80%
    score = 0
    if vp_ratio < 0.4:
        score = 70
    elif vp_ratio < 0.5:
        score = 50
    elif vp_ratio < 0.6:
        score = 30
    else:
        score = 10
    
    if score < 30:
        status = "✅"
    elif score < 60:
        status = "⚠️"
    else:
        status = "🔴"
    
    return DimensionResult(
        name="Verified Purchase Ratio",
        name_zh="VPRatio",
        score=round(score, 1),
        status=status,
        detail=f"{vp_ratio*100:.0f}% verified purchase ({vp_count}/{len(vp_reviews)})",
        detail_zh=f"{vp_ratio*100:.0f}%  already VerifyPurchase ({vp_count}/{len(vp_reviews)})",
        weight=0.15
    )


def detect_review_length(reviews: List[Review]) -> DimensionResult:
    """DetectionReviewDegreeDistribution"""
    lengths = [len(r.content) for r in reviews]
    
    if not lengths:
        return DimensionResult(
            name="Review Length",
            name_zh="ReviewDegree",
            score=0,
            status="✅",
            detail="No reviews",
            detail_zh="noReview",
            weight=0.05
        )
    
    avg_length = sum(lengths) / len(lengths)
    short_count = sum(1 for l in lengths if l < 50)
    short_ratio = short_count / len(lengths)
    
    # DetectionTemplate（DegreeHighDegreeConsistent）
    if len(lengths) > 5:
        length_std = (sum((l - avg_length) ** 2 for l in lengths) / len(lengths)) ** 0.5
        cv = length_std / avg_length if avg_length > 0 else 0  # Variation coefficient
    else:
        cv = 1
    
    score = 0
    if short_ratio > 0.7:
        score += 40
    if cv < 0.3 and len(lengths) > 10:  # DegreeToo consistent
        score += 40
    
    score = min(100, score)
    
    if score < 30:
        status = "✅"
    elif score < 60:
        status = "⚠️"
    else:
        status = "🔴"
    
    return DimensionResult(
        name="Review Length",
        name_zh="ReviewDegree",
        score=round(score, 1),
        status=status,
        detail=f"Avg length: {avg_length:.0f} chars, {short_ratio*100:.0f}% short (<50)",
        detail_zh=f"AverageDegree: {avg_length:.0f} Character, {short_ratio*100:.0f}%   short  (<50)",
        weight=0.05
    )


def detect_keywords(reviews: List[Review]) -> DimensionResult:
    """DetectionFake ordersKeywords"""
    # Common fake ordersReviewKeywords
    suspicious_keywords = [
        'received free', 'free product', 'in exchange', 'honest review',
        'discount code', 'promotional', 'gifted', 'complimentary',
        'five stars', '5 stars', 'best ever', 'amazing product',
        'highly recommend', 'must buy', 'perfect product',
        #  in text
        'Positive ReviewCashback', 'Five starPositive Review', 'Free trial', 'Gift'
    ]
    
    keyword_hits = 0
    for review in reviews:
        content_lower = review.content.lower()
        for keyword in suspicious_keywords:
            if keyword.lower() in content_lower:
                keyword_hits += 1
                break
    
    hit_ratio = keyword_hits / len(reviews) if reviews else 0
    score = min(100, hit_ratio * 200)
    
    if score < 30:
        status = "✅"
    elif score < 60:
        status = "⚠️"
    else:
        status = "🔴"
    
    return DimensionResult(
        name="Suspicious Keywords",
        name_zh="SuspiciousKeywords",
        score=round(score, 1),
        status=status,
        detail=f"{keyword_hits} reviews contain suspicious keywords",
        detail_zh=f"{keyword_hits} itemReviewPackagecontainSuspiciousKeywords",
        weight=0.05
    )


def identify_suspicious_reviews(reviews: List[Review], dimensions: List[DimensionResult]) -> List[SuspiciousReview]:
    """IdentifyHigh RiskReview"""
    suspicious = []
    
    for review in reviews:
        risk_score = 0
        reasons = []
        reasons_zh = []
        
        #  short Review
        if len(review.content) < 30:
            risk_score += 20
            reasons.append("Very short review")
            reasons_zh.append("Review  short ")
        
        # non VP
        if review.has_vp and not review.verified_purchase:
            risk_score += 25
            reasons.append("Not verified purchase")
            reasons_zh.append("nonVerifyPurchase")
        
        # ExtremeRating + TemplateContent
        if review.has_rating and review.rating == 5:
            generic_phrases = ['great', 'amazing', 'perfect', 'love it', 'best', 'excellent']
            if any(p in review.content.lower() for p in generic_phrases) and len(review.content) < 100:
                risk_score += 30
                reasons.append("Generic 5-star template")
                reasons_zh.append("Template5starPositive Review")
        
        # SuspiciousKeywords
        suspicious_keywords = ['received free', 'in exchange', 'honest review', 'discount']
        if any(k in review.content.lower() for k in suspicious_keywords):
            risk_score += 35
            reasons.append("Contains incentivized review keywords")
            reasons_zh.append("Contains incentiveReviewKeywords")
        
        if risk_score >= 40:
            suspicious.append(SuspiciousReview(
                content=review.content[:100] + "..." if len(review.content) > 100 else review.content,
                risk_score=min(100, risk_score),
                reasons=reasons,
                reasons_zh=reasons_zh
            ))
    
    #  by RiskScoreSort
    suspicious.sort(key=lambda x: x.risk_score, reverse=True)
    return suspicious[:10]  # Top 10


def determine_analysis_level(reviews: List[Review]) -> Tuple[AnalysisLevel, List[str], List[str]]:
    """ConfirmAnalyzelayerLevelAndMissing field"""
    available = ["content"]
    missing = []
    
    has_rating = any(r.has_rating for r in reviews)
    has_date = any(r.has_date for r in reviews)
    has_vp = any(r.has_vp for r in reviews)
    has_reviewer = any(r.reviewer_name for r in reviews)
    
    if has_rating:
        available.append("rating")
    else:
        missing.append("rating")
    
    if has_date:
        available.append("date")
    else:
        missing.append("date")
    
    if has_vp:
        available.append("verified_purchase")
    else:
        missing.append("verified_purchase")
    
    if has_reviewer:
        available.append("reviewer_info")
    else:
        missing.append("reviewer_info")
    
    # Determine levelLevel
    if has_rating and has_date and has_vp:
        level = AnalysisLevel.L4_FULL
    elif has_rating and has_date:
        level = AnalysisLevel.L3_SCORED
    elif has_date:
        level = AnalysisLevel.L2_TIMED
    else:
        level = AnalysisLevel.L1_BASIC
    
    return level, available, missing


def generate_deepening_hints(missing: List[str]) -> Tuple[List[str], List[str]]:
    """GenerateDeepenHint"""
    hints_en = []
    hints_zh = []
    
    hint_map = {
        "rating": (
            "Add star ratings → Unlock 'Rating Distribution Analysis'",
            "Add starLevelRating → Unlock「Rating DistributionAnalyze」"
        ),
        "date": (
            "Add review dates → Unlock 'Time Clustering Detection'",
            "SupplementReviewDayPeriod → Unlock「TimeAggregateDetection」"
        ),
        "verified_purchase": (
            "Add VP status → Unlock 'Verified Purchase Analysis'",
            "SupplementVPStatus → Unlock「PurchaseVerifyAnalyze」"
        ),
        "reviewer_info": (
            "Add reviewer info → Unlock 'Account Profile Analysis'",
            "SupplementReviewpersonInformation → Unlock「Account profileAnalyze」"
        ),
    }
    
    for field in missing:
        if field in hint_map:
            hints_en.append(hint_map[field][0])
            hints_zh.append(hint_map[field][1])
    
    return hints_en, hints_zh


def analyze_reviews(reviews: List[Review], asin: str = "UNKNOWN") -> AnalysisResult:
    """MainAnalyzeFunction"""
    if not reviews:
        return AnalysisResult(
            asin=asin,
            total_reviews=0,
            analysis_level=AnalysisLevel.L1_BASIC,
            authenticity_score=50,
            risk_level=RiskLevel.MEDIUM,
            dimensions=[],
            suspicious_reviews=[],
            available_fields=[],
            missing_fields=["content"],
            deepening_hints=["Please provide review data"],
            deepening_hints_zh=["Please provideReview Countdata"],
            summary="No reviews to analyze",
            summary_zh="noReview Countdata"
        )
    
    # ConfirmAnalyzelayerLevel
    level, available, missing = determine_analysis_level(reviews)
    hints_en, hints_zh = generate_deepening_hints(missing)
    
    # ExecuteDetection
    dimensions = []
    
    # L1: BasicDetection (AlwaysExecute)
    dimensions.append(detect_content_similarity(reviews))
    dimensions.append(detect_review_length(reviews))
    dimensions.append(detect_keywords(reviews))
    
    # L2+: TimeDetection
    time_result = detect_time_clustering(reviews)
    if time_result:
        dimensions.append(time_result)
    
    # L3+: RatingDetection
    rating_result = detect_rating_distribution(reviews)
    if rating_result:
        dimensions.append(rating_result)
    
    # L4: VPDetection
    vp_result = detect_vp_ratio(reviews)
    if vp_result:
        dimensions.append(vp_result)
    
    # CalculateComprehensiveScore (WeightedAverageSuspiciousDegree，ThenConvert toAuthenticityScore)
    total_weight = sum(d.weight for d in dimensions)
    if total_weight > 0:
        suspicion_score = sum(d.score * d.weight for d in dimensions) / total_weight
    else:
        suspicion_score = 50
    
    authenticity_score = max(0, min(100, 100 - suspicion_score))
    
    # ConfirmRiskGrade
    if authenticity_score >= 70:
        risk_level = RiskLevel.LOW
    elif authenticity_score >= 50:
        risk_level = RiskLevel.MEDIUM
    elif authenticity_score >= 30:
        risk_level = RiskLevel.HIGH
    else:
        risk_level = RiskLevel.CRITICAL
    
    # IdentifySuspiciousReview
    suspicious = identify_suspicious_reviews(reviews, dimensions)
    
    # GenerateSummary
    risk_text = {
        RiskLevel.LOW: ("Low risk - Reviews appear authentic", "Low Risk - ReviewLooks likeReal"),
        RiskLevel.MEDIUM: ("Medium risk - Some concerns detected", " in  etcRisk - FoundSomeConcern"),
        RiskLevel.HIGH: ("High risk - Multiple red flags", "High Risk -  many DangerSignal"),
        RiskLevel.CRITICAL: ("Critical risk - Likely fake reviews", "CriticalRisk - MayExistLarge amountFakeReview"),
    }
    
    summary_en = f"{risk_text[risk_level][0]}. Analyzed {len(reviews)} reviews at {level.value} level."
    summary_zh = f"{risk_text[risk_level][1]}。Analyze  {len(reviews)} itemReview，AnalyzelayerLevel {level.value}。"
    
    return AnalysisResult(
        asin=asin,
        total_reviews=len(reviews),
        analysis_level=level,
        authenticity_score=round(authenticity_score),
        risk_level=risk_level,
        dimensions=dimensions,
        suspicious_reviews=suspicious,
        available_fields=available,
        missing_fields=missing,
        deepening_hints=hints_en,
        deepening_hints_zh=hints_zh,
        summary=summary_en,
        summary_zh=summary_zh
    )


# ============================================================
# OutputFormat
# ============================================================

def format_report(result: AnalysisResult, lang: str = "en") -> str:
    """FormatReport"""
    risk_icons = {
        RiskLevel.LOW: "✅",
        RiskLevel.MEDIUM: "⚠️",
        RiskLevel.HIGH: "🔴",
        RiskLevel.CRITICAL: "💀",
    }
    
    if lang == "zh":
        lines = [
            f"📊 **ReviewAuthenticityAnalyzeReport**",
            f"",
            f"**ASIN**: {result.asin}",
            f"**Review Count**: {result.total_reviews}",
            f"**AnalyzelayerLevel**: {result.analysis_level.value}",
            f"",
            f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━",
            f"",
            f"## AuthenticityRating: {result.authenticity_score}/100 {risk_icons[result.risk_level]}",
            f"",
            f"{result.summary_zh}",
            f"",
            f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━",
            f"",
            f"## DetectiondimensionDegree",
            f"",
        ]
        
        for d in result.dimensions:
            lines.append(f"{d.status} **{d.name_zh}**: {d.score:.0f}/100")
            lines.append(f"   {d.detail_zh}")
            lines.append("")
        
        if result.suspicious_reviews:
            lines.append("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
            lines.append("")
            lines.append(f"## High RiskReview (Top {len(result.suspicious_reviews)})")
            lines.append("")
            for i, sr in enumerate(result.suspicious_reviews[:5], 1):
                lines.append(f"**{i}. Risk {sr.risk_score:.0f}%**")
                lines.append(f'   "{sr.content}"')
                lines.append(f"   Reason: {', '.join(sr.reasons_zh)}")
                lines.append("")
        
        if result.deepening_hints_zh:
            lines.append("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
            lines.append("")
            lines.append("🔍 **Want more accurateAnalyze？Supplement with following info:**")
            lines.append("")
            for hint in result.deepening_hints_zh:
                lines.append(f"• {hint}")
    else:
        lines = [
            f"📊 **Review Authenticity Report**",
            f"",
            f"**ASIN**: {result.asin}",
            f"**Reviews**: {result.total_reviews}",
            f"**Analysis Level**: {result.analysis_level.value}",
            f"",
            f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━",
            f"",
            f"## Authenticity Score: {result.authenticity_score}/100 {risk_icons[result.risk_level]}",
            f"",
            f"{result.summary}",
            f"",
            f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━",
            f"",
            f"## Detection Dimensions",
            f"",
        ]
        
        for d in result.dimensions:
            lines.append(f"{d.status} **{d.name}**: {d.score:.0f}/100")
            lines.append(f"   {d.detail}")
            lines.append("")
        
        if result.suspicious_reviews:
            lines.append("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
            lines.append("")
            lines.append(f"## Suspicious Reviews (Top {len(result.suspicious_reviews)})")
            lines.append("")
            for i, sr in enumerate(result.suspicious_reviews[:5], 1):
                lines.append(f"**{i}. Risk {sr.risk_score:.0f}%**")
                lines.append(f'   "{sr.content}"')
                lines.append(f"   Reasons: {', '.join(sr.reasons)}")
                lines.append("")
        
        if result.deepening_hints:
            lines.append("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
            lines.append("")
            lines.append("🔍 **Want more accurate analysis? Add the following:**")
            lines.append("")
            for hint in result.deepening_hints:
                lines.append(f"• {hint}")
    
    return "\n".join(lines)


# ============================================================
# Parse
# ============================================================

def parse_simple_reviews(text: str) -> List[Review]:
    """ParseSimpleTextFormatReview"""
    reviews = []
    
    # TrySplit by paragraph
    paragraphs = re.split(r'\n\n+', text.strip())
    
    for para in paragraphs:
        para = para.strip()
        if len(para) > 10:
            review = Review(content=para)
            
            # TryExtract starLevel
            star_match = re.search(r'(\d)\s*(?:star|★|⭐)', para, re.I)
            if star_match:
                review.rating = int(star_match.group(1))
            
            # TryExtractDayPeriod
            date_match = re.search(r'(\d{4}-\d{2}-\d{2}|\w+\s+\d{1,2},?\s+\d{4})', para)
            if date_match:
                review.date = date_match.group(1)
            
            # TryExtract VP
            if 'verified purchase' in para.lower() or 'VP' in para:
                review.verified_purchase = True
            elif 'not verified' in para.lower():
                review.verified_purchase = False
            
            reviews.append(review)
    
    return reviews


# ============================================================
# CLI Entry Point
# ============================================================

def main():
    """CLI Entry Point"""
    # TestData
    test_reviews = [
        Review(content="Great product! Works perfectly. Highly recommend to everyone.", rating=5, verified_purchase=True, date="2024-01-15"),
        Review(content="Amazing! Best purchase ever. Love it!", rating=5, verified_purchase=False, date="2024-01-15"),
        Review(content="Great product! Works perfectly. Must buy!", rating=5, verified_purchase=False, date="2024-01-16"),
        Review(content="Received free product in exchange for honest review. It's good.", rating=5, verified_purchase=False, date="2024-01-16"),
        Review(content="Excellent quality, fast shipping. Very satisfied with purchase.", rating=5, verified_purchase=True, date="2024-01-20"),
        Review(content="Not as described. Cheap quality.", rating=1, verified_purchase=True, date="2024-02-01"),
        Review(content="Perfect!", rating=5, verified_purchase=False, date="2024-01-16"),
        Review(content="Good value for money. Does what it says.", rating=4, verified_purchase=True, date="2024-02-15"),
        Review(content="Five stars! Amazing product!", rating=5, verified_purchase=False, date="2024-01-17"),
        Review(content="Decent product for the price.", rating=3, verified_purchase=True, date="2024-03-01"),
    ]
    
    # If command lineInput
    if len(sys.argv) > 1:
        arg = sys.argv[1]
        if arg == "--demo":
            pass  # UseTestData
        elif arg.startswith('['):
            # JSON group
            try:
                data = json.loads(arg)
                test_reviews = [Review(**r) for r in data]
            except:
                pass
        else:
            # pureText
            test_reviews = parse_simple_reviews(arg)
    
    result = analyze_reviews(test_reviews, asin="B08XXXXX")
    
    # Default ChineseOutput
    lang = "zh" if "--zh" in sys.argv else "en"
    print(format_report(result, lang))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Review Parser - Multi-format Input Parser
ReviewParse -  many FormatInputParse

SupportFormat:
- pureText (Paragraph separator)
- JSON
- CSV
- TSV
- Markdown Table

Version: 1.0.0
"""

import json
import csv
import re
from typing import List, Optional
from dataclasses import dataclass
from io import StringIO


@dataclass
class Review:
    """Single itemReview"""
    content: str
    rating: Optional[int] = None
    date: Optional[str] = None
    reviewer_name: Optional[str] = None
    verified_purchase: Optional[bool] = None
    helpful_votes: Optional[int] = None
    reviewer_reviews_count: Optional[int] = None


class ReviewParser:
    """ many FormatReviewParse"""
    
    @staticmethod
    def detect_format(text: str) -> str:
        """DetectionInputFormat"""
        text = text.strip()
        
        # JSON group
        if text.startswith('[') and text.endswith(']'):
            try:
                json.loads(text)
                return 'json'
            except:
                pass
        
        # JSON Object
        if text.startswith('{'):
            try:
                json.loads(text)
                return 'json_single'
            except:
                pass
        
        # CSV (CheckComma separated with header)
        lines = text.split('\n')
        if len(lines) > 1:
            first_line = lines[0].lower()
            if ',' in first_line and any(h in first_line for h in ['content', 'review', 'rating', 'date', 'text']):
                return 'csv'
        
        # TSV
        if len(lines) > 1 and '\t' in lines[0]:
            return 'tsv'
        
        # Markdown Table
        if '|' in text and '---' in text:
            return 'markdown'
        
        # DefaultpureText
        return 'text'
    
    @staticmethod
    def parse(text: str) -> List[Review]:
        """ParseReview"""
        format_type = ReviewParser.detect_format(text)
        
        parsers = {
            'json': ReviewParser.parse_json,
            'json_single': ReviewParser.parse_json_single,
            'csv': ReviewParser.parse_csv,
            'tsv': ReviewParser.parse_tsv,
            'markdown': ReviewParser.parse_markdown,
            'text': ReviewParser.parse_text,
        }
        
        parser = parsers.get(format_type, ReviewParser.parse_text)
        return parser(text)
    
    @staticmethod
    def parse_json(text: str) -> List[Review]:
        """Parse JSON group"""
        try:
            data = json.loads(text)
            reviews = []
            for item in data:
                review = Review(
                    content=item.get('content', item.get('text', item.get('review', ''))),
                    rating=item.get('rating', item.get('stars', item.get('star'))),
                    date=item.get('date', item.get('review_date')),
                    reviewer_name=item.get('reviewer_name', item.get('author', item.get('name'))),
                    verified_purchase=item.get('verified_purchase', item.get('vp', item.get('verified'))),
                    helpful_votes=item.get('helpful_votes', item.get('helpful')),
                )
                if review.content:
                    reviews.append(review)
            return reviews
        except:
            return []
    
    @staticmethod
    def parse_json_single(text: str) -> List[Review]:
        """ParseSingle JSON Object"""
        try:
            item = json.loads(text)
            review = Review(
                content=item.get('content', item.get('text', item.get('review', ''))),
                rating=item.get('rating', item.get('stars')),
                date=item.get('date'),
                reviewer_name=item.get('reviewer_name', item.get('author')),
                verified_purchase=item.get('verified_purchase', item.get('vp')),
            )
            return [review] if review.content else []
        except:
            return []
    
    @staticmethod
    def parse_csv(text: str) -> List[Review]:
        """Parse CSV"""
        reviews = []
        try:
            reader = csv.DictReader(StringIO(text))
            for row in reader:
                # TryMultipleField name
                content = (
                    row.get('content') or row.get('Content') or
                    row.get('review') or row.get('Review') or
                    row.get('text') or row.get('Text') or
                    row.get('review_text') or row.get('body') or ''
                )
                
                rating_str = (
                    row.get('rating') or row.get('Rating') or
                    row.get('stars') or row.get('Stars') or
                    row.get('star') or ''
                )
                rating = None
                if rating_str:
                    try:
                        rating = int(float(rating_str))
                    except:
                        pass
                
                date = (
                    row.get('date') or row.get('Date') or
                    row.get('review_date') or row.get('created_at') or None
                )
                
                vp_str = (
                    row.get('verified_purchase') or row.get('vp') or
                    row.get('verified') or row.get('VP') or ''
                )
                vp = None
                if vp_str:
                    vp = vp_str.lower() in ['true', 'yes', '1', 'y', 'verified']
                
                reviewer = (
                    row.get('reviewer_name') or row.get('author') or
                    row.get('name') or row.get('reviewer') or None
                )
                
                if content:
                    reviews.append(Review(
                        content=content,
                        rating=rating,
                        date=date,
                        reviewer_name=reviewer,
                        verified_purchase=vp,
                    ))
        except Exception as e:
            print(f"CSV parse error: {e}")
        
        return reviews
    
    @staticmethod
    def parse_tsv(text: str) -> List[Review]:
        """Parse TSV"""
        # Convert to CSV FormatProcess
        csv_text = text.replace('\t', ',')
        return ReviewParser.parse_csv(csv_text)
    
    @staticmethod
    def parse_markdown(text: str) -> List[Review]:
        """Parse Markdown Table"""
        reviews = []
        lines = text.strip().split('\n')
        
        # find to Tablehead
        header_line = None
        data_start = 0
        for i, line in enumerate(lines):
            if '|' in line and '---' not in line:
                header_line = line
                data_start = i + 2  # Skip separator lines
                break
        
        if not header_line:
            return []
        
        # ParseTablehead
        headers = [h.strip().lower() for h in header_line.split('|') if h.strip()]
        
        # ParseDatarow
        for line in lines[data_start:]:
            if '|' not in line:
                continue
            
            cells = [c.strip() for c in line.split('|') if c.strip()]
            if len(cells) < len(headers):
                continue
            
            row = dict(zip(headers, cells))
            
            content = row.get('content', row.get('review', row.get('text', '')))
            rating = None
            rating_str = row.get('rating', row.get('stars', ''))
            if rating_str:
                try:
                    rating = int(float(rating_str.replace('★', '').strip()))
                except:
                    pass
            
            if content:
                reviews.append(Review(
                    content=content,
                    rating=rating,
                    date=row.get('date'),
                    reviewer_name=row.get('author', row.get('reviewer')),
                ))
        
        return reviews
    
    @staticmethod
    def parse_text(text: str) -> List[Review]:
        """ParsepureText"""
        reviews = []
        
        # TrySplit by paragraph
        paragraphs = re.split(r'\n\n+', text.strip())
        
        for para in paragraphs:
            para = para.strip()
            if len(para) < 10:
                continue
            
            review = Review(content=para)
            
            # TryExtract starLevel
            star_patterns = [
                r'(\d)\s*(?:star|stars|★|⭐)',
                r'(?:star|stars|rating)[:\s]*(\d)',
                r'^(\d)★',
                r'^(\d)\s*star',
            ]
            for pattern in star_patterns:
                match = re.search(pattern, para, re.I)
                if match:
                    review.rating = int(match.group(1))
                    break
            
            # TryExtractDayPeriod
            date_patterns = [
                r'(\d{4}-\d{2}-\d{2})',
                r'(\w+\s+\d{1,2},?\s+\d{4})',
                r'(\d{1,2}/\d{1,2}/\d{4})',
            ]
            for pattern in date_patterns:
                match = re.search(pattern, para)
                if match:
                    review.date = match.group(1)
                    break
            
            # TryExtract VP Status
            if re.search(r'verified\s*purchase', para, re.I):
                review.verified_purchase = True
            elif re.search(r'not\s*verified', para, re.I):
                review.verified_purchase = False
            
            reviews.append(review)
        
        return reviews


def parse_reviews(text: str) -> List[Review]:
    """ConvenientFunction"""
    return ReviewParser.parse(text)


# CLI
if __name__ == "__main__":
    import sys
    
    if len(sys.argv) > 1:
        text = sys.argv[1]
    else:
        # TestData
        text = """
content,rating,date,verified_purchase
"Great product!",5,2024-01-15,true
"Not good",2,2024-01-16,false
"Amazing!",5,2024-01-17,true
"""
    
    reviews = parse_reviews(text)
    print(f"Parsed {len(reviews)} reviews:")
    for r in reviews:
        print(f"  - {r.rating}★ | VP:{r.verified_purchase} | {r.content[:50]}...")

#!/usr/bin/env python3
"""
Review Analysis HTML Report Generator
ReviewAnalyze HTML ReportGenerate

Features:
- CanViewTable (Chart.js)
- Responsive layout
-  deep colorMaintopic
- InteractiveDataDisplay

Version: 1.0.0
"""

import json
from typing import List, Dict, Any
from datetime import datetime


def generate_html_report(
    asin: str,
    authenticity_score: int,
    risk_level: str,
    dimensions: List[Dict[str, Any]],
    suspicious_reviews: List[Dict[str, Any]],
    total_reviews: int,
    analysis_level: str,
    summary: str,
    output_path: str = "review_analysis_report.html"
) -> str:
    """Generate HTML Report"""
    
    # RiskGradeColor
    risk_colors = {
        "low": "#10b981",      # Green
        "medium": "#f59e0b",   # Yellow
        "high": "#ef4444",     # Red
        "critical": "#7c2d12", #  deep red
    }
    
    risk_color = risk_colors.get(risk_level.lower(), "#6b7280")
    
    # dimensionDegreeData (use at chartTable)
    dimension_labels = json.dumps([d.get('name', d.get('name_zh', '')) for d in dimensions])
    dimension_scores = json.dumps([d.get('score', 0) for d in dimensions])
    
    # SuspiciousReview JSON
    suspicious_json = json.dumps(suspicious_reviews[:10], ensure_ascii=False, default=str)
    
    html = f"""<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Review Analysis Report - {asin}</title>
    <script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
    <style>
        * {{
            margin: 0;
            padding: 0;
            box-sizing: border-box;
        }}
        
        body {{
            font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
            background: linear-gradient(135deg, #1a1a2e 0%, #16213e 100%);
            color: #e4e4e7;
            min-height: 100vh;
            padding: 20px;
        }}
        
        .container {{
            max-width: 1200px;
            margin: 0 auto;
        }}
        
        .header {{
            text-align: center;
            padding: 40px 20px;
            background: rgba(255,255,255,0.05);
            border-radius: 16px;
            margin-bottom: 30px;
        }}
        
        .header h1 {{
            font-size: 2rem;
            background: linear-gradient(90deg, #a78bfa, #818cf8);
            -webkit-background-clip: text;
            -webkit-text-fill-color: transparent;
            margin-bottom: 10px;
        }}
        
        .header .asin {{
            color: #a1a1aa;
            font-size: 1.1rem;
        }}
        
        .score-card {{
            background: rgba(255,255,255,0.08);
            border-radius: 16px;
            padding: 40px;
            text-align: center;
            margin-bottom: 30px;
            border: 1px solid rgba(255,255,255,0.1);
        }}
        
        .score-number {{
            font-size: 5rem;
            font-weight: bold;
            color: {risk_color};
            line-height: 1;
        }}
        
        .score-label {{
            font-size: 1.2rem;
            color: #a1a1aa;
            margin-top: 10px;
        }}
        
        .risk-badge {{
            display: inline-block;
            padding: 8px 20px;
            border-radius: 20px;
            background: {risk_color};
            color: white;
            font-weight: 600;
            margin-top: 15px;
            text-transform: uppercase;
        }}
        
        .grid {{
            display: grid;
            grid-template-columns: repeat(auto-fit, minmax(300px, 1fr));
            gap: 20px;
            margin-bottom: 30px;
        }}
        
        .card {{
            background: rgba(255,255,255,0.08);
            border-radius: 12px;
            padding: 24px;
            border: 1px solid rgba(255,255,255,0.1);
        }}
        
        .card h3 {{
            font-size: 1.1rem;
            margin-bottom: 20px;
            color: #a78bfa;
        }}
        
        .stat {{
            display: flex;
            justify-content: space-between;
            padding: 12px 0;
            border-bottom: 1px solid rgba(255,255,255,0.1);
        }}
        
        .stat:last-child {{
            border-bottom: none;
        }}
        
        .stat-label {{
            color: #a1a1aa;
        }}
        
        .stat-value {{
            font-weight: 600;
        }}
        
        .chart-container {{
            position: relative;
            height: 300px;
            width: 100%;
        }}
        
        .dimension-item {{
            display: flex;
            align-items: center;
            padding: 12px 0;
            border-bottom: 1px solid rgba(255,255,255,0.1);
        }}
        
        .dimension-icon {{
            width: 30px;
            font-size: 1.2rem;
        }}
        
        .dimension-name {{
            flex: 1;
        }}
        
        .dimension-score {{
            font-weight: 600;
            width: 80px;
            text-align: right;
        }}
        
        .suspicious-item {{
            background: rgba(239, 68, 68, 0.1);
            border: 1px solid rgba(239, 68, 68, 0.3);
            border-radius: 8px;
            padding: 16px;
            margin-bottom: 12px;
        }}
        
        .suspicious-header {{
            display: flex;
            justify-content: space-between;
            margin-bottom: 8px;
        }}
        
        .suspicious-risk {{
            color: #ef4444;
            font-weight: 600;
        }}
        
        .suspicious-content {{
            color: #d4d4d8;
            font-style: italic;
            margin-bottom: 8px;
        }}
        
        .suspicious-reasons {{
            color: #a1a1aa;
            font-size: 0.9rem;
        }}
        
        .summary {{
            background: rgba(167, 139, 250, 0.1);
            border: 1px solid rgba(167, 139, 250, 0.3);
            border-radius: 12px;
            padding: 20px;
            margin-bottom: 30px;
        }}
        
        .footer {{
            text-align: center;
            padding: 20px;
            color: #71717a;
            font-size: 0.9rem;
        }}
        
        @media (max-width: 768px) {{
            .score-number {{
                font-size: 3rem;
            }}
            .grid {{
                grid-template-columns: 1fr;
            }}
        }}
    </style>
</head>
<body>
    <div class="container">
        <div class="header">
            <h1>📊 Review Authenticity Analysis</h1>
            <div class="asin">ASIN: {asin}</div>
        </div>
        
        <div class="score-card">
            <div class="score-number">{authenticity_score}</div>
            <div class="score-label">Authenticity Score (0-100)</div>
            <div class="risk-badge">{risk_level.upper()} RISK</div>
        </div>
        
        <div class="summary">
            <strong>Summary:</strong> {summary}
        </div>
        
        <div class="grid">
            <div class="card">
                <h3>📈 Overview</h3>
                <div class="stat">
                    <span class="stat-label">Total Reviews</span>
                    <span class="stat-value">{total_reviews}</span>
                </div>
                <div class="stat">
                    <span class="stat-label">Analysis Level</span>
                    <span class="stat-value">{analysis_level}</span>
                </div>
                <div class="stat">
                    <span class="stat-label">Suspicious Reviews</span>
                    <span class="stat-value">{len(suspicious_reviews)}</span>
                </div>
                <div class="stat">
                    <span class="stat-label">Generated</span>
                    <span class="stat-value">{datetime.now().strftime('%Y-%m-%d %H:%M')}</span>
                </div>
            </div>
            
            <div class="card">
                <h3>🎯 Detection Dimensions</h3>
                <div class="chart-container">
                    <canvas id="dimensionChart"></canvas>
                </div>
            </div>
        </div>
        
        <div class="card" style="margin-bottom: 30px;">
            <h3>🔍 Dimension Details</h3>
            {"".join([f'''
            <div class="dimension-item">
                <div class="dimension-icon">{d.get('status', '❓')}</div>
                <div class="dimension-name">{d.get('name', d.get('name_zh', ''))}</div>
                <div class="dimension-score" style="color: {'#10b981' if d.get('score', 0) < 30 else '#f59e0b' if d.get('score', 0) < 60 else '#ef4444'}">{d.get('score', 0):.0f}/100</div>
            </div>
            <div style="color: #a1a1aa; font-size: 0.9rem; padding-left: 30px; padding-bottom: 12px;">{d.get('detail', d.get('detail_zh', ''))}</div>
            ''' for d in dimensions])}
        </div>
        
        <div class="card">
            <h3>⚠️ Suspicious Reviews (Top {min(len(suspicious_reviews), 5)})</h3>
            <div id="suspiciousReviews"></div>
        </div>
        
        <div class="footer">
            <p>Generated by Amazon Review Checker | Nexscope AI</p>
            <p>This analysis is for reference only. Results may not be 100% accurate.</p>
        </div>
    </div>
    
    <script>
        // dimensionDegreeRadar chart
        const ctx = document.getElementById('dimensionChart').getContext('2d');
        new Chart(ctx, {{
            type: 'radar',
            data: {{
                labels: {dimension_labels},
                datasets: [{{
                    label: 'Suspicion Score',
                    data: {dimension_scores},
                    backgroundColor: 'rgba(167, 139, 250, 0.2)',
                    borderColor: 'rgba(167, 139, 250, 1)',
                    borderWidth: 2,
                    pointBackgroundColor: 'rgba(167, 139, 250, 1)',
                }}]
            }},
            options: {{
                responsive: true,
                maintainAspectRatio: false,
                scales: {{
                    r: {{
                        beginAtZero: true,
                        max: 100,
                        ticks: {{
                            color: '#a1a1aa',
                            backdropColor: 'transparent'
                        }},
                        grid: {{
                            color: 'rgba(255,255,255,0.1)'
                        }},
                        pointLabels: {{
                            color: '#e4e4e7',
                            font: {{ size: 11 }}
                        }}
                    }}
                }},
                plugins: {{
                    legend: {{
                        display: false
                    }}
                }}
            }}
        }});
        
        // SuspiciousReview
        const suspicious = {suspicious_json};
        const container = document.getElementById('suspiciousReviews');
        
        if (suspicious.length === 0) {{
            container.innerHTML = '<p style="color: #10b981;">No highly suspicious reviews detected.</p>';
        }} else {{
            suspicious.slice(0, 5).forEach((review, index) => {{
                container.innerHTML += `
                    <div class="suspicious-item">
                        <div class="suspicious-header">
                            <span>#${{index + 1}}</span>
                            <span class="suspicious-risk">Risk: ${{review.risk_score?.toFixed(0) || 'N/A'}}%</span>
                        </div>
                        <div class="suspicious-content">"${{review.content}}"</div>
                        <div class="suspicious-reasons">Reasons: ${{(review.reasons || review.reasons_zh || []).join(', ')}}</div>
                    </div>
                `;
            }});
        }}
    </script>
</body>
</html>
"""
    
    # SaveFile
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write(html)
    
    print(f"✅ HTML report saved to: {output_path}")
    return output_path


# CLI
if __name__ == "__main__":
    # TestData
    generate_html_report(
        asin="B08XXXXX",
        authenticity_score=66,
        risk_level="medium",
        dimensions=[
            {"name": "Content Similarity", "name_zh": "Similar contentDegree", "score": 24, "status": "✅", "detail": "Found 0 similar review pairs"},
            {"name": "Time Clustering", "name_zh": "TimeAggregate", "score": 70, "status": "🔴", "detail": "6 reviews in 48h window"},
            {"name": "Rating Distribution", "name_zh": "Rating Distribution", "score": 0, "status": "✅", "detail": "Normal distribution"},
            {"name": "VP Ratio", "name_zh": "VPRatio", "score": 30, "status": "⚠️", "detail": "50% verified purchase"},
            {"name": "Review Length", "name_zh": "ReviewDegree", "score": 0, "status": "✅", "detail": "Normal length distribution"},
            {"name": "Suspicious Keywords", "name_zh": "SuspiciousKeywords", "score": 80, "status": "🔴", "detail": "4 reviews contain suspicious keywords"},
        ],
        suspicious_reviews=[
            {"content": "Perfect!", "risk_score": 75, "reasons": ["Very short", "Not VP", "Generic template"]},
            {"content": "Five stars! Amazing product!", "risk_score": 75, "reasons": ["Very short", "Not VP", "Generic template"]},
            {"content": "Received free product in exchange for honest review...", "risk_score": 60, "reasons": ["Not VP", "Incentivized review"]},
        ],
        total_reviews=10,
        analysis_level="L4",
        summary="Medium risk - Some concerns detected. Analyzed 10 reviews at L4 level.",
        output_path="review_analysis_report.html"
    )