123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313 |
- import random
- import jieba
- import numpy as np
- import os
- import pickle
- import sys
- import torch
- from fastapi import APIRouter
- from loguru import logger
- import traceback
- import hashlib
- import os
- import sys
- import traceback
- import uuid
- from datetime import date, datetime
- from datetime import timedelta
- from io import BytesIO
- from typing import Optional, List
- from urllib.parse import quote
- import pandas as pd
- import yaml
- from fastapi import APIRouter
- from fastapi.responses import StreamingResponse
- from loguru import logger
- from pangres import upsert
- from pydantic import BaseModel, Field
- curr_path = os.path.abspath(os.path.dirname(__file__))
- project_root_path = curr_path[:curr_path.find("video_to_word") + len("video_to_word")]
- sys.path.append(project_root_path)
- from config.url_and_db import toutiao_static_video_url, ai_word_engine
- with open('/data/pythonProject/video_to_word/script_score/pkl/vocab.pkl', 'rb') as f:
- vocab = pickle.load(f)
- with open('/data/pythonProject/video_to_word/script_score/pkl/word_to_idx.pkl', 'rb') as f:
- word_to_idx = pickle.load(f)
- log_score_min = -19.266664505004883
- log_score_max = 0
- def split_by_jieba(x):
- seg_list = jieba.cut(x)
- seg_list = ','.join(seg_list)
- seg_list = seg_list.split(",")
- seg_list = [v for v in seg_list if v != ',']
- return seg_list
- def encode_samples(tokenized_samples):
- features = []
- for sample in tokenized_samples:
- feature = []
- for token in sample[0]:
- if token in word_to_idx:
- feature.append(word_to_idx[token])
- else:
- feature.append(0)
- features.append(feature)
- return features
- def pad_samples(features, maxlen=113, PAD=0):
- padded_features = []
- for feature in features:
- if len(feature) >= maxlen:
- padded_feature = feature[:maxlen]
- else:
- padded_feature = feature
- while len(padded_feature) < maxlen:
- padded_feature.append(PAD)
- padded_features.append(padded_feature)
- return padded_features
- def log_and_map_min_max_score(score):
- # log 变换
- log_score = np.log(score)
- # 映射到 0-100 分
- script_score = 0 + (100 - 0) / (log_score_max - log_score_min) * (log_score - log_score_min)
- script_score = int(script_score)
- # 映射到 等级 (优:(95,100], 良:(85,95], 低质:(0,85])
- # 1:'优', 2:'良', 3:'低质'
- script_level = 3 if script_score <= 85 else (2 if script_score <= 95 else 1)
- high_quality_prob = 0.1905 if script_score <= 85 else (0.4330 if script_score <= 95 else 0.6916)
- return script_score, script_level, high_quality_prob
- router = APIRouter(tags=['script_score_server'])
- class BaseResponse(BaseModel):
- message: str = Field(..., description='消息')
- success: bool = Field(True, description='true or false')
- code: int = Field(0, description='')
- class GetScriptScoreRequest(BaseModel):
- script: str = Field(..., description="脚本内容")
- user_name: str = Field(..., description="用户名")
- user_id: str = Field(..., description="用户id")
- class Config:
- schema_extra = {
- "example": {
- "script": "十六块二十六块,只要二十六块包邮到家,这么大一件派克服,现在不要两百,不要一百二十六块就给你包邮到家,真的太划算了,咱们工厂现在为了扩大销售渠道,所以特地拿出一批货在桃树上做活动,这款派克服寒气版型,特别时尚,抽绳收腰设计,修身显瘦,加绒内里还保暖毛领,精致又洋气,喜欢的朋友赶紧点击视频下方链接进入操作即可就可以购买啦。",
- "user_name": "管理员",
- "user_id": "e9ca23d68d884d4ebb19d07889727dae"
- }
- }
- class ScriptScoreInfo(BaseModel):
- unique_id: str = Field(..., description="脚本配置id")
- script_text: str = Field(..., description="脚本内容")
- script_score: int = Field(..., description="脚本得分")
- script_level: int = Field(..., description="脚本评分等级{1:'优', 2:'良', 3:'低质'}")
- high_quality_prob: float = Field(..., description="跑量的概率")
- script_hash: str = Field(..., description="脚本内容哈希值")
- user_name: str = Field(..., description="用户名")
- user_id: str = Field(..., description="用户id")
- start_time: datetime = Field(datetime.now().strftime('%Y-%m-%dT%H:%M:%S'), description="时间")
- class GetScriptScoreLstResponse(BaseResponse):
- total_num: int = Field(0, description="总个数")
- page_num: int = Field(1, description="第几页")
- page_size: int = Field(10, description="每页个数")
- result: List[ScriptScoreInfo] = Field([], description="结果详情")
- class GetSingleScriptScore(BaseResponse):
- result: ScriptScoreInfo = Field(None, description="结果详情")
- class ScriptScoreLstRequest(BaseModel):
- start_date: Optional[date] = Field(date.today() + timedelta(days=-29), description="开始日期-用于查询")
- end_date: Optional[date] = Field(date.today(), description="结束日期-用于查询")
- search_word: Optional[str] = Field('', description="查询词")
- user_name: Optional[str] = Field('', description="用户名称")
- script_level: Optional[List[int]] = Field([], description="评分等级{1:'优', 2:'良', 3:'低质'}")
- page_num: int = Field(1, description="第几页")
- page_size: int = Field(10, description="每页的大小")
- class DeleteScriptScoreRequest(BaseModel):
- unique_id: str = Field(..., description="唯一标识")
- user_name: str = Field(..., description="用户名")
- user_id: str = Field(..., description="用户id")
- class Config:
- schema_extra = {
- "example": {
- "unique_id": "89d4d72a-8c7d-4c89-8b65-258d7206cd6b",
- "user_name": "管理员",
- "user_id": "e9ca23d68d884d4ebb19d07889727dae"
- }
- }
- @logger.catch()
- @router.post("/delete_script_score",
- description="删除脚本评级",
- summary="删除脚本评级",
- response_model=BaseResponse)
- def delete_script_score(item: DeleteScriptScoreRequest):
- response = BaseResponse(code=0, message='delete success', success=True)
- try:
- logger.info(f"request body: {item}")
- sql = f"select * from ctop_ai_script_score where unique_id = '{item.unique_id}' " \
- f"and operate_type = 1 " \
- f"and end_time = '9999-12-31'"
- df = pd.read_sql(sql, ai_word_engine)
- if df.empty:
- response = BaseResponse(code=-2, message='不存在该条记录', success=False)
- return response
- # 判断删除者是否为该条记录的创建者,如果不是,则没用权限删除该条记录
- creator_id = df['user_id'].values[0]
- if creator_id != item.user_id:
- response = BaseResponse(code=-1, message='没有删除权限', success=False)
- else:
- # 数据拉链表方式
- # 修改历史记录的 end_time, 使其失效
- # 新增删除记录,用于记录删除的时间
- update_df = df.copy(deep=True)
- update_df['end_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
- update_df.set_index('unique_id', drop=True, inplace=True)
- upsert(engine=ai_word_engine,
- df=update_df,
- table_name='ctop_ai_script_score',
- if_row_exists='update')
- add_df = df.copy(deep=True)
- add_df['operate_type'] = 3
- add_df['start_time'] = datetime.today().strftime('%Y-%m-%d %H:%M:%S')
- add_df.set_index('unique_id', drop=True, inplace=True)
- upsert(engine=ai_word_engine,
- df=add_df,
- table_name='ctop_ai_script_score',
- if_row_exists='update')
- logger.info(f"request body: {item}, response body: {response}")
- except:
- response.code = -1
- response.message = traceback.format_exc()
- response.success = False
- logger.error(f"request body: {item}, response: {response}")
- return response
- @logger.catch()
- @router.post("/get_script_score_lst",
- description="获取脚本评级列表",
- summary="获取脚本评级列表",
- response_model=GetScriptScoreLstResponse)
- def get_script_score_lst(item: ScriptScoreLstRequest):
- response = GetScriptScoreLstResponse(message="查询成功")
- try:
- end_date = item.end_date + timedelta(days=1)
- script_level_lst = [-1, -2] if len(item.script_level) == 0 else (item.script_level * 2 if len(item.script_level) == 1 else item.script_level)
- sql = f"select * from ctop_ai_script_score " \
- f"where ('{item.search_word}' = '' or script_text like '%%{item.search_word}%%' ) " \
- f"and ('{item.user_name}' = '' or user_name like '%%{item.user_name}%%' ) " \
- f"and ('{item.script_level}' = '[]' or script_level in {tuple(script_level_lst)} ) " \
- f"and operate_type = 1 and end_time ='9999-12-31' " \
- f"and start_time >= '{item.start_date}' and start_time < '{end_date}' "
- org_df = pd.read_sql(sql, ai_word_engine)
- if not org_df.empty:
- org_df.sort_values(by='start_time', ascending=False, inplace=True)
- org_df['number'] = list(range(1, len(org_df) + 1))
- total_num = org_df.shape[0]
- detail = org_df.iloc[(item.page_num - 1) * item.page_size: item.page_num * item.page_size].to_dict('records')
- response.result = detail
- response.total_num = total_num
- response.page_num = item.page_num
- response.page_size = item.page_size
- else:
- response.message = "没有符合条件的数据"
- response.page_num = item.page_num
- response.page_size = item.page_size
- logger.info(f"request body: {item}, response body: {response}")
- except:
- response.code = -1
- response.message = traceback.format_exc()
- response.success = False
- logger.error(f"request body: {item}, response body: {response}")
- return response
- @logger.catch()
- @router.post("/get_script_score",
- description="脚本质量评级",
- summary="脚本质量评级",
- response_model=GetSingleScriptScore)
- def get_script_score(item: GetScriptScoreRequest):
- response = GetSingleScriptScore(message="脚本评分完成")
- try:
- script_split_lst = split_by_jieba(item.script)
- feature = torch.tensor(pad_samples(encode_samples([[script_split_lst]])))
- from script_score.lstm_network import SentimentNet
- net = torch.load('/data/pythonProject/video_to_word/script_score/pkl/epoch52_test_auc_0.790_train_auc_0.815.pth',
- map_location='cpu')
- with torch.no_grad():
- score = net(feature)
- script_score, script_level, high_quality_prob = log_and_map_min_max_score(score[0][1].item())
- unique_id = str(uuid.uuid4())
- response.result = {"script_level": script_level,
- "script_score": script_score,
- "high_quality_prob": high_quality_prob,
- "script_hash": hashlib.md5(item.script.encode('utf-8')).hexdigest(),
- "script_text": item.script,
- "unique_id": unique_id,
- "user_id": item.user_id,
- "user_name": item.user_name,
- "operate_type": 1,
- }
- # write to db
- script_info_df = pd.DataFrame([response.result])
- script_info_df.to_sql(name="ctop_ai_script_score",
- con=ai_word_engine,
- if_exists='append',
- index=False)
- logger.info(f"request body: {item}, response body: {response}")
- except:
- response.code = -1
- response.success = False
- response.message = traceback.format_exc()
- logger.error(f"request body: {item}, response body: {response}")
- return response
- if __name__ == '__main__':
- text = "十六块二十六块,只要二十六块包邮到家,这么大一件派克服,现在不要两百,不要一百二十六块就给你包邮到家,真的太划算了,咱们工厂现在为了扩大销售渠道,所以特地拿出一批货在桃树上做活动,这款派克服寒气版型,特别时尚,抽绳收腰设计,修身显瘦,加绒内里还保暖毛领,精致又洋气,喜欢的朋友赶紧点击视频下方链接进入操作即可就可以购买啦。"
- print(get_script_score(text))
|