comment_analytics.py

import random
import sys
# sys.path.append("/home/yswg/amazon_selection/")
import time
import traceback
import pandas as pd
import re
import nltk
# from utils.utils import Utils
from collections import Counter
from nltk.stem import WordNetLemmatizer
from nltk.corpus import stopwords
from nltk.sentiment import SentimentIntensityAnalyzer  # 情绪打分模型
from app.context import create_engine_db

pd.set_option('expand_frame_repr', False)  # 显示所有列
pd.set_option('display.max_colwidth', 100)  # 显示打印的字符数


class CommentAnalytics():

    def __init__(self, site_name='uk', asin='', phrases_counts=20):
        # super(CommentAnalytics, self).__init__()
        self.site_name = site_name
        self.asin = asin
        self.phrases_counts = phrases_counts
        # self.engine_pg = self.class_db.db_connection(db_type="pg", db_node="h15", site_name=self.site_name)
        self.engine_pg = create_engine_db(site_name=self.site_name)
        self.table_name_read = self.get_table_name_read_by_asin()
        self.table_name_save = f"{self.site_name}_asin_comment_analytics"
        self.df_asin = pd.DataFrame()
        self.df_save = pd.DataFrame()

    def init_refresh(self, site_name, asin, phrases_counts):
        self.site_name = site_name
        self.asin = asin
        self.phrases_counts = phrases_counts
        # self.engine_pg = self.class_db.db_connection(db_type="pg", db_node="h15", site_name=self.site_name)
        self.engine_pg = create_engine_db(site_name=self.site_name)
        self.table_name_read = self.get_table_name_read_by_asin()
        self.table_name_save = f"{self.site_name}_asin_comment_analytics"
        self.df_asin = pd.DataFrame()
        self.df_save = pd.DataFrame()

    # @staticmethod
    def _hd_clean_and_extract(self, text):
        # 去掉标点符号和表情包
        text = re.sub(r'[^\w\s]|_+', '', text)
        # 将文本转换成小写
        text = text.lower()
        # 加载停用词
        stop_words = set(stopwords.words('english'))
        # 将文本分词，并过滤停用词
        words = nltk.word_tokenize(text)
        words = [word for word in words if word not in stop_words]
        # 将词性标注统一为nltk库的标注方式
        pos_tags = []
        for word, tag in nltk.pos_tag(words):
            if tag.startswith('NN'):
                pos_tags.append((word, 'n'))
            elif tag.startswith('VB'):
                pos_tags.append((word, 'v'))
            elif tag.startswith('JJ'):
                pos_tags.append((word, 'a'))
            elif tag.startswith('R'):
                pos_tags.append((word, 'r'))
            else:
                pos_tags.append((word, None))
        # 使用WordNetLemmatizer进行单复数合并
        lemmatizer = WordNetLemmatizer()
        pos_tags = [(lemmatizer.lemmatize(word, pos) if pos else word, pos) for word, pos in pos_tags]
        # 使用nltk库的ngram函数进行短语提取，提取2-gram和3-gram
        phrases_list = []
        for i in range(2, 5):
            ngrams = nltk.ngrams(pos_tags, i)
            for ngram in ngrams:
                # 过滤掉不合法的短语，只保留名词短语和形容词短语
                if ngram[-1][1] == 'n' or ngram[-1][1] == 'a':
                    phrase = ' '.join([word[0] for word in ngram])
                    phrases_list.append(phrase)
        phrases_counter = Counter(phrases_list)
        # 按数量降序排序
        sorted_items = sorted(phrases_counter.items(), key=lambda x: x[1], reverse=True)
        keys_list, values_list = [], []
        for k, v in sorted_items:
            keys_list.append(k)
            values_list.append(str(v))
        if len(keys_list) >= self.phrases_counts:
            keys_list, values_list = keys_list[:self.phrases_counts], values_list[:self.phrases_counts]
        phrases_str = ",".join(keys_list)
        phrases_counts_str = ",".join(values_list)
        # 求和
        phrases_counts_total = sum(map(int, values_list))
        # 计算每个元素针对求和值的占比
        percentages = [(int(x) / phrases_counts_total) for x in values_list]
        # 将每个占比保留2位小数
        phrases_counts_rate_list = [str(round(x, 4)) for x in percentages]
        phrases_counts_rate_str = ",".join(phrases_counts_rate_list)
        # print("总计词频数量:", phrases_counts_total)
        # print("每个词组占总计词频的占比:", phrases_counts_rate_str)
        return phrases_list, phrases_str, phrases_counts_str, phrases_counts_rate_str

    def get_table_name_read_by_asin(self):
        if self.asin.lower()[:3] in ['b01', 'b02', 'b03', 'b04', 'b05', 'b06']:
            table_name_read = f'{self.site_name}_asin_comment_b01_b06'
        elif self.asin.lower()[:3] in ['b00', 'b07', 'b08', 'b09']:
            table_name_read = f'{self.site_name}_asin_comment_{self.asin.lower()[:3]}'
        else:
            table_name_read = f'{self.site_name}_asin_comment_other'
        print("asin, table_name_read:", self.asin, table_name_read)
        return table_name_read

    def read_data(self):
        sql = f"select asin, content, rating from {self.table_name_read} where asin='{self.asin}';"
        print("sql:", sql)
        self.df_asin = pd.read_sql(sql, con=self.engine_pg)
        print(self.df_asin.shape)
        print(self.df_asin.head())

    def handle_data(self):
        # 合并评论
        df_asin_good = self.df_asin.loc[self.df_asin.rating.isin([4, 5])].groupby(['asin']).agg(
            {"content": " ".join}).reset_index()
        df_asin_bad = self.df_asin.loc[self.df_asin.rating.isin([1, 2])].groupby(['asin']).agg(
            {"content": " ".join}).reset_index()
        df_asin_neu = self.df_asin.loc[self.df_asin.rating.isin([3])].groupby(['asin']).agg(
            {"content": " ".join}).reset_index()
        df_asin_mix = self.df_asin.groupby(['asin']).agg({"content": " ".join}).reset_index()
        df_asin_good['rating_level'] = '4~5'
        df_asin_bad['rating_level'] = '1~2'
        df_asin_neu['rating_level'] = '3'
        df_asin_mix['rating_level'] = '1~5'
        self.df_save = pd.concat([df_asin_mix, df_asin_good, df_asin_bad, df_asin_neu])
        # self.df_save = self.df_asin.groupby(['asin']).agg({"content": " ".join}).reset_index()
        # self.df_save['rating'] = '1~5'
        # 清洗英文评论文本+提取关键词组+统计词频+词频占比
        self.df_save[['phrases_list', 'phrases_str', 'phrases_counts_str', 'phrases_counts_rate_str']] = self.df_save.content.apply(self._hd_clean_and_extract).apply(pd.Series)
        print(self.df_save.head())
        print(self.df_save.shape, self.df_save.columns)
        self._hd_emotional_analysis()

    def _hd_emotional_analysis(self):
        # 创建SentimentIntensityAnalyzer对象
        sia = SentimentIntensityAnalyzer()
        self.df_save['sentiment_score'] = self.df_save.content.apply(lambda x: sia.polarity_scores(x))
        self.df_save['sentiment_score'] = self.df_save['sentiment_score'].astype("U")

    def save_data(self):
        print("开始存储评论分析的结果数据: 先删除已经存在的，再存储")
        with self.engine_pg.begin() as conn:
            sql = f"delete from {self.table_name_save} where asin='{self.asin}' and phrases_counts={self.phrases_counts}"
            conn.execute(sql)
        self.df_save = self.df_save.loc[:, ['asin', 'rating_level', 'phrases_str', 'phrases_counts_str', 'phrases_counts_rate_str', 'sentiment_score']]
        self.df_save['phrases_counts'] = self.phrases_counts
        print(self.df_save.head())
        self.df_save.to_sql(self.table_name_save, con=self.engine_pg, if_exists="append", index=False)

    def run(self, site_name, asin, phrases_counts):
        while True:
            try:
                self.init_refresh(site_name=site_name, asin=asin, phrases_counts=phrases_counts)
                # 先判断asin+phrases_counts是否近24h出现，如果出现直接结束
                sql = f"select asin, phrases_counts from {self.table_name_save} where asin='{self.asin}' and phrases_counts={self.phrases_counts} and created_time >= now() - interval '24 hour';"
                print("sql:", sql)
                df_judge = pd.read_sql(sql, con=self.engine_pg)
                if df_judge.shape[0] > 0:
                    # 如果数据库近24h内查询出来有记录，则退出
                    print("数据库近24h内查询出来有记录，则退出")
                else:
                    self.read_data()
                    self.handle_data()
                    self.save_data()
                break
            except Exception as e:
                print("error:", traceback.format_exc(e))
                print("随机等待10-30s:", time.sleep(random.randint(10, 30)))
                # self.engine_pg = self.class_db.db_connection(db_type="pg", db_node="h15", site_name=self.site_name)
                self.engine_pg = create_engine_db(site_name=self.site_name)
                continue


def get_comment_api(data):
    import requests
    import json
    # url = 'http://192.168.2.77:5000/analyze_comments'
    # url = 'http://127.0.0.1:5000/analyze_comments'
    url = 'http://113.100.143.162:5000/analyze_comments'
    # url = 'http://192.168.10.225:5000/analyze_comments'
    # url = 'http://192.168.2.41:5000/analyze_comments'
    # url = 'http://192.168.10.221:5000/analyze_comments'
    headers = {'Content-type': 'application/json'}
    # data = {'arg1': 'value1', 'arg2': 'value2'}  # 示例
    response = requests.post(url, headers=headers, data=json.dumps(data))
    # {"stat{self.site_name}_code": 200, "message": "success"}.json()
    print(response.json())


if __name__ == '__main__':
    # handle_obj = CommentAnalytics(site_name='us', asin='B00CBAWIIY', phrases_counts=20)  # 默认最多保留20个关键词
    # handle_obj.run()

    # site_name = sys.argv[1]  # 参数1：站点
    # asin = sys.argv[2]  # 参数2：asin码
    # phrases_counts = sys.argv[3]  # 参数3：提取关键词的数量
    # data = {"site_name": site_name, "asin": asin, "phrases_counts": phrases_counts}
    data = {"site_name": 'us', "asin": 'B00061MWH2', "phrases_counts": '40'}
    # data = {"site_name": 'us', "asin": 'B0C8BJL1D3', "phrases_counts": '40'}
    get_comment_api(data=data)