调试

common/gpt4o_mini_help.py

@@ -146,5 +146,91 @@ class GPT4oMini:
         except Exception as e:
             return "这个视频，分享给我的老友，祝愿您能幸福安康"
 
+    @classmethod
+    def get_ai_mini_test_title(cls, title):
+        url = "http://aigc-api.cybertogether.net//aigc/dev/test/gpt"
+        payload = json.dumps({
+            "imageList": [],
+            "model": "gpt-4o-mini-2024-07-18",
+            "prompt": (
+                "1. 标题开头包含醒目emoji🔴，整体字符长度必须控制在10-25个字"
+                "2. 优化方向是面向老年人群体，群友们、退休人员、50~70后 等元素，可以适当在标题中表现出来。"
+                "3. 标题中可增加对内容的肯定的观点或态度，示例：听完觉得太对了，太香了、老外至今难以相信"
+                "4. 标题结尾可以根据标题内容增加一些引导语，格式参考示例但不必局限于示例。示例：你们见过吗、你听对不对、说的太好了、请听、太神奇了、一字一句都是金、句句难以辩驳、说得真好"
+                "5. 对于包含#话题和@人的标题，则去除@人元素后，利用剩下的全部信息生成标题。"
+                "6. 禁止编造任何信息，如原标题没有养老金，不能在生成的标题中出现养老金。"
+                "7. 标题内不能使用强引导分享点击的词句，如：快来看看、大家都听一听、值得一看、都看看吧、你也来看看吧、大家注意、都听听等。"
+                "8. 不能使用无实质信息和强烈诱导点击、紧急、夸张、震惊的描述，避免使用“震惊国人”、“速看”、“太震撼了”等类似描述。"
+                "9. 标题需要简洁、清晰，不要使用年轻一代的网络流行语，如：太燃了、佛系、躺平、内卷、炸裂等。"
+                "10、用语要符合50岁以上老年人的阅读习惯，例如：太好了、太真实了、涨知识了、看呆了、精彩十足、实话实说、通透、拍案叫绝、难以相信、句句在理、说的太好了、发人深省、说的非常好、句句大实话、泪流不止、一睹为快、瞠目结舌等"
+                "输出要求："
+                "请基于优秀标题要求及示例，生成吸引用户注意的标题"
+
+                "优秀标题示例"
+                "🔴街头采访趣事，朋友们你认为说的对不对"
+                "终于找到了这首歌，献给你！"
+                "这八大建议一经提出，手都拍红了"
+                "⭕六七十年代的《忠字舞》，你们见过吗？"
+                "老朋友你好吗？我想你了"
+                "农村顺口溜大实话"
+                "🔴逗逗你，看你开心不？"
+                "澳门的黄金发财树!太美了"
+                "各位退休的同学，请听！"
+                "🔴国庆为什么要放假，看完你就明白了！"
+                "农村大实话，说的太好了"
+                "越听越舒心，发给群友听~"
+                "🔴天地之间有杆称，说的真好❗"
+                "⭕幽默大实话，你看你开心"
+                "原来这就是中国如此强大的原因！"
+                "🔴如果微信倒闭了，存的钱怎么办？大家注意！"
+                "🔴真实的台湾，原来是这个样子的！"
+                "今天这三首歌，送给有缘的你！"
+                "不结婚不生孩子，老了之后会是什么下场？"
+                "⭕老两口深情演绎，感动全场！"
+                "🔴令人拍案叫绝的实话实说，句句难以辩驳！"
+                "我的天啊，太美了"
+                "🔥📢🔴十分通透的一段话，写得太棒了，请朋友们收好！"
+                "街头骗局，出门在外要提高警惕"
+                "🔴淄博再次爆火，老百姓乐坏了！"
+                "💖人山人海的景区里，一组图看看到底挤了多少人?"
+                "穿什么裤子最显年轻？"
+                "🔴淄博再次因为他而火！到底好不好，你来说了算！"
+                "🔴这个视频勾起多少人童年的回忆，你还记得吗"
+                "⭕人民币大实话，句句在理！"
+                "1900架无人机表演，龙飞凤舞堪比春晚"
+                "~~涨知识了，日本唯一不敢攻打的城市是这里"
+                f"请严格按照Goodcase示例输出，需要优化的标题为：{title}，返回新的标题。"
+            ),
+            "responseFormat": {
+                "type": "json_schema",
+                "json_schema": {
+                    "strict": True,
+                    "name": "share_script_result",
+                    "schema": {
+                        "type": "object",
+                        "properties": {
+                            "新标题": {
+                                "type": "string",
+                                "description": "生成新的标题"
+                            }
+                        },
+                        "required": ["新标题"],
+                        "additionalProperties": False
+                    }
+                }
+            }
+        })
+        headers = {'Content-Type': 'application/json'}
+        try:
+            response = requests.post(url, headers=headers, data=payload)
+            response_data = response.json()
+
+            data = json.loads(response_data.get('data', '{}'))
+            new_title = data["新标题"]
+            return new_title
+        except Exception as e:
+            return "这个视频，分享给我的老友，祝愿您能幸福安康"
+
 if __name__ == '__main__':
-    GPT4oMini.get_ai_mini_title("🔴这位美女说的太好了！这就是我们的大中国")
+    title = GPT4oMini.get_ai_mini_test_title("🔴这位美女说的太好了！这就是我们的大中国")
+    print(title)

video_rewriting/video_processor.py

@@ -139,7 +139,6 @@ class VideoProcessor:
 
                 cls.remove_files(mark)
                 file_path = cls.create_folders(mark)
-                new_title = cls.generate_title(video, title)
                 v_id = video["video_id"]
                 cover = video["cover"]
                 video_url = video["video_url"]
@@ -281,190 +280,201 @@ class VideoProcessor:
                                       "【 机器改造通知 】")
                     continue
 
-
-                # 上传视频和封面，并更新数据库
-                code = cls.upload_video_and_thumbnail(video_path, cover, v_id, new_title, task_mark, name, piaoquan_id,
-                                                   file_path, mark, channel_id, url, old_title, rule, video)
-                logger.info(f"[+][+][+][+][+][+] {name}的{task_mark}下的ID{url} 改造成功")
-
-                # 更新已使用的视频号状态
-                pq_url = f'https://admin.piaoquantv.com/cms/post-detail/{code}/detail'  # 站内视频链接
-                if name == "单点视频":
-                    sphdd_status = sqlCollect.update_shp_dd_vid(v_id)
-                    if sphdd_status == 1:
-                        from_user_name = video['from_user_name']  # 来源用户
-                        from_group_name = video['from_group_name']  # 来源群组
-                        source = video['source']  # 渠道
-                        channel_id = source
-                        text = (
-                            f"**站内视频链接**: {pq_url}\n"
-                            f"**渠道**: {source}\n"
-                            f"**来源用户**: {from_user_name}\n"
-                            f"**来源群组**: {from_group_name}\n"
-                            f"**原视频链接**: {video['video_url']}\n"
-                            f"**原视频封面**: {video['cover']}\n"
-                            f"**原视频标题**: {video['old_title']}\n"
-                        )
-                        Feishu.finish_bot(text, "https://open.feishu.cn/open-apis/bot/v2/hook/d2f751a8-5b0a-49ca-a306-1fda142707a9", "【 有一条新的内容改造成功 】")
-                if name == "快手推荐流" or name == "视频号推荐流":
-                    feed_status = sqlCollect.update_feed_vid(v_id)
-                if channel_id == "快手历史" or channel_id == "抖音历史" or channel_id == "视频号历史":
-                    explain = "历史爆款"
+                # 上传视频和封面，
+                oss_object_key, jpg = cls.oss_video_path(video_path, cover, v_id, task_mark, name,
+                                   file_path, channel_id, url, video)
+                if not oss_object_key:
+                    continue
+                new_old_title = cls.generate_title(video, title)
+                if channel_id == '快手搜索':
+                    new_test_title = cls.generate_test_title(video, title)
+                    title_list = [new_old_title, new_test_title]
                 else:
-                    explain = "新供给"
-                current_time = datetime.now()
-                formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
-                if name == "品类关键词搜索":
-                    first_category = task["first_category"]
-                    keyword_principal = task["keyword_name"]
-                    tag_first = f"一级品类_{first_category}"
-                    tag_keyword = f"关键词_{url}"
-                    if channel_id == "抖音搜索":
-                        tag_channel = "来源_抖音关键词"
-                    elif channel_id == "快手搜索":
-                        tag_channel = "来源_快手关键词"
-                    elif channel_id == "视频号搜索":
-                        tag_channel = "来源_视频号关键词"
-                    tag = f"{tag_first},{tag_keyword},{tag_channel}"
-                    tag_status = Tag.video_tag(code, tag)
-                    if tag_status == 0:
-                        logger.info(f"[+] {name}的{task_mark}下的ID{url}下的票圈视频{code},写入标签成功")
-
-                    secondary_category = task["secondary_category"]
-                    log_data = f"user:{url},,video_id:{v_id},,video_url:{video_url},,ai_title:{new_title},,voice:{voice},,first_category:{first_category},,secondary_category:{secondary_category},,keyword_principal:{keyword_principal},,tag:{tag}"
-                    values = [
-                        [
-                            name,
-                            task_mark,
-                            channel_id,
-                            url,
-                            str(v_id),
-                            piaoquan_id,
-                            old_title,
-                            title if title in ["原标题", "AI标题"] else "",
-                            new_title,
-                            str(code),
-                            formatted_time,
-                            str(rule),
-                            explain,
-                            voice,
-                            first_category,
-                            secondary_category,
-                            keyword_principal,
-                            pq_url
+                    title_list = [new_old_title]
+                for new_title in title_list:
+                    # 更新数据库
+                    code = cls.upload_video_and_thumbnail(oss_object_key, jpg, v_id, new_title, task_mark, name, piaoquan_id,
+                                                                mark, channel_id, url, old_title, rule, video)
+                    logger.info(f"[+][+][+][+][+][+] {name}的{task_mark}下的ID{url} 改造成功")
+
+                    # 更新已使用的视频号状态
+                    pq_url = f'https://admin.piaoquantv.com/cms/post-detail/{code}/detail'  # 站内视频链接
+                    if name == "单点视频":
+                        sphdd_status = sqlCollect.update_shp_dd_vid(v_id)
+                        if sphdd_status == 1:
+                            from_user_name = video['from_user_name']  # 来源用户
+                            from_group_name = video['from_group_name']  # 来源群组
+                            source = video['source']  # 渠道
+                            channel_id = source
+                            text = (
+                                f"**站内视频链接**: {pq_url}\n"
+                                f"**渠道**: {source}\n"
+                                f"**来源用户**: {from_user_name}\n"
+                                f"**来源群组**: {from_group_name}\n"
+                                f"**原视频链接**: {video['video_url']}\n"
+                                f"**原视频封面**: {video['cover']}\n"
+                                f"**原视频标题**: {video['old_title']}\n"
+                            )
+                            Feishu.finish_bot(text, "https://open.feishu.cn/open-apis/bot/v2/hook/d2f751a8-5b0a-49ca-a306-1fda142707a9", "【 有一条新的内容改造成功 】")
+                    if name == "快手推荐流" or name == "视频号推荐流":
+                        sqlCollect.update_feed_vid(v_id)
+                    if channel_id == "快手历史" or channel_id == "抖音历史" or channel_id == "视频号历史":
+                        explain = "历史爆款"
+                    else:
+                        explain = "新供给"
+                    current_time = datetime.now()
+                    formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
+                    if name == "品类关键词搜索":
+                        first_category = task["first_category"]
+                        keyword_principal = task["keyword_name"]
+                        tag_first = f"一级品类_{first_category}"
+                        tag_keyword = f"关键词_{url}"
+                        if channel_id == "抖音搜索":
+                            tag_channel = "来源_抖音关键词"
+                        elif channel_id == "快手搜索":
+                            tag_channel = "来源_快手关键词"
+                        elif channel_id == "视频号搜索":
+                            tag_channel = "来源_视频号关键词"
+                        tag = f"{tag_first},{tag_keyword},{tag_channel}"
+                        tag_status = Tag.video_tag(code, tag)
+                        if tag_status == 0:
+                            logger.info(f"[+] {name}的{task_mark}下的ID{url}下的票圈视频{code},写入标签成功")
+
+                        secondary_category = task["secondary_category"]
+                        log_data = f"user:{url},,video_id:{v_id},,video_url:{video_url},,ai_title:{new_title},,voice:{voice},,first_category:{first_category},,secondary_category:{secondary_category},,keyword_principal:{keyword_principal},,tag:{tag}"
+                        values = [
+                            [
+                                name,
+                                task_mark,
+                                channel_id,
+                                url,
+                                str(v_id),
+                                piaoquan_id,
+                                old_title,
+                                title if title in ["原标题", "AI标题"] else "",
+                                new_title,
+                                str(code),
+                                formatted_time,
+                                str(rule),
+                                explain,
+                                voice,
+                                first_category,
+                                secondary_category,
+                                keyword_principal,
+                                pq_url
+                            ]
                         ]
-                    ]
-                elif name == "抖音品类账号-1" or name == "抖音品类账号" or name == "视频号品类账号" or name == "快手品类账号":
-                    first_category = task["first_category"]
-                    tag_first = f"一级品类_{first_category}"
-                    if channel_id == "抖音" or channel_id == "抖音历史":
-                        tag_channel = "来源_抖音品类账号"
-                    elif channel_id == "快手" or channel_id == "快手历史":
-                        tag_channel = "来源_快手品类账号"
-                    elif channel_id == "视频号" or channel_id == "视频号历史":
-                        tag_channel = "来源_视频号品类账号"
-                    tag = f"{tag_first},{tag_channel}"
-                    tag_status = Tag.video_tag( code, tag )
-                    if tag_status == 0:
-                        logger.info(f"{name}的{task_mark}下的ID{url}下的票圈视频{code},写入标签成功")
-                    log_data = f"user:{url},,video_id:{v_id},,video_url:{video_url},,ai_title:{new_title},,voice:{voice},,tag:{tag}"
-                    values = [
-                        [
-                            name,
-                            task_mark,
-                            channel_id,
-                            url,
-                            str( v_id ),
-                            piaoquan_id,
-                            old_title,
-                            title if title in ["原标题", "AI标题"] else "",
-                            new_title,
-                            str( code ),
-                            formatted_time,
-                            str( rule ),
-                            explain,
-                            voice,
-                            first_category,
-                            pq_url
+                    elif name == "抖音品类账号-1" or name == "抖音品类账号" or name == "视频号品类账号" or name == "快手品类账号":
+                        first_category = task["first_category"]
+                        tag_first = f"一级品类_{first_category}"
+                        if channel_id == "抖音" or channel_id == "抖音历史":
+                            tag_channel = "来源_抖音品类账号"
+                        elif channel_id == "快手" or channel_id == "快手历史":
+                            tag_channel = "来源_快手品类账号"
+                        elif channel_id == "视频号" or channel_id == "视频号历史":
+                            tag_channel = "来源_视频号品类账号"
+                        tag = f"{tag_first},{tag_channel}"
+                        tag_status = Tag.video_tag( code, tag )
+                        if tag_status == 0:
+                            logger.info(f"{name}的{task_mark}下的ID{url}下的票圈视频{code},写入标签成功")
+                        log_data = f"user:{url},,video_id:{v_id},,video_url:{video_url},,ai_title:{new_title},,voice:{voice},,tag:{tag}"
+                        values = [
+                            [
+                                name,
+                                task_mark,
+                                channel_id,
+                                url,
+                                str( v_id ),
+                                piaoquan_id,
+                                old_title,
+                                title if title in ["原标题", "AI标题"] else "",
+                                new_title,
+                                str( code ),
+                                formatted_time,
+                                str( rule ),
+                                explain,
+                                voice,
+                                first_category,
+                                pq_url
+                            ]
                         ]
-                    ]
 
-                else:
-                    log_data = f"user:{url},,video_id:{v_id},,video_url:{video_url},,ai_title:{new_title},,voice:{voice}"
-                    values = [
-                        [
-                            name,
-                            task_mark,
-                            channel_id,
-                            url,
-                            str(v_id),
-                            piaoquan_id,
-                            old_title,
-                            title if title in ["原标题", "AI标题"] else "",
-                            new_title,
-                            str(code),
-                            formatted_time,
-                            str(rule),
-                            explain,
-                            voice
+                    else:
+                        log_data = f"user:{url},,video_id:{v_id},,video_url:{video_url},,ai_title:{new_title},,voice:{voice}"
+                        values = [
+                            [
+                                name,
+                                task_mark,
+                                channel_id,
+                                url,
+                                str(v_id),
+                                piaoquan_id,
+                                old_title,
+                                title if title in ["原标题", "AI标题"] else "",
+                                new_title,
+                                str(code),
+                                formatted_time,
+                                str(rule),
+                                explain,
+                                voice
+                            ]
                         ]
-                    ]
-                AliyunLogger.logging(channel_id, name, url, v_id, "视频改造成功", "1000", log_data, str(code))
-                text = (
-                    f"**通知类型**: 视频改造成功\n"
-                    f"**站内视频链接**: {pq_url}\n"
-                    f"**负责人**: {name}\n"
-                    f"**渠道**: {channel_id}\n"
-                    f"**视频主页ID**: {url}\n"
-                    f"**视频Video_id**: {v_id}\n"
-                    f"**使用音频音色**: {voice}\n"
-                )
-                Feishu.finish_bot(text,
-                                  "https://open.feishu.cn/open-apis/bot/v2/hook/e7697dc6-5254-4411-8b59-3cd0742bf703",
-                                  "【 机器改造通知 】")
-                if tags:
-                    Tag.video_tag(code, tags)
-                if values:
-                    if name == "王雪珂":
-                        sheet = "vfhHwj"
-                    elif name == "鲁涛":
-                        sheet = "FhewlS"
-                    elif name == "范军":
-                        sheet = "B6dCfS"
-                    elif name == "余海涛":
-                        sheet = "mfBrNT"
-                    elif name == "罗情":
-                        sheet = "2J3PwN"
-                    elif name == "王玉婷":
-                        sheet = "bBHFwC"
-                    elif name == "刘诗雨":
-                        sheet = "fBdxIQ"
-                    elif name == "信欣":
-                        sheet = "lPe1eT"
-                    elif name == "快手创作者版品类推荐流":
-                        sheet = "k7l7nQ"
-                    elif name == "抖音品类账号":
-                        sheet = "ZixHmf"
-                    elif name == "抖音品类账号-1":
-                        sheet = "61kvW7"
-                    elif name == "视频号品类账号":
-                        sheet = "b0uLWw"
-                    elif name == "单点视频":
-                        sheet = "ptgCXW"
-                    elif name == "快手品类账号":
-                        sheet = "ibjoMx"
-                    elif name == "品类关键词搜索":
-                        sheet = "rBAJT8"
-                    elif name == "快手推荐流":
-                        sheet = "9Ii8lw"
-                    elif name == "视频号推荐流":
-                        sheet = "hMBv7T"
-                    elif name == "快手小程序":
-                        sheet = "GeDT6Q"
-                    Feishu.insert_columns("ILb4sa0LahddRktnRipcu2vQnLb", sheet, "ROWS", 1, 2)
-                    time.sleep(0.5)
-                    Feishu.update_values("ILb4sa0LahddRktnRipcu2vQnLb", sheet, "A2:Z2", values)
-                    continue
+                    AliyunLogger.logging(channel_id, name, url, v_id, "视频改造成功", "1000", log_data, str(code))
+                    text = (
+                        f"**通知类型**: 视频改造成功\n"
+                        f"**站内视频链接**: {pq_url}\n"
+                        f"**负责人**: {name}\n"
+                        f"**渠道**: {channel_id}\n"
+                        f"**视频主页ID**: {url}\n"
+                        f"**视频Video_id**: {v_id}\n"
+                        f"**使用音频音色**: {voice}\n"
+                    )
+                    Feishu.finish_bot(text,
+                                      "https://open.feishu.cn/open-apis/bot/v2/hook/e7697dc6-5254-4411-8b59-3cd0742bf703",
+                                      "【 机器改造通知 】")
+                    if tags:
+                        Tag.video_tag(code, tags)
+                    if values:
+                        if name == "王雪珂":
+                            sheet = "vfhHwj"
+                        elif name == "鲁涛":
+                            sheet = "FhewlS"
+                        elif name == "范军":
+                            sheet = "B6dCfS"
+                        elif name == "余海涛":
+                            sheet = "mfBrNT"
+                        elif name == "罗情":
+                            sheet = "2J3PwN"
+                        elif name == "王玉婷":
+                            sheet = "bBHFwC"
+                        elif name == "刘诗雨":
+                            sheet = "fBdxIQ"
+                        elif name == "信欣":
+                            sheet = "lPe1eT"
+                        elif name == "快手创作者版品类推荐流":
+                            sheet = "k7l7nQ"
+                        elif name == "抖音品类账号":
+                            sheet = "ZixHmf"
+                        elif name == "抖音品类账号-1":
+                            sheet = "61kvW7"
+                        elif name == "视频号品类账号":
+                            sheet = "b0uLWw"
+                        elif name == "单点视频":
+                            sheet = "ptgCXW"
+                        elif name == "快手品类账号":
+                            sheet = "ibjoMx"
+                        elif name == "品类关键词搜索":
+                            sheet = "rBAJT8"
+                        elif name == "快手推荐流":
+                            sheet = "9Ii8lw"
+                        elif name == "视频号推荐流":
+                            sheet = "hMBv7T"
+                        elif name == "快手小程序":
+                            sheet = "GeDT6Q"
+                        Feishu.insert_columns("ILb4sa0LahddRktnRipcu2vQnLb", sheet, "ROWS", 1, 2)
+                        time.sleep(0.5)
+                        Feishu.update_values("ILb4sa0LahddRktnRipcu2vQnLb", sheet, "A2:Z2", values)
+                continue
         except Exception as e:
             AliyunLogger.logging(channel_id, name, url, v_id, f"改造失败{traceback.format_exc()}", "3001")
             logger.error(f"[+] {name}的{task_mark}任务处理失败:{traceback.format_exc()}")
@@ -536,6 +546,34 @@ class VideoProcessor:
             new_title = random.choice(titles)
         return new_title
 
+    @classmethod
+    def generate_test_title(cls, video, title):
+        """
+        生成新标题
+        """
+        if video['old_title']:
+            new_title = video['old_title'].strip().replace("\n", "") \
+                .replace("/", "").replace("\\", "").replace("\r", "") \
+                .replace(":", "").replace("*", "").replace("？", "") \
+                .replace("?", "").replace('"', "").replace("<", "") \
+                .replace(">", "").replace("|", "").replace(" ", "") \
+                .replace("&NBSP", "").replace(".", "。").replace(" ", "") \
+                .replace("'", "").replace("#", "").replace("Merge", "")
+        else:
+            return '这个视频，分享给我的老友，祝愿您能幸福安康'
+        if title == "原标题":
+            if not new_title:
+                new_title = '这个视频，分享给我的老友，祝愿您能幸福安康'
+        elif title == "AI标题":
+            if not new_title:
+                new_title = '这个视频，分享给我的老友，祝愿您能幸福安康'
+            else:
+                new_title = GPT4oMini.get_ai_mini_test_title(new_title)
+        else:
+            titles = title.split('/') if '/' in title else [title]
+            new_title = random.choice(titles)
+        return new_title
+
     @classmethod
     def download_and_process_video(cls, channel_id, video_url, video_path_url, v_id, video):
         """
@@ -556,20 +594,15 @@ class VideoProcessor:
         return new_video_path
 
 
-
-
-
     @classmethod
-    def upload_video_and_thumbnail(cls, new_video_path: str, cover: str, v_id, new_title: str, task_mark: str, name: str, piaoquan_id,
-                                   video_path_url: str, mark: str, channel_id: str, url: str, old_title: str, rule: str, video):
-        """
-        上传视频和封面到OSS，并更新数据库
+    def oss_video_path(cls, new_video_path: str, cover: str, v_id, task_mark: str, name: str,
+                                   video_path_url: str, channel_id: str, url: str, video):
+        oss_id = cls.random_id()
+        logger.info(f"[+] {name}的{task_mark}下的ID{url} 开始发送oss")
         """
+                上传视频和封面到OSS，并更新数据库
+                """
         try:
-
-            oss_id = cls.random_id()
-            logger.info(f"[+] {name}的{task_mark}下的ID{url} 开始发送oss")
-
             oss_object_key = Oss.stitching_sync_upload_oss(new_video_path, oss_id)  # 视频发送OSS
             logger.info(f"[+] {name}的{task_mark}下的ID{url} 发送oss成功{oss_object_key}")
 
@@ -581,28 +614,42 @@ class VideoProcessor:
                 if channel_id == "快手历史" or channel_id == "快手" or channel_id == '快手搜索' or channel_id == '视频号':
                     jpg = None
                 elif channel_id == "视频号历史":
-                    jpg_path = Oss.download_sph_ls( cover, video_path_url, v_id )
+                    jpg_path = Oss.download_sph_ls(cover, video_path_url, v_id)
                 elif channel_id == '单点视频':
                     if video['source'] != "快手":
-                        jpg_path = PQ.download_video_jpg( cover, video_path_url, v_id )  # 下载视频封面
-                if jpg_path and os.path.isfile( jpg_path ):
-                    oss_jpg_key = Oss.stitching_fm_upload_oss( jpg_path, oss_id )  # 封面上传OSS
-                    jpg = oss_jpg_key.get( "oss_object_key" )
+                        jpg_path = PQ.download_video_jpg(cover, video_path_url, v_id)  # 下载视频封面
+                if jpg_path and os.path.isfile(jpg_path):
+                    oss_jpg_key = Oss.stitching_fm_upload_oss(jpg_path, oss_id)  # 封面上传OSS
+                    jpg = oss_jpg_key.get("oss_object_key")
                 else:
                     jpg = None
-                code = PQ.insert_piaoquantv(oss_object_key, new_title, jpg, piaoquan_id)
-                logger.info(f"[+] {name}的{task_mark}下的视频ID{v_id}写入票圈后台成功")
-                sqlCollect.insert_task(task_mark, v_id, mark, channel_id)  # 插入数据库
-                current_time = datetime.now()
-                formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
-                if name == "单点视频":
-                    url = str(rule)
-                sqlCollect.insert_machine_making_data(name, task_mark, channel_id, url, v_id, piaoquan_id, new_title, code,
-                                                      formatted_time, old_title, oss_object_key)
-
-                return code
+                return oss_object_key, jpg
+            return None, None
+        except Exception as e:
+            logger.error(f"[+] {name}的{task_mark}上传视频和封面到OSS失败:{traceback.format_exc()}")
+            AliyunLogger.logging(channel_id, name, url, video["video_id"], "改造失败-上传视频和封面到OSS", "3001")
+            return None, None
+
+
+    @classmethod
+    def upload_video_and_thumbnail(cls, oss_object_key: str, jpg: str, v_id, new_title: str, task_mark: str, name: str, piaoquan_id,
+                                                            mark: str, channel_id: str, url: str, old_title: str, rule: str, video):
+        """
+        上传视频和封面到OSS，并更新数据库
+        """
+        try:
+            code = PQ.insert_piaoquantv(oss_object_key, new_title, jpg, piaoquan_id)
+            logger.info(f"[+] {name}的{task_mark}下的视频ID{v_id}写入票圈后台成功")
+            sqlCollect.insert_task(task_mark, v_id, mark, channel_id)  # 插入数据库
+            current_time = datetime.now()
+            formatted_time = current_time.strftime("%Y-%m-%d %H:%M:%S")
+            if name == "单点视频":
+                url = str(rule)
+            sqlCollect.insert_machine_making_data(name, task_mark, channel_id, url, v_id, piaoquan_id, new_title, code,
+                                                  formatted_time, old_title, oss_object_key)
+            return code
         except Exception as e:
-            logger.error(f"[+] {name}的{task_mark}上传视频和封面到OSS，并更新数据库失败:{traceback.format_exc()}")
+            logger.error(f"[+] {name}的{task_mark}更新数据库失败:{traceback.format_exc()}")
             AliyunLogger.logging(channel_id, name, url, video["video_id"], "改造失败-上传视频和封面到OSS", "3001")
             return