ai.py

import asyncio
from datetime import datetime

from flask import Blueprint, request, render_template, Response, current_app
from sqlalchemy import text

from app.context import exchange_db, create_engine_db
from app.models.Resp import Resp
from app.util.common_util import CommonUtil
from app.util.openai_session import GptChatSession, num_tokens_from_messages

ai_bp = Blueprint('ai_bp', __name__, url_prefix="/ai")


def save_ai_log(asin, type, question, reply, cost_token, cost_time, site_name):
    if reply is None:
        return
    db = create_engine_db("us")
    sql = """
    insert into open_ai_log (asin, type, question, reply, cost_token, cost_time, site_name)
    values (:asin, :type, :question, :reply, :cost_token, :cost_time, :site_name);
    """
    db.execute(text(sql), {
        "asin": asin,
        "type": type,
        "question": question,
        "reply": reply,
        "cost_token": cost_token,
        "cost_time": cost_time,
        "site_name": site_name
    })
    pass


def save_user_request_ai_log(user_id, request_api, request_type, request_param, replay_param):
    db = create_engine_db("us")
    sql = """
    insert into user_request_ai_log(user_id, request_api, request_type, request_param, replay_param)
    values (:user_id, :request_api, :request_type, :request_param, :replay_param);
    """
    db.execute(text(sql), {
        "user_id": user_id,
        "request_api": request_api,
        "request_type": request_type,
        "request_param": request_param,
        "replay_param": replay_param,
    })
    pass


def get_exist(asin, type):
    db = exchange_db("us")
    sql = """
        select reply,updated_time 
        from open_ai_log
        where asin = :asin
          and type = :type
        order by created_time desc
        limit 1
    """
    result = db.session.execute(text(sql), {"asin": asin, "type": type})
    row = result.fetchone()
    if row is not None:
        # 获取当前时间
        current_time = datetime.now()
        created_time = row['created_time']
        # 判断答复日志的效期是否超过15天，如超过则返回None，重新请求GPT获取最新答复
        if CommonUtil.date_day_diff(created_time, current_time) > 30:
            return None
        return row['reply']
    return None


@ai_bp.route('', methods=['GET', 'POST'])
def common_question():
    """一般性问题直接查询
### args
|  args | required | request type | type |  remarks |
|-------|----------|--------------|------|----------|
| question |  true    |    form_data      | str  | 问题    |

### request
```json
{"question": "xxx"}
```
### return
```html
返回html
```
"""
    if request.method == 'POST':
        if len(request.form['question']) < 1:
            return render_template('message.html', question="NULL", res="输入的问题不能为空")
        question = request.form['question']
        # print("======================================")
        # print("接收到的问题为:", question)

        session = GptChatSession()
        reply, cost_token, cost_time = session.send_msg(question)

        # print("Q：\n", question)
        # print("A：\n", reply)

        return render_template('message.html', question=question, res=str(reply))
    return render_template('message.html', question=0)


@ai_bp.route("/token_num", methods=['POST'])
def token_num():
    """检查token数量
### args
|  args | required | request type | type |  remarks |
|-------|----------|--------------|------|----------|
| text  |  true    |    body      | str  | 字符串 |

### request
```json
{"str": "xxx"}
```

### return
```json
{"status": "success", "code": "200", "data": null}
```
"""
    data: dict = request.json
    assert data is not None, "data 不能为空！"
    # 评论分析
    text = data.get('text')
    assert text is not None, "text 不能为空！"

    messages = [
        {"role": "system", "content": "You are a research assistant."},
    ]
    messages.append(
        {"role": "user", "content": text},
    )
    model = "gpt-3.5-turbo"
    num = num_tokens_from_messages(messages, model)
    return Resp.ok(num)


@ai_bp.route("/comment_analysis", methods=['POST'])
def comment_analysis():
    """评论分析
   ### args
   |  args | required | request type | type |  remarks |
   |-------|----------|--------------|------|----------|
   | asin |  true    |    body      | str  | asin    |
   | site_name  |  true    |    body      | str  | 站点 |

   ### request
   ```json
   {"asin": "xxx", "site_name": "xxx"}
   ```

   ### return
   ```json
   {"status": "success", "code": "200", "data": null}
   ```
   """
    start = CommonUtil.current_time()
    data: dict = request.json
    assert data is not None, "data 不能为空！"
    # 评论分析
    asin = data.get('asin')
    site_name = data.get('site_name')
    assert site_name is not None, "站点不能为空"
    assert asin is not None, "asin不能为空"
    asin = str(asin).strip()
    site_name = site_name.lower()
    type = 'comment_analysis'
    reply = get_exist(asin, type)
    if reply is not None:
        return Resp.ok(reply)
    else:
        db = exchange_db(site_name)
        sql = """select table_name ,alias_name from us_asin_association where alias_name = lower(substr(:asin, 1, 3));"""
        results = db.session.execute(text(sql), params={"asin": asin}).fetchone()
        # 增加判断，不存在asin_association表中的asin规则全归类到asin_comment_other
        if CommonUtil.notNone(results):
            # from builtins import type
            # print(type(results))
            table_name = results['table_name']
        else:
            table_name = "_asin_comment_other"
        table_name = f"{site_name}{table_name}"
        #  评分5.0和评分1.0各取10条
        sql = f"""    
        select *
        from (
                 select content
                 from {table_name}
                 where asin = :asin
                   and is_vp = 1
          		   and OCTET_LENGTH(content) > 50
		           and OCTET_LENGTH(content) < 500
                   and rating >= 4
                 order by agree_num desc
                 limit 15
             ) tmp
        union all
        (
            select content
            from {table_name}
            where asin = :asin
              and is_vp = 1
              and OCTET_LENGTH(content) > 50
              and OCTET_LENGTH(content) < 500
              and rating <= 2
            order by agree_num desc
            limit 15
        )
    """
        rows = db.session.execute(text(sql), {"asin": asin})
        comment = []
        for row in rows:
            comment.append(row['content'])

        assert len(comment) > 0, "当前商品可供抽样的review太少无法分析，请切换商品进行分析。"

        # 构建评论发送模板 使用五条任务
        format_str = "\n".join(list(map(lambda it: f"{it[0] + 1}.{it[1]}", enumerate(comment))))
        send = f"""Complete the analysis of the following six tasks based on this product reviews information, Note that each answer must include the title of the current task :
Task 1:Summarize the top 5 motivations customers used to buy this product and give the reviews proportion
Task 2:Summarize the top 5 reasons why customers like this product and give the reviews proportion
Task 3:Summarize the top 5 reasons why customers dislike this product and give the reviews proportion
Task 4:Summarize the top 5 most common usage scenarios and give the reviews proportion
Task 5:Summarize 5 suggestions for customer desired improvements and give the reviews proportion
Task 6:Summarize the user portrait of this product
Below is a list of comments:
{format_str}"""
        session = GptChatSession()
        reply, cost_token, cost_time = session.send_msg(send)

        if reply.count("Task") > 5:
            save_ai_log(
                asin=asin,
                type=type,
                question=send,
                reply=reply,
                cost_token=cost_token,
                cost_time=cost_time,
                site_name=site_name
            )
        else:
            assert False, "gpt资源繁忙，请重试！"
        end = CommonUtil.current_time()
        print(end - start)
        return Resp.ok(reply)


@ai_bp.route("/comment_analysis_v2", methods=['POST'])
def comment_analysis_v2():
    """评论分析
   ### args
   |  args | required | request type | type |  remarks |
   |-------|----------|--------------|------|----------|
   | asin |  true    |    body      | str  | asin    |
   | asin_list |  true    |    body      | str  | asin以及变体    |
   | site_name  |  true    |    body      | str  | 站点 |

   ### request
   ```json
   {"asin":xxx,"asin_list": ["xxx","xxx"], "site_name": "xxx"}
   ```

   ### return
   ```json
   {"status": "success", "code": "200", "data": null}
   ```
   """
    start = CommonUtil.current_time()
    data: dict = request.json
    assert data is not None, "data 不能为空！"
    # 评论分析
    asin = data.get('asin')
    asin_list = data.get('asin_list')
    site_name = data.get('site_name')
    user_id = data.get('user_id')
    assert CommonUtil.notBlank(site_name), "站点传参不能为空"
    assert CommonUtil.listNotNone(asin_list), "asin_list传参不能为空"
    assert CommonUtil.notBlank(asin), "asin传参不能为空"
    assert CommonUtil.notBlank(user_id), "user_id传参不能为空"
    asin = str(asin).strip()
    site_name = site_name.lower()
    type = 'comment_analysis'
    reply = get_exist(asin, type)
    if reply is not None:
        save_user_request_ai_log(
            user_id=user_id,
            request_api="comment_analysis_v2",
            request_type="post",
            request_param=str(data),
            replay_param=str(Resp.ok(reply).json)
        )
        return Resp.ok(reply)
    else:
        db = exchange_db(site_name)
        #  评分超过4.0 按好评提取样本，频分低于2.0按差评提取样本；---提取所有主体和变体后再进一步按照评分和赞同数进行分析
        sql = f"""
        select content
        from (select content
              from (select asin, content, rating, agree_num, is_vp
                    from {site_name}_asin_comment_other
                    union all
                    (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b00)
                    union all
                    (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b01_b06)
                    union all
                    (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b07)
                    union all
                    (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b08)
                    union all
                    (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b09)) t1
              where asin in ({CommonUtil.list_to_insql(asin_list)})
                and is_vp = 1
                and rating >= 4
                and OCTET_LENGTH(content) > 50
                and OCTET_LENGTH(content) < 500
              order by agree_num desc
              limit 20) as great_cont
        union all
        (select content
         from (select asin, content, rating, agree_num, is_vp
               from {site_name}_asin_comment_other
               union all
               (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b00)
               union all
               (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b01_b06)
               union all
               (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b07)
               union all
               (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b08)
               union all
               (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b09)) t1
         where asin in ({CommonUtil.list_to_insql(asin_list)})
           and is_vp = 1
           and rating <= 2
           and OCTET_LENGTH(content) > 50
           and OCTET_LENGTH(content) < 500
         order by agree_num desc
         limit 20)
    """
        try:
            rows = db.engines[site_name].execute(text(sql), {"asin": asin})
        except:
            raise Exception(f"asin{asin}评论查询失败！请稍后重试或联系管理员。")
        comment = []
        for row in rows:
            comment.append(row['content'])

        assert len(comment) > 5, "当前商品可供抽样的review太少无法分析，请切换商品或重新爬取商品评论后进行分析。"

        # 构建评论发送模板 使用五条任务
        format_str = "\n".join(list(map(lambda it: f"{it[0] + 1}.{it[1]}", enumerate(comment))))
        send = f"""Complete the analysis of the following six tasks based on this product reviews information, Note that each answer must include the title of the current task:
Task 1:Summarize the top 5 motivations customers used to buy this product and give the percentage of reviews
Task 2:Summarize the top 5 reasons why customers like this product and give the percentage of reviews
Task 3:Summarize the top 5 reasons why customers dislike this product and give the percentage of reviews
Task 4:Summarize the top 5 most common usage scenarios and give the percentage of reviews
Task 5:Summarize 5 suggestions for customer desired improvements and give the percentage of reviews
Task 6:Summarize the user profile of the product from the aspects of (who, where, why, what)
Below is a list of comments:
{format_str}"""
        session = GptChatSession()
        reply, cost_token, cost_time = session.send_message_qwen(send)
        #  此处去掉着重符号
        reply = reply.replace("**", "")
        if reply.count("Task") > 5:
            start_save_time = CommonUtil.current_time()
            save_ai_log(
                asin=asin,
                type=type,
                question=send,
                reply=reply,
                cost_token=cost_token,
                cost_time=cost_time,
                site_name=site_name
            )
            save_user_request_ai_log(
                user_id=user_id,
                request_api="comment_analysis_v2",
                request_type="post",
                request_param=str(data),
                replay_param=str(Resp.ok(reply).json)
            )
            end_save_time = CommonUtil.current_time()
            print("储存数据耗时：", end_save_time - start_save_time)
        else:
            assert False, "请求资源繁忙，请重试！"
        end = CommonUtil.current_time()
        print("评论接口完整耗时：", end - start)
        return Resp.ok(reply)


@ai_bp.route("/optimize_listing", methods=['POST'])
def optimize_listing():
    """listing 优化建议

   ### args
   |  args | required | request type | type |  remarks |
   |-------|----------|--------------|------|----------|
   | asin |  true    |    body      | str  | asin    |
   | content  |  true    |    body      | str  | asin标题 |

   ### request
   ```json
   {"asin": "xxx", "content": "xxx"}
   ```

   ### return
   ```json
   {"status": "success", "code": "200", "data": null}
   ```
   """
    data: dict = request.json
    assert data is not None, "data 不能为空！"
    content = data.get('content')
    asin = data.get('asin')
    site_name = data.get('site_name')
    assert asin is not None, "asin不能为空"
    assert content is not None, "content不能为空"
    type = 'optimize_listing'
    reply = get_exist(asin, type)
    if reply is not None:
        return Resp.ok(reply)
    else:
        session = GptChatSession()
        send = f"""Please propose Listing optimization suggestions for amazon product titles，list the top five points,the result Keyword should be capitalized,here is the title:"{content}"
"""
        reply, cost_token, cost_time = session.send_msg(send)

        save_ai_log(
            asin=asin,
            type=type,
            question=send,
            reply=reply,
            cost_token=cost_token,
            cost_time=cost_time,
            site_name=site_name
        )
    return Resp.ok(reply)


@ai_bp.route("/qa_analysis", methods=['POST'])
def qa_analysis():
    """ QA分析

   ### args
   |  args | required | request type | type |  remarks |
   |-------|----------|--------------|------|----------|
   | asin |  true    |    body      | str  | asin    |
   | site_name  |  true    |    body      | str  | 站点 |

   ### request
   ```json
   {"asin": "xxx", "site_name": "xxx"}
   ```

   ### return
   ```json
   {"status": ”success“, "code": "200", "data": null}
   ```
   """
    data: dict = request.json
    assert data is not None, "data 不能为空！"
    site_name = data.get('site_name')
    asin = data.get('asin')
    assert asin is not None, "asin不能为空"
    assert site_name is not None, "站点不能为空"
    asin = str(asin).strip()
    site_name = site_name.lower()

    type = "qa_analysis"
    reply = get_exist(asin, type)
    if reply is not None:
        return Resp.ok(reply)
    else:
        db = exchange_db(site_name)
        sql = """select replace(table_name,'comment','qa') as table_name from us_asin_association where alias_name = lower(substr(:asin, 1, 3));"""
        # table_name = db.session.execute(text(sql), {"asin": asin}).fetchone()['table_name']
        results = db.session.execute(text(sql), {"asin": asin}).fetchone()
        # 增加判断，不存在asin_association表中的asin规则全归类到asin_qa_other
        if CommonUtil.notNone(results):
            table_name = results['table_name']
        else:
            table_name = "_asin_qa_other"
        table_name = f"{site_name}{table_name}"
        sql = f"""select question, answer from {table_name} where asin = :asin limit 20;"""
        rows = db.session.execute(text(sql), {"asin": asin})
        qa_list_str = ""
        for row in rows:
            question = row['question']
            answer = row['answer']
            qa_list_str += "\n"
            qa_list_str += f"Question:{question}" + "\n" + f"Answer:{answer}"
            qa_list_str += "\n"
        send = f"""Complete the following tasks according to the information at the end of this article，Note that each answer must include the title of the current task：
Task 1:Organize the text information in terms of Question type, Keywords and Mention Times, and output it in table form
Task 2:List the top 5 most frequently asked question type
Task 3:According to these question answer dialogue information summed up What customers care about most and the  product optimization suggestions
Here is the transcript of the conversation, which consists of Question (Q) and Answer (A):
{qa_list_str}
       """
        print(send)
        session = GptChatSession()
        reply, cost_token, cost_time = session.send_msg(send)
        save_ai_log(
            asin=asin,
            type=type,
            question=send,
            reply=reply,
            cost_token=cost_token,
            cost_time=cost_time,
            site_name=site_name
        )
        return Resp.ok(reply)


@ai_bp.route("/comment_analysis_v3", methods=['POST'])
def comment_analysis_v3():
    """评论分析
   ### args
   |  args | required | request type | type |  remarks |
   |-------|----------|--------------|------|----------|
   | asin |  true    |    body      | str  | asin    |
   | asin_list |  true    |    body      | str  | asin以及变体    |
   | site_name  |  true    |    body      | str  | 站点 |

   ### request
   ```json
   {"asin":xxx,"asin_list": ["xxx","xxx"], "site_name": "xxx", "user_id":"xxx"}
   ```

   ### return
   ```json
   {"status": "success", "code": "200", "data": null}
   ```
   """
    data: dict = request.json
    args: dict = request.args
    assert data is not None, "data 不能为空！"
    # 评论分析
    asin = data.get('asin')
    asin_list = data.get('asin_list')
    site_name = data.get('site_name')
    user_id = data.get('user_id') or args.get("user_id")
    assert CommonUtil.notBlank(site_name), "站点传参不能为空"
    assert CommonUtil.listNotNone(asin_list), "asin_list传参不能为空"
    assert CommonUtil.notBlank(asin), "asin传参不能为空"
    assert CommonUtil.notBlank(user_id), "user_id传参不能为空"
    asin = str(asin).strip()
    site_name = site_name.lower()
    type = 'comment_analysis'
    # 判断15天内，是否存储过对应asin的评论请求
    reply = get_exist(asin, type)
    if reply is not None:
        save_user_request_ai_log(
            user_id=user_id,
            request_api="comment_analysis_v3",
            request_type="post",
            request_param=str(data),
            replay_param=str(Resp.ok(reply).json)
        )
        return Resp.ok(reply)
    else:
        db = exchange_db(site_name)
        #  评分超过4.0 按好评提取样本，频分低于2.0按差评提取样本；---提取所有主体和变体后再进一步按照评分和赞同数进行分析
        sql = f"""
        select content
        from (select content
              from (select asin, content, rating, agree_num, is_vp
                    from {site_name}_asin_comment_other
                    union all
                    (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b00)
                    union all
                    (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b01_b06)
                    union all
                    (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b07)
                    union all
                    (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b08)
                    union all
                    (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b09)) t1
              where asin in ({CommonUtil.list_to_insql(asin_list)})
                and is_vp = 1
                and rating >= 4
                and OCTET_LENGTH(content) > 50
                and OCTET_LENGTH(content) < 500
              order by agree_num desc
              limit 20) as great_cont
        union all
        (select content
         from (select asin, content, rating, agree_num, is_vp
               from {site_name}_asin_comment_other
               union all
               (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b00)
               union all
               (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b01_b06)
               union all
               (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b07)
               union all
               (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b08)
               union all
               (select asin, content, rating, agree_num, is_vp from {site_name}_asin_comment_b09)) t1
         where asin in ({CommonUtil.list_to_insql(asin_list)})
           and is_vp = 1
           and rating <= 2
           and OCTET_LENGTH(content) > 50
           and OCTET_LENGTH(content) < 500
         order by agree_num desc
         limit 20)
    """
        try:
            rows = db.engines[site_name].execute(text(sql), {"asin": asin})
        except:
            raise Exception(f"asin{asin}评论查询失败！请稍后重试或联系管理员。")
        comment = []
        for row in rows:
            comment.append(row['content'])

        assert len(comment) > 5, "当前商品可供抽样的review太少无法分析，请切换商品或重新爬取商品评论后进行分析。"

        # 构建评论发送模板 使用五条任务
        format_str = "\n".join(list(map(lambda it: f"{it[0] + 1}.{it[1]}", enumerate(comment))))
        send = f"""Complete the analysis of the following six tasks based on this product reviews information, Note that each answer must include the title of the current task:
Task 1:Summarize the top 5 motivations customers used to buy this product and give the percentage of reviews
Task 2:Summarize the top 5 reasons why customers like this product and give the percentage of reviews
Task 3:Summarize the top 5 reasons why customers dislike this product and give the percentage of reviews
Task 4:Summarize the top 5 most common usage scenarios and give the percentage of reviews
Task 5:Summarize 5 suggestions for customer desired improvements and give the percentage of reviews
Task 6:Summarize the user profile of the product from the aspects of (who, where, why, what)
Below is a list of comments:
{format_str}"""

        # 流式调用
        def qwen_stream_chat(send, app_context):
            from http import HTTPStatus
            session = GptChatSession()
            start_save_time = CommonUtil.current_time()
            output = None
            cost_token = None
            try:
                res_generator = session.send_message_qwen_stream(send)
                last_content = None
                for response in res_generator:
                    if response.status_code == HTTPStatus.OK:
                        output = response.output.get("choices")[0].get("message").get("content")
                        cost_token = response.usage.get("output_tokens")
                        if last_content is None:
                            add_content = output
                        else:
                            add_content = output[len(last_content):]
                        last_content = output
                        print(last_content)
                        yield str(add_content).encode("utf-8")
                    else:
                        code = response.code
                        message = response.message
                        yield f"通义千问调用失败 code: {code}, message:{message}，请重试或联系管理员！！".encode("utf-8")
                        pass
            finally:
                # 入栈否则会db失效
                app_context.push()
                end_save_time = CommonUtil.current_time()
                cost_time = end_save_time - start_save_time
                save_ai_log(
                    asin=asin,
                    type=type,
                    question=send,
                    reply=output,
                    cost_token=cost_token,
                    cost_time=cost_time,
                    site_name=site_name
                )

                save_user_request_ai_log(
                    user_id=user_id,
                    request_api="comment_analysis_v3",
                    request_type="post",
                    request_param=str(data),
                    replay_param=str(Resp.ok(output).json)
                )
                end_save_time = CommonUtil.current_time()
                print("储存数据耗时：", end_save_time - start_save_time)
                pass

        return Response(qwen_stream_chat(send, current_app.app_context()), mimetype="text/event-stream")