更好的阅读体验见飞书文档 长视频洗稿程序自动化——站在13天开通YouTube YPP，从0到1月入过万的自动化实操记录Zero大佬的肩膀上 https://rlwxa0zgke.feishu.cn/docx/NKA9dzov2o6rTnxW0FBcoD4...
在分享前先感谢大佬Zero的无私分享，Zero分享的这个帖子是年度最佳的帖子，如同轮船需要灯塔，那么人生也是需要引路人。
Zero大佬的帖子是2025年4月3号发布的，我在2025年5月12号对Zero大佬的帖子做了回应，指出关于剪辑自动化的出处
出于程序员的敏感，我去github也就找到了 轻量、灵活、易上手的Python剪映草稿生成及导出工具，构建全自动化视频剪辑/混剪流水线。
本项目的CapCut版本正于 https://github.com/GuanYixuan/pyCapCut 内开发，github地址 https://github.com/GuanYixuan/pyJianYingDraft
既然有了开源免费的github项目，那么说干就干
视频自动化剪辑.exe 链接: https://pan.baidu.com/s/1PbZc8RFD4dmL221O9wUgSw?pwd=assq 提取码: assq
这是第一个从 Zero 大佬得到启发做的软件，其实不算开发，github有现成的源码
Zero 大佬分享的帖子13天开通YouTube YPP，从0到1月入过万的自动化实操记录 我没有看到相关软件，代码确实是有分享，对于小白来说，就光有代码也用不了。
由于是辛苦开发的软件，我留的程序，但是同样有私心，免费授权一年给生财圈友，5个软件机器码都是一样的 生财圈友机器码专用问卷
统一给大家一年的软件授权使用，如下图，统一是365天，最大使用次数-1 为不限制使用
操作视频如下
长视频的做好后，就是YouTube Al 助手.exe 链接: https://pan.baidu.com/s/1a52CLPfH7tNfD6esPry6PQ?pwd=fx6i 提取码: fx6i ，操作视频如下
后面是关于长视频洗稿的三个软件
智能伪原创工具.exe
链接: https://pan.baidu.com/s/1_xpcWjeqOgK-cXkMKP2AcQ?pwd=w4eb 提取码: w4eb
通过网盘分享的文件：配音自动化.exe
链接: https://pan.baidu.com/s/1c_kMw8xp7KEqXGtR1f02Xw?pwd=9s4v 提取码: 9s4v
通过网盘分享的文件：AI图片生成器.exe
链接: https://pan.baidu.com/s/1ZTJnRVSUAK57DqBz82RMXw?pwd=2t2k 提取码: 2t2k
智能伪原创工具说明：
买ChatGPT API KEY网址 点击蓝色链接即可达，默认选 gpt-4o-mini，统一用 OpenAI模型，这里说明一下，建议还需要一些人工来处理，人工对原文或是洗稿后的文章再修改后再伪原创一遍
Zero大佬用的是Finetune Model，没整明白，就用gpt-4o-mini 来处理
配音自动化说明：
这个图片这里已经有提示了，点链接进去
配音这一块，那就是直接借鉴了，用的 minimaxi.com 来处理，其实核心原则也不难，就是调官方的接口
照着官方文档来调API，那么配音这一块就完成了
AI图片生成器说明
https://my.runware.ai/login 注册账号进行充值，消耗的是API KEY的钱，前面的OpenAI API Key前面有提过去哪买了
出图这一块也是同配音一样，也是调用API接口就好
付费这一块，实测我发风向标这个卡是可以付的
这里给出示例，程序需要不断完善
总的来说，一个事做出来了就不难，如果没做出来那就会很难，感谢生财，再次感谢Zero 分享，
最后无需要加我微信，授权后的授权码我会留在评价区长视频洗稿程序自动化——站在13天开通YouTube YPP，从0到1月入过万的自动化实操记录Zero大佬的肩膀上
在分享前先感谢大佬Zero的无私分享，Zero分享的这个帖子是年度最佳的帖子，如同轮船需要灯塔，那么人生也是需要引路人。
image.png
Zero大佬的帖子是2025年4月3号发布的，我在2025年5月12号对Zero大佬的帖子做了回应，指出关于剪辑自动化的出处
image.png
image.png
image.png
出于程序员的敏感，我去github也就找到了 轻量、灵活、易上手的Python剪映草稿生成及导出工具，构建全自动化视频剪辑/混剪流水线。
本项目的CapCut版本正于 https://github.com/GuanYixuan/pyCapCut 内开发，github地址 https://github.com/GuanYixuan/pyJianYingDraft
既然有了开源免费的github项目，那么说干就干
视频自动化剪辑.exe 链接: https://pan.baidu.com/s/1PbZc8RFD4dmL221O9wUgSw?pwd=assq 提取码: assq
这是第一个从 Zero 大佬得到启发做的软件，其实不算开发，github有现成的源码
Zero 大佬分享的帖子13天开通YouTube YPP，从0到1月入过万的自动化实操记录 我没有看到相关软件，代码确实是有分享，对于小白来说，就光有代码也用不了。
由于是辛苦开发的软件，我留的程序，但是同样有私心，免费授权一年给生财圈友，5个软件机器码都是一样的 生财圈友机器码专用问卷
image.png
image.png
统一给大家一年的软件授权使用，如下图，统一是365天，最大使用次数-1 为不限制使用
image.png
操作视频如下
一键生成视频工具演示效果.mp4
长视频的做好后，就是YouTube Al 助手.exe 链接: https://pan.baidu.com/s/1a52CLPfH7tNfD6esPry6PQ?pwd=fx6i 提取码: fx6i ，操作视频如下
YouTube Al 助手 操作教程.mp4
后面是关于长视频洗稿的三个软件
智能伪原创工具.exe
链接: https://pan.baidu.com/s/1_xpcWjeqOgK-cXkMKP2AcQ?pwd=w4eb 提取码: w4eb
通过网盘分享的文件：配音自动化.exe
链接: https://pan.baidu.com/s/1c_kMw8xp7KEqXGtR1f02Xw?pwd=9s4v 提取码: 9s4v
通过网盘分享的文件：AI图片生成器.exe
链接: https://pan.baidu.com/s/1ZTJnRVSUAK57DqBz82RMXw?pwd=2t2k 提取码: 2t2k
智能伪原创工具说明：
买ChatGPT API KEY网址 点击蓝色链接即可达，默认选 gpt-4o-mini，统一用 OpenAI模型，这里说明一下，建议还需要一些人工来处理，人工对原文或是洗稿后的文章再修改后再伪原创一遍
image.png
image.png
Zero大佬用的是Finetune Model，没整明白，就用gpt-4o-mini 来处理
image.png
配音自动化说明：
这个图片这里已经有提示了，点链接进去
image.png
image.png
image.png
配音这一块，那就是直接借鉴了，用的 minimaxi.com 来处理，其实核心原则也不难，就是调官方的接口
image.png
image.png
照着官方文档来调API，那么配音这一块就完成了
AI图片生成器说明
https://my.runware.ai/login 注册账号进行充值，消耗的是API KEY的钱，前面的OpenAI API Key前面有提过去哪买了
image.png
出图这一块也是同配音一样，也是调用API接口就好
image.png
付费这一块，实测我发风向标这个卡是可以付的
image.png
这里给出示例，程序需要不断完善
image.png
1_1.png
2_4.png
6_1.png
总的来说，一个事做出来了就不难，如果没做出来那就会很难，感谢生财，再次感谢Zero 分享，
最后无需要加我微信，授权后的授权码我会留在评价区
【从0开始做油管】13天开通YouTube YPP，从0到1月入过万的自动化实操记录
我在生财总共发过4篇帖 4篇全部都是精华帖
这一篇精华帖 是生财有术全站文章历史锚点第一名 拿下龙珠一颗
他承载的 是我2025年2月到4月的故事 记录我如何在YouTube从0月入过万（RMB）
在这一篇之后 还有三篇连载 地址如下
第二篇精华帖
他承载的 是我2025年5月到9月下旬的故事 详情可见生财有术·会员主题贴
记录我在两个高级YPP被封禁 50万粉丝人间蒸发后 如何重新起航 12天用新号拿下2000万播放的重生故事
第三篇精华帖
他承载的 是我从今年9月中旬到现在10月的故事【从0开始做YouTube】17天开通两个高级YPP 月入9000刀的实操记录
记录的是我 从0开始 分别用17天和24天 从0开通两个高级YPP 达成月入美刀过万目标的故事
第四篇精华帖
他承载的 是25年10月开始到10月底的故事【从0开始做YouTube】10月播放量4个亿 我做AI Shorts的爆款公式
这篇的主题是 高筑墙 广积粮 耐心做基建 以量变待质变
欢迎各位追连载 一定比小说都精彩
零：写在前面
今天做这个分享，
是给过去的自己做一个总结，
是给想做AI+YouTube的普通人一点参考，
是想拿个龙珠，希望能看到更多的前路和可能性
这篇帖有近4万字，是一篇大长篇，你将跟随我的视角，
完整的和我回顾我过往的四个月，是如何使用AI在Youtube从0到1赚到几万块钱的真实人生，
我会告诉你我在Youtube如何从0到1的全部经历，以及我迭代了几十个版本的各种自动化代码
看完这篇的人，你将得到
一个完整的AI+YouTube变现实操路径：如何从0粉0播新号到月入过万
从0到1的自动化心得及全部代码：我是如何将每天十几个小时的工作量压缩到1分钟以内的
我从0到1的做事心法：这才是重中之重
在开始下面的故事以前，我要说一句掏心窝子的话：
2025年是我心中真正的AI编程元年，这是真正的全民平权的工业革命
从现在开始，你必须要学会使用AI编程，这一点也不难，你甚至不需要学1个代码，
你只需学会具象化你的需求，也就是学会和AI对话的方式，
我从23年就买过GPTPlus，用到今天，7788的AI我都用过，
到了24年 虽然有了Chatgpt4和Claude3.5 但仍不够好用，
今天的Claude3.7，虽然也没有那么好用，但已经非常能用了，
今后的AI，还会数倍强于现在，所以你必须现在就用起来，
在大部份人都还不会用 不知道用的时候，做第一批吃上红利的人，
我常打的比方，就是只要现在你开始学习AI编程，
就等于在Iphone4发布的前夜，你买了苹果的股票，
你明白这是多大的含金量吗？
新一轮的工业革命在即，全新的金矿即将面世，你必须先做好准备，
接下来，让我们正式开启正文...
下面是过去几个月，我自己做Youtube纯AI赛道的一点成绩：
24年11月1日，我从0开始，发第一个作品，13天，账号1达成千粉4000小时，开通高级Ypp；
image.png
仍然是从0开始，账号2从24年11月25、账号3从12月3日发第一个作品，这两个账号也达成了高级Ypp，耗时长一点，一个35天，一个85天；
image.png
image.png
目前跑的最高的一个作品，就是两周前的事，给我赚了1000多刀，至今仍然在给量；
image.png
很多人觉得这有点nb，其实在我看来没什么特别的，
这些成绩，并非来自朝夕一瞬，而是过往的厚积薄发。
23年我做AI知识付费，从0开始，47天，抖音3.5万粉，做私域原创7万字抖音+AI副业课程，纯利7万+
image.png
image.png
24年我做红包封面，从0开始，做小红书店铺用MJ做红包封面，1个月粉丝1.5万，纯利3万+
image.png
image.png
再往前，我做母婴号，私域卖自己整理的十万字孕产攻略表格，纯利20余万
image.png
另外还有一些鸡毛蒜皮的，卖教辅资料赚过几百块、写AI公众号赚过几千块，这些就不谈了
上面这些成绩，很普通，我也就只是一个普通人，
我没有什么nb的学历，大款的背景，我没赚到什么大钱，没什么超然的本事，
我所有的成绩，都是：
0成本、冷启动、跨行业、1个人、1台Mac、以副业的形式做出来的，还要上班带娃
把这几个词连在一起来做事，本应该是十死无生的，
但我仍不止一次的，能做出些许成绩
很多人都说我nb，但我不以为然，
我的经历 是每个普通人都能达到的境界
就算你是个0基础0起步的 最普通的人 你也都能达到我这般的成绩
这不是站着说话不腰疼，
这并不困难，也不难理解，这就是这个世界的运行规律，
因为
生财 确实有术
一：生财“术”之一：只做你感兴趣的领域
一切都是来源于此，兴趣，就是万物的原点，
你绝做不成你不感兴趣的事情
以AI+Youtube这件事为例，在我开始做这个之前 我甚至都不看YouTube
我为什么在24年11月选择从0开始，突然做这么个玩意？
我感兴趣的点在于：
自媒体赚钱，不受地点时间限制（上班累的要死）
“可以不受国内繁文缛节的管控”（之前做AI副业被抖音管控的要死）
我想做“不出镜、不出声、不找素材、不写文案” 还能赚钱的东西（之前出镜做母婴号累的要死）
那什么才是可以来钱、还能不出镜、不出声、不找素材、不写文案，还不受国内管控的事情呢？
YouTube+AI视频+AI配音+AI文本，完美适配
但这些东西在24年以前，是不太现实的东西
但从24年开始 这些东西已经可以投入实战了
要说为什么我会选择去看Youtube，还是多亏了亦仁老兄说的那个超级风向标
我在YouTube上随便刷了刷 偶然碰到的一个博主 就是下面这位
奇幻研究所https://www.youtube.com/@qihuanyanjiu/videos
24年11月的时候 他大部分的视频播放，都有5-10万+
image.png
image.png
他的视频，全部都是AI图片轮播+AI配音
image.png
image.png
image.png
通过他 我刷找到了更多的同类型账号 以及整个生态的文案来源
就是下面这种露脸的真人口播号 播放数据更是高到离谱的几十万甚至上百万
https://www.youtube.com/@seeker7603
https://www.youtube.com/@wenzhaostudio
https://www.youtube.com/@emmashram
image.png
刷到他们真的不难 我相信只要你看得懂中文 稍微玩一会YouTube 你就一定刷到过类似的视频
你多点几个关注 他们就都出来了 真的非常好找 真的
看得多了，我发现他们的视频当中，有两个领域是播放量最高的，
宗教类，是佛教和基督教
预言类，是一些危言耸听的道听途说
所以，我的三个Ypp账号，
第一个，就是专精佛学内容；第二个，就是专精预言型内容；第三个，就是专精基督教内容
他们也先后都开通了Ypp变现资格
二：生财“术”之二：抄出一个MVP
所有生意的本质都是抄，别想着创什么新，
抄生二、二生三、三生万物，抄就完了
接下来 我就要发作品对标他们近期播放量最高的视频了
我的理想状态是什么呢？
我的作品=AI文本+AI视频+AI配音 最好能有个什么东西 能我一键就把他们都给生成就好了
这就是最理想的形态
但 最开始的时候 别想那么多有的 没的 好的 坏的
用最笨的方法来试错
做 就是真理 哪怕做的是最烂的东西也可以
只有实践 才出真知
岸上学游泳 永远学不会 扔到水坑里 呛几次你就会了
我最开始的作品 文本是我自己手工伪原创的 3000字的稿件，我需要1小时才能改好
再后来 我甚至还招了总结了SOP招了兼职还帮我写，感兴趣的可以直接看我的SOP改写细节
AI 配音是找的免费的https://ttsmaker.cn/
他好一点的音色 转一次有500文本的限制 转一次需要几分钟
我是一段段手工剪切粘贴进去 然后一段段下载出来的
AI视频 那时候还没有AI视频 都是AI图片轮播 我是一段段文本翻译成英文然后一段段送给MJ
然后一张张下载下来的
最后 还要把这些都手动都拖到剪映里 一段段的校准上百张图片的展示长度...
这就是我最开始的工作流 全是tmd手工
做一个10分钟左右的视频 要花上tmd几个小时
就这么笨的方法 我强行更了1个礼拜 每天更新
那段时间真的做得想死
11月1号更到11月7号 我的视频头一次突破几千播放，现在看起来多了一些
image.png
次日 我又有视频突破万播
image.png
次次次日 我的视频突破了好几万播
image.png
到此为止 MVP已成
我知道了 这件事走得通
三：生财“术”之三：具象化你的需求
到了这里，你就要具象化你的需求了，
什么是具象化的需求？
比如说你有一段文本，你需要隔几句话就均匀一些的将一小段文本另起一行，
这算是一个需求，但这不是具象化的需求，
因为你会遇到很多逻辑上的漏洞，比如说：
“均匀”是多少？是按照字数算还是标点算？
如果按字数算，多少字合适？
如果到了既定的字数，直接另起一行导致一句完整的话被截断怎么办？
如果按标点算，要从哪些标点另起一行？
有的标点段落长有的标点段落短，搞下来长短不一怎么办？
这些就是逻辑漏洞，所以如果具象化的需求 ，那就是
在每100字的位置，另起一行，
如果100字的位置不是标点符号，就向后顺延，直到找到下列标点的任意一个“。！？，；” 再另起一行
这样才是具象化的需求，如果你还不会这样说话，你务必需要从现在开始刻意练习
自动化，是因人而异的，每个人/每个项目的需求都不一样，
你需要将你的需求列出来，然后一个个的具象化，一步步的用AI来实现
对我来说，我的需求：
我想要能一键批量下载YouTube，因为他每天消耗我半小时
我想要能一键伪原创成千上万的文本，因为他每天消耗我两三个小时
我想要能一键给文本配好几百上千张配图，因为他每天消耗我两三个小时
我想要能一键给文本配好配音，因为他每天消耗我一两个小时
我想要能一键剪辑让声音和图片匹配，还要有关键帧动画过度的效果，因为他每天消耗我三四个小时
这就是我的需求，这些工序原本每天要消耗我七八个小时甚至更多，
你要先有了这些需求，才有后续的内容，
你连问题都不会提，你就不会有答案，
那么如何来具象化你的需求？
逻辑就是，你要的到底是什么？
然后再问，你要的到底是什么？
然后再问，你要的到底是什么？
直到你问的清晰到不能再细化了为止
比如说，我想要一键剪辑，
让我的文本声音和图片匹配，图片还要有关键帧和动画转场的效果，
那么问自己
你要的是怎样匹配？要的是怎样的关键帧和转场效果？
一张图可以配一两句话，但也不绝对，句子太长了也不行，太短了也不行，要合适一点
关键帧就是那种让图片可以匀速的上下左右移动，外加放大缩小效果的东西
动画转场就是闪黑 闪白 翻页 淡入 淡出 那种图片和图片之间的过渡效果
那么再问自己
什么是合适？什么是匀速？到底是要向上下左右哪个方向移动？到底是要放大还是缩小？放多大缩多小？你要的是哪些转场效果？转场效果的持续时长是多少？
首先，什么是合适呢？在仔细观察对标之后，你能发现，每张图片的展示时长是有区间的，有的长有的短，但图片的开始展示和结束展示时间 一定和字幕的开始和结束时间一致，
也就是说，一张图片的开始展示，一定是和一段字幕一起开始，一张图片结束展示，也一定是和一段字幕一起结束，图片切换会发生在字幕的自然断点处，这样观感上会好。
那么什么是合适？我的“合适”的逻辑就是：
每张图片的最小展示时间为 5 秒
先找到5秒后第一个字幕的结束点，他就是第一个图片的结束展示时间
无缝衔接展示下一张图片，然后重复上面两点
那么关键帧呢？到底是要向上下左右哪个方向移动？到底是要放大还是缩小？放多大缩多小？
给每个图片都打上两个关键帧点，一个在开始时间，一个在结束时间
位移范围设置为 ±0.21（约为画面宽度/高度的 21%），这样就是放大和缩小区间，
随机选择左、右、上、下的任意一个方向
这样所有的图片就都有了匀速的移动和放大缩小
那么要的是哪些转场效果？转场效果的持续时长是多少？
选择渐显、缩放、弹跳、雨刷四种效果
给每张图片之间都随机加上四种效果之一，持续时间是0.7秒
这样所有的图片就都有了动画过渡
这就是具象化的需求
四：生财“术”之四：开始提效
要想提效 招兼职 定SOP招人帮我处理固然是一个办法 但来的人实在是水平太差（主要是发不起工资）
经常分包出去 结果还要返工 效率还不如我自己来 非常的不靠谱
所以在具象了你的需求之后，就该自动化登场了
一句话概括我的代码，他可以将我一天10个小时甚至更多的工作量，降低到1分钟以内
不管它是脑力工作，还是机械操作，
是的，就是这么变态。
我需要提效的关键节点 其实就是下面这些
YouTube文件批量下载
伪原创改写数万字的文本
将数万字的文本对应批量出图
数万字文本的配音生成
数百张图片和字幕BGM的剪辑拼接
在你具象化你的需求之后，你就可以拿着你的需求去找AI了
这些东西，我虽然貌似知道了想要怎样的效果，但如何才能实现自动化？
对当时的我来说，我完全不知该怎么办
这里 就是我命运的分水岭
因为我想到了询问AI 我问AI
“有没有什么方法 能帮我这样那样的提升效率啊”
这一个问题 打开了我新世界的大门
他告诉我了一个东西叫做python
你要知道 在这个时间点之前 我一行代码也没有写过
你明白当我看到他吐出一行行代码的时候那是什么感觉吗？
是的 是想死的感觉 他写的东西我一个字也看不懂 他写的代码我也一行都不想看
但我为了能提效，硬着头皮一行一行的问他那是什么 要怎么做
于是我在AI的指导下
人生中第一次打开了一个东西叫做terminal
打出了人生中第一个python+空格
然后就收到了人生中第一个error报错
使出了我最熟练的复制粘贴把error发给AI
然后就收到了更多的error报错
.....
现在回头看 我知道这个东西叫做AI编程 叫做cursor 叫做windsurf
但在那时候 我p也不知道
时至今日 仅仅几个月的时间 我在AI编程买的会员费超过100刀
还不算我换了6个邮箱硬薅cursor免费额度的羊毛直到他BAN了我的IP
接下来关于自动化的内容 对没有需求的人来说 枯燥且乏味
但对需要它的人来说 无异于是金山银山
为什么说你一定要学会AI编程 因为每个人每个项目的需求 一定是不一样的
要想数倍的提效 你必须学会AI编程 来根据你的需求定制化 打造你属于你自己的兵器
然后在使用过程中不停迭代不停迭代不停迭代，他就会变得越来越好用，越来越趁手
你的武功才能最大限度的发挥
AI编程基本等于没有门槛 只要你多用用就好
下面的代码 如果你不会用/看不懂/有报错
请你去问AI 就能解决99%的问题
YouTube 批量下载自动化：从耗时半小时降低到小于1分钟
首先 YouTube下载，我本来是用这种网页 https://yt1d.com/en306/
问题是下载的又慢 又要看广告 只能1个1个的弄 还经常不稳定失效 特别的烦
你接下来看到的 是我迭代了几十个版本后的YouTube下载py
极其稳定 可批量下载 及视频的封面图
只需要新建一个word 把你想要的视频链接塞到里面 运行即可 真的非常好用
当然你需要安装一些依赖 修改一些路径
如果YouTube更改规则导致你下载失败 只需要一行简单的
pip install -U yt-dlp
就能升级到最新版 解决99%下载不了的问题
import os
import sys
import time
import random
from pathlib import Path
from datetime import datetime
from docx import Document
import yt_dlp
def get_current_time():
"""获取当前时间的格式化字符串"""
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def log_message(message, level="INFO"):
"""打印带时间戳的日志"""
print(f"[{get_current_time()}] [{level}] {message}")
def read_youtube_links():
"""从桌面剪辑文件夹中的youtube.docx读取视频链接"""
try:
获取桌面剪辑文件夹路径
desktop_path = str(Path.home() / "Desktop")
clips_folder = os.path.join(desktop_path, "Youtube")
docx_path = os.path.join(clips_folder, "youtube.docx")
if not os.path.exists(docx_path):
log_message(f"文件不存在: {docx_path}", "ERROR")
return None
读取文档
doc = Document(docx_path)
links = []
从每个段落中提取链接
for para in doc.paragraphs:
text = para.text.strip()
if "youtube.com" in text or "youtu.be" in text:
links.append(text)
log_message(f"从文档中读取到 {len(links)} 个链接")
return links
except Exception as e:
log_message(f"读取文件失败: {str(e)}", "ERROR")
return None
def download_videos(links):
"""下载YouTube视频"""
if not links:
log_message("没有找到要下载的链接", "WARNING")
return
创建下载目录在剪辑文件夹中
desktop_path = str(Path.home() / "Desktop")
output_dir = os.path.join(desktop_path, "Youtube", "YouTube下载")
os.makedirs(output_dir, exist_ok=True)
yt-dlp配置
ydl_opts = {
'format': 'bestaudio/best',
'paths': {'home': output_dir},
'postprocessors': [{
'key': 'FFmpegExtractAudio',
'preferredcodec': 'mp3',
'preferredquality': '192',
}, {
'key': 'FFmpegThumbnailsConvertor',
'format': 'jpg'
}],
'writethumbnail': True, # 下载缩略图
'outtmpl': {
'default': '%(title)s.%(ext)s',
'thumbnail': '%(title)s.%(ext)s' # 缩略图文件名格式
},
'cookiesfrombrowser': ('chrome',),
'proxy': 'http://127.0.0.1:7890',
'verbose': True,
'no_warnings': False,
'extract_flat': False,
添加重试和错误处理选项
'retries': 10,
'fragment_retries': 10,
'file_access_retries': 5,
'retry_sleep': True,
'sleep_interval': 3,
'max_sleep_interval': 7,
'sleep_interval_requests': 1,
添加网络相关选项
'socket_timeout': 30,
'http_chunk_size': 10485760, # 10MB
'buffersize': 1024,
添加地理位置绕过
'geo_bypass': True,
'geo_bypass_country': 'US',
添加请求头
'http_headers': {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Accept': '*/*',
'Accept-Language': 'en-US,en;q=0.9',
'Accept-Encoding': 'gzip, deflate',
'Origin': 'https://www.youtube.com',
'Referer': 'https://www.youtube.com/',
'Connection': 'keep-alive'
}
}
total = len(links)
success = 0
for i, url in enumerate(links, 1):
try:
添加随机延时，避免被限制
delay = random.uniform(2, 5)
log_message(f"等待 {delay:.1f} 秒后开始下载...")
time.sleep(delay)
log_message(f"正在处理第 {i}/{total} 个视频...")
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
error_code = ydl.download([url])
if error_code == 0:
success += 1
log_message(f"视频下载成功: {url}")
else:
log_message(f"视频下载失败: {url}", "ERROR")
except Exception as e:
log_message(f"下载失败: {str(e)}", "ERROR")
log_message(f"跳过视频: {url}", "WARNING")
log_message(f"下载完成! 成功: {success}/{total}")
def main():
"""主函数"""
try:
log_message("开始运行YouTube下载程序")
读取链接
links = read_youtube_links()
if links:
log_message(f"成功读取 {len(links)} 个链接")
下载视频
download_videos(links)
else:
log_message("未能读取到任何链接", "ERROR")
except KeyboardInterrupt:
log_message("用户中断下载", "WARNING")
sys.exit(1)
except Exception as e:
log_message(f"程序执行出错: {str(e)}", "ERROR")
sys.exit(1)
if __name__ == "__main__":
main()
至此 YouTube批量下载的问题解决了
我再也不用一个个的操作了 只需要批量粘贴对标的视频到word里
然后运行代码 工作量直接从半小时降到了小于1分钟
所有的对标就都下载到我指定的文件夹里了
伪原创文本自动化：从耗时1小时降低到小于1分钟
这是最难的部分 因为这种AI视频 文本就是灵魂
特别是宗教类的内容 出处 释义 读法 文言文 每一样都不容易
chatgpt4 4o mini claude3.5sonnect haiku 还有什么豆包 火山 MCP
只要有的我全都试过 不论你怎么调教prompt 都没有办法写出合格的伪原创
所以我又问了AI 到底该怎么办
他给了我一个方案：Finetune Model
这个东西 说是这个项目的灵魂也不为过
Finetune Model 就是微调模型 具体的你可以看官方文档
https://platform.openai.com/docs/guides/fine-tuning
简单来说 在这个项目里 他的作用
就是你给他1篇原文+1篇你伪原创的改文 他就能学会你的文笔
你以后用这个模型 给他原文 他就能按照你的改法 给出伪原创的文章来
这个样本好搞 毕竟我之前人工精校了N篇伪原创
所以我给了他30组对照样本 训练费用只花了5刀
至于怎么训练这个模型 Openai的官网有详细教程（教程我一个字也看不懂也不需要懂 不会就问AI）
这样改写出来的效果 可以说是全网最强 没有之一
然后 将下载好的文本 通过API调用的方式发送给模型 等待他写完返回 就可以了
下面是改写文本的代码
from openai import OpenAI
import os
from docx import Document
import string
import time
import tiktoken
import openai
import re
import traceback
class TextRewriter:
def __init__(self):
"""初始化改写器"""
初始化API客户端
直接在代码中设置 API key
api_key = "" # 请将此处替换为您的实际API密钥
self.openai_client = OpenAI(
base_url="https://api.openai.com/v1",
api_key=api_key
)
====== 配置项 ======
文本处理相关配置
self.MAX_TEXT_LENGTH = 12000 # 单次处理的最大文本长度
self.MAX_RETRIES = 5 # 最大重试次数
self.ENABLE_SECOND_ROUND = True # 是否启用第二轮改写
改写幅度配置
self.rewrite_ratio = {
'min': 60, # 最小改写幅度（百分比）
'max': 85 # 最大改写幅度（百分比）
}
翻译模式字符比例配置
self.translation_ratio = {
'min': 100, # 最小字符比例（百分比）
'max': 400 # 最大字符比例（百分比）
}
英文到中文翻译字符比例配置
self.en_to_cn_ratio = {
'min': 20, # 最小字符比例（百分比）
'max': 100 # 最大字符比例（百分比）
}
模型配置
self.MODEL_CONFIG = {
微调模型配置
'ft_model': {
'name': "ft:gpt-4o-mini-2024-07-18:personal:", # 模型名称
'temperature': 0.9 # 温度参数
},
GPT-4模型配置
'gpt4o_model': {
'name': "gpt-4o-mini-2024-07-18", # 模型名称
'temperature': {
'default': 0.75, # 默认温度
'translation': 0.7, # 翻译模式温度
'min': 0.2, # 最小温度
'step': 0.1 # 每次重试减少的温度值
},
'max_length': 16000, # 最大长度
'max_tokens': 16000 # 最大token数
}
}
API配置
self.claude_client = OpenAI(
base_url="",
api_key=""
)
====== 提示词配置 ======
英文基督教内容提示词
self.en_christian_prompt = """You are a professional Christian content editor. Your task is to rewrite the given English text with the following requirements:
1. Content Requirements:
- Maintain biblical accuracy
- Make the language more vivid and easy to understand
2. Improve the clarity and fluency of the article, which can be adjusted from the original text, and must be significantly different from the original text
3. Correct all typos and grammatical errors, without any content in parentheses
4. Only provide the rewritten content, without explanation"""
中文基督教内容提示词
self.cn_christian_prompt = """你是一位基督教专家。你的任务是改写给定的文本，要求：
1. 使语言更加生动易懂
2. 改善文章的清晰度和流畅性，可以对原文进行删减和调整，要和原文明显不一样
3. 修正所有错别字和语法错误，不要有括号内容
4. 只提供改写后的内容，无需解释"""
通用内容提示词
self.general_prompt = """你是一位内容编辑专家。你的任务是改写给定的文本，要求：
1. 使语言更加生动易懂
2. 改善文章的清晰度和流畅性，可对原文进行删减和调整，要和原文明显不一样，相似度不能高于5%
3. 修正所有错别字和语法错误，不要有括号内容
4. 只提供改写后的内容，无需解释"""
提取主题改写提示词
self.topic_extraction_prompt = """你是一位内容编辑专家。你的任务是改写给定的文本，要求：
1. 将文本的大纲进行提取，要保证日期时间和引用内容的准确性
2. 根据文章大纲进行重写，文章逻辑要清晰流畅
3. 不要有错别字和语法错误，不要有括号内容
4. 只提供改写后的内容，无需解释"""
西班牙语翻译提示词
self.spanish_prompt = """Por favor, traduce el siguiente texto al español. Requisitos:
1. Mantén el significado original y todos los detalles del texto
2. Asegúrate de que la traducción sea precisa y completa
3. Mantén un tono profesional y natural
4. Conserva toda la terminología técnica y específica
5. Mantén la misma estructura y organización del texto original
6. La traducción debe ser fiel al original, sin omitir ni agregar información
7. Utiliza expresiones naturales en español manteniendo el mismo nivel de formalidad
Importante: La traducción debe ser detallada y mantener todos los matices del texto original."""
中文到英文翻译提示词
self.chinese_to_english_prompt = """You are a professional translator. Your task is to translate the given Chinese text into English with the following requirements:
1. Translation Requirements:
- Make the language vivid and easy to understand
- Improve the clarity and fluency while maintaining the original meaning
- Use natural and idiomatic English expressions
2. Content Requirements:
- Maintain accuracy of all details and technical terms
- Remove any content in parentheses or special symbols
- Fix any typos or grammatical errors
3. Only provide the translated content in English, without any explanations or notes"""
英文到中文翻译提示词
self.english_to_chinese_prompt = """你是一位翻译改写专家。你的任务是将给定的文本翻译并微调，要求：
1. 使翻译后的语言更加生动易懂 符合中文语境
2. 改善文章的清晰度和流畅性
3. 修正所有错别字和语法错误，不要有括号内容
4. 只提供改写后的内容，无需解释"""
GPT-4 本地化改写提示词
self.gpt4_localization_prompt = """You are a professional content editor. Your task is to rewrite the given Chinese text with the following requirements:
1. Content Requirements:
- Replace Chinese place names and person names noun concepts in the text,with common American names for better understanding by American audience
- Make the language more vivid and easy to understand
2. Improve the clarity and fluency of the article, which can be adjusted from the original text
3. Correct all typos and grammatical errors, without any content in parentheses
4. 在满足上述条件后，将中文全部翻译成英文
5. Only provide the rewritten content, without explanation"""
第二轮改写提示词
self.general_prompt_round2 = """你是一位内容编辑专家。你的任务是将给定的文本进行纠错，要求：
1. 使语言更加易懂，修正所有错别字和语法错误
2. 只提供改写后的内容，无需解释
"""
第二轮基督教内容提示词
self.cn_christian_prompt_round2 = """你是一位基督教专家。你的任务是纠错给定的文本，要求：
1. 使语言更加易懂，修正所有错别字和语法错误
2. 只提供改写后的内容，无需解释
"""
def read_docx(self, file_path):
"""读取docx文件内容"""
try:
doc = Document(file_path)
paragraphs = []
for para in doc.paragraphs:
text = para.text.strip()
if text: # 只添加非空段落
paragraphs.append(text)
return ''.join(paragraphs)
except Exception as e:
print(f"读取文件时出错: {str(e)}")
return None
def save_to_docx(self, text, file_path):
"""保存内容到docx文件"""
try:
doc = Document()
将文本按换行符分割成段落
paragraphs = text.split('')
for para in paragraphs:
if para.strip(): # 只添加非空段落
doc.add_paragraph(para.strip())
doc.save(file_path)
return True
except Exception as e:
print(f"保存文件时出错: {str(e)}")
return False
def get_unique_filename(self, base_path):
"""获取唯一的文件名，如果文件存在则在文件名后加上数字"""
try:
确保输出到"改写"文件夹
rewrite_folder = os.path.join(os.path.expanduser("~/Desktop"), "剪辑/改写")
if not os.path.exists(rewrite_folder):
os.makedirs(rewrite_folder)
获取原始文件名和扩展名
filename = os.path.basename(base_path)
name, ext = os.path.splitext(filename)
生成新的文件路径
counter = 1
new_path = os.path.join(rewrite_folder, f"{name}{ext}")
while os.path.exists(new_path):
new_path = os.path.join(rewrite_folder, f"{name}_{counter}{ext}")
counter += 1
return new_path
except Exception as e:
print(f"生成文件名时出错: {str(e)}")
return None
def get_docx_files(self):
"""获取改写文件夹中的docx文件"""
try:
确保桌面上的"改写"文件夹存在
rewrite_folder = os.path.join(os.path.expanduser("~/Desktop"), "剪辑/改写")
if not os.path.exists(rewrite_folder):
os.makedirs(rewrite_folder)
print(f'已创建文件夹：{rewrite_folder}')
print('请将要处理的文件放入此文件夹，然后重新运行程序')
return []
获取文件夹中的所有非隐藏的docx文件
docx_files = []
for file in os.listdir(rewrite_folder):
跳过隐藏文件和临时文件
if (not file.startswith('.') and not file.startswith('~')) and file.endswith('.docx'):
file_path = os.path.join(rewrite_folder, file)
if os.path.isfile(file_path):
docx_files.append((file, file_path))
if not docx_files:
print(f'在 {rewrite_folder} 中没有找到.docx文件')
print('请将要处理的文件放入此文件夹，然后重新运行程序')
return []
按自然排序对文件进行排序
docx_files.sort(key=lambda x: self.natural_sort_key(x[0]))
显示找到的文件
print("找到以下文件：")
for i, (filename, _) in enumerate(docx_files, 1):
print(f"{i}. {filename}")
print()
return docx_files
except Exception as e:
print(f"获取文件列表时出错: {str(e)}")
traceback.print_exc() # 打印详细的错误信息
return []
def count_tokens(self, text):
"""使用tiktoken准确计算token数量"""
if text is None:
return 0
try:
encoding = tiktoken.get_encoding("cl100k_base")
return len(encoding.encode(str(text))) # 确保text是字符串
except Exception as e:
print(f"Token计算失败: {str(e)}")
如果tiktoken失败，使用简单估算（每4个字符约1个token）
return len(str(text)) // 4
def calculate_cost(self, input_tokens, output_tokens, is_finetune=False):
"""计算API调用成本"""
if input_tokens is None or output_tokens is None:
return 0
if is_finetune:
微调模型的价格（每1M token）
input_price = 0.3 / 1000000 # 输入价格 ($0.3/1M tokens)
output_price = 1.2 / 1000000 # 输出价格 ($1.2/1M tokens)
else:
GPT-4的价格（每1M token）
input_price = 0.15 / 1000000 # 输入价格 ($0.15/1M tokens)
output_price = 0.6 / 1000000 # 输出价格 ($0.6/1M tokens)
return (input_tokens * input_price + output_tokens * output_price)
def split_text_by_length(self, text, max_length=None):
"""将文本按长度分段"""
if max_length is None:
max_length = self.MODEL_CONFIG['gpt4o_model']['max_length']
segments = []
current_segment = ""
for paragraph in text.split(''):
if len(current_segment) + len(paragraph) + 1 <= max_length:
current_segment += (paragraph + '')
else:
if current_segment:
segments.append(current_segment.strip())
current_segment = paragraph + ''
if current_segment:
segments.append(current_segment.strip())
return segments
def process_segment(self, text, mode):
"""处理单个文本段落"""
import time
确保mode是整数
mode = int(mode)
生成提示词
if mode == 1: # 通用模式
if hasattr(self, 'is_second_round') and self.is_second_round:
prompt = f"{self.general_prompt_round2}请改写以下文本：{text}"
else:
prompt = self.general_prompt + "" + text
elif mode == 2: # 中文到英文翻译
prompt = f"{self.chinese_to_english_prompt}请翻译以下文本：{text}"
elif mode == 3: # 英文本地化人地名改写
prompt = f"{self.gpt4_localization_prompt}请改写以下文本：{text}"
elif mode == 4: # 英文到中文翻译
prompt = f"{self.english_to_chinese_prompt}请翻译以下英文文本：{text}"
elif mode == 5: # 提取主题改写
if hasattr(self, 'is_second_round') and self.is_second_round:
prompt = f"{self.general_prompt_round2}请改写以下文本：{text}"
else:
prompt = f"{self.topic_extraction_prompt}请改写以下文本：{text}"
else:
print(f"错误：不支持的处理模式 {mode}")
return None, 0
try:
根据重试次数调整temperature
current_temp = max(
self.MODEL_CONFIG['gpt4o_model']['temperature']['min'],
self.MODEL_CONFIG['gpt4o_model']['temperature']['default'] -
0 * self.MODEL_CONFIG['gpt4o_model']['temperature']['step']
)
print("GPT-4处理中...")
start_time = time.time()
调用API
response = self.openai_client.chat.completions.create(
model=self.MODEL_CONFIG['gpt4o_model']['name'],
messages=[{"role": "user", "content": prompt}],
temperature=current_temp,
max_tokens=self.MODEL_CONFIG['gpt4o_model']['max_tokens']
)
process_time = time.time() - start_time
print(f"GPT-4处理完成，耗时: {process_time:.2f}秒")
获取生成的内容
content = response.choices[0].message.content.strip()
计算token和成本
input_tokens = response.usage.prompt_tokens
output_tokens = response.usage.completion_tokens
cost = self.calculate_cost(input_tokens, output_tokens)
print(f"第 {1} 次尝试:")
print(f"使用模型: {self.MODEL_CONFIG['gpt4o_model']['name']}")
print(f"本次调用成本: ${cost:.4f}")
print(f"Token估算 - 输入: {input_tokens}, 输出: {output_tokens}")
return content, cost
except Exception as e:
print(f"处理出错: {str(e)}")
return None, 0
def process_file(self, filepath, mode):
"""处理单个文件"""
print(f"开始处理文件: {filepath}")
total_cost = 0
检查文件
if not os.path.exists(filepath):
print("文件不存在")
return False, 0, 0
print(f"文件是否存在: {os.path.exists(filepath)}")
print(f"文件是否可读: {os.access(filepath, os.R_OK)}")
print(f"文件大小: {os.path.getsize(filepath)} 字节")
try:
print("尝试读取文件内容...")
content = self.read_docx(filepath)
if not content:
print("文件内容为空")
return False, 0, 0
original_chars = len(content)
print(f"成功读取文件，字符数: {original_chars}")
print(f"内容预览（前100字符）: {content[:100]}")
处理逻辑
if int(mode) in [1, 5]: # 通用模式和提取主题改写模式
print(f"开始微调模型改写，文本长度: {len(content)} 字符")
第一轮：微调模型处理
first_result = self.process_with_finetune(content, int(mode))
if not first_result:
print("微调模型处理失败")
if int(mode) == 1: # 如果是模式1，且5次生成都失败，则不进行二次改写
print("模式1中5次生成都失败，不进行二次改写，处理结束")
return False, total_cost, 0
else:
print("跳过微调模型，直接使用GPT-4处理")
base_length = len(content) # 使用原始长度作为基准
else:
if self.ENABLE_SECOND_ROUND:
print("开始GPT-4第二轮改写...")
self.is_second_round = True # 标记为第二轮
base_length = len(first_result) # 使用微调模型处理后的长度作为基准
content = first_result
else:
print("第二轮改写已禁用，将使用微调模型的处理结果")
content = first_result
保存微调模型处理结果
output_path = self.save_to_docx(content, self.get_unique_filename(filepath))
if output_path:
print(f"处理完成，结果已保存至: {output_path}")
return True, total_cost, len(content)
else:
base_length = len(content) # 其他模式使用原始长度作为基准
分段处理
segments = self.split_text_by_length(content)
print(f"文本已分为 {len(segments)} 段")
processed_segments = []
for i, segment in enumerate(segments, 1):
print(f"开始处理第 {i}/{len(segments)} 段:")
processed_text, cost = self.process_segment(segment, mode)
if processed_text:
processed_segments.append(processed_text)
total_cost += cost
else:
print(f"处理第 {i} 段失败")
return False, total_cost, 0
合并处理后的内容
final_content = "".join(processed_segments)
保存结果
output_path = self.save_to_docx(final_content, self.get_unique_filename(filepath))
if output_path:
print(f"处理完成，结果已保存至: {output_path}")
print(f"总成本: ${total_cost:.4f}")
return True, total_cost, len(final_content)
return False, total_cost, 0
except Exception as e:
print(f"处理文件时出错: {str(e)}")
return False, total_cost, 0
def select_mode(self, file_name):
"""为单个文件选择处理模式"""
while True:
print(f"请为文件 {file_name} 选择处理模式:")
print("1. 通用模式（先微调后GPT-4）")
print("2. 中文到英文翻译（仅GPT-4）")
print("3. 英文本地化人地名改写（仅GPT-4）")
print("4. 英文到中文翻译（仅GPT-4）")
print("5. 提取主题改写（先微调后GPT-4）")
choice = input("请输入模式编号(1-5): ").strip()
if choice in ['1', '2', '3', '4', '5']:
print(f"已选择模式 {choice}")
return choice
else:
print("无效的选择，请重试")
def natural_sort_key(self, file_tuple):
import re
file_name = file_tuple[0]
convert = lambda text: int(text) if text.isdigit() else text.lower()
alphanum_key = lambda key: [convert(c) for c in re.split('([0-9]+)', key)]
return alphanum_key(file_name)
def process_files(self, files, modes):
"""处理多个文件"""
results = []
total_chars = 0
total_rewritten_chars = 0
total_cost = 0
success_count = 0
for i, (filename, filepath) in enumerate(files, 1):
print(f"处理文件 {i}/{len(files)}: {filename}")
try:
mode = modes.get(filename)
if not mode:
print(f"未找到文件 {filename} 的处理模式")
continue
print(f"完整文件路径: {filepath}")
print(f"选择的处理模式: {mode}")
success, cost, chars = self.process_file(filepath, mode)
if success:
success_count += 1
total_chars += chars
获取改写后的字符数
output_path = self.get_latest_output_path()
if output_path and os.path.exists(output_path):
try:
使用python-docx读取文件内容
doc = Document(output_path)
rewritten_text = ''.join([paragraph.text for paragraph in doc.paragraphs])
rewritten_chars = len(rewritten_text)
total_rewritten_chars += rewritten_chars
results.append({
'filename': filename,
'original_chars': chars,
'rewritten_chars': rewritten_chars,
'success': True
})
except Exception as e:
print(f"读取输出文件时出错: {str(e)}")
results.append({
'filename': filename,
'original_chars': chars,
'rewritten_chars': 0,
'success': False
})
else:
print(f"无法读取输出文件: {output_path}")
results.append({
'filename': filename,
'original_chars': chars,
'rewritten_chars': 0,
'success': False
})
else:
results.append({
'filename': filename,
'original_chars': chars,
'rewritten_chars': 0,
'success': False
})
total_cost += cost
except Exception as e:
print(f"处理文件时出错: {str(e)}")
traceback.print_exc()
results.append({
'filename': filename,
'original_chars': 0,
'rewritten_chars': 0,
'success': False
})
打印处理结果
print("处理完成!")
print(f"成功: {success_count}/{len(files)}")
打印字符统计表格
print("字符统计:")
print("-" * 100)
print(f"{'文件名':<40} {'原文字符':<16} {'改写字符':<16} {'变化比例':<12} {'状态':<8}")
print("-" * 100)
for result in results:
filename = result['filename']
original = result['original_chars']
rewritten = result['rewritten_chars']
ratio = (rewritten / original * 100) if original > 0 else 0.0
status = "成功" if result['success'] else "失败"
print(f"{filename:<40} {original:<16} {rewritten:<16} {ratio:.1f}% {status:<8}")
print("-" * 100)
print("总计:")
print(f"原文总字符: {total_chars}")
print(f"改写总字符: {total_rewritten_chars}")
total_ratio = (total_rewritten_chars / total_chars * 100) if total_chars > 0 else 0.0
print(f"总体变化比例: {total_ratio:.1f}%")
print(f"总成本: ${total_cost:.4f}")
def get_latest_output_path(self):
"""获取最新的输出文件路径"""
try:
rewrite_folder = os.path.join(os.path.expanduser("~/Desktop"), "改写")
if not os.path.exists(rewrite_folder):
return None
files = [f for f in os.listdir(rewrite_folder) if f.endswith('.docx')]
if not files:
return None
files.sort(key=lambda x: os.path.getmtime(os.path.join(rewrite_folder, x)), reverse=True)
return os.path.join(rewrite_folder, files[0])
except Exception as e:
print(f"获取最新输出文件路径时出错: {str(e)}")
return None
def generate_prompt(self, text, mode):
"""根据模式生成提示词"""
mode = int(mode)
if mode == 1: # 通用模式
if hasattr(self, 'is_second_round') and self.is_second_round:
return f"{self.general_prompt_round2}请改写以下文本：{text}"
else:
return self.general_prompt + "" + text
elif mode == 2: # 中文到英文翻译
return f"{self.chinese_to_english_prompt}Please translate the following Chinese text into English:{text}"
elif mode == 3: # 英文本地化人地名改写
return f"{self.gpt4_localization_prompt}Please rewrite the following text:{text}"
elif mode == 4: # 英文到中文翻译
return f"{self.english_to_chinese_prompt}请翻译以下英文文本：{text}"
elif mode == 5: # 提取主题改写
return f"{self.topic_extraction_prompt}请改写以下文本：{text}"
def process_with_finetune(self, content, mode):
"""使用微调模型处理内容"""
total_cost = 0
for attempt in range(self.MAX_RETRIES):
try:
print(f"第 {attempt + 1} 次尝试使用微调模型...")
生成提示词
prompt = self.generate_prompt(content, mode)
调用API
response = self.openai_client.chat.completions.create(
model=self.MODEL_CONFIG['ft_model']['name'],
messages=[{"role": "user", "content": prompt}],
temperature=self.MODEL_CONFIG['ft_model']['temperature']
)
获取生成的内容
result = response.choices[0].message.content.strip()
计算token和成本
input_tokens = response.usage.prompt_tokens
output_tokens = response.usage.completion_tokens
cost = self.calculate_cost(input_tokens, output_tokens, is_finetune=True)
total_cost += cost
print(f"使用模型: {self.MODEL_CONFIG['ft_model']['name']}")
print(f"Token使用 - 输入: {input_tokens}, 输出: {output_tokens}")
print(f"本次成本: ${cost:.4f}")
print(f"累计成本: ${total_cost:.4f}")
检查长度
original_length = len(content)
result_length = len(result)
min_length = int(original_length * (self.rewrite_ratio['min'] / 100)) # 将百分比转换为小数
max_length = int(original_length * (self.rewrite_ratio['max'] / 100))
print(f"原文长度: {original_length} 字符")
print(f"改写长度: {result_length} 字符")
print(f"长度要求: {min_length}-{max_length} 字符")
print(f"达标率: {(result_length / original_length * 100):.1f}%")
if result_length < min_length or result_length > max_length:
if attempt < self.MAX_RETRIES - 1:
print(f"生成的内容长度{'不足' if result_length < min_length else '超出限制'}，正在重试...")
continue
else:
print(f"已达到最大重试次数 ({self.MAX_RETRIES})，微调模型处理失败")
return None
print("微调模型处理成功")
return result
except Exception as e:
print(f"微调模型处理出错: {str(e)}")
if attempt < self.MAX_RETRIES - 1:
print("正在重试...")
continue
else:
print(f"已达到最大重试次数 ({self.MAX_RETRIES})，微调模型处理失败")
return None
return None
def run(self):
"""主运行方法"""
while True:
获取所有docx文件
docx_files = self.get_docx_files()
if not docx_files:
return
显示当前配置并让用户选择是否启用第二轮改写
print("是否启用第二轮改写? (y/n)")
self.ENABLE_SECOND_ROUND = input().lower() == 'y'
print(f"已{'启用' if self.ENABLE_SECOND_ROUND else '禁用'}第二轮改写")
获取用户选择
choice = input("请输入要处理的文件编号（多个文件用空格分隔，输入q退出）: ").strip()
if choice.lower() == 'q':
break
try:
解析用户选择
selected_indices = [int(x) - 1 for x in choice.split()]
selected_files = []
modes = {}
验证选择的有效性
for idx in selected_indices:
if 0 <= idx < len(docx_files):
selected_files.append(docx_files[idx])
else:
print(f"无效的文件编号: {idx + 1}")
continue
if not selected_files:
print("未选择任何文件")
continue
print("为每个文件选择处理模式...")
print("文件 1/1")
为每个选中的文件选择处理模式
for filename, filepath in selected_files:
while True:
print(f"为文件 {filename} 选择处理模式：")
print("1. 通用模式（先微调后GPT-4）")
print("2. 中文到英文翻译（仅GPT-4）")
print("3. 英文本地化人地名改写（仅GPT-4）")
print("4. 英文到中文翻译（仅GPT-4）")
print("5. 提取主题改写（先微调后GPT-4）")
print("q. 跳过此文件")
mode = input("请输入模式编号（1-5，或q跳过）: ").strip()
if mode.lower() == 'q':
break
if mode in ['1', '2', '3', '4', '5']:
modes[filename] = mode # 使用filename作为键
break
print("无效的选择，请重试")
显示选择的文件和模式
print("已选择 {} 个文件处理，{} 个文件跳过".format(
len(modes), len(selected_files) - len(modes)))
if modes:
print("选择的处理模式：")
for filename, _ in selected_files:
if filename in modes:
print(f"{filename}: {modes[filename]}")
确认开始处理
if input("确认开始处理？(y/n): ").lower() == 'y':
self.process_files(selected_files, modes)
break # 处理完成后退出循环
else:
print("已取消处理")
except ValueError:
print("输入无效，请重试")
except Exception as e:
print(f"发生错误: {str(e)}")
traceback.print_exc()
if __name__ == "__main__":
rewriter = TextRewriter()
rewriter.run()
这套代码 我迭代了至少50次 为达到最好的效果 我得出几个极其重要的点
微调模型产出的token数量 要在原文的60%-85%之间
改写 依然免不了抽卡 在抽卡几百次之后 我总结出的这个区间 是改写效果最好的
超过85% 则重复度过高 低于60%则删减的太多
在脚本中 我设置了5次重试机制 如果5次都没能得到这个区间 就会失败 不会无限调用
双保险机制
这个微调模型 本质上是4omini 这玩意本来是一点谱也不靠的 他产出的东西很有可能胡说八道
所以我加上了第二层保险 就是在将4omini返回的结果 再跑一遍其他模型 让他纠错
不管你是跑claude3.7也行 chatgpt4turbo也行 我都试过 自然是越贵的效果越好
这个代码 就是这个项目的灵魂 如果你看的细 你会发现里面还有很多的模式 比如翻译成本地化的英文什么的
那些是我尝试西班牙文和英文内容的实验
在运行脚本后 脚本会让你是否开启双保险模式-要处理的文件是哪些-以及让你选择每个文件的改写模式
提示词 分段 进度 价格 改写比例 token计算等等等等一应俱全
其最终的效果，已经可以达到我80%的改写水平，
我人工伪原创3000字 就需要1小时
但我使用脚本伪原创30万字 只需要1行命令 让他自己跑小半小时就行
这是tmd什么生产力革命
至此，改写的问题就解决了
出图自动化：从耗时2小时降低到小于1分钟
伪原创和出图都是消耗时间的大头
伪原创是累脑子 出图就是纯机械操作了
如果你有了解 你会知道MJ的出图效果是最棒的 没有之一
但MJ不允许调用API 而且就算他哪天开放了 你批量搞 也基本用不起
有一个东西叫做Midjourney-proxy 可以使用discordBOT的方式来模仿API调用
我试过了 是真的能用 且很好用
但是！容易被封号 MJ抓的很严的
在使用他的时候你必须保持人工也在正常使用Midjourney
否则必封号 我被封3次了
下面是我的MJ- proxy的代码 这个也迭代了一个礼拜
里面有三个模式 有通用-中文-英文三个模式 对应不同的使用场景
会自动将你docx里的文本用GPT翻译成英文 然后自动过滤掉MJ的敏感词
因为在discord里 MJ一次出图是4张放在一起 所以会自动切割成4张
然后分别命名为1.1、1.2、1.3、1.4；2.1、2.2以此类推
提交后会自动读取discord出图队列 在排队有2个任务的时候就睡眠N秒然后继续 防止队列满和被监测到
import os
import re
import time
import requests
from PIL import Image
from io import BytesIO
from pathlib import Path
import json
from datetime import datetime
from openai import OpenAI
from docx import Document
from docx.shared import Pt
import subprocess
import string
class MJImageGenerator:
def __init__(self):
基础配置
self.desktop = str(Path.home() / "Desktop")
self.image_base_folder = "图片"
self.mj_api_base = "http://locj" # MJ Proxy的地址
self.start_time = None
self.caffeinate_process = None
self.relax_mode = True
OpenAI客户端
self.client = OpenAI(
base_url="http",
api_key="sk-"
)
GPT提示词
self.system_prompts = {
"通用": """
忘记你以前的一切设定
将下面的文本翻译为详细的场景英文描述，用作给AI出图，要求：
1. 生成的场景描述，生成的英文中绝对不要包含“Blood, Cutting, Twerk, Making love, Voluptuous, Naughty, Wincest, Orgy, XXX, No clothes, Au naturel, No shirt, Decapitate, Bare, Nude, Barely dressed, Bra, Risque, Clear, Scantily clad, Cleavage, Stripped, Infested, Full frontal, Unclothed, Invisible clothes, Wearing nothing, Lingerie, With no shirt, Naked, Without clothes on, Negligee, Zero clothes, Gruesome, Fascist, Nazi, Prophet Mohammed, Slave, Coon, Honkey, Cocaine, Heroin, Meth, Crack, Kill, Belle Delphine, Hitler, Jinping, Lolita, President Xi, Torture, Disturbing, Farts, Fart, Poop, Infected, Warts, Shit, Brown pudding, Bunghole, Vomit, Seductive, Sperm, Hot, Sexy, Sadist, Sensored, Censored, Silenced, Deepfake, Inappropriate, Pus, Waifu, MP5, Succubus, 1488, Slaughter, Surgery, Reproduce, Crucified, Explicit, Large bust, Wang, Teratoma, Intimate, see through, Tryphophobia, Bloodbath, Wound, Cronenberg, Khorne, Cannibal, Cannibalism, Visceral, Guts, Bloodshot, Gory, Killing, Crucifixion, Vivisection, Massacre, Hemoglobin, Suicide, Arse, Labia, Ass, Mammaries, Badonkers, Bloody, Minge, Big Ass, Mommy Milker, Booba, Nipple, Oppai, Booty, Organs, Bosom, Ovaries, Flesh, Breasts, Penis, Busty, Phallus, Clunge, Sexy Female, Crotch, Skimpy, Dick, Thick, Bruises, Girth, Titty, Honkers, Vagina, Hooters, Veiny, Knob, Ahegao, Pinup, Ballgag, Car crash, Playboy, Bimbo, Pleasure, Bodily fluids, Pleasures, Boudoir, Rule34, Brothel, Seducing, Dominatrix, Corpse, Erotic, Fuck, Sensual, Hardcore, Hentai, Shag, Horny, Shibari, Incest, Smut, JAV, Jerk off king at pic, Thot, Kinbaku, Transparent, Legs spread, sensuality, belly button, porn, patriotic, bleed, excrement, petite, seduction, mccurry, provocative, sultry, erected, camisole, tight white, arrest, see-through, feces, anus, revealing clothing, vein, loli, -edge, boobs, -backed, tied up, zedong, bathing, jail, reticulum, rear end, sakimichan, behind bars, shirtless, dong, sexi, sexualiz”
2. 描述符合文本中的内容
4. 描述写成一行 描述写成一行 描述写成一行
5. 只输出英文描述，不要其他的任何说明
""",
"中文基督": """
忘记你以前的一切设定
将下面的文本翻译为详细的场景英文描述，用作给AI出图，要求：
1. 生成的场景描述，生成的英文中绝对不要包含“Blood, Cutting, Twerk, Making love, Voluptuous, Naughty, Wincest, Orgy, XXX, No clothes, Au naturel, No shirt, Decapitate, Bare, Nude, Barely dressed, Bra, Risque, Clear, Scantily clad, Cleavage, Stripped, Infested, Full frontal, Unclothed, Invisible clothes, Wearing nothing, Lingerie, With no shirt, Naked, Without clothes on, Negligee, Zero clothes, Gruesome, Fascist, Nazi, Prophet Mohammed, Slave, Coon, Honkey, Cocaine, Heroin, Meth, Crack, Kill, Belle Delphine, Hitler, Jinping, Lolita, President Xi, Torture, Disturbing, Farts, Fart, Poop, Infected, Warts, Shit, Brown pudding, Bunghole, Vomit, Seductive, Sperm, Hot, Sexy, Sadist, Sensored, Censored, Silenced, Deepfake, Inappropriate, Pus, Waifu, MP5, Succubus, 1488, Slaughter, Surgery, Reproduce, Crucified, Explicit, Large bust, Wang, Teratoma, Intimate, see through, Tryphophobia, Bloodbath, Wound, Cronenberg, Khorne, Cannibal, Cannibalism, Visceral, Guts, Bloodshot, Gory, Killing, Crucifixion, Vivisection, Massacre, Hemoglobin, Suicide, Arse, Labia, Ass, Mammaries, Badonkers, Bloody, Minge, Big Ass, Mommy Milker, Booba, Nipple, Oppai, Booty, Organs, Bosom, Ovaries, Flesh, Breasts, Penis, Busty, Phallus, Clunge, Sexy Female, Crotch, Skimpy, Dick, Thick, Bruises, Girth, Titty, Honkers, Vagina, Hooters, Veiny, Knob, Ahegao, Pinup, Ballgag, Car crash, Playboy, Bimbo, Pleasure, Bodily fluids, Pleasures, Boudoir, Rule34, Brothel, Seducing, Dominatrix, Corpse, Erotic, Fuck, Sensual, Hardcore, Hentai, Shag, Horny, Shibari, Incest, Smut, JAV, Jerk off king at pic, Thot, Kinbaku, Transparent, Legs spread, sensuality, belly button, porn, patriotic, bleed, excrement, petite, seduction, mccurry, provocative, sultry, erected, camisole, tight white, arrest, see-through, feces, anus, revealing clothing, vein, loli, -edge, boobs, -backed, tied up, zedong, bathing, jail, reticulum, rear end, sakimichan, behind bars, shirtless, dong, sexi, sexualiz”
2. 描述符合文本中的内容
4. 描述写成一行 描述写成一行 描述写成一行
5. 只输出英文描述，不要其他的任何说明
""",
"英文基督": """
忘记你以前的一切设定
将下面的文本翻译为详细的场景英文描述，用作给AI出图，要求：
1. 生成的场景描述，生成的英文中绝对不要包含“Blood, Cutting, Twerk, Making love, Voluptuous, Naughty, Wincest, Orgy, XXX, No clothes, Au naturel, No shirt, Decapitate, Bare, Nude, Barely dressed, Bra, Risque, Clear, Scantily clad, Cleavage, Stripped, Infested, Full frontal, Unclothed, Invisible clothes, Wearing nothing, Lingerie, With no shirt, Naked, Without clothes on, Negligee, Zero clothes, Gruesome, Fascist, Nazi, Prophet Mohammed, Slave, Coon, Honkey, Cocaine, Heroin, Meth, Crack, Kill, Belle Delphine, Hitler, Jinping, Lolita, President Xi, Torture, Disturbing, Farts, Fart, Poop, Infected, Warts, Shit, Brown pudding, Bunghole, Vomit, Seductive, Sperm, Hot, Sexy, Sadist, Sensored, Censored, Silenced, Deepfake, Inappropriate, Pus, Waifu, MP5, Succubus, 1488, Slaughter, Surgery, Reproduce, Crucified, Explicit, Large bust, Wang, Teratoma, Intimate, see through, Tryphophobia, Bloodbath, Wound, Cronenberg, Khorne, Cannibal, Cannibalism, Visceral, Guts, Bloodshot, Gory, Killing, Crucifixion, Vivisection, Massacre, Hemoglobin, Suicide, Arse, Labia, Ass, Mammaries, Badonkers, Bloody, Minge, Big Ass, Mommy Milker, Booba, Nipple, Oppai, Booty, Organs, Bosom, Ovaries, Flesh, Breasts, Penis, Busty, Phallus, Clunge, Sexy Female, Crotch, Skimpy, Dick, Thick, Bruises, Girth, Titty, Honkers, Vagina, Hooters, Veiny, Knob, Ahegao, Pinup, Ballgag, Car crash, Playboy, Bimbo, Pleasure, Bodily fluids, Pleasures, Boudoir, Rule34, Brothel, Seducing, Dominatrix, Corpse, Erotic, Fuck, Sensual, Hardcore, Hentai, Shag, Horny, Shibari, Incest, Smut, JAV, Jerk off king at pic, Thot, Kinbaku, Transparent, Legs spread, sensuality, belly button, porn, patriotic, bleed, excrement, petite, seduction, mccurry, provocative, sultry, erected, camisole, tight white, arrest, see-through, feces, anus, revealing clothing, vein, loli, -edge, boobs, -backed, tied up, zedong, bathing, jail, reticulum, rear end, sakimichan, behind bars, shirtless, dong, sexi, sexualiz”
2. 描述符合文本中的内容
4. 描述写成一行 描述写成一行 描述写成一行
5. 只输出英文描述，不要其他的任何说明
"""
}
初始化结果列表
self.gpt_results = []
def get_current_time(self):
"""获取当前时间的格式化字符串"""
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def log_message(self, message, level="INFO"):
"""添加日志记录功能"""
timestamp = self.get_current_time()
log_entry = f"[{timestamp}] [{level}] {message}"
print(log_entry)
def check_sensitive_words(self, text):
"""检查是否包含敏感词"""
sensitive_words = [
"Blood", "Cutting", "Twerk", "Making love", "Voluptuous", "Naughty",
"Wincest", "Orgy", "XXX", "No clothes", "Au naturel", "No shirt",
"Decapitate", "Bare", "Nude", "Barely dressed", "Bra", "Risque",
"Clear", "Scantily clad", "Cleavage", "Stripped", "Infested",
"Full frontal", "Unclothed", "Invisible clothes", "Wearing nothing",
"Lingerie", "With no shirt", "Naked", "Without clothes on", "Negligee",
"Zero clothes", "Gruesome", "Fascist", "Nazi", "Prophet Mohammed",
"Slave", "Coon", "Honkey", "Cocaine", "Heroin", "Meth", "Crack",
"Kill", "Belle Delphine", "Hitler", "Jinping", "Lolita", "President Xi",
"Torture", "Disturbing", "Farts", "Fart", "Poop", "Infected", "Warts",
"Shit", "Brown pudding", "Bunghole", "Vomit", "Seductive", "Sperm",
"Hot", "Sexy", "Sadist", "Sensored", "Censored", "Silenced", "Deepfake",
"Inappropriate", "Pus", "Waifu", "MP5", "Succubus", "1488", "Slaughter",
"Surgery", "Reproduce", "Crucified", "Explicit", "Large bust", "Wang",
"Teratoma", "Intimate", "see through", "Tryphophobia", "Bloodbath",
"Wound", "Cronenberg", "Khorne", "Cannibal", "Cannibalism", "Visceral",
"Guts", "Bloodshot", "Gory", "Killing", "Crucifixion", "Vivisection",
"Massacre", "Hemoglobin", "Suicide", "Arse", "Labia", "Ass", "Mammaries",
"Badonkers", "Bloody", "Minge", "Big Ass", "Mommy Milker", "Booba",
"Nipple", "Oppai", "Booty", "Organs", "Bosom", "Ovaries", "Flesh",
"Breasts", "Penis", "Busty", "Phallus", "Clunge", "Sexy Female",
"Crotch", "Skimpy", "Dick", "Thick", "Bruises", "Girth", "Titty",
"Honkers", "Vagina", "Hooters", "Veiny", "Knob", "Ahegao", "Pinup",
"Ballgag", "Car crash", "Playboy", "Bimbo", "Pleasure", "Bodily fluids",
"Pleasures", "Boudoir", "Rule34", "Brothel", "Seducing", "Dominatrix",
"Corpse", "Erotic", "Fuck", "Sensual", "Hardcore", "Hentai", "Shag",
"Horny", "Shibari", "Incest", "Smut", "JAV", "Jerk off king at pic",
"Thot", "Kinbaku", "Transparent", "Legs spread", "sensuality",
"belly button", "porn", "patriotic", "bleed", "excrement", "petite",
"seduction", "mccurry", "provocative", "sultry", "erected", "camisole",
"tight white", "arrest", "see-through", "feces", "anus",
"revealing clothing", "vein", "loli", "-edge", "boobs", "-backed",
"tied up", "zedong", "bathing", "jail", "reticulum", "rear end",
"sakimichan", "behind bars", "shirtless", "dong", "sexi", "sexualiz"
]
将文本转换为小写进行检查
text_lower = text.lower()
for word in sensitive_words:
if word.lower() in text_lower:
return word
return None
def submit_task(self, prompt):
"""提交任务到MJ"""
while True: # 使用循环代替递归
try:
检查并移除敏感词
sensitive_word = self.check_sensitive_words(prompt)
if sensitive_word:
self.log_message(f"提示词中包含敏感词: {sensitive_word}，已移除")
prompt = prompt.replace(sensitive_word, "").strip()
data = {
"prompt": f"{prompt} --ar 16:9{' --relax' if self.relax_mode else ''}", # 根据relax_mode决定是否添加relax参数
"notifyHook": ""
}
response = requests.post(f"{self.mj_api_base}/submit/imagine", json=data)
检查响应状态码
if response.status_code != 200:
self.log_message(f"API请求失败，状态码: {response.status_code}")
return None
try:
result = response.json()
except ValueError:
self.log_message(f"API返回的不是有效的JSON: {response.text}")
return None
if not isinstance(result, dict):
self.log_message(f"API返回格式错误: {result}")
return None
code = result.get("code")
if code == 1: # 成功
task_id = result.get("result")
if isinstance(task_id, dict):
task_id = task_id.get("taskId")
self.log_message(f"任务提交成功，ID: {task_id}")
return task_id
elif code == 22: # 进入队列
检查当前队列状态
_, queue_count = self.check_queue_status()
if queue_count >= 3:
self.log_message(f"队列中已有{queue_count}个任务，等待90秒后重试")
time.sleep(90) # 等待90秒
else:
self.log_message(f"队列中有{queue_count}个任务，等待10秒后重试")
time.sleep(10) # 等待10秒
continue # 继续循环，重新尝试提交
elif code == 23: # 队列已满
self.log_message("队列已满，等待2分钟后重试")
time.sleep(120) # 等待2分钟
continue # 继续循环，重新尝试提交
elif code == 24: # API返回敏感词
properties = result.get("properties", {})
if isinstance(properties, dict):
banned_word = properties.get("bannedWord", "unknown")
else:
banned_word = "unknown"
self.log_message(f"API检测到敏感词: {banned_word}，尝试移除后重新提交")
移除API检测到的敏感词
prompt = prompt.replace(banned_word, "").strip()
continue # 继续循环，重新尝试提交
else:
self.log_message(f"提交任务失败: {result}")
return None
except Exception as e:
self.log_message(f"提交任务时出错: {str(e)}")
return None
def submit_mj_task(self, prompt, index):
"""提交MJ任务"""
try:
根据模式添加参数
prompt_with_params = f"{prompt} --ar 16:9"
if self.relax_mode:
prompt_with_params += " --relax"
data = {
"prompt": prompt_with_params,
"notifyHook": ""
}
response = requests.post(f"{self.mj_api_base}/submit/imagine", json=data)
result = response.json()
code = result.get("code")
if code in [1, 21, 22]: # 成功、已存在、进入队列
return result.get("result")
elif code == 23: # 队列已满
self.log_message("队列已满，等待重试")
return None
elif code == 24: # 敏感词
banned_word = result.get("properties", {}).get("bannedWord", "unknown")
self.log_message(f"包含敏感词: {banned_word}")
return None
else:
self.log_message(f"提交失败: {result.get('description', '未知错误')}")
return None
except Exception as e:
self.log_message(f"提交任务失败: {str(e)}", "ERROR")
return None
def save_image_with_index(self, img, x, y):
"""保存图，理同名文件"""
while True:
filename = f"{x}.{y}.png"
filepath = os.path.join(self.image_folder, filename)
if not os.path.exists(filepath):
img.save(filepath)
self.log_message(f"保存图片: {filename}")
break
y += 1
def split_and_save_image(self, image_url, index):
"""将图片切割成4份并保存"""
try:
response = requests.get(image_url)
img = Image.open(BytesIO(response.content))
获取原图尺寸
width, height = img.size
因为是16:9，所以直接平均切成4份即可
part_width = width // 2
part_height = height // 2
定义个区域
regions = [
(0, 0, part_width, part_height), # 左上
(part_width, 0, width, part_height), # 右上
(0, part_height, part_width, height), # 左下
(part_width, part_height, width, height) # 右下
]
割并保存
for i, region in enumerate(regions, 1):
part = img.crop(region)
self.save_image_with_index(part, index, i)
except Exception as e:
self.log_message(f"切割保存图片失败: {str(e)}", "ERROR")
def read_from_docx(self, file_name):
"""从docx文件中读取内容"""
try:
如果文件名已经包含.docx扩展名，就不再添加
if not file_name.endswith('.docx'):
file_name = f"{file_name}.docx"
file_path = os.path.join(self.desktop, file_name)
self.log_message(f"尝试读取文件: {file_path}")
if not os.path.exists(file_path):
self.log_message("错误: 找不到文件", "ERROR")
return None
doc = Document(file_path)
content = []
for paragraph in doc.paragraphs:
if paragraph.text.strip(): # 只添加非空段落
content.append(paragraph.text.strip())
return "".join(content) if content else None
except Exception as e:
self.log_message(f"读取文件失败: {str(e)}", "ERROR")
return None
def split_into_paragraphs(self, text):
"""将文本分段"""
try:
paragraphs = []
lines = [line.strip() for line in text.split('') if line.strip()]
text = ''.join(lines)
start_pos = 0
while start_pos < len(text):
if len(text) - start_pos <= 100:
last_paragraph = text[start_pos:].strip()
if last_paragraph:
paragraphs.append(last_paragraph)
break
check_pos = start_pos + 100
found_punctuation = False
for i in range(check_pos, min(check_pos + 15, len(text))):
if text[i] in '".。，,':
paragraphs.append(text[start_pos:i + 1].strip())
start_pos = i + 1
found_punctuation = True
break
if not found_punctuation:
paragraphs.append(text[start_pos:start_pos + 120].strip())
start_pos += 120
return paragraphs
except Exception as e:
self.log_message(f"分段处理出错: {str(e)}", "ERROR")
return None
def check_queue_status(self):
"""检查当前队列状态"""
try:
检查正在运行的任务
running_response = requests.get(f"{self.mj_api_base}/task/running")
running_tasks = running_response.json()
检查队列中的任务
queue_response = requests.get(f"{self.mj_api_base}/task/queue")
queue_tasks = queue_response.json()
准备进度信息
progress_info = []
处理运行中的任务
for task in running_tasks:
task_id = task.get("id", "未知")
progress = task.get("progress", "0%")
status = task.get("status", "未知")
progress_info.append(f"任务ID: {task_id} | 状态: {status} | 进度: {progress}")
处理队列中的任务
for task in queue_tasks:
task_id = task.get("id", "未知")
progress_info.append(f"任务ID: {task_id} | 状态: 排队中")
打印进度信息
self.log_message("=== 当前任务状态 ===")
if progress_info:
for info in progress_info:
self.log_message(info)
self.log_message("=" * 30)
return len(running_tasks), len(queue_tasks)
except Exception as e:
self.log_message(f"检查队列状态失败: {str(e)}", "ERROR")
return 0, 0
def check_tasks_progress(self):
"""检查所有任务的进度"""
try:
检查运行任务
running_response = requests.get(f"{self.mj_api_base}/task/running")
running_tasks = running_response.json().get("result", [])
检查队列中的任务
queue_response = requests.get(f"{self.mj_api_base}/task/queue")
queued_tasks = queue_response.json().get("result", [])
获取进度信息
progress_info = []
处理运行中的任务
for task in running_tasks:
task_id = task.get("id")
progress = task.get("progress", "未知")
prompt = task.get("prompt", "未知")
status = "运行中"
progress_info.append(f"任务ID: {task_id} | 状态: {status} | 进度: {progress}% | 提示词: {prompt[:30]}...")
处理队列中的任务
for task in queued_tasks:
task_id = task.get("id")
prompt = task.get("prompt", "未知")
status = "排队中"
progress_info.append(f"任务ID: {task_id} | 状态: {status} | 提示词: {prompt[:30]}...")
打印进度信息
print(f"[{self.get_current_time()}] === 当前任务状态 ===")
print(f"运行中任务数: {len(running_tasks)}")
print(f"排队中任务数: {len(queued_tasks)}")
if progress_info:
print("详细进度:")
for info in progress_info:
print(info)
print("=" * 50)
return len(running_tasks), len(queued_tasks)
except Exception as e:
print(f"[{self.get_current_time()}] 检查任务进度失败: {str(e)}")
return 0, 0
def check_task_status(self, task_id):
"""检查任务状态"""
try:
response = requests.get(f"{self.mj_api_base}/task/{task_id}/fetch")
result = response.json()
if result.get("status") == "SUCCESS":
return True, result.get("imageUrl")
elif result.get("status") == "FAILURE":
self.log_message(f"任务 {task_id} 失败: {result.get('failReason')}", "ERROR")
return False, None
elif result.get("status") in ["NOT_START", "SUBMITTED", "IN_PROGRESS"]:
progress = result.get("progress", "0%")
self.log_message(f"任务 {task_id} 执行中 - 进度: {progress}")
return None, None
else:
self.log_message(f"未知状态: {result.get('status')}")
return None, None
except Exception as e:
self.log_message(f"检查任务状态失败: {str(e)}", "ERROR")
return None, None
def start_caffeinate(self):
"""启动防睡眠"""
self.caffeinate_process = subprocess.Popen(['caffeinate', '-d'])
self.log_message("防睡眠已启动")
def stop_caffeinate(self):
"""停止防睡眠"""
if self.caffeinate_process:
self.caffeinate_process.terminate()
self.caffeinate_process = None
self.log_message("防睡眠已关闭")
def select_files(self):
"""选择要处理的文件"""
获取桌面上的所有非隐藏的docx文件
docx_files = [f for f in os.listdir(self.desktop)
if f.endswith('.docx')
and not f.startswith('.') # 非隐藏文件
and not f.startswith('~')] # 非临时文件
if not docx_files:
self.log_message("未找到docx文件")
return None
while True:
print("=== 可用的文件 ===")
for i, file in enumerate(docx_files, 1):
print(f"{i}. {file}")
print("请选择要处理的文件编号（多个文件用空格分隔，输入0退出）：")
try:
choice = input().strip()
if choice == '0':
return None
indices = [int(x) - 1 for x in choice.split()]
if all(0 <= i < len(docx_files) for i in indices):
return indices
else:
print("无效的选择，请重新输入")
except ValueError:
print("输入格式错误，请重新输入")
def select_mode(self, file_name):
"""为文件选择处理模式"""
while True:
print(f"为文件 {file_name} 选择处理模式：")
print("1. 通用模式")
print("2. 中文基督模式")
print("3. 英文基督模式")
print("0. 返回上一步")
mode = input("请输入选项编号 (0/1/2/3): ").strip()
if mode == '0':
return None
elif mode in ['1', '2', '3']:
mode_map = {
"1": "通用",
"2": "中文基督",
"3": "英文基督"
}
return mode_map[mode]
else:
print("输入无效，请重新输入")
def run(self):
"""主运行方法"""
try:
启动防睡眠
self.start_caffeinate()
self.log_message("开始运行MJ图片生成器")
选择要处理的文件
indices = self.select_files()
if indices is None: # 用户选择退出
return
获取桌面上的所有docx文件
docx_files = [f for f in os.listdir(self.desktop)
if f.endswith('.docx')
and not f.startswith('.')
and not f.startswith('~')]
为每个选中的文件选择处理模式
file_modes = {}
for i in indices:
file_name = docx_files[i]
mode = self.select_mode(file_name)
if mode is None: # 用户选择返回
return
file_modes[file_name] = mode
处理每个文件
for file_name, mode in file_modes.items():
self.current_mode = mode # 设置当前模式
self.log_message(f"开始处理文件: {file_name}，模式: {mode}")
读取文件内容
content = self.read_from_docx(file_name)
if not content:
continue
分段处理
paragraphs = self.split_into_paragraphs(content)
if not paragraphs:
continue
使用GPT处理每段文本
self.gpt_results = []
for i, paragraph in enumerate(paragraphs, 1):
try:
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": self.system_prompts[mode]},
{"role": "user", "content": paragraph}
]
)
result = response.choices[0].message.content
self.gpt_results.append(result)
self.log_message(f"GPT处理进度: {i}/{len(paragraphs)}")
except Exception as e:
self.log_message(f"GPT处理失败: {str(e)}", "ERROR")
continue
开始生成图片
total_tasks = len(self.gpt_results)
self.log_message(f"需要生成的图片任务总数: {total_tasks}")
开始提交任务
i = 0 # 当前处理的任务索引
submitted_tasks = {} # 存储已提交的任务
completed_tasks = set() # 成功完成的任务
failed_tasks = set() # 失败的任务
image_folder = os.path.join(self.desktop, self.image_base_folder, file_name)
if not os.path.exists(image_folder):
os.makedirs(image_folder)
self.image_folder = image_folder
while i < total_tasks or len(completed_tasks) + len(failed_tasks) < total_tasks:
检查已提交任务的状态
for task_index, task_id in list(submitted_tasks.items()):
if task_id not in completed_tasks and task_id not in failed_tasks:
status, image_url = self.check_task_status(task_id)
if status is True and image_url: # 任务成功
self.split_and_save_image(image_url, task_index + 1)
completed_tasks.add(task_id)
self.log_message(f"任务进度 - 成功: {len(completed_tasks)}/{total_tasks}, "
f"失败: {len(failed_tasks)}, "
f"进行中: {len(submitted_tasks) - len(completed_tasks) - len(failed_tasks)}")
elif status is False: # 任务失败
failed_tasks.add(task_id)
self.log_message(f"任务 {task_id} 失败")
如果还有任务未提交，继续提交新任务
if i < total_tasks:
提交新任务
prompt = self.gpt_results[i]
task_id = self.submit_task(prompt)
if task_id:
submitted_tasks[i] = task_id
i += 1
time.sleep(10) # 成功提交后等待10秒再提交下一个
else:
time.sleep(60) # 提交失败后等待60秒再重试
else:
所有任务已提交，等待完成
time.sleep(10)
输出当前文件的处理结果
self.log_message(f"{file_name}处理完成！"
f"总任务数: {total_tasks}"
f"成功完成: {len(completed_tasks)}"
f"失败任务: {len(failed_tasks)}")
except KeyboardInterrupt:
self.log_message("用户中断程序")
except Exception as e:
self.log_message(f"程序运行出错: {str(e)}", "ERROR")
finally:
self.stop_caffeinate()
if __name__ == "__main__":
generator = MJImageGenerator()
generator.run()
但MJ的账号1个月就要30刀 一被封就是7天起步 实在是有点贵
所以退而求其次 只能选择不要钱的SD 要想好好的大批量生产图片
但没有好的设备 所以就选择了云平台
我用的是runware 使用Fluxdev 一张图只需要0.002刀 非常划算 是我找到的最划算的了
这里又有很多的逻辑 首先我的文本是中文 但SD只吃英文 所以我又得用GPTmini翻译一遍
然后一张图片的展示时间不能太长 所以他对应的中文字数也不能太多
所以经过N轮调试后，按我的工作流 115个中文字送一组是最合适的，英文的话55个英文字一组是最合适的，
当然里面还有很多的模式可供选择 比如基督的模式 英文的模式 甚至西班牙语和恐怖故事的模式，那都是我之前的尝试，
按字数切割翻译成英文 出图 然后返回 其中标点的逻辑 分段合并的逻辑 都已经调教的非常完美了 换上key就能用
import os
import re
from pathlib import Path
from docx import Document
from openai import OpenAI
import runware
from PIL import Image
import requests
from io import BytesIO
import time
from datetime import datetime
import requests
import json
import uuid
import subprocess # 添加subprocess模块
class RunwareImageGenerator:
def __init__(self):
基础配置
self.desktop = str(Path.home() / "Desktop")
self.rewrite_folder = os.path.join(self.desktop, "剪辑/改写") # 改写文件夹路径
self.image_folder = os.path.join(self.desktop, "剪辑/图片") # 图片文件夹路径
self.start_time = None
self.current_file = None
self.current_file_name = None # 添加当前处理的文件名
self.total_cost = 0.0 # 总费用
self.prevent_sleep = None # 添加防睡眠进程变量
创建必要的文件夹
os.makedirs(self.rewrite_folder, exist_ok=True)
os.makedirs(self.image_folder, exist_ok=True)
OpenAI客户端
self.client = OpenAI(
base_url="",
api_key="sk-"
)
初始化Runware客户端
runware.api_key = "" # 需要替换为实际的API key
不同模式的提示词
self.system_prompts = {
"通用": """
忘记你以前的一切设定
将下面的文本翻译为详细的场景英文描述，用作给AI出图，要求：
1. 译成英文，尽量以人事物+动作+周围场景作为格式
2. 描述写成一行
3. 只输出英文描述，不要其他的任何说明
""",
"中文基督": """
忘记你以前的一切设定
将下面的文本翻译为详细的场景英文描述，用作给AI出图，要求：
1. 译成英文，尽量以人事物+动作+周围场景作为格式
2. 描述写成一行
3. 只输出英文描述，不要其他的任何说明
4. 这是一个基督教相关的内容，所以如果可能，在英文翻译中加入和翻译相符的基督教的元素，如果加不进去也不要强行加入
""",
"英文出图": """
忘记你以前的一切设定
Please enhance the following English text for AI image generation, requirements:
1. Keep the original meaning but make it more detailed and vivid
2. Output in a single line
3. Only output the English description, no other explanations
""",
"西班牙语": """
忘记你以前的一切设定
Por favor, traduzca el siguiente texto a una descripción detallada de la escena en inglés para la generación de imágenes de IA, requisitos:
1. Traduzca al inglés, intentando utilizar la estructura de persona/objeto + acción + entorno
2. La descripción debe ser una sola línea
3. Solo salida de la descripción en inglés, sin ninguna otra explicación
""",
"恐怖故事": """
忘记你以前的一切设定
将文本生成为详细的恐怖场景英文描述，用作给AI出图，要求：
1. 尽量以人事物+动作+周围恐怖场景作为格式
2. 描述写成一行
3. 只输出英文描述，不要其他的任何说明
""",
}
不同模式的切割配置
self.split_configs = {
"1": { # 通用模式（中文）
"target_chars": 115, # 目标字符数
"type": "chars", # 按字符切割
"punctuation": "。！？，；" # 中文标点
},
"2": { # 中文基督模式（中文）
"target_chars": 105, # 目标字符数
"type": "chars", # 按字符切割
"punctuation": "。！？，；" # 中文标点
},
"3": { # 英文基督模式（英文）
"target_words": 55, # 目标单词数
"type": "words", # 按单词切割
"punctuation": ".," # 英文标点
},
"4": { # 西班牙语模式（西班牙文）
"target_words": 70, # 目标单词数
"type": "words", # 按单词切割
"punctuation": ".,!?¡¿" # 西班牙文标点
},
"5": { # 恐怖故事模式（英文）
"target_words": 55, # 目标单词数
"type": "words", # 按单词切割
"punctuation": ".," # 英文标点
}
}
当前使用的模式
self.current_mode = None
初始化结果列表
self.gpt_results = []
def get_current_time(self):
"""获取当前时间的格式化字符
[⚠️ 飞书文档内容过长，已截断，仅展示前 8 万字，完整正文请通过帖子详情页 detailUrl 查看]
