分表开发

applications/deal/__init__.py

@@ -1,9 +1,6 @@
 """
 @author: luojunhui
+接口信息
 """
-from .search_deal import SearchDeal
-from .re_search_deal import ReSearchDeal
-from .process_deal import ProcessDeal
-from .search_deal_v2 import SearchDeal2
-from .get_done_content_deal import ProcessDeal2
-from .recall_deal import RecallDeal
+from .record import Record
+from .response import Response

applications/deal/re_search_deal.py

@@ -1,79 +0,0 @@
-"""
-@author: luojunhui
-"""
-import json
-import asyncio
-
-from applications.schedule import re_search_videos
-
-
-class ReSearchDeal(object):
-    """
-    重新搜索逻辑
-    obj = {
-        "ori_title": params['title'],
-        "content_title": params['kimi_summary'],
-        "content_keys": params['kimi_keys'],
-        "trace_id": params['trace_id']
-    }
-    """
-
-    def __init__(self, params):
-        self.gh_id = None
-        self.trace_id = None
-        self.kimi_keys = None
-        self.kimi_summary = None
-        self.title = None
-        self.params = params
-
-    def check_params(self):
-        """
-        check params
-        :return:
-        """
-        try:
-            self.title = self.params['title']
-            self.kimi_summary = self.params['kimi_summary']
-            self.kimi_keys = json.loads(self.params['kimi_keys'])
-            self.trace_id = self.params['trace_id']
-            self.gh_id = self.params['ghId']
-            return None
-        except AttributeError as e:
-            result = {
-                "status": "fail",
-                "code": 1,
-                "message": str(e),
-                "info": "params check error"
-            }
-            return result
-        
-    async def deal(self):
-        """
-        research function deal
-        :return: 
-        """
-        params_error = self.check_params()
-        if params_error:
-            return params_error
-        else:
-            try:
-                asyncio.ensure_future(
-                    re_search_videos(
-                        self.params,
-                        self.trace_id,
-                        self.gh_id)
-                )
-                res = {
-                    "status": "success",
-                    "code": 0,
-                    "traceId": self.trace_id
-                }
-                return res
-            except Exception as e:
-                res = {
-                    "status": "fail",
-                    "code": 1,
-                    "message": str(e)
-                }
-                return res
-            

applications/deal/search_deal_v2.py → applications/deal/record.py

@@ -6,10 +6,10 @@ import time
 from uuid import uuid4
 
 from applications.functions.log import logging
-from applications.static.config import db_article
+from static.config import db_article
 
 
-class SearchDeal2(object):
+class Record(object):
     """
     搜索接口处理逻辑
     """

applications/deal/recall_deal.py → applications/deal/response.py

@@ -9,11 +9,11 @@ import hashlib
 import urllib.parse
 
 from applications.functions.log import logging
-from applications.static.config import db_article
+from static.config import db_article
 from applications.functions.common import request_for_info
 
 
-class RecallDeal(object):
+class Response(object):
     """
     召回逻辑
     """

applications/deal/search_deal.py

@@ -1,216 +0,0 @@
-"""
-@author: luojunhui
-"""
-import time
-import asyncio
-
-from uuid import uuid4
-
-from applications.functions.log import logging
-from applications.static.config import db_article
-from applications.schedule import search_videos
-
-
-class SearchDeal(object):
-    """
-    搜索接口处理逻辑
-    """
-
-    def __init__(self, params, mysql_client):
-        self.content_id = None
-        self.account_name = None
-        self.contents = None
-        self.title = None
-        self.gh_id = None
-        self.params = params
-        self.mysql_client = mysql_client
-        self.trace_id = "search-{}-{}".format(str(uuid4()), str(int(time.time())))
-
-    def check_params(self):
-        """
-        检查请求params
-        :return:
-        """
-        try:
-            self.gh_id = self.params['ghId']
-            self.title = self.params['title'].split("@@")[-1].replace("'", "")
-            self.contents = self.params['content'].replace("'", "")
-            self.account_name = self.params['accountName'].replace("'", "")
-            self.content_id = self.params['articleId']
-            logging(
-                code="1001",
-                info="搜索视频内容接口请求成功, 参数校验成功",
-                port="title_to_search",
-                trace_id=self.trace_id,
-                data=self.params
-            )
-            return None
-        except Exception as e:
-            result = {
-                "status": "fail",
-                "code": 1,
-                "message": str(e),
-                "info": "params check error"
-            }
-            logging(
-                code="4001",
-                info="搜索视频内容接口请求成功, 参数校验失败",
-                port="title_to_search",
-                trace_id=self.trace_id,
-                data=self.params
-            )
-            return result
-
-    async def record(self):
-        """
-        把数据插入
-        :return:
-        """
-        insert_sql = f"""
-                        INSERT INTO {db_article}
-                            (trace_id, gh_id, article_title, article_text, account_name, content_id)
-                        VALUES 
-                            (%s, %s, %s, %s, %s, %s);
-                            """
-        await self.mysql_client.async_insert(
-            sql=insert_sql,
-            params=(self.trace_id, self.gh_id, self.title, self.contents, self.account_name, self.content_id)
-        )
-        logging(
-            code="1002",
-            info="成功记录请求数据到mysql中",
-            trace_id=self.trace_id
-        )
-
-    async def process_video_id(self):
-        """
-        如果video_id在标题中，则做特殊处理
-        :return:
-        """
-        video_id = self.title.split("video_id=")[-1]
-        update_sql = f"""
-            UPDATE  
-                {db_article}
-            SET 
-                recall_video_id1 = %s
-            WHERE  
-                trace_id = %s;"""
-        await self.mysql_client.async_insert(
-            sql=update_sql,
-            params=(video_id, self.trace_id)
-        )
-        res = {
-            "status": "success",
-            "code": 0,
-            "traceId": self.trace_id
-        }
-        logging(
-            code="1003",
-            info="视频生成文本服务请求，video_id = {}".format(video_id),
-            trace_id=self.trace_id
-        )
-        return res
-
-    async def insert_history_contents_videos(self, vid1, vid2, vid3, kimi_title):
-        """
-        插入历史视频id
-        :return:
-        """
-        update_sql = f"""
-        UPDATE {db_article}
-        SET 
-            kimi_title=%s,
-            recall_video_id1=%s, 
-            recall_video_id2=%s, 
-            recall_video_id3=%s
-        WHERE  trace_id = %s
-        """
-        print(update_sql)
-        await self.mysql_client.async_insert(
-            sql=update_sql,
-            params=(
-                kimi_title,
-                vid1,
-                {"NULL" if vid2 is None else vid2},
-                {"NULL" if vid3 is None else vid3},
-                self.trace_id
-            )
-        )
-
-    async def get_history_contents(self):
-        """
-        check whether the content id exists
-        :return:
-
-        """
-        select_sql = f"""
-            SELECT recall_video_id1, recall_video_id2, recall_video_id3, kimi_title
-            FROM {db_article}
-            WHERE content_id = '{self.content_id}' and trace_id != '{self.trace_id}'
-            ORDER BY id DESC;
-        """
-        result = await self.mysql_client.async_select(select_sql)
-        if result:
-            for item in result:
-                video_1, video_2, video_3, kimi_title = item
-                if video_1 and kimi_title:
-                    return [video_1, video_2, video_3, kimi_title]
-                else:
-                    continue
-            return None
-        else:
-            return None
-
-    async def deal(self):
-        """
-        deal
-        :return:
-        """
-        params_error = self.check_params()
-        if params_error:
-            return params_error
-        else:
-            # 记录
-            await self.record()
-            if "video_id=" in self.title:
-                return await self.process_video_id()
-            else:
-                video_ids = await self.get_history_contents()
-                if video_ids:
-                    logging(
-                        code="1004",
-                        info="获取历史到文章视频",
-                        data=video_ids,
-                        trace_id=self.trace_id
-                    )
-                    await self.insert_history_contents_videos(
-                        video_ids[0],
-                        video_ids[1],
-                        video_ids[2],
-                        video_ids[3]
-                    )
-                    return {"status": "success", "code": 0, "traceId": self.trace_id}
-                else:
-                    # search from the Internet
-                    try:
-                        asyncio.ensure_future(
-                            search_videos(
-                                params={"title": self.title, "content": self.contents, "trace_id": self.trace_id},
-                                trace_id=self.trace_id,
-                                gh_id=self.gh_id,
-                                mysql_client=self.mysql_client
-                            )
-                        )
-                        res = {
-                            "status": "success",
-                            "code": 0,
-                            "traceId": self.trace_id
-                        }
-                        return res
-                    except Exception as e:
-                        res = {
-                            "status": "fail",
-                            "code": 1,
-                            "message": str(e)
-                        }
-                        return res

applications/routes.py

@@ -3,7 +3,7 @@
 """
 from quart import Blueprint, jsonify, request
 
-from applications.deal import SearchDeal2, ProcessDeal, RecallDeal
+from applications.deal import Response, Record
 
 my_blueprint = Blueprint('LongArticles', __name__)
 
@@ -28,7 +28,7 @@ def Routes(mysql_client):
         :return:
         """
         params = await request.get_json()
-        SD = SearchDeal2(params=params, mysql_client=mysql_client)
+        SD = Record(params=params, mysql_client=mysql_client)
         result = await SD.deal()
         return jsonify(result)
 
@@ -41,18 +41,8 @@ def Routes(mysql_client):
         data = await request.get_json()
         trace_id = data['traceId']
         minigram_type = data['miniprogramUseType']
-        RD = RecallDeal(trace_id=trace_id, mini_program_type=minigram_type, mysql_client=mysql_client)
+        RD = Response(trace_id=trace_id, mini_program_type=minigram_type, mysql_client=mysql_client)
         response = await RD.deal()
         return jsonify(response)
 
-    @my_blueprint.route("/task")
-    async def schedule_task():
-        """
-        执行代码
-        :return:
-        """
-        PD = ProcessDeal(mysql_client=mysql_client)
-        await PD.deal()
-        return jsonify({"success": "true"})
-
     return my_blueprint

applications/schedule/process_schedule.py

@@ -6,7 +6,7 @@ import json
 
 from applications.functions.common import request_for_info, create_gzh_path, create_gzh_path_v2
 from applications.functions.log import logging
-from applications.static.config import db_article
+from static.config import db_article
 
 
 async def return_info_v2(video_id, trace_id, mysql_client):

applications/schedule/search_schedule.py

@@ -7,7 +7,7 @@ import time
 
 from applications.match_algorithm.rank import title_similarity_rank
 from applications.search import *
-from applications.static.config import gh_id_dict, db_article
+from static.config import gh_id_dict, db_article
 from applications.functions.log import logging
 from applications.functions.video_item import VideoProducer
 from applications.functions.kimi import KimiServer

dev/task_test.py

@@ -1,3 +0,0 @@
-"""
-@author: luojunhui
-"""

task.py → match_video_task.py

@@ -6,9 +6,7 @@ import datetime
 import asyncio
 
 import aiomysql
-
-from applications.deal import ProcessDeal
-from applications.deal.get_done_content_deal import ProcessDeal2
+from tasks import MatchTask1, MatchTask2
 
 
 class TaskMySQLClient(object):
@@ -58,6 +56,7 @@ class TaskMySQLClient(object):
     async def async_insert(self, sql, params):
         """
         insert and update method
+        :param params:
         :param sql:
         :return:
         """
@@ -74,7 +73,7 @@ async def main():
     """
     TMC = TaskMySQLClient()
     await TMC.init_pool()
-    PD = ProcessDeal(TMC)
+    PD = MatchTask1(TMC)
     await PD.deal()
 
 
@@ -85,7 +84,7 @@ async def main2():
     """
     TMC = TaskMySQLClient()
     await TMC.init_pool()
-    PD = ProcessDeal2(TMC)
+    PD = MatchTask2(TMC)
     await PD.deal()
 
 

tasks/__init__.py

@@ -0,0 +1,6 @@
+"""
+@author: luojunhui
+定时任务
+"""
+from .task1 import MatchTask1
+from .task2 import MatchTask2

applications/deal/process_deal.py → tasks/task1.py

@@ -3,13 +3,13 @@
 """
 import asyncio
 
-from applications.static.config import db_article
+from static.config import db_article
 from applications.schedule import search_videos
 from applications.functions.log import logging
-from applications.static.config import spider_coroutines
+from static.config import spider_coroutines
 
 
-class ProcessDeal(object):
+class MatchTask1(object):
     """
     定时执行任务
     """
@@ -200,87 +200,74 @@ class ProcessDeal(object):
             )
         )
         try:
-            # 判断标题中是否包含video_id
-            if "video_id=" in params['title']:
+            await search_videos(
+                params={"title": params['title'], "content": params['text'], "trace_id": params['trace_id']},
+                trace_id=params['trace_id'],
+                gh_id=params['gh_id'],
+                mysql_client=self.mysql_client
+            )
+            # 执行完成之后，判断是否存在视频id
+            select_sql = f"""
+                SELECT recall_video_id1, recall_video_id2, recall_video_id3
+                FROM {db_article}
+                WHERE trace_id = '{params["trace_id"]}';
+            """
+            result = await self.mysql_client.async_select(sql=select_sql)
+            vid1, vid2, vid3 = result[0]
+            if vid1 or vid2 or vid3:
+                update_sql2 = f"""
+                    UPDATE {db_article}
+                    SET 
+                       content_status = %s,
+                       process_times = %s
+                       WHERE trace_id = %s;
+                """
+                await self.mysql_client.async_insert(
+                    sql=update_sql2,
+                    params=(
+                        2, {int(params['process_times']) + 1}, params['trace_id']
+                    )
+                )
                 logging(
-                    code="9006",
-                    info="视频生成文本测试",
+                    code="9008",
+                    info="视频搜索成功， 状态修改为2",
                     trace_id=params['trace_id']
                 )
-                await self.process_video_id(
-                    title=params['title'],
-                    trace_id=params['trace_id'],
-                    process_times=params['process_times']
-                )
             else:
-                await search_videos(
-                    params={"title": params['title'], "content": params['text'], "trace_id": params['trace_id']},
-                    trace_id=params['trace_id'],
-                    gh_id=params['gh_id'],
-                    mysql_client=self.mysql_client
-                )
-                # 执行完成之后，判断是否存在视频id
-                select_sql = f"""
-                    SELECT recall_video_id1, recall_video_id2, recall_video_id3
-                    FROM {db_article}
-                    WHERE trace_id = '{params["trace_id"]}';
-                """
-                result = await self.mysql_client.async_select(sql=select_sql)
-                vid1, vid2, vid3 = result[0]
-                if vid1 or vid2 or vid3:
-                    update_sql2 = f"""
+                if int(params['process_times']) < 5:
+                    update_sql3 = f"""
                         UPDATE {db_article}
                         SET 
                            content_status = %s,
                            process_times = %s
-                           WHERE trace_id = %s;
-                    """
+                        WHERE trace_id = %s;
+                                    """
                     await self.mysql_client.async_insert(
-                        sql=update_sql2,
-                        params=(
-                            2, {int(params['process_times']) + 1}, params['trace_id']
-                        )
+                        sql=update_sql3,
+                        params=(0, int(params['process_times']) + 1, params['trace_id'])
                     )
                     logging(
-                        code="9008",
-                        info="视频搜索成功， 状态修改为2",
+                        code="9018",
+                        info="视频搜索失败，回退状态为0",
                         trace_id=params['trace_id']
                     )
                 else:
-                    if int(params['process_times']) < 5:
-                        update_sql3 = f"""
-                            UPDATE {db_article}
-                            SET 
-                               content_status = %s,
-                               process_times = %s
-                            WHERE trace_id = %s;
-                                        """
-                        await self.mysql_client.async_insert(
-                            sql=update_sql3,
-                            params=(0, int(params['process_times']) + 1, params['trace_id'])
-                        )
-                        logging(
-                            code="9018",
-                            info="视频搜索失败，回退状态为0",
-                            trace_id=params['trace_id']
-                        )
-                    else:
-                        update_sql3 = f"""
-                            UPDATE {db_article}
-                            SET 
-                               content_status = %s,
-                               process_times = %s
-                            WHERE trace_id = %s;
-                                        """
-                        await self.mysql_client.async_insert(
-                            sql=update_sql3,
-                            params=(3, int(params['process_times']) + 1, params['trace_id'])
-                        )
-                        logging(
-                            code="9019",
-                            info="视频多次搜索失败，状态修改为3",
-                            trace_id=params['trace_id']
-                        )
+                    update_sql3 = f"""
+                        UPDATE {db_article}
+                        SET 
+                           content_status = %s,
+                           process_times = %s
+                        WHERE trace_id = %s;
+                                    """
+                    await self.mysql_client.async_insert(
+                        sql=update_sql3,
+                        params=(3, int(params['process_times']) + 1, params['trace_id'])
+                    )
+                    logging(
+                        code="9019",
+                        info="视频多次搜索失败，状态修改为3",
+                        trace_id=params['trace_id']
+                    )
         except Exception as e:
             if int(params['process_times']) < 5:
                 logging(

applications/deal/get_done_content_deal.py → tasks/task2.py

@@ -3,14 +3,13 @@
 """
 import asyncio
 
-from applications.static.config import db_article
+from static.config import db_article
 from applications.schedule import search_videos
 from applications.functions.log import logging
-from applications.static.config import mysql_coroutines
+from static.config import mysql_coroutines
 
 
-
-class ProcessDeal2(object):
+class MatchTask2(object):
     """
     定时执行任务
     """