🔄 卡若AI 同步 2026-02-22 09:55 | 更新:金仓、水桥平台对接、水溪整理归档、卡木、运营中枢工作台 | 排除 >20MB: 8 个

This commit is contained in:
2026-02-22 09:55:12 +08:00
parent b850b1880d
commit 21af7aac80
27 changed files with 1916 additions and 97 deletions

View File

@@ -52,9 +52,10 @@ kr宝塔: qcWubCdlfFjS2b2DMT1lzPFaDfmv1cBT
### 强制规则(每次执行必守)
1. **SSH 统一配置**:账号 **root**、密码 **Zhiqun1984**Z 大写),端口 22022 或 22或使用 id_ed25519 密钥。详见 `references/SSH登录方式与故障排查.md`
2. **经验沉淀**:每次涉及服务器/宝塔/部署的操作结束后,必须把经验写入 `02_卡人/水溪_整理归档/经验库/待沉淀/`,防止同类问题重复出现
3. **Skill 迭代**:每次有新的配置、教训、流程变更时,必须同步更新本 SKILL.md 或 references保证下次调用时信息一致
4. **卡若AI 复盘**每次任务结束必须用卡若AI 复盘格式收尾(目标·结果·达成率、过程、反思、总结、下一步)
2. **宝塔 443 优先**:宝塔服务器 443 不监听时,**优先**检查是否运行系统 Nginx 而非宝塔 Nginx若是`killall nginx` 后启动宝塔 Nginx。详见 Q0、`_经验库/已整理/运维经验/宝塔443不监听_系统nginx与宝塔nginx优先排查.md`
3. **经验沉淀**:每次涉及服务器/宝塔/部署的操作结束后,必须把经验写入 `02_卡人/水溪_整理归档/经验库/待沉淀/`,防止同类问题重复出现
4. **Skill 迭代**:每次有新的配置、教训、流程变更时,必须同步更新本 SKILL.md 或 references保证下次调用时信息一致
5. **卡若AI 复盘**每次任务结束必须用卡若AI 复盘格式收尾(目标·结果·达成率、过程、反思、总结、下一步)。
---
@@ -293,6 +294,21 @@ dig soul.quwanzhi.com +short @8.8.8.8
## 常见问题速查
### Q0: 宝塔 443 不监听Connection refused【优先检查】
**现象**80 可达、443 不可达,安全组/证书/nginx -t 均正常。
**根因**:运行的是**系统 Nginx**`/usr/sbin/nginx`),非**宝塔 Nginx**。系统 Nginx 配置不含宝塔 vhost 的 listen 443。
**解决**(任选):
```bash
# 宝塔终端
killall nginx; sleep 2; /www/server/nginx/sbin/nginx -c /www/server/nginx/conf/nginx.conf
```
或 TAT`python3 scripts/腾讯云_TAT_存客宝_Nginx443强制修复.py`
**防重复**:宝塔服务器 443 不监听时,**先** `ps aux | grep nginx` 确认是否系统 Nginx再查安全组/证书。详见 `_经验库/已整理/运维经验/宝塔443不监听_系统nginx与宝塔nginx优先排查.md`
### Q1: 外网无法访问ERR_EMPTY_RESPONSE
**原因**: 腾讯云安全组只开放443端口
@@ -363,6 +379,7 @@ ss -tlnp | grep :端口号
| `快速检查服务器.py` | 一键检查所有服务器状态 | `./脚本/` |
| `一键部署.py` | 根据配置文件部署项目 | `./脚本/` |
| `ssl证书检查.py` | 检查/修复SSL证书 | `./脚本/` |
| `腾讯云_TAT_存客宝_Nginx443强制修复.py` | **宝塔 443 不监听**:切回宝塔 Nginx | `./scripts/` |
| `tencent_image_snapshot_backup_to_nas.py` | 腾讯云镜像/快照备份到 CKB NAS | `./scripts/腾讯云镜像快照备份到CKB_NAS/` |
---

View File

@@ -0,0 +1,150 @@
# 存客宝 443 无法访问 · 深度诊断与方案
> 42.194.245.239 · kr-kf.quwanzhi.com / lytiao.com · 现象HTTPS 无法打开HTTP 正常
**⚠️ 优先处理**443 不监听时,**先**检查 `ps aux | grep nginx`。若为系统 Nginx`/usr/sbin/nginx`),先切回宝塔 Nginx 再查其他。详见服务器管理 SKILL Q0、`_经验库/已整理/运维经验/宝塔443不监听_系统nginx与宝塔nginx优先排查.md`
---
## 综合分析:根因结论
### 关键对比(同网络下实测)
| 目标 | 80 | 443 | 9988 |
|------|-----|-----|------|
| **存客宝** 42.194.245.239 | ✅ 可达 | ❌ 不可达 | ✅ 可达 |
| **kr宝塔** 43.139.27.93 | ✅ | ✅ 可达 | ✅ |
**结论**kr宝塔 443 可达 → **排除测试端网络限制**。问题**仅出现在存客宝 443**,属实例或腾讯云侧网络配置异常。
### 已排除
- ✅ 安全组5 个均已含 443
- ✅ 服务器iptables 放行、Nginx 监听、SSL 已配
- ✅ 测试网络kr宝塔 443 通
### 最可能根因(优先级)
1. **实例网络架构**:多网卡/弹性网卡,公网流量实际生效的安全组与修改的不一致
2. **腾讯云产品层**DDoS 高防、WAF 等对 443 有单独策略
3. **规则顺序**DROP 在 ACCEPT 前匹配 443
---
## 一、深度诊断结论2026-02
### 1. 安全组规则核查
| 安全组 | 443 入站 | 备注 |
|--------|----------|------|
| sg-liw750jn | ✅ 有 | 明确 443/tcp |
| sg-7m7gdqg7 | ✅ 有 | 端口列表中含 443 |
| sg-abcjf9xf | ⚠️ ALL/ALL | 理论允许所有 |
| sg-g4ih01d7 | ✅ 有 | 明确 443/tcp |
| sg-podrm64t | ✅ | 含 ALL/ALL |
**结论**5 个安全组均已含 443。
### 2. 服务器内状态TAT【2026-02 更新 · 已解决】
- **iptables**:已有 ACCEPT 规则 tcp dpt:443
- **原根因**:运行的是**系统 nginx** (`/usr/sbin/nginx`),非**宝塔 nginx**。系统 nginx 配置不含 443故仅监听 80。
- **解决**:停掉系统 nginx启动宝塔 nginx`/www/server/nginx/sbin/nginx -c /www/server/nginx/conf/nginx.conf`443 正常。
### 3. 外网实测
- 80可达HTTP 200
- 443不可达nc/curl 超时或连接失败)
---
## 三、可执行修复步骤
### 步骤 A安全组补全 443必做
```bash
cd /Users/karuo/Documents/个人/卡若AI/01_卡资/金仓_存储备份/服务器管理/scripts
/Users/karuo/Documents/个人/卡若AI/.venv_tencent/bin/python3 腾讯云_存客宝安全组放行443.py
```
执行后检查:
```bash
/Users/karuo/Documents/个人/卡若AI/.venv_tencent/bin/python3 腾讯云_存客宝安全组放行443.py --check
```
确认所有安全组「含443: ✅」。
### 步骤 B腾讯云控制台深度核对定位根因
1. [云服务器 CVM](https://console.cloud.tencent.com/cvm/instance) → 找到存客宝 42.194.245.239
2. **实例详情** → 查看:弹性公网 IP 绑定方式、主网卡/弹性网卡、各网卡绑定的安全组
3. **安全组** → 逐个检查 5 个安全组:入站规则顺序、是否存在 DROP 在 ACCEPT 前
4. **安全产品**:检查是否开启 DDoS 高防、WAF、云防火墙等查看 443 相关策略
5. **对比 kr宝塔**:与 43.139.27.93 实例的网络配置、安全组绑定方式对比差异
### 步骤 C**修复 Nginx 443 监听根因·2026-02 已定位)**
**根因**:运行的是**系统 nginx**`/usr/sbin/nginx`),非**宝塔 nginx**。系统 nginx 配置不含 443仅监听 80。
**修复步骤(任选)**
1. **TAT 脚本**`python3 scripts/腾讯云_TAT_存客宝_Nginx443强制修复.py`
2. **宝塔终端**`killall nginx; sleep 2; /www/server/nginx/sbin/nginx -c /www/server/nginx/conf/nginx.conf`
3. **宝塔面板**:软件商店 → Nginx → 重启
### 步骤 D宝塔防火墙
1. https://42.194.245.239:9988 → 安全 → 防火墙
2. 确认 443 在放行列表
### 步骤 E多网络验证
- 手机 4G 访问https://kr-kf.quwanzhi.com
- 或使用在线端口检测https://www.yougetsignal.com/tools/open-ports/
输入 42.194.245.239,端口 443
---
## 四、替代方案(若 443 仍不通)
### 方案 1临时用 HTTP
- 访问 http://kr-kf.quwanzhi.com 或 http://www.lytiao.com
- 适合内部分发或过渡期
### 方案 2CDN 代理(推荐)
使用 腾讯云 CDN 或 Cloudflare
- 域名解析到 CDN
- CDN 回源协议选 **HTTP80**
- SSL 在 CDN 侧终止,用户访问 HTTPS
这样不依赖服务器 443 开放。
### 方案 3kr宝塔反代推荐立即可用
若 kr宝塔 43.139.27.93 的 443 可访问,可将 kr-kf 解析到 kr宝塔由 kr宝塔 Nginx 反代到存客宝 HTTP 80。
**操作步骤:**
1. 在 kr宝塔 宝塔面板 → 网站 → 添加站点 → 域名 `kr-kf.quwanzhi.com`
2. 站点设置 → 反向代理 → 添加反向代理:
- 代理目录:`/`
- 目标 URL`http://42.194.245.239`
3. 域名解析:将 kr-kf.quwanzhi.com 的 A 记录改为 `43.139.27.93`(原为 42.194.245.239
4. kr宝塔 为该站点配置 SSLLet's Encrypt
**效果**:用户访问 https://kr-kf.quwanzhi.com → kr宝塔 443 → 反代到 存客宝 80。
---
## 五、相关脚本
| 脚本 | 用途 |
|------|------|
| 腾讯云_存客宝安全组放行443.py | 安全组添加 443 |
| 腾讯云_存客宝安全组放行443.py --check | 检查安全组规则 |
| 腾讯云_TAT_存客宝诊断443.py | 服务器内诊断 |
| 腾讯云_TAT_存客宝放行443本地防火墙.py | iptables/宝塔防火墙放行 |
| **存客宝_443放行_宝塔终端执行.sh** | 在宝塔【终端】粘贴执行,确保防火墙 443 放行 |
| **存客宝_443深度诊断.py** | 实例网络+安全组+服务器内全量诊断 |
| **腾讯云_TAT_存客宝_Nginx443修复.py** | 诊断 Nginx 不监听 443 的原因(证书/配置) |
| **腾讯云_TAT_存客宝_Nginx443强制修复.py** | **修复根因**:切回宝塔 nginx恢复 443 监听 |

View File

@@ -27,7 +27,18 @@
## 站点无法访问ERR_CONNECTION_CLOSED
若 kr-kf.quwanzhi.com、lytiao.com 等无法打开:先查 **443 端口**常见为腾讯云安全组未放行 443或 Nginx 未监听 443。详见 `references/存客宝_站点无法访问_ERR_CONNECTION_CLOSED修复.md`
若 kr-kf.quwanzhi.com、lytiao.com 等无法打开:先查 **443 端口**
**⚠️ 优先处理(防重复)**443 不监听时,**先**检查是否运行**系统 Nginx** 而非宝塔 Nginx
```bash
ps aux | grep nginx # 若 /usr/sbin/nginx → 有问题
killall nginx; sleep 2; /www/server/nginx/sbin/nginx -c /www/server/nginx/conf/nginx.conf
```
或 TAT`python3 scripts/腾讯云_TAT_存客宝_Nginx443强制修复.py`
其他:安全组未放行 443、证书未部署等。详见
- `references/存客宝_站点无法访问_ERR_CONNECTION_CLOSED修复.md`
- **深度诊断与替代方案**`references/存客宝_443无法访问_深度诊断与方案.md`
---

View File

@@ -3,6 +3,9 @@
> 现象ERR_CONNECTION_CLOSED面板显示 运行中
> 诊断结果:**443 端口外网不可达**80 正常)
**⚠️ 优先处理2026-02 补充)**:若 443 不监听,**先**检查是否运行系统 Nginx 而非宝塔 Nginx。宝塔终端执行
`ps aux | grep nginx` → 若为 `/usr/sbin/nginx`,则执行 `killall nginx; sleep 2; /www/server/nginx/sbin/nginx -c /www/server/nginx/conf/nginx.conf`。详见 `references/存客宝_443无法访问_深度诊断与方案.md`
---
## 复盘2026-02-22 检查结果)

View File

@@ -0,0 +1,45 @@
#!/bin/bash
# 存客宝 443 放行 · 在宝塔面板【终端】复制整段粘贴执行
# 用于确保 宝塔防火墙 和 系统防火墙 均放行 443
echo "========== 存客宝 443 放行 =========="
echo ""
echo "[1] iptables 添加 443"
iptables -I INPUT -p tcp --dport 443 -j ACCEPT 2>/dev/null && echo " OK" || echo " 已存在或失败"
echo ""
echo "[2] 宝塔 firewall.json 添加 443"
python3 << 'PY'
import json, os
p = "/www/server/panel/data/firewall.json"
if os.path.isfile(p):
with open(p) as f:
d = json.load(f)
ps = d.get("ports", "") or ""
lst = [x.strip() for x in ps.split(",") if x.strip()]
if "443" not in lst:
lst.append("443")
d["ports"] = ",".join(lst)
with open(p, "w") as f:
json.dump(d, f, ensure_ascii=False, indent=2)
print(" 已添加 443")
else:
print(" 已有 443")
PY
echo ""
echo "[3] 重要:宝塔面板 安全 → 防火墙 → 手动添加 443"
echo " 若列表中无 443请点击「添加端口」添加 443协议 TCP策略放行"
echo ""
echo "[4] 若 443 未监听,切回宝塔 Nginx"
if ! ss -tlnp | grep -q ':443 '; then
echo " 443 未监听,重启为宝塔 Nginx..."
killall nginx 2>/dev/null || true
sleep 2
/www/server/nginx/sbin/nginx -c /www/server/nginx/conf/nginx.conf 2>/dev/null && echo " 宝塔 Nginx 已启动" || /etc/init.d/nginx start
sleep 1
fi
echo ""
echo "[5] 检查监听"
ss -tlnp | grep -E ':80 |:443 ' || true
echo ""
echo "========== 完成 =========="
echo "访问 https://kr-kf.quwanzhi.com 测试"

View File

@@ -0,0 +1,264 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
存客宝 443 深度诊断:实例网络、安全组、弹性网卡、服务器内防火墙
定位 42.194.245.239 的 443 不可达根因
"""
import base64
import json
import os
import re
import sys
import time
CKB_IP = "42.194.245.239"
CKB_INSTANCE_ID = "ins-ciyv2mxa"
REGION = "ap-guangzhou"
REGIONS = ["ap-guangzhou", "ap-beijing", "ap-shanghai"]
# 增强 TAT 诊断firewalld、fail2ban、完整 iptables、规则顺序
TAT_CMD = """
echo "=== 1. iptables INPUT 完整链(含行号) ==="
iptables -L INPUT -n -v --line-numbers 2>/dev/null | head -60
echo ""
echo "=== 2. iptables 是否存在 443 DROP ==="
iptables -L INPUT -n -v 2>/dev/null | grep -E '443|dpt:443' || echo "(无 443 相关)"
echo ""
echo "=== 3. firewalld 状态 ==="
systemctl is-active firewalld 2>/dev/null || echo "未安装/未运行"
firewall-cmd --list-all 2>/dev/null | head -20 || true
echo ""
echo "=== 4. fail2ban 是否封 443 ==="
fail2ban-client status 2>/dev/null || echo "fail2ban 未运行"
fail2ban-client status nginx-http-auth 2>/dev/null || true
fail2ban-client status sshd 2>/dev/null || true
echo ""
echo "=== 5. 80/443 监听 ==="
ss -tlnp | grep -E ':80 |:443 '
echo ""
echo "=== 6. 宝塔 firewall.json ==="
cat /www/server/panel/data/firewall.json 2>/dev/null
echo ""
echo "=== 7. 本机 curl 127.0.0.1:443 ==="
curl -sI -o /dev/null -w 'HTTP:%{http_code}' --connect-timeout 3 https://127.0.0.1 -k 2>/dev/null || echo "curl_fail"
echo ""
echo "DONE"
"""
def _find_root():
d = os.path.dirname(os.path.abspath(__file__))
for _ in range(6):
if os.path.basename(d) == "卡若AI" or (os.path.isdir(os.path.join(d, "运营中枢")) and os.path.isdir(os.path.join(d, "01_卡资"))):
return d
d = os.path.dirname(d)
return None
def _read_creds():
root = _find_root()
if not root:
return None, None
path = os.path.join(root, "运营中枢", "工作台", "00_账号与API索引.md")
if not os.path.isfile(path):
return None, None
with open(path, "r", encoding="utf-8") as f:
text = f.read()
sid = skey = None
in_t = False
for line in text.splitlines():
if "### 腾讯云" in line:
in_t = True
continue
if in_t and line.strip().startswith("###"):
break
if not in_t:
continue
m = re.search(r"\|\s*[^|]*(?:SecretId|密钥)[^|]*\|\s*`([^`]+)`", line, re.I)
if m and m.group(1).strip().startswith("AKID"):
sid = m.group(1).strip()
m = re.search(r"\|\s*SecretKey\s*\|\s*`([^`]+)`", line, re.I)
if m:
skey = m.group(1).strip()
return sid or os.environ.get("TENCENTCLOUD_SECRET_ID"), skey or os.environ.get("TENCENTCLOUD_SECRET_KEY")
def main():
secret_id, secret_key = _read_creds()
if not secret_id or not secret_key:
print("❌ 未配置腾讯云 SecretId/SecretKey")
return 1
try:
from tencentcloud.common import credential
from tencentcloud.cvm.v20170312 import cvm_client, models as cvm_models
from tencentcloud.vpc.v20170312 import vpc_client, models as vpc_models
from tencentcloud.tat.v20201028 import tat_client, models as tat_models
except ImportError as e:
print("请安装: pip install tencentcloud-sdk-python-common tencentcloud-sdk-python-cvm tencentcloud-sdk-python-vpc tencentcloud-sdk-python-tat")
return 1
cred = credential.Credential(secret_id, secret_key)
result = {"instance": None, "enis": [], "security_groups": {}, "tat_output": None}
print("=" * 60)
print(" 存客宝 42.194.245.239 443 深度诊断")
print("=" * 60)
# ---------- 1. CVM 实例 ----------
print("\n【1. CVM 实例】")
instance_id = None
region = None
for r in REGIONS:
try:
c = cvm_client.CvmClient(cred, r)
req = cvm_models.DescribeInstancesRequest()
req.Limit = 100
resp = c.DescribeInstances(req)
for ins in (getattr(resp, "InstanceSet", None) or []):
if CKB_IP in list(getattr(ins, "PublicIpAddresses", None) or []):
instance_id = getattr(ins, "InstanceId", None)
region = r
result["instance"] = {
"id": instance_id,
"region": region,
"name": getattr(ins, "InstanceName", ""),
"private_ips": list(getattr(ins, "PrivateIpAddresses", None) or []),
"sg_ids": list(getattr(ins, "SecurityGroupIds", None) or []),
}
break
except Exception as e:
print(" %s: %s" % (r, e))
if instance_id:
break
if not instance_id:
print(" 未找到实例 %s" % CKB_IP)
else:
print(" 实例ID: %s 地域: %s" % (instance_id, region))
print(" 名称: %s" % (result["instance"].get("name") or "-"))
print(" 内网IP: %s" % ", ".join(result["instance"].get("private_ips") or []))
print(" 实例绑定安全组: %s" % ", ".join(result["instance"].get("sg_ids") or []))
# ---------- 2. 弹性网卡含公网IP、isWanIpBlocked----------
print("\n【2. 弹性网卡】")
if region:
try:
vc = vpc_client.VpcClient(cred, region)
req = vpc_models.DescribeNetworkInterfacesRequest()
f = vpc_models.Filter()
f.Name = "attachment.instance-id"
f.Values = [instance_id or CKB_INSTANCE_ID]
req.Filters = [f]
resp = vc.DescribeNetworkInterfaces(req)
enis = getattr(resp, "NetworkInterfaceSet", None) or []
for eni in enis:
ni = {
"eni_id": getattr(eni, "NetworkInterfaceId", ""),
"primary": getattr(eni, "Primary", False),
"private_ips": [],
"wan_ip": None,
"is_wan_blocked": None,
"sg_ids": [],
}
addrs = getattr(eni, "PrivateIpAddressSet", None) or []
for a in addrs:
pip = getattr(a, "PrivateIpAddress", "")
ni["private_ips"].append(pip)
wan = getattr(a, "PublicIpAddress", None) or getattr(a, "WanIp", None)
if wan == CKB_IP or (wan and CKB_IP in str(wan)):
ni["wan_ip"] = wan
blocked = getattr(a, "IsWanIpBlocked", None)
if blocked is not None:
ni["is_wan_blocked"] = blocked
for g in (getattr(eni, "GroupSet", None) or []):
sg_id = getattr(g, "SecurityGroupId", None) or getattr(g, "sgId", None)
if sg_id:
ni["sg_ids"].append(sg_id)
result["enis"].append(ni)
print(" ENI %s (主:%s) | 内网:%s | 公网:%s | 封堵:%s | 安全组:%s" % (
ni["eni_id"], ni["primary"], ni["private_ips"],
ni["wan_ip"] or "-", ni["is_wan_blocked"], ni["sg_ids"]))
if not enis:
print(" (无弹性网卡或 API 无此字段可能为普通公网IP)")
except Exception as e:
print(" DescribeNetworkInterfaces 异常: %s" % e)
# ---------- 3. 安全组规则(含 443、规则顺序----------
print("\n【3. 安全组 443 规则核查】")
sg_ids = result["instance"].get("sg_ids") if result["instance"] else []
if not sg_ids and result["enis"]:
for eni in result["enis"]:
sg_ids.extend(eni.get("sg_ids", []))
sg_ids = list(dict.fromkeys(sg_ids))
if sg_ids and region:
vc = vpc_client.VpcClient(cred, region)
for sg_id in sg_ids:
try:
req = vpc_models.DescribeSecurityGroupPoliciesRequest()
req.SecurityGroupId = sg_id
resp = vc.DescribeSecurityGroupPolicies(req)
s = resp.SecurityGroupPolicySet
ing = (s.Ingress or []) if hasattr(s, "Ingress") else []
rules = []
for i, x in enumerate(ing):
port = getattr(x, "Port", "") or ""
proto = getattr(x, "Protocol", "") or ""
action = getattr(x, "Action", "") or ""
rules.append((i, port, proto, action))
acc443 = [r for r in rules if "443" in str(r[1]) and str(r[3]).upper() == "ACCEPT"]
drop443 = [r for r in rules if "443" in str(r[1]) and str(r[3]).upper() == "DROP"]
result["security_groups"][sg_id] = {"acc443": len(acc443), "drop443": len(drop443), "rules": rules[:15]}
print(" %s: 443 ACCEPT %d 条 | 443 DROP %d 条 | 总规则 %d" % (
sg_id, len(acc443), len(drop443), len(rules)))
if drop443:
print(" ⚠️ 存在 443 DROP顺序: %s" % drop443)
if acc443 and not drop443:
print(" ✅ 443 仅 ACCEPT")
except Exception as e:
print(" %s: %s" % (sg_id, e))
# ---------- 4. TAT 服务器内诊断 ----------
print("\n【4. TAT 服务器内诊断】")
try:
tat = tat_client.TatClient(cred, REGION)
req = tat_models.RunCommandRequest()
req.Content = base64.b64encode(TAT_CMD.encode()).decode()
req.InstanceIds = [CKB_INSTANCE_ID]
req.CommandType = "SHELL"
req.Timeout = 60
req.CommandName = "CKB_443_DeepDiagnose"
resp = tat.RunCommand(req)
inv_id = resp.InvocationId
print(" 已下发,等待 25s...")
time.sleep(25)
req2 = tat_models.DescribeInvocationTasksRequest()
f = tat_models.Filter()
f.Name = "invocation-id"
f.Values = [inv_id]
req2.Filters = [f]
resp2 = tat.DescribeInvocationTasks(req2)
for t in (resp2.InvocationTaskSet or []):
tr = getattr(t, "TaskResult", None)
if tr:
try:
j = json.loads(tr) if isinstance(tr, str) else tr
out = j.get("Output", "")
if out:
try:
import base64 as b64
out = b64.b64decode(out).decode("utf-8", errors="replace")
except Exception:
pass
result["tat_output"] = out
print("\n--- 服务器输出 ---\n%s\n---" % out[:4500])
except Exception:
pass
except Exception as e:
print(" TAT 异常: %s" % e)
print("\n" + "=" * 60)
print("【诊断完成】")
print("=" * 60)
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,142 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
存客宝:通过宝塔 API 为所有站点批量申请 Let's Encrypt SSL并重载 Nginx
解决 443 无法访问Nginx 未监听 443问题。
使用python3 存客宝_宝塔API_全站SSL申请.py
需将执行机 IP 加入存客宝宝塔 API 白名单。
"""
import hashlib
import json
import sys
import time
import urllib.request
import urllib.parse
import ssl
PANEL_URL = "https://42.194.245.239:9988"
API_KEY = "TNKjqDv5N1QLOU20gcmGVgr82Z4mXzRi"
ssl._create_default_https_context = ssl._create_unverified_context
def sign():
t = int(time.time())
s = str(t) + hashlib.md5(API_KEY.encode()).hexdigest()
return {"request_time": t, "request_token": hashlib.md5(s.encode()).hexdigest()}
def post(endpoint, data):
url = PANEL_URL + endpoint
payload = sign()
payload.update(data)
req = urllib.request.Request(url, data=urllib.parse.urlencode(payload).encode(), method="POST")
req.add_header("Content-Type", "application/x-www-form-urlencoded")
try:
with urllib.request.urlopen(req, timeout=30) as resp:
return json.loads(resp.read().decode())
except Exception as e:
return {"status": False, "msg": str(e)}
def main():
print("=" * 60)
print(" 存客宝 · 全站 SSL 批量申请与修复")
print("=" * 60)
# 1. 获取站点列表
sites_r = post("/data?action=getData", {"table": "sites", "limit": 100, "p": 1})
if sites_r.get("status") is not True:
print("❌ 获取站点列表失败:", sites_r)
return 1
site_list = sites_r.get("data") or []
if isinstance(site_list, dict):
site_list = site_list.get("data", site_list) if isinstance(site_list.get("data"), list) else []
if not isinstance(site_list, list):
site_list = []
print("[1] 共 %d 个站点" % len(site_list))
# 2. 逐个处理:未配置 SSL 的尝试申请 Let's Encrypt
applied = []
skipped = []
failed = []
for s in site_list:
name = s.get("name", "")
sid = s.get("id")
ssl_status = s.get("ssl", 0)
if not name or not sid:
continue
domain = name.split()[0] if " " in name else name
if "localhost" in domain or domain == "default":
continue
if ssl_status:
skipped.append(name)
continue
# 申请 Let's Encrypt
print(" [申请] %s (id=%s) ..." % (domain, sid))
r = post("/acme?action=apply_cert", {
"domains": json.dumps([domain]),
"auth_type": "http",
"id": sid,
})
if r.get("status") is True:
applied.append(domain)
print(" ✅ 已申请")
else:
msg = r.get("msg", str(r))
if "already" in str(msg).lower() or "" in str(msg):
applied.append(domain)
print(" ✅ 已有证书")
else:
failed.append((domain, msg))
print("", msg[:80])
time.sleep(2)
# 3. 对已有 SSL 的站点,尝试续期/更新SetSSL 确保开启)
for s in site_list:
name = s.get("name", "")
sid = s.get("id")
ssl_status = s.get("ssl", 0)
if not name or not sid or not ssl_status:
continue
domain = name.split()[0] if " " in name else name
# 确保 SSL 已启用
r = post("/site?action=SetSSL", {
"id": sid,
"type": "1",
})
if r.get("status") is True:
print(" [确认] %s SSL 已启用" % domain)
time.sleep(0.5)
# 4. 重载 Nginx
print("\n[2] 重载 Nginx ...")
reload_r = post("/system?action=ServiceAdmin", {"name": "nginx", "type": "reload"})
if reload_r.get("status") is True:
print(" ✅ Nginx 已重载")
else:
print(" ❌ 重载失败:", reload_r.get("msg", reload_r))
# 5. 汇总
print("\n" + "=" * 60)
print(" 汇总")
print("=" * 60)
print(" 已申请/确认: %s" % ", ".join(applied) if applied else "(无)")
print(" 已有 SSL 跳过: %d" % len(skipped))
if failed:
print(" 申请失败:")
for d, m in failed:
print(" - %s: %s" % (d, m[:60]))
print("\n 请访问 https://kr-kf.quwanzhi.com 等域名测试")
return 0 if not failed else 1
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,132 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
腾讯云 TAT 在存客宝上诊断并修复 Nginx 不监听 443 的问题
根因Nginx 未监听 443 → Connection refused
"""
import base64
import os
import re
import sys
import time
import json
CKB_INSTANCE_ID = "ins-ciyv2mxa"
REGION = "ap-guangzhou"
CMD = """
echo "=== 1. 含 listen 443 的配置 ==="
grep -l 'listen.*443' /www/server/panel/vhost/nginx/*.conf 2>/dev/null || echo "(无)"
echo ""
echo "=== 2. 各站点 SSL 证书路径 ==="
for f in /www/server/panel/vhost/nginx/*.conf; do
if grep -q 'listen.*443' "$f" 2>/dev/null; then
echo "--- $f ---"
grep -E 'ssl_certificate|listen.*443' "$f" | head -6
fi
done
echo ""
echo "=== 3. 证书文件是否存在 ==="
for cert in /www/server/panel/vhost/cert/*/fullchain.pem; do
[ -f "$cert" ] && echo "$cert: $(openssl x509 -in "$cert" -noout -dates -subject 2>/dev/null | tr '\\n' ' ')" || true
done
echo ""
echo "=== 4. Nginx 配置测试 ==="
nginx -t 2>&1
echo ""
echo "=== 5. 重载 Nginx ==="
nginx -s reload 2>&1
echo ""
echo "=== 6. 重载后 443 监听 ==="
ss -tlnp | grep -E ':80 |:443 ' || true
echo ""
echo "DONE"
"""
def _find_root():
d = os.path.dirname(os.path.abspath(__file__))
for _ in range(6):
if os.path.basename(d) == "卡若AI" or (os.path.isdir(os.path.join(d, "运营中枢")) and os.path.isdir(os.path.join(d, "01_卡资"))):
return d
d = os.path.dirname(d)
return None
def _read_creds():
root = _find_root()
path = os.path.join(root or "", "运营中枢", "工作台", "00_账号与API索引.md")
if not os.path.isfile(path):
return None, None
with open(path, "r", encoding="utf-8") as f:
text = f.read()
sid = skey = None
in_t = False
for line in text.splitlines():
if "### 腾讯云" in line:
in_t = True
continue
if in_t and line.strip().startswith("###"):
break
if not in_t:
continue
m = re.search(r"\|\s*[^|]*(?:SecretId|密钥)[^|]*\|\s*`([^`]+)`", line, re.I)
if m and m.group(1).strip().startswith("AKID"):
sid = m.group(1).strip()
m = re.search(r"\|\s*SecretKey\s*\|\s*`([^`]+)`", line, re.I)
if m:
skey = m.group(1).strip()
return sid or os.environ.get("TENCENTCLOUD_SECRET_ID"), skey or os.environ.get("TENCENTCLOUD_SECRET_KEY")
def main():
secret_id, secret_key = _read_creds()
if not secret_id or not secret_key:
print("❌ 未配置腾讯云 SecretId/SecretKey")
return 1
try:
from tencentcloud.common import credential
from tencentcloud.tat.v20201028 import tat_client, models
except ImportError:
print("pip install tencentcloud-sdk-python-common tencentcloud-sdk-python-tat")
return 1
cred = credential.Credential(secret_id, secret_key)
client = tat_client.TatClient(cred, REGION)
req = models.RunCommandRequest()
req.Content = base64.b64encode(CMD.encode()).decode()
req.InstanceIds = [CKB_INSTANCE_ID]
req.CommandType = "SHELL"
req.Timeout = 60
req.CommandName = "CKB_Nginx443_Fix"
resp = client.RunCommand(req)
inv_id = resp.InvocationId
print("⏳ TAT 已下发 Nginx 443 诊断与修复,等待 25s...")
time.sleep(25)
try:
req2 = models.DescribeInvocationTasksRequest()
f = models.Filter()
f.Name = "invocation-id"
f.Values = [inv_id]
req2.Filters = [f]
resp2 = client.DescribeInvocationTasks(req2)
for t in (resp2.InvocationTaskSet or []):
tr = getattr(t, "TaskResult", None)
if tr:
try:
j = json.loads(tr) if isinstance(tr, str) else tr
out = j.get("Output", "")
if out:
try:
import base64 as b64
out = b64.b64decode(out).decode("utf-8", errors="replace")
except Exception:
pass
print("\n--- 服务器输出 ---\n%s\n---" % out[:5000])
except Exception:
print(" TaskResult:", str(tr)[:800])
except Exception as e:
print(" 查询异常:", e)
print("\n若 443 仍未监听,需在宝塔 → 网站 → kr-kf/lytiao → SSL 中申请或配置证书后重载")
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,114 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
TAT 在存客宝上:确认实际运行的 Nginx 及其配置,强制重载并开放 443
根因推测:宝塔 Nginx 与系统 Nginx 可能不同,需用宝塔的 nginx 重载
"""
import base64
import json
import os
import re
import sys
import time
CKB_INSTANCE_ID = "ins-ciyv2mxa"
REGION = "ap-guangzhou"
CMD = r'''
echo "=== 1. Nginx 进程 ==="
ps aux | grep -E "nginx|PID" | grep -v grep
echo ""
echo "=== 2. 宝塔 Nginx 测试 ==="
BT_NGINX="/www/server/nginx/sbin/nginx"
BT_CONF="/www/server/nginx/conf/nginx.conf"
[ -x "$BT_NGINX" ] && $BT_NGINX -t -c $BT_CONF 2>&1
echo ""
echo "=== 3. 用宝塔 Nginx 重启(覆盖系统 nginx==="
# 停掉可能的系统 nginx
/etc/init.d/nginx stop 2>/dev/null || killall nginx 2>/dev/null || true
sleep 2
# 启动宝塔 nginx
[ -x "$BT_NGINX" ] && $BT_NGINX -c $BT_CONF 2>&1 && echo "宝塔 Nginx 已启动" || /etc/init.d/nginx start 2>&1
sleep 2
echo ""
echo "=== 4. 443 监听 ==="
ss -tlnp | grep -E ':80 |:443 '
echo ""
echo "=== 5. curl 127.0.0.1:443 ==="
curl -sk -o /dev/null -w "HTTP:%{http_code}" --connect-timeout 3 https://127.0.0.1 -k 2>/dev/null || echo "fail"
echo ""
echo "DONE"
'''
def _read_creds():
d = os.path.dirname(os.path.abspath(__file__))
for _ in range(6):
root = d
if os.path.basename(d) == "卡若AI":
break
d = os.path.dirname(d)
p = os.path.join(root, "运营中枢", "工作台", "00_账号与API索引.md")
if not os.path.isfile(p):
return None, None
with open(p, "r", encoding="utf-8") as f:
t = f.read()
sid = skey = None
in_t = False
for line in t.splitlines():
if "### 腾讯云" in line:
in_t = True
continue
if in_t and line.strip().startswith("###"):
break
if not in_t:
continue
m = re.search(r"SecretId[^|]*\|\s*`([^`]+)`", line, re.I)
if m and m.group(1).strip().startswith("AKID"):
sid = m.group(1).strip()
m = re.search(r"SecretKey[^|]*\|\s*`([^`]+)`", line, re.I)
if m:
skey = m.group(1).strip()
return sid or os.environ.get("TENCENTCLOUD_SECRET_ID"), skey or os.environ.get("TENCENTCLOUD_SECRET_KEY")
def main():
sid, skey = _read_creds()
if not sid or not skey:
print("❌ 未配置腾讯云凭证")
return 1
from tencentcloud.common import credential
from tencentcloud.tat.v20201028 import tat_client, models
cred = credential.Credential(sid, skey)
client = tat_client.TatClient(cred, REGION)
req = models.RunCommandRequest()
req.Content = base64.b64encode(CMD.encode()).decode()
req.InstanceIds = [CKB_INSTANCE_ID]
req.CommandType = "SHELL"
req.Timeout = 60
req.CommandName = "CKB_Nginx443_ForceFix"
resp = client.RunCommand(req)
print("⏳ TAT 已下发 Nginx 443 强制修复,等待 25s...")
time.sleep(25)
req2 = models.DescribeInvocationTasksRequest()
f = models.Filter()
f.Name = "invocation-id"
f.Values = [resp.InvocationId]
req2.Filters = [f]
r2 = client.DescribeInvocationTasks(req2)
for t in (r2.InvocationTaskSet or []):
tr = getattr(t, "TaskResult", None)
if tr:
try:
jj = json.loads(tr) if isinstance(tr, str) else tr
out = jj.get("Output", "")
if out:
out = base64.b64decode(out).decode("utf-8", errors="replace")
print("\n--- 服务器输出 ---\n%s\n---" % out[:4000])
except Exception:
print(str(tr)[:600])
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,164 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
TAT 在存客宝上:诊断 Nginx 443 不监听原因,并尝试通过宝塔 API127.0.0.1)批量申请 SSL
服务器内调用 127.0.0.1 无需白名单。
"""
import base64
import json
import os
import re
import sys
import time
CKB_INSTANCE_ID = "ins-ciyv2mxa"
REGION = "ap-guangzhou"
API_KEY = "TNKjqDv5N1QLOU20gcmGVgr82Z4mXzRi"
CMD = r'''
echo "=== 1. nginx -t 完整输出 ==="
nginx -t 2>&1 || true
echo ""
echo "=== 2. 检查各 listen 443 站点的 ssl_certificate 文件 ==="
for f in /www/server/panel/vhost/nginx/*.conf; do
[ -f "$f" ] || continue
if grep -q "listen.*443" "$f" 2>/dev/null; then
cert=$(grep "ssl_certificate " "$f" 2>/dev/null | grep -v "#" | head -1 | awk '{print $2}' | tr -d ';')
key=$(grep "ssl_certificate_key " "$f" 2>/dev/null | grep -v "#" | head -1 | awk '{print $2}' | tr -d ';')
echo "--- $(basename $f) ---"
[ -n "$cert" ] && echo " cert: $cert" && ([ -f "$cert" ] && echo " 存在" || echo " 不存在")
[ -n "$key" ] && echo " key: $key" && ([ -f "$key" ] && echo " 存在" || echo " 不存在")
fi
done
echo ""
echo "=== 3. 宝塔 Python 批量申请 Let'\''s Encrypt未配置站点==="
/www/server/panel/pyenv/bin/python3 << 'PYEND'
import hashlib, json, time, urllib.request, urllib.parse, ssl
ssl._create_default_https_context = ssl._create_unverified_context
API_KEY = "TNKjqDv5N1QLOU20gcmGVgr82Z4mXzRi"
BASE = "https://127.0.0.1:9988"
def sign():
t = int(time.time())
s = str(t) + hashlib.md5(API_KEY.encode()).hexdigest()
return {"request_time": t, "request_token": hashlib.md5(s.encode()).hexdigest()}
def post(path, data):
payload = sign()
payload.update(data)
req = urllib.request.Request(BASE + path, data=urllib.parse.urlencode(payload).encode())
try:
with urllib.request.urlopen(req, timeout=15) as r:
return json.loads(r.read().decode())
except Exception as e:
return {"status": False, "msg": str(e)}
r = post("/data?action=getData", {"table": "sites", "limit": 100, "p": 1})
sites = r.get("data") or []
if isinstance(sites, dict): sites = sites.get("data", []) or []
if not isinstance(sites, list): sites = []
for s in sites:
if s.get("ssl", 0) or not s.get("name"): continue
name = str(s.get("name", "")).split()[0]
if "localhost" in name: continue
sid = s.get("id")
print(" 申请 %s (id=%s) ..." % (name, sid))
r2 = post("/acme?action=ApplyCert", {"id": sid, "domains": json.dumps([name]), "auth_type": "http"})
print(" ", r2.get("msg", r2) if not r2.get("status") else "OK")
time.sleep(4)
PYEND
echo ""
echo "=== 4. 重载 Nginx ==="
nginx -s reload 2>&1 || /etc/init.d/nginx reload 2>&1
echo ""
echo "=== 5. 443 监听 ==="
ss -tlnp | grep -E ':80 |:443 ' || true
echo "DONE"
'''
def _find_root():
d = os.path.dirname(os.path.abspath(__file__))
for _ in range(6):
if os.path.basename(d) == "卡若AI" or (os.path.isdir(os.path.join(d, "运营中枢")) and os.path.isdir(os.path.join(d, "01_卡资"))):
return d
d = os.path.dirname(d)
return None
def _read_creds():
root = _find_root()
p = os.path.join(root or "", "运营中枢", "工作台", "00_账号与API索引.md")
if not os.path.isfile(p):
return None, None
with open(p, "r", encoding="utf-8") as f:
t = f.read()
sid = skey = None
in_t = False
for line in t.splitlines():
if "### 腾讯云" in line:
in_t = True
continue
if in_t and line.strip().startswith("###"):
break
if not in_t:
continue
m = re.search(r"\|\s*[^|]*(?:SecretId|密钥)[^|]*\|\s*`([^`]+)`", line, re.I)
if m and m.group(1).strip().startswith("AKID"):
sid = m.group(1).strip()
m = re.search(r"\|\s*SecretKey\s*\|\s*`([^`]+)`", line, re.I)
if m:
skey = m.group(1).strip()
return sid or os.environ.get("TENCENTCLOUD_SECRET_ID"), skey or os.environ.get("TENCENTCLOUD_SECRET_KEY")
def main():
sid, skey = _read_creds()
if not sid or not skey:
print("❌ 未配置腾讯云凭证")
return 1
try:
from tencentcloud.common import credential
from tencentcloud.tat.v20201028 import tat_client, models
except ImportError:
print("pip install tencentcloud-sdk-python-tat")
return 1
cred = credential.Credential(sid, skey)
client = tat_client.TatClient(cred, REGION)
req = models.RunCommandRequest()
req.Content = base64.b64encode(CMD.encode()).decode()
req.InstanceIds = [CKB_INSTANCE_ID]
req.CommandType = "SHELL"
req.Timeout = 120
req.CommandName = "CKB_SSL_DiagFix"
resp = client.RunCommand(req)
inv_id = resp.InvocationId
print("⏳ TAT 已下发 SSL 诊断与修复(约 60s...")
time.sleep(60)
try:
req2 = models.DescribeInvocationTasksRequest()
f = models.Filter()
f.Name = "invocation-id"
f.Values = [inv_id]
req2.Filters = [f]
r2 = client.DescribeInvocationTasks(req2)
for t in (r2.InvocationTaskSet or []):
tr = getattr(t, "TaskResult", None)
if tr:
try:
jj = json.loads(tr) if isinstance(tr, str) else tr
out = jj.get("Output", "")
if out:
try:
out = base64.b64decode(out).decode("utf-8", errors="replace")
except Exception:
pass
print("\n--- 服务器输出 ---\n%s\n---" % out[:6000])
except Exception:
print(str(tr)[:800])
except Exception as e:
print("查询:", e)
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,145 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
腾讯云 TAT存客宝 lytiao Docker 诊断与修复(容器未在面板显示时执行)
"""
import base64
import os
import re
import sys
CKB_INSTANCE_ID = "ins-ciyv2mxa"
REGION = "ap-guangzhou"
def _find_karuo_ai_root():
d = os.path.dirname(os.path.abspath(__file__))
for _ in range(6):
if os.path.basename(d) == "卡若AI" or (
os.path.isdir(os.path.join(d, "运营中枢")) and os.path.isdir(os.path.join(d, "01_卡资"))
):
return d
d = os.path.dirname(d)
return None
def _read_creds():
root = _find_karuo_ai_root()
if not root:
return None, None
path = os.path.join(root, "运营中枢", "工作台", "00_账号与API索引.md")
if not os.path.isfile(path):
return None, None
with open(path, "r", encoding="utf-8") as f:
text = f.read()
secret_id = secret_key = None
in_tencent = False
for line in text.splitlines():
if "### 腾讯云" in line:
in_tencent = True
continue
if in_tencent and line.strip().startswith("###"):
break
if not in_tencent:
continue
m = re.search(r"\|\s*[^|]*(?:SecretId|密钥)[^|]*\|\s*`([^`]+)`", line, re.I)
if m:
val = m.group(1).strip()
if val.startswith("AKID"):
secret_id = val
m = re.search(r"\|\s*SecretKey\s*\|\s*`([^`]+)`", line, re.I)
if m:
secret_key = m.group(1).strip()
return secret_id or None, secret_key or None
# 诊断 + 修复:检查 lytiao 容器,不存在则重新部署
CMD = """echo "=== 1. 所有容器(含已停止) ==="
docker ps -a
echo ""
echo "=== 2. lytiao 相关容器 ==="
docker ps -a --filter "name=lytiao" 2>/dev/null || true
echo ""
echo "=== 3. /opt/lytiao_docker 是否存在 ==="
ls -la /opt/lytiao_docker 2>/dev/null || echo "目录不存在"
echo ""
echo "=== 4. 修复:启动或重新部署 lytiao ==="
if [ -d /opt/lytiao_docker ] && [ -f /opt/lytiao_docker/docker-compose.yml ]; then
cd /opt/lytiao_docker
docker compose up -d
echo ""
echo "=== 5. 修复后容器列表 ==="
docker ps -a
else
echo "目录或 compose 文件缺失,需要重新部署。执行部署脚本..."
mkdir -p /opt/lytiao_docker
SRC="/www/wwwroot/www.lytiao.com"
if [ ! -d "$SRC" ]; then
echo "错误: 网站源 $SRC 不存在"
exit 1
fi
cat > /opt/lytiao_docker/Dockerfile << 'DFEND'
FROM php:7.1-apache
RUN a2enmod rewrite
RUN apt-get update && apt-get install -y libpng-dev libjpeg-dev libzip-dev zip unzip \\
&& docker-php-ext-configure gd --with-png-dir=/usr --with-jpeg-dir=/usr \\
&& docker-php-ext-install -j$(nproc) gd mysqli pdo pdo_mysql zip \\
&& apt-get clean && rm -rf /var/lib/apt/lists/*
WORKDIR /var/www/html
EXPOSE 80
DFEND
cat > /opt/lytiao_docker/docker-compose.yml << 'DCEND'
version: "3.8"
services:
lytiao-web:
build: .
container_name: lytiao-www
ports:
- "8080:80"
volumes:
- ./www:/var/www/html:ro
restart: unless-stopped
environment:
- TZ=Asia/Shanghai
DCEND
rm -rf /opt/lytiao_docker/www
cp -a "$SRC" /opt/lytiao_docker/www
cd /opt/lytiao_docker
docker compose up -d --build
echo ""
echo "=== 部署完成,容器列表 ==="
docker ps -a
fi
echo ""
echo "=== DONE ==="
"""
def main():
secret_id = os.environ.get("TENCENTCLOUD_SECRET_ID")
secret_key = os.environ.get("TENCENTCLOUD_SECRET_KEY")
if not secret_id or not secret_key:
sid, skey = _read_creds()
secret_id = secret_id or sid
secret_key = secret_key or skey
if not secret_id or not secret_key:
print("❌ 未配置腾讯云 SecretId/SecretKey")
return 1
try:
from tencentcloud.common import credential
from tencentcloud.tat.v20201028 import tat_client, models
except ImportError:
print("请安装: pip install tencentcloud-sdk-python-common tencentcloud-sdk-python-tat")
return 1
cred = credential.Credential(secret_id, secret_key)
client = tat_client.TatClient(cred, REGION)
req = models.RunCommandRequest()
req.Content = base64.b64encode(CMD.encode()).decode()
req.InstanceIds = [CKB_INSTANCE_ID]
req.CommandType = "SHELL"
req.Timeout = 120
req.CommandName = "CKB_lytiao_DockerDiagnose"
resp = client.RunCommand(req)
print("✅ 诊断与修复指令已下发 InvocationId:", resp.InvocationId)
print(" 约 12 分钟后到 宝塔 → Docker → 总览 点击「刷新容器列表」即可看到 lytiao-www")
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,142 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
TAT 在存客宝上通过 宝塔 API本机 127.0.0.1 无需白名单)添加防火墙 443 端口
"""
import base64
import os
import re
import sys
import time
CKB_INSTANCE_ID = "ins-ciyv2mxa"
REGION = "ap-guangzhou"
# 在服务器上执行:宝塔防火墙放行 443 + 重载防火墙
CMD = r'''
set -e
echo "=== 1. 宝塔 firewall.json 添加 443 ==="
python3 << 'PY'
import json, os
p = "/www/server/panel/data/firewall.json"
if os.path.isfile(p):
with open(p) as f:
d = json.load(f)
ps = d.get("ports", "") or ""
lst = [x.strip() for x in ps.split(",") if x.strip()]
if "443" not in lst:
lst.append("443")
d["ports"] = ",".join(lst)
with open(p, "w") as f:
json.dump(d, f, ensure_ascii=False, indent=2)
print(" firewall.json 443 已添加")
else:
print(" firewall.json 已有 443")
PY
echo ""
echo "=== 2. 调用宝塔防火墙重载 ==="
/www/server/panel/pyenv/bin/python -c "
import sys
sys.path.insert(0, '/www/server/panel/class')
try:
from firewallModel import firewallModel
f = firewallModel()
f.add_accept_port('443', '443', 'tcp')
print(' add_accept_port 443 OK')
except Exception as e:
print(' add_accept_port:', e)
" 2>/dev/null || true
echo ""
echo "=== 3. iptables 确保 443 ACCEPT ==="
iptables -C INPUT -p tcp --dport 443 -j ACCEPT 2>/dev/null || iptables -I INPUT -p tcp --dport 443 -j ACCEPT
echo " iptables 443 OK"
echo ""
echo "DONE"
'''
def _find_root():
d = os.path.dirname(os.path.abspath(__file__))
for _ in range(6):
if os.path.basename(d) == "卡若AI" or (os.path.isdir(os.path.join(d, "运营中枢")) and os.path.isdir(os.path.join(d, "01_卡资"))):
return d
d = os.path.dirname(d)
return None
def _read_creds():
root = _find_root()
if not root:
return None, None
p = os.path.join(root, "运营中枢", "工作台", "00_账号与API索引.md")
if not os.path.isfile(p):
return None, None
with open(p, "r", encoding="utf-8") as f:
t = f.read()
sid = skey = None
in_t = False
for line in t.splitlines():
if "### 腾讯云" in line:
in_t = True
continue
if in_t and line.strip().startswith("###"):
break
if not in_t:
continue
m = re.search(r"\|\s*[^|]*(?:SecretId|密钥)[^|]*\|\s*`([^`]+)`", line, re.I)
if m and m.group(1).strip().startswith("AKID"):
sid = m.group(1).strip()
m = re.search(r"\|\s*SecretKey\s*\|\s*`([^`]+)`", line, re.I)
if m:
skey = m.group(1).strip()
return sid or os.environ.get("TENCENTCLOUD_SECRET_ID"), skey or os.environ.get("TENCENTCLOUD_SECRET_KEY")
def main():
sid, skey = _read_creds()
if not sid or not skey:
print("❌ 未配置凭证")
return 1
try:
from tencentcloud.common import credential
from tencentcloud.tat.v20201028 import tat_client, models
except ImportError:
print("pip install tencentcloud-sdk-python-tat")
return 1
cred = credential.Credential(sid, skey)
client = tat_client.TatClient(cred, REGION)
req = models.RunCommandRequest()
req.Content = base64.b64encode(CMD.encode()).decode()
req.InstanceIds = [CKB_INSTANCE_ID]
req.CommandType = "SHELL"
req.Timeout = 45
req.CommandName = "CKB_BT_Firewall443"
resp = client.RunCommand(req)
print("✅ TAT 已下发 宝塔防火墙 443")
time.sleep(20)
try:
import json as j
import base64 as b64
req2 = models.DescribeInvocationTasksRequest()
f = models.Filter()
f.Name = "invocation-id"
f.Values = [resp.InvocationId]
req2.Filters = [f]
r2 = client.DescribeInvocationTasks(req2)
for t in (r2.InvocationTaskSet or []):
tr = getattr(t, "TaskResult", None)
if tr:
try:
jj = j.loads(tr) if isinstance(tr, str) else tr
out = jj.get("Output", "")
if out:
try:
out = b64.b64decode(out).decode("utf-8", errors="replace")
except Exception:
pass
print(out[:2500])
except Exception:
print(str(tr)[:600])
except Exception as e:
print("查询:", e)
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -108,8 +108,10 @@ def main():
print("%s 已添加 443/TCP 入站" % sg_id)
added += 1
except Exception as e:
if "RuleAlreadyExists" in str(e) or "已存在" in str(e):
err = str(e)
if "RuleAlreadyExists" in err or "已存在" in err or "duplicate" in err.lower():
print("%s 443 规则已存在" % sg_id)
added += 1
else:
print("%s: %s" % (sg_id, e))
@@ -152,7 +154,10 @@ def check_rules():
resp = vc.DescribeSecurityGroupPolicies(req)
s = resp.SecurityGroupPolicySet
ing = (s.Ingress or []) if hasattr(s, "Ingress") else []
print(" %s 入站: %s" % (sg_id, [(getattr(x,"Port",""), getattr(x,"Protocol","")) for x in ing[:8]]))
rules = [(getattr(x,"Port",""), getattr(x,"Protocol",""), getattr(x,"Action","")) for x in ing]
has443_acc = any("443" in str(r[0]) and str(r[2]).upper()=="ACCEPT" for r in rules)
drop443 = [r for r in rules if "443" in str(r[0]) and str(r[2]).upper()=="DROP"]
print(" %s 入站(共%d条) | 443 ACCEPT: %s | 443 DROP: %s" % (sg_id, len(rules), "" if has443_acc else "", "⚠️有" if drop443 else ""))
except Exception as e:
print(" %s: %s" % (sg_id, e))
return 0

View File

@@ -0,0 +1,226 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
飞书妙记 · 按剪辑方案图片切片
============================
按「视频剪辑方案」图片中的 高峰时刻 + 想象的内容 整理切片,
去语助词、去空格、关键词高亮、加速10%
用法:
1. 先手动在飞书妙记页点击下载视频
2. python3 feishu_image_slice.py --video "下载的视频.mp4"
或指定飞书链接(会打开链接,待你下载后监控 ~/Downloads
python3 feishu_image_slice.py --url "https://cunkebao.feishu.cn/minutes/xxx"
"""
import argparse
import json
import os
import re
import subprocess
import sys
import time
from pathlib import Path
# 图片方案高峰时刻7段约10分钟内
# 来源:视频剪辑方案 | 智能剪辑 - 提取高峰时刻
PEAK_MOMENTS = [
{"start": "00:00:05", "end": "00:00:30", "title": "引出问题建立共鸣", "hook_3sec": "这个问题你遇到过吗?", "cta_ending": "关注我,每天学一招"},
{"start": "00:01:15", "end": "00:02:00", "title": "解决方案核心价值", "hook_3sec": "核心方法来了", "cta_ending": "想学更多评论区扣1"},
{"start": "00:03:40", "end": "00:04:20", "title": "案例分享增加信任", "hook_3sec": "真实案例告诉你", "cta_ending": "关注获取完整案例"},
{"start": "00:05:00", "end": "00:05:45", "title": "未来展望激发行动", "hook_3sec": "接下来这样做", "cta_ending": "行动起来,评论区见"},
{"start": "00:06:30", "end": "00:07:10", "title": "痛点强调促成转化", "hook_3sec": "这个坑千万别踩", "cta_ending": "收藏避坑"},
{"start": "00:08:00", "end": "00:08:50", "title": "福利展示刺激购买", "hook_3sec": "福利限时放送", "cta_ending": "私信领取"},
{"start": "00:09:30", "end": "00:10:15", "title": "权威背书打消疑虑", "hook_3sec": "专业背书可信", "cta_ending": "关注我,每天学一招私域干货"},
]
# 想象的内容:用于 AI 识别时的关键词(若用转录)
IMAGINED_KEYWORDS = ["痛点", "解决方案", "场景", "案例", "数据", "情感", "号召"]
SCRIPT_DIR = Path(__file__).resolve().parent
SOUL_SLICE = Path("/Users/karuo/Documents/个人/卡若AI/03_卡木/木叶_视频内容/视频切片/脚本")
OUTPUT_BASE = Path.home() / "Downloads" / "feishu_image_clips"
def get_video_duration(path: Path) -> float:
"""获取视频时长(秒)"""
cmd = ["ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "default=noprint_wrappers=1:nokey=1", str(path)]
r = subprocess.run(cmd, capture_output=True, text=True)
return float(r.stdout.strip()) if r.returncode == 0 else 0
def parse_time(s: str) -> float:
"""HH:MM:SS 或 MM:SS -> 秒"""
parts = s.strip().split(":")
if len(parts) == 3:
return int(parts[0]) * 3600 + int(parts[1]) * 60 + float(parts[2])
if len(parts) == 2:
return int(parts[0]) * 60 + float(parts[1])
return float(parts[0]) if parts else 0
def format_time(sec: float) -> str:
h = int(sec // 3600)
m = int((sec % 3600) // 60)
s = int(sec % 60)
return f"{h:02d}:{m:02d}:{s:02d}"
def scale_highlights(highlights: list, duration_sec: float, template_end_sec: float = 615) -> list:
"""按视频时长缩放高峰时刻"""
if duration_sec >= template_end_sec:
return [{**h, "start_time": h["start"], "end_time": h["end"]}
for h in highlights if parse_time(h["end"]) <= duration_sec]
scale = duration_sec / template_end_sec
out = []
for h in highlights:
start_sec = parse_time(h["start"]) * scale
end_sec = parse_time(h["end"]) * scale
if end_sec - start_sec < 15:
continue
out.append({
**h,
"start_time": format_time(start_sec),
"end_time": format_time(end_sec),
})
return out
def find_recent_video(after_time: float) -> Path | None:
down = Path.home() / "Downloads"
cands = []
for f in list(down.glob("*.mp4")) + list(down.glob("*.mov")):
if f.stat().st_mtime > after_time:
cands.append((f, f.stat().st_mtime))
return max(cands, key=lambda x: x[1])[0] if cands else None
def main():
parser = argparse.ArgumentParser(description="飞书妙记 · 按剪辑方案图片切片")
parser.add_argument("--video", "-v", help="视频文件路径")
parser.add_argument("--url", "-u", help="飞书妙记链接(会打开,待手动下载)")
parser.add_argument("--output", "-o", help="输出目录")
args = parser.parse_args()
video_path = None
if args.video:
video_path = Path(args.video).resolve()
elif args.url:
print("📌 正在打开飞书链接,请在页面中点击【下载】按钮")
subprocess.run(["open", args.url], check=True)
start = time.time()
print("⏳ 监控 ~/Downloads检测到新视频后自动继续...")
for _ in range(120):
time.sleep(3)
v = find_recent_video(start)
if v and v.stat().st_size > 1_000_000:
video_path = v
print(f"✅ 检测到: {v.name}")
break
if not video_path:
print("❌ 10分钟内未检测到新视频请下载后使用 --video 指定路径")
return
if not video_path or not video_path.exists():
print("❌ 请提供 --video 路径,或使用 --url 并完成下载")
return
duration = get_video_duration(video_path)
print(f"📹 视频: {video_path.name}")
print(f" 时长: {duration/60:.1f} 分钟")
# 高峰时刻(按需缩放)
template_end = parse_time(PEAK_MOMENTS[-1]["end"])
highlights = scale_highlights(PEAK_MOMENTS, duration, template_end)
for h in highlights:
h["start_time"] = h.get("start_time") or h["start"]
h["end_time"] = h.get("end_time") or h["end"]
highlights = [h for h in highlights if parse_time(h["end_time"]) <= duration + 5]
print(f" 切片: {len(highlights)} 段(高峰时刻方案)")
# 输出目录
out_dir = Path(args.output) if args.output else OUTPUT_BASE / video_path.stem
out_dir.mkdir(parents=True, exist_ok=True)
base_dir = out_dir / (video_path.stem + "_output")
base_dir.mkdir(parents=True, exist_ok=True)
transcript_path = base_dir / "transcript.srt"
highlights_path = base_dir / "highlights.json"
clips_dir = base_dir / "clips"
enhanced_dir = base_dir / "clips_enhanced"
# 保存 highlights
with open(highlights_path, "w", encoding="utf-8") as f:
json.dump(highlights, f, ensure_ascii=False, indent=2)
# 1. 转录(可选,用于 soul_enhance 字幕;无则跳过增强的字幕烧录)
if not transcript_path.exists():
print("\n📝 提取音频 + MLX 转录...")
audio_path = base_dir / "audio.wav"
subprocess.run(
["ffmpeg", "-y", "-i", str(video_path), "-vn", "-ar", "16000", "-ac", "1", str(audio_path)],
capture_output=True, check=True
)
try:
subprocess.run([
"mlx_whisper", str(audio_path),
"--model", "mlx-community/whisper-small-mlx", "--language", "zh",
"--output-format", "srt", "--output-dir", str(base_dir), "--output-name", "transcript"
], capture_output=True, text=True, timeout=900)
except Exception as e:
print(f" ⚠ 转录跳过: {e}")
# 2. 批量切片
print("\n✂️ 批量切片...")
clips_dir.mkdir(parents=True, exist_ok=True)
batch_clip = SOUL_SLICE / "batch_clip.py"
subprocess.run([
sys.executable, str(batch_clip),
"--input", str(video_path),
"--highlights", str(highlights_path),
"--output", str(clips_dir),
"--prefix", "peak",
], check=True)
# 3. 增强(封面+字幕+加速10%)— 需 transcript
if transcript_path.exists():
soul_enhance = SOUL_SLICE / "soul_enhance.py"
enhanced_dir.mkdir(parents=True, exist_ok=True)
print("\n🎨 增强加速10%+字幕+关键词高亮)...")
subprocess.run([
sys.executable, str(soul_enhance),
"--clips", str(clips_dir),
"--highlights", str(highlights_path),
"--transcript", str(transcript_path),
"--output", str(enhanced_dir),
], capture_output=True)
result_dir = enhanced_dir
else:
# 仅加速10%(无字幕)
print("\n⚡ 加速10%(无转录,跳过字幕)...")
enhanced_dir.mkdir(parents=True, exist_ok=True)
for f in sorted(clips_dir.glob("*.mp4")):
out_f = enhanced_dir / f.name.replace(".mp4", "_enhanced.mp4")
subprocess.run([
"ffmpeg", "-y", "-i", str(f),
"-filter_complex", "[0:v]setpts=0.909*PTS[v];[0:a]atempo=1.1[a]",
"-map", "[v]", "-map", "[a]",
"-c:v", "libx264", "-preset", "fast", "-crf", "23",
"-c:a", "aac", "-b:a", "128k", str(out_f)
], capture_output=True)
result_dir = enhanced_dir
print("\n" + "=" * 50)
print("✅ 完成")
print("=" * 50)
print(f"📂 切片: {clips_dir}")
print(f"📂 增强: {result_dir}")
print(f"📋 方案: {highlights_path}")
if __name__ == "__main__":
main()

View File

@@ -18,13 +18,24 @@ python3 /Users/karuo/Documents/个人/卡若AI/02_卡人/飞书管理/s
"飞书群webhook可选默认产宁团队群"
```
### 按剪辑方案图片切片(高峰时刻+想象的内容)
按「视频剪辑方案」图片整理7 段高峰时刻 + 加速 10% + 去语助词 + 关键词高亮。
```bash
# 1. 打开飞书链接,点击下载视频
# 2. 下载完成后运行(或直接用 --video 指定已有视频)
python3 /Users/karuo/Documents/个人/卡若AI/02_卡人/水桥_平台对接/飞书管理/脚本/feishu_image_slice.py --url "https://cunkebao.feishu.cn/minutes/obcnzs51k1j754643vx138sx"
# 若已下载,直接指定视频路径
python3 脚本/feishu_image_slice.py --video "~/Downloads/xxx.mp4"
```
### 示例
```bash
# 示例产研团队第20场会议
python3 /Users/karuo/Documents/个人/卡若AI/02_卡人/飞书管理/scripts/feishu_one_click.py \
"https://cunkebao.feishu.cn/minutes/obcnjnsx2mz7vj5q843172p8" \
--clips 5
python3 feishu_one_click.py "https://cunkebao.feishu.cn/minutes/obcnjnsx2mz7vj5q843172p8" --clips 5
```
---

View File

@@ -0,0 +1,65 @@
# 宝塔 443 不监听 · 系统 Nginx 与宝塔 Nginx 优先排查
> 来源卡若AI 服务器管理 | 存客宝 42.194.245.239 | 2026-02
## 现象
- 80 可达443 不可达
- 外网连接 443 返回 **Connection refused**
- 安全组、iptables、证书、nginx -t 均正常
## 根因(优先检查)
**运行的是系统 Nginx`/usr/sbin/nginx`),非宝塔 Nginx。**
- 系统 Nginx 使用 `/etc/nginx/` 配置,通常不含宝塔的 vhost含 listen 443 ssl
- 宝塔 Nginx 使用 `/www/server/nginx/conf/nginx.conf`,包含所有站点的 SSL 配置
- 若系统 Nginx 先启动或接管,则只监听 80不监听 443
## 诊断命令
```bash
# 1. 查看实际运行的 Nginx 进程
ps aux | grep nginx
# 若显示 /usr/sbin/nginx → 系统 Nginx有问题
# 若显示 /www/server/nginx/sbin/nginx → 宝塔 Nginx正常
# 2. 检查 443 是否监听
ss -tlnp | grep 443
# 无输出则 443 未监听
```
## 修复(任选其一)
### 方式一TAT 脚本(本机执行,免 SSH
```bash
cd "01_卡资/金仓_存储备份/服务器管理/scripts"
.venv_tencent/bin/python3 腾讯云_TAT_存客宝_Nginx443强制修复.py
```
### 方式二:宝塔面板终端
```bash
killall nginx
sleep 2
/www/server/nginx/sbin/nginx -c /www/server/nginx/conf/nginx.conf
```
### 方式三:宝塔面板
软件商店 → Nginx → 重启
## 防重复
**宝塔服务器 443 不监听时,优先执行:**
1. `ps aux | grep nginx` 确认是否系统 Nginx
2. 若是,直接执行上述修复,再查安全组/证书
## 相关文档
- `references/存客宝_443无法访问_深度诊断与方案.md`
- `scripts/腾讯云_TAT_存客宝_Nginx443强制修复.py`
- `scripts/存客宝_443放行_宝塔终端执行.sh`(含自动切回宝塔 Nginx 逻辑)

View File

@@ -1,21 +1,23 @@
---
name: 视频切片
description: Soul派对视频切片处理。触发词视频切片、Soul切片、派对切片、处理视频。
description: Soul派对视频切片处理。触发词视频切片、Soul切片、派对切片、处理视频。所有文档与字幕输出统一简体中文。
group: 木
triggers: 视频剪辑、切片发布、字幕烧录
owner: 木叶
version: "1.0"
updated: "2026-02-16"
version: "1.1"
updated: "2026-02-22"
---
# 视频切片
> **语言**:所有文档、字幕、封面文案统一使用**简体中文**。soul_enhance 自动繁转简。
## ⭐ Soul派对切片流程默认
```
原始视频 → MLX转录 → 精彩片段识别 → 批量切片 → 增强处理 → 输出成片
封面+字幕+加速10%+去语气词
原始视频 → MLX转录 → 字幕转简体 → 高光识别(Ollama→规则) → 批量切片 → soul_enhance → 输出成片
提取后立即繁转简+修正错误 封面+字幕(已简体)+加速10%+去语气词
```
### 一键命令Soul派对专用
@@ -26,9 +28,12 @@ updated: "2026-02-16"
cd 03_卡木/木叶_视频内容/视频切片/脚本
conda activate mlx-whisper
python3 soul_slice_pipeline.py --video "/path/to/soul派对会议第57场.mp4" --clips 6
# 仅重新烧录(字幕转简体后重跑增强)
python3 soul_slice_pipeline.py -v "视频.mp4" -n 6 --skip-transcribe --skip-highlights --skip-clips
```
流程:**转录 → 高光识别(AI/规则) → 批量切片 → 增强(Hook+CTA)**
流程:**转录 → 字幕转简体 → 高光识别 → 批量切片 → 增强**
#### 分步命令
@@ -38,24 +43,23 @@ eval "$(~/miniforge3/bin/conda shell.zsh hook)"
conda activate mlx-whisper
mlx_whisper audio.wav --model mlx-community/whisper-small-mlx --language zh --output-format all
# 2. 高光识别Ollama → Groq → 规则,不依赖 Gemini
# 2. 高光识别Ollama → 规则;流水线会在读取 transcript 前自动转简体
python3 identify_highlights.py -t transcript.srt -o highlights.json -n 6
# 3. 切片
python3 batch_clip.py -i 视频.mp4 -l highlights.json -o clips/
python3 batch_clip.py -i 视频.mp4 -l highlights.json -o clips/ -p soul
# 4. 增强处理(Hook+CTA 封面文字
python3 enhance_clips.py -c clips/ -l highlights.json -o clips_enhanced/
# 4. 增强处理(封面+字幕+加速soul_enhance
python3 soul_enhance.py -c clips/ -l highlights.json -t transcript.srt -o clips_enhanced/
```
### 增强功能说明
| 功能 | 说明 |
|------|------|
| **封面贴片** | 前2.5秒显示Hook文字,毛玻璃背景 |
| **字幕烧录** | 关键词金黄高亮,思源黑体 |
| **封面贴片** | 前2.5秒 Hook,苹方/思源黑体 |
| **字幕烧录** | 关键词加粗加大亮金黄突出,去语助词+去空格 |
| **加速10%** | 节奏更紧凑,适合短视频 |
| **去语气词** | 清理"嗯、啊、那个"等语气词 |
### 时间预估
@@ -203,12 +207,13 @@ python3 scripts/burn_subtitles_clean.py -i enhanced.mp4 -s clean.srt -o 成片.m
| 脚本 | 功能 | 使用频率 |
|------|------|---------|
| **one_video.py** | 一键处理,输出单个成片 | ⭐⭐⭐ 最常用 |
| burn_subtitles_clean.py | 字幕烧录(无阴影) | ⭐⭐ |
| burn_subtitles_pro.py | 字幕烧录(有阴影 | ⭐ |
| fix_subtitles.py | 字幕清洗(繁转简) | ⭐⭐ |
| batch_clip.py | 批量切片 | ⭐ |
| soul_clip.py | Soul派对专用 | ⭐ |
| **soul_slice_pipeline.py** | Soul 切片一体化流水线 | ⭐⭐⭐ 最常用 |
| **soul_enhance.py** | 封面+字幕(简体)+加速+去语气词 | ⭐⭐ |
| identify_highlights.py | 高光识别Ollama→规则 | ⭐ |
| batch_clip.py | 批量切片 | ⭐⭐ |
| one_video.py | 单视频一键成片 | ⭐ |
| burn_subtitles_clean.py | 字幕烧录(无阴影) | ⭐ |
| fix_subtitles.py | 字幕清洗(繁转简) | ⭐ |
---
@@ -219,7 +224,8 @@ python3 scripts/burn_subtitles_clean.py -i enhanced.mp4 -s clean.srt -o 成片.m
- **MLX Whisper**: `~/miniforge3/envs/mlx-whisper`**默认转录引擎**
- Apple Silicon优化比CPU Whisper快10倍+
- 2.5小时视频转录仅需3分钟
- **字体**: `03_卡木/视频切片/fonts/`
- **字体**: `03_卡木/木叶_视频内容/视频切片/fonts/`(优先)
- **字幕**: 统一简体中文soul_enhance 自动繁转简)
### 转录命令(默认)

View File

@@ -24,8 +24,8 @@ pyJianYingDraft>=0.2.5 # 程序化生成剪映草稿
# playwright>=1.40.0 # 浏览器自动化
# 安装后需要playwright install chromium
# 高光识别可选Groq 免费 APIOllama 失败时使用
# groq>=1.0.0
# 字幕简体中文soul_enhance 繁转简
opencc-python-reimplemented>=0.1.0
# 可选依赖(高级功能)
# pyannote.audio>=3.1.0 # 说话人分离需要HuggingFace Token

View File

@@ -8,7 +8,7 @@
| 组件 | 实现 | 说明 |
|------|------|------|
| **转录** | MLX Whisper | 本地、快速,无 API 依赖 |
| **高光识别** | Ollama → Groq → 规则 | 级联,不依赖 Gemini |
| **高光识别** | Ollama → 规则 | 级联,不依赖 Gemini/Groq |
| **切片** | FFmpeg batch_clip | 标准工具 |
| **增强** | soul_enhancePillow 封面+字幕)| 无需 drawtext |
@@ -54,8 +54,7 @@
| 转录 | MLX Whisper / Ollama Whisper | ✅ 可用 |
| 高光识别 | Ollama qwen2.5:1.5b | ✅ 可用 |
| 切片 | FFmpeg | ✅ 可用 |
| 封面/Hook | FFmpeg drawtext | 需 libfreetype |
| 无 drawtext | 复制原始切片 | 流水线自动降级 |
| 封面/字幕 | soul_enhancePillow| 简体中文,自动繁转简 |
### 云端可选
@@ -69,7 +68,7 @@
- 默认 `brew install ffmpeg` 可能无 `drawtext`(缺 libfreetype
- 可选:`brew install ffmpeg@7` 或自行编译 `--enable-libfreetype`
- 当前增强enhance_clips 用 drawtext无 drawtext 时流水线直接复制切片
- 当前增强:soul_enhance 用 Pillow封面+字幕均为简体中文
---
@@ -84,13 +83,15 @@ Pillow
# 高光识别
# Ollama + qwen2.5:1.5b(本地,无需 pip
groq # 可选pip install groq
# 字幕简体中文(可选,推荐)
opencc-python-reimplemented # 繁转简soul_enhance 可用
```
---
## 五、推荐组合(最简)
1. **本地全流程**MLX Whisper → Ollama → FFmpeg(无 drawtext 则复制切片
2. **提升质量**:设置 `GROQ_API_KEY`Ollama 失败时自动用 Groq
3. **无需 Gemini**:当前方案已完全脱离 Gemini 依赖
1. **本地全流程**MLX Whisper → Ollama → FFmpeg → soul_enhance简体中文
2. **无需 Gemini/Groq**:只用 Ollama + 规则
3. **字幕**:自动繁转简,统一简体中文

View File

@@ -1,5 +1,7 @@
# 高光识别提示词
> 所有输出Hook、CTA、标题统一**简体中文**。
用于分析视频文字稿识别最有传播力的片段并生成前3秒Hook和结尾CTA。
## 完整提示词

View File

@@ -100,7 +100,7 @@ def _build_prompt(transcript: str, clip_count: int) -> str:
- transcript_excerpt: 片段内容前50字
- reason: 推荐理由
时长 {MIN_DURATION}-{MAX_DURATION}相邻间隔30秒。只输出 JSON 数组,不要其他文字或```包裹。
时长 {MIN_DURATION}-{MAX_DURATION}相邻间隔30秒。输出必须使用简体中文。只输出 JSON 数组,不要其他文字或```包裹。
视频文字稿:
---

View File

@@ -39,23 +39,67 @@ SPEED_FACTOR = 1.10 # 加速10%
SILENCE_THRESHOLD = -40 # 静音阈值(dB)
SILENCE_MIN_DURATION = 0.5 # 最短静音时长(秒)
# 语气词列表(需要清理的
# 繁转简OpenCC 优先,否则用映射
_OPENCC = None
def _get_opencc():
global _OPENCC
if _OPENCC is None:
try:
from opencc import OpenCC
_OPENCC = OpenCC('t2s')
except ImportError:
_OPENCC = False
return _OPENCC
def _to_simplified(text: str) -> str:
"""转为简体中文"""
cc = _get_opencc()
if cc:
return cc.convert(text)
# 常用繁简映射(无 opencc 时)
trad_simp = {
'': '', '': '', '': '', '': '', '': '',
'': '', '': '', '': '', '': '', '': '',
'': '', '': '', '': '', '': '', '': '',
}
for t, s in trad_simp.items():
text = text.replace(t, s)
return text
# 常见转录错误修正(与 one_video 一致)
CORRECTIONS = {
'私余': '私域', '统安': '同安', '信一下': '线上', '头里': '投入',
'幅画': '负责', '施育': '私域', '经历论': '净利润', '成于': '乘以',
'马的': '码的', '猜济': '拆解', '巨圣': '矩阵', '货客': '获客',
}
# 语助词列表(需清理,含常见口头禅)
FILLER_WORDS = [
'', '', '', '', '', '', '', '',
'那个', '就是', '然后', '这个', '所以说', '怎么说',
'对吧', '是吧', '好吧', '行吧', '', '',
'', '', '', '', '', '', '', '', '', '',
'那个', '就是', '然后', '这个', '所以说', '怎么说', '怎么说呢',
'对吧', '是吧', '好吧', '行吧', '', '', '就是那个',
'其实', '那么', '然后呢', '还有就是', '以及', '另外', '等等',
'怎么说呢', '你知道吗', '我跟你说', '', '', 'OK', 'ok',
]
# 关键词高亮
# 关键词高亮(重点突出,按长度排序避免短词覆盖长词)
KEYWORDS = [
'100万', '30万', '50万', '10万', '5万', '1万',
'私域', 'AI', '自动化', '矩阵', 'SOP', 'IP',
'获客', '变现', '分润', '转化', '复购', '裂变',
'阿米巴', '电商', '创业', '项目', '收益',
'抖音', 'Soul', '微信', '美团',
'100万', '50万', '30万', '10万', '5万', '1万',
'存客宝', '私域', '自动化', '阿米巴', '矩阵', '获客', '变现',
'分润', '转化', '复购', '裂变', 'AI', 'SOP', 'IP',
'电商', '创业', '项目', '收益', '流量', '引流',
'抖音', 'Soul', '微信', '美团', '方法', '技巧', '干货',
'核心', '关键', '重点', '赚钱', '收入', '利润',
]
# 样式配置
# 字体优先级Mac 优先苹方,更清晰)
FONT_PRIORITY = [
"/System/Library/Fonts/PingFang.ttc", # 苹方-简Mac 默认,清晰)
"/System/Library/Fonts/STHeiti Medium.ttc",
"/Library/Fonts/Arial Unicode.ttf",
]
# 样式配置(字体更大、关键词更突出)
STYLE = {
'cover': {
'bg_blur': 30,
@@ -63,31 +107,33 @@ STYLE = {
'duration': 2.5,
},
'hook': {
'font_size': 72,
'font_size': 76,
'color': (255, 255, 255),
'outline_color': (40, 40, 80),
'outline_width': 4,
'outline_color': (30, 30, 50),
'outline_width': 5,
},
'subtitle': {
'font_size': 42,
'font_size': 44,
'color': (255, 255, 255),
'outline_color': (30, 30, 30),
'outline_width': 3,
'keyword_color': (255, 215, 0), # 金黄
'bg_color': (20, 20, 40, 180),
'margin_bottom': 60,
'outline_color': (25, 25, 25),
'outline_width': 4,
'keyword_color': (255, 200, 50), # 金黄
'keyword_outline': (80, 50, 0), # 深黄描边
'keyword_size_add': 4, # 关键词字号+4
'bg_color': (15, 15, 35, 200),
'margin_bottom': 70,
}
}
# ============ 工具函数 ============
def get_font(font_path, size):
"""获取字体,带回退"""
for path in [font_path, FONT_BOLD, FALLBACK_FONT]:
if os.path.exists(path):
"""获取字体,优先苹方/系统字体"""
for path in [font_path, FONT_BOLD] + FONT_PRIORITY + [FALLBACK_FONT]:
if path and os.path.exists(path):
try:
return ImageFont.truetype(path, size)
except:
except Exception:
continue
return ImageFont.load_default()
@@ -109,14 +155,21 @@ def draw_text_with_outline(draw, pos, text, font, color, outline_color, outline_
draw.text((x, y), text, font=font, fill=color)
def clean_filler_words(text):
"""清理语气词"""
"""清理语助词 + 去除多余空格"""
result = text
for word in FILLER_WORDS:
# 只清理独立的语气词(前后有空格或标点或在开头结尾)
result = re.sub(rf'^{word}[,,、]?\s*', '', result)
result = re.sub(rf'\s*[,,、]?{word}$', '', result)
result = re.sub(rf'\s+{word}\s+', ' ', result)
return result.strip()
# 按长度降序,先删长词避免残留
for word in sorted(FILLER_WORDS, key=len, reverse=True):
if not word:
continue
result = re.sub(rf'^{re.escape(word)}[,,、\s]*', '', result)
result = re.sub(rf'[,,、\s]*{re.escape(word)}$', '', result)
result = re.sub(rf'\s+{re.escape(word)}\s+', ' ', result)
result = re.sub(rf'[,,、]+{re.escape(word)}[,,、\s]*', '', result)
# 合并多余空格、去除首尾空格
result = re.sub(r'\s+', ' ', result)
result = re.sub(r'\s*[,]\s*', '', result)
result = re.sub(r'[,]+', '', result).strip(' ,')
return result
def parse_srt_for_clip(srt_path, start_sec, end_sec):
"""解析SRT提取指定时间段的字幕"""
@@ -143,7 +196,10 @@ def parse_srt_for_clip(srt_path, start_sec, end_sec):
rel_start = max(0, sub_start - start_sec)
rel_end = sub_end - start_sec
# 清理语气词
# 繁转简 + 修正错误 + 清理语气词
text = _to_simplified(text)
for w, c in CORRECTIONS.items():
text = text.replace(w, c)
cleaned_text = clean_filler_words(text)
if len(cleaned_text) > 1: # 过滤太短的
subtitles.append({
@@ -184,7 +240,8 @@ def get_video_info(video_path):
# ============ 封面生成 ============
def create_cover_image(hook_text, width, height, output_path, video_path=None):
"""创建封面贴片"""
"""创建封面贴片(简体中文)"""
hook_text = _to_simplified(str(hook_text))
style = STYLE['cover']
hook_style = STYLE['hook']
@@ -265,13 +322,16 @@ def create_cover_image(hook_text, width, height, output_path, video_path=None):
# ============ 字幕图片生成 ============
def create_subtitle_image(text, width, height, output_path):
"""创建字幕图片(关键词高亮"""
"""创建字幕图片(关键词加粗加大突出"""
style = STYLE['subtitle']
img = Image.new('RGBA', (width, height), (0, 0, 0, 0))
draw = ImageDraw.Draw(img)
font = get_font(FONT_BOLD, style['font_size'])
base_size = style['font_size']
kw_size = base_size + style.get('keyword_size_add', 4)
font = get_font(FONT_BOLD, base_size)
kw_font = get_font(FONT_HEAVY, kw_size) # 关键词用粗体+大字
text_w, text_h = get_text_size(draw, text, font)
base_x = (width - text_w) // 2
@@ -293,15 +353,18 @@ def create_subtitle_image(text, width, height, output_path):
img = Image.alpha_composite(img, bg_layer)
draw = ImageDraw.Draw(img)
# 识别关键词位置
# 识别关键词位置(按长度降序,长词优先避免短词截断)
highlights = []
for keyword in KEYWORDS:
for keyword in sorted(KEYWORDS, key=len, reverse=True):
start = 0
while True:
pos = text.find(keyword, start)
if pos == -1:
break
highlights.append((pos, pos + len(keyword)))
# 避免重叠
overlap = any(s <= pos < e or s < pos + len(keyword) <= e for s, e in highlights)
if not overlap:
highlights.append((pos, pos + len(keyword)))
start = pos + 1
highlights = sorted(highlights, key=lambda x: x[0])
@@ -320,15 +383,17 @@ def create_subtitle_image(text, width, height, output_path):
break
if in_keyword:
# 绘制整个关键词
# 关键词:粗体+大字+亮金黄+深色描边
keyword_text = text[char_idx:keyword_end]
kw_outline = style.get('keyword_outline', (60, 40, 0))
kw_ow = style.get('outline_width', 3) + 1
draw_text_with_outline(
draw, (current_x, base_y), keyword_text, font,
draw, (current_x, base_y - 1), keyword_text, kw_font,
style['keyword_color'],
style['outline_color'],
style['outline_width']
kw_outline,
kw_ow
)
kw_w, _ = get_text_size(draw, keyword_text, font)
kw_w, _ = get_text_size(draw, keyword_text, kw_font)
current_x += kw_w
char_idx = keyword_end
else:

View File

@@ -4,7 +4,7 @@
Soul 切片一体化流水线
视频制作(封面/Hook格式+ 视频切片
流程:转录 → 高光识别(AI) → 批量切片 → 增强(封面+Hook+CTA)
流程:转录 → 字幕转简体 → 高光识别(AI) → 批量切片 → 增强(封面+字幕+CTA)
"""
import argparse
import json
@@ -17,6 +17,53 @@ SCRIPT_DIR = Path(__file__).resolve().parent
SKILL_DIR = SCRIPT_DIR.parent
FONTS_DIR = SKILL_DIR / "fonts"
# 常见转录错误修正
CORRECTIONS = {
'私余': '私域', '统安': '同安', '信一下': '线上', '头里': '投入',
'幅画': '负责', '施育': '私域', '经历论': '净利润', '成于': '乘以',
'马的': '码的', '猜济': '拆解', '巨圣': '矩阵', '货客': '获客',
}
# 语助词(转录后去除)
FILLER_WORDS = [
'', '', '', '', '', '', '', '', '', '',
'那个', '就是', '然后', '这个', '所以说', '怎么说', '怎么说呢',
'对吧', '是吧', '好吧', '行吧', '其实', '那么', '以及', '另外',
]
def transcript_to_simplified(srt_path: Path) -> bool:
"""转录后立即处理:繁转简+修正错误+去语助词+去多余空格"""
import re
try:
from opencc import OpenCC
cc = OpenCC('t2s')
except ImportError:
cc = None
with open(srt_path, 'r', encoding='utf-8') as f:
content = f.read()
def clean_line(line: str) -> str:
if not line.strip() or line.strip().isdigit() or '-->' in line:
return line
s = cc.convert(line) if cc else line
for w, c in CORRECTIONS.items():
s = s.replace(w, c)
for w in sorted(FILLER_WORDS, key=len, reverse=True):
s = re.sub(rf'^{re.escape(w)}[,,、\s]*', '', s)
s = re.sub(rf'[,,、\s]*{re.escape(w)}$', '', s)
s = re.sub(rf'\s+{re.escape(w)}\s+', ' ', s)
s = re.sub(r'\s+', ' ', s).strip(' ,')
return s
lines = content.split('\n')
out = []
for line in lines:
out.append(clean_line(line))
with open(srt_path, 'w', encoding='utf-8') as f:
f.write('\n'.join(out))
return True
def run(cmd: list, desc: str = "", check: bool = True, timeout: int = 600) -> bool:
if desc:
@@ -44,6 +91,7 @@ def main():
parser.add_argument("--clips", "-n", type=int, default=8, help="切片数量")
parser.add_argument("--skip-transcribe", action="store_true", help="跳过转录(已有 transcript.srt")
parser.add_argument("--skip-highlights", action="store_true", help="跳过高光识别(已有 highlights.json")
parser.add_argument("--skip-clips", action="store_true", help="跳过切片(已有 clips/,仅重新增强)")
args = parser.parse_args()
video_path = Path(args.video).resolve()
@@ -102,6 +150,10 @@ def main():
print(f"❌ 需要 transcript.srt请先完成转录: {transcript_path}")
sys.exit(1)
# 1.5 字幕转简体(提取后立即处理,繁转简+修正错误)
transcript_to_simplified(transcript_path)
print(" ✓ 字幕已转简体")
# 2. 高光识别
if not args.skip_highlights:
run(
@@ -133,18 +185,22 @@ def main():
# 3. 批量切片
clips_dir.mkdir(parents=True, exist_ok=True)
run(
[
sys.executable,
str(SCRIPT_DIR / "batch_clip.py"),
"--input", str(video_path),
"--highlights", str(highlights_path),
"--output", str(clips_dir),
"--prefix", "soul",
],
"批量切片",
timeout=300,
)
if not args.skip_clips:
run(
[
sys.executable,
str(SCRIPT_DIR / "batch_clip.py"),
"--input", str(video_path),
"--highlights", str(highlights_path),
"--output", str(clips_dir),
"--prefix", "soul",
],
"批量切片",
timeout=300,
)
elif not list(clips_dir.glob("*.mp4")):
print("❌ clips/ 为空,请去掉 --skip-clips 或先完成切片")
sys.exit(1)
# 4. 增强(封面+字幕+加速soul_enhancePillow无需 drawtext
enhanced_dir.mkdir(parents=True, exist_ok=True)

View File

@@ -25,3 +25,20 @@ Soul 视频切片流水线已统一为:转录(MLX Whisper)→高光(Ollama→
**▶ 下一步执行**
后续 Soul 视频可直接跑 `python3 soul_slice_pipeline.py --video "xxx.mp4" --clips 6`,无需配置 Groq/Gemini。
---
## [卡若复盘] 文档与字幕简体中文优化2026-02
**🎯 目标·结果·达成率**
目标:所有文档与字幕统一简体中文,参考 one_video 优化完整流程。结果soul_enhance 增加繁转简+修正错误SKILL/参考资料全部标注简体中文,达成率 100%。
**📌 过程**
1. **soul_enhance**:增加 `_to_simplified`OpenCC 优先,无则映射)+ `CORRECTIONS`(与 one_video 一致),字幕与封面文案自动繁转简。
2. **SKILL.md**:分步命令改为 soul_enhance脚本列表更新注明「简体中文」。
3. **参考资料**AI视频切片_GitHub替代方案.md、高光识别提示词.md 补充简体中文说明。
4. **identify_highlights**prompt 增加「输出必须使用简体中文」。
**💡 反思**
1. 繁转简优先用 OpenCC无依赖时用映射兜底。
2. 文档明确标注「简体中文」可减少歧义。

View File

@@ -0,0 +1,34 @@
# 宝塔 443 不监听 · 系统 Nginx 与宝塔 Nginx 优先排查
> 来源:存客宝 42.194.245.239 | 2026-02
## 现象
- 80 可达443 不可达;外网 443 返回 Connection refused
- 安全组、iptables、证书、nginx -t 均正常
## 根因(优先检查)
**运行的是系统 Nginx`/usr/sbin/nginx`),非宝塔 Nginx。**
- 系统 Nginx`/etc/nginx/`,不含宝塔 vhost 的 listen 443
- 宝塔 Nginx`/www/server/nginx/conf/nginx.conf`,含全部 SSL
## 诊断
```bash
ps aux | grep nginx # 若 /usr/sbin/nginx → 有问题
ss -tlnp | grep 443 # 无输出则未监听
```
## 修复
```bash
killall nginx
sleep 2
/www/server/nginx/sbin/nginx -c /www/server/nginx/conf/nginx.conf
```
## 防重复
**宝塔服务器 443 不监听时,先查 `ps aux | grep nginx`,若为系统 Nginx 则按上修复。**

View File

@@ -62,3 +62,4 @@
| 2026-02-22 09:06:58 | 🔄 卡若AI 同步 2026-02-22 09:06 | 更新:总索引与入口、金仓、运营中枢工作台 | 排除 >20MB: 8 个 |
| 2026-02-22 09:12:03 | 🔄 卡若AI 同步 2026-02-22 09:12 | 更新:总索引与入口、火种知识模型、运营中枢参考资料、运营中枢工作台 | 排除 >20MB: 8 个 |
| 2026-02-22 09:20:12 | 🔄 卡若AI 同步 2026-02-22 09:20 | 更新:金仓、卡木、运营中枢参考资料、运营中枢工作台 | 排除 >20MB: 8 个 |
| 2026-02-22 09:27:09 | 🔄 卡若AI 同步 2026-02-22 09:27 | 更新:金仓、卡木、运营中枢工作台 | 排除 >20MB: 8 个 |

View File

@@ -65,3 +65,4 @@
| 2026-02-22 09:06:58 | 成功 | 成功 | 🔄 卡若AI 同步 2026-02-22 09:06 | 更新:总索引与入口、金仓、运营中枢工作台 | 排除 >20MB: 8 个 | [仓库](http://open.quwanzhi.com:3000/fnvtk/karuo-ai) [百科](http://open.quwanzhi.com:3000/fnvtk/karuo-ai/wiki) |
| 2026-02-22 09:12:03 | 成功 | 成功 | 🔄 卡若AI 同步 2026-02-22 09:12 | 更新:总索引与入口、火种知识模型、运营中枢参考资料、运营中枢工作台 | 排除 >20MB: 8 个 | [仓库](http://open.quwanzhi.com:3000/fnvtk/karuo-ai) [百科](http://open.quwanzhi.com:3000/fnvtk/karuo-ai/wiki) |
| 2026-02-22 09:20:12 | 成功 | 成功 | 🔄 卡若AI 同步 2026-02-22 09:20 | 更新:金仓、卡木、运营中枢参考资料、运营中枢工作台 | 排除 >20MB: 8 个 | [仓库](http://open.quwanzhi.com:3000/fnvtk/karuo-ai) [百科](http://open.quwanzhi.com:3000/fnvtk/karuo-ai/wiki) |
| 2026-02-22 09:27:09 | 成功 | 成功 | 🔄 卡若AI 同步 2026-02-22 09:27 | 更新:金仓、卡木、运营中枢工作台 | 排除 >20MB: 8 个 | [仓库](http://open.quwanzhi.com:3000/fnvtk/karuo-ai) [百科](http://open.quwanzhi.com:3000/fnvtk/karuo-ai/wiki) |