eric/gitlab-instance-0a899031_douyinyoutube
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

's'

Browse Source
This commit is contained in:
eric
2026-03-28 02:39:29 -05:00
parent 243b9d7acb
commit 4b3c6cbee4
23 changed files with 1468 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

351
web.py
View File

@@ -1,45 +1,97 @@
from fastapi import FastAPI, Query, Request
from fastapi.responses import FileResponse, JSONResponse
from pathlib import Path
import asyncio
import re
from typing import Optional
from fastapi.middleware.cors import CORSMiddleware
from __future__ import annotations
app = FastAPI(title="多进程录制控制台")
import asyncio
import sys
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Optional
from fastapi import FastAPI, Query, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from src.web_process_backend import WebProcessBackend, force_kill_ffmpeg, strip_ansi_codes
# ---------------- 配置（根目录结构） ----------------
BASE_DIR = Path(__file__).parent
BASE_DIR = Path(__file__).parent.resolve()
CONFIG_DIR = BASE_DIR / "config"
DEFAULT_LOG_DIR = BASE_DIR / ".pm2" / "logs"
CONSOLE_OUT = BASE_DIR / "web-console" / "out"
MAX_LOG_LINES = 300
process_backend = WebProcessBackend(BASE_DIR, DEFAULT_LOG_DIR)
def _label_short(prefix: str) -> str:
return {"tiktok": "TikTok", "youtube": "YouTube", "obs": "OBS", "web": "Web 控制台"}.get(prefix, prefix)
return {
"tiktok": "TikTok",
"youtube": "YouTube",
"douyin_youtube": "抖音→YouTube",
"obs": "OBS",
"web": "Web 控制台",
}.get(prefix, prefix)
def _sort_script_entries(entries: list[dict], web_script_name: str) -> list[dict]:
"""下拉列表顺序：抖音→YouTube → YouTube → TikTok → OBS → Web 控制台（默认首选 YouTube 业务）。"""
def sort_key(e: dict) -> tuple:
script = e["script"]
st = Path(script).stem.lower()
if st.startswith("douyin_youtube"):
return (0, script)
if script.endswith(".py") and st.startswith("youtube"):
return (1, script)
if st.startswith("tiktok"):
return (2, script)
if st.startswith("obs"):
return (3, script)
if script == web_script_name or st == "web":
return (99, script)
return (50, script)
return sorted(entries, key=sort_key)
def discover_script_entries() -> tuple:
"""扫描项目根目录自动识别入口：tiktok*.py、youtube*.py、obs*.sh，以及当前本文件（Web 控制台）。PM2 名 = 主文件名（无扩展名）。"""
"""扫描根目录：tiktok*.py、youtube*.py、douyin_youtube*.py；OBS 在 Windows 用 obs*.bat/.cmd，在类 Unix 用 obs*.sh。"""
base = BASE_DIR
web_path = Path(__file__).resolve()
entries: list[dict] = []
seen_stems: set[str] = set()
def add(script_name: str, label_key: str) -> None:
stem = Path(script_name).stem
if stem in seen_stems:
return
seen_stems.add(stem)
entries.append({"script": script_name, "label": _label_short(label_key)})
for path in sorted(base.glob("tiktok*.py")):
if path.is_file() and path.resolve() != web_path:
entries.append({"script": path.name, "label": _label_short("tiktok")})
add(path.name, "tiktok")
for path in sorted(base.glob("youtube*.py")):
if path.is_file() and path.resolve() != web_path:
entries.append({"script": path.name, "label": _label_short("youtube")})
add(path.name, "youtube")
for path in sorted(base.glob("obs*.sh")):
if path.is_file():
entries.append({"script": path.name, "label": _label_short("obs")})
for path in sorted(base.glob("douyin_youtube*.py")):
if path.is_file() and path.resolve() != web_path:
add(path.name, "douyin_youtube")
if sys.platform == "win32":
for path in sorted(list(base.glob("obs*.bat")) + list(base.glob("obs*.cmd"))):
if path.is_file():
add(path.name, "obs")
else:
for path in sorted(base.glob("obs*.sh")):
if path.is_file():
add(path.name, "obs")
entries.append({"script": web_path.name, "label": _label_short("web")})
entries = _sort_script_entries(entries, web_path.name)
return tuple(entries)
@@ -55,7 +107,10 @@ def _stem(script: str) -> str:
def is_youtube_script(script: str) -> bool:
return script.endswith(".py") and _stem(script).startswith("youtube")
if not script.endswith(".py"):
return False
s = _stem(script).lower()
return s.startswith("youtube") or s.startswith("douyin_youtube")
def is_tiktok_script(script: str) -> bool:
@@ -89,7 +144,15 @@ def script_for_pm2(pm2: str) -> Optional[str]:
return p["script"]
return None
# CORS
@asynccontextmanager
async def lifespan(app: FastAPI):
await process_backend.refresh_mode()
yield
app = FastAPI(title="多进程录制控制台", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
@@ -97,50 +160,12 @@ app.add_middleware(
allow_headers=["*"],
)
# ---------------- 工具函数 ----------------
def strip_ansi_codes(text: str) -> str:
return re.sub(r'\x1b\[[0-9;]*m', '', text)
async def run_pm2(cmd: str, timeout: float = 15.0) -> str:
process = await asyncio.create_subprocess_shell(
f"pm2 {cmd}",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
try:
stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
result = (stdout or stderr).decode(errors="ignore").strip()
return result if result else "命令执行完成（无输出）"
except asyncio.TimeoutError:
process.kill()
await process.wait()
return "ERROR: PM2 命令超时，已强制终止"
async def force_kill_ffmpeg(process_name: str):
"""尝试强杀与该进程相关的 ffmpeg"""
cmds = [
# 优先杀带有进程名的 ffmpeg（最精准）
f"pkill -f 'ffmpeg.*{process_name}' || true",
# 再杀所有 ffmpeg（兜底，比较暴力）
"killall -9 ffmpeg 2>/dev/null || true",
]
for cmd in cmds:
proc = await asyncio.create_subprocess_shell(
cmd,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
await proc.wait()
await asyncio.sleep(0.4) # 稍微间隔，避免竞争
async def read_log_path(path: str, lines: int) -> str:
async def read_log_path(path: Optional[str], lines: int) -> str:
if not path:
return "无日志路径\n"
if "/dev/null" in path:
norm = path.replace("\\", "/").lower()
if "/dev/null" in path or norm.endswith("/nul") or norm == "nul":
return "日志输出被禁用（PM2 配置为无日志）\n"
log_file = Path(path)
if not log_file.exists():
@@ -149,127 +174,144 @@ async def read_log_path(path: str, lines: int) -> str:
with open(log_file, "r", encoding="utf-8", errors="ignore") as f:
content_lines = f.readlines()
return "".join(content_lines[-lines:]) if content_lines else "无日志内容\n"
except Exception as e:
except OSError as e:
return f"读取日志失败 ({path}): {str(e)}\n"
async def get_process_info(process: str) -> dict:
describe_raw = await run_pm2(f"describe {process}", timeout=10.0)
describe_clean = strip_ansi_codes(describe_raw)
return await process_backend.parse_pm2_describe(process)
if any(x in describe_clean.lower() for x in ["no such process", "unknown process", "not found"]) or not describe_clean.strip():
return {
"process_status": "not_found",
"out_path": None,
"err_path": None,
}
status = "unknown"
out_path = None
err_path = None
# ---------------- 服务信息（前端可展示 PM2 / 本地后端） ----------------
@app.get("/health")
async def health():
return {"status": "ok"}
for line in describe_raw.splitlines():
line_clean = strip_ansi_codes(line)
if line_clean.count('│') < 2:
continue
parts = [p.strip() for p in line_clean.split('│') if p.strip()]
if len(parts) >= 2:
key = parts[0].lower()
value = parts[1]
if key == "status":
status = value.lower()
elif key == "out_log_path":
out_path = value
elif key == "err_log_path":
err_path = value
if not out_path:
out_path = str(DEFAULT_LOG_DIR / f"{process}-out.log")
if not err_path:
err_path = str(DEFAULT_LOG_DIR / f"{process}-error.log")
@app.get("/server_info")
async def server_info(refresh: bool = Query(False, description="为 true 时重新检测 PM2 是否可用")):
if refresh:
await process_backend.refresh_mode()
else:
await process_backend.ensure_mode()
return {
"process_status": status,
"out_path": out_path,
"err_path": err_path,
"platform": sys.platform,
"process_backend": process_backend.mode,
"console_built": (CONSOLE_OUT / "index.html").is_file(),
"message": "PM2 可用时使用 PM2；否则使用本地 PID 表（无需 npm i -g pm2）。",
"deployment": {
"runtime_env_example": "config/runtime.env.example",
"runtime_env_active": (CONFIG_DIR / "runtime.env").is_file(),
"docker_compose_default_port": 8101,
"process_dropdown_order": "douyin_youtube → youtube → tiktok → obs → web",
},
}
# ---------------- 配置路由（youtube.ini，仅 youtube） ----------------
@app.get("/get_config")
async def get_config(process: str = Query(..., description="进程名")):
sc = script_for_pm2(process)
if not sc or not is_youtube_script(sc):
return JSONResponse({"error": "仅 youtube*.py 对应进程支持 youtube.ini 编辑"}, status_code=400)
return JSONResponse(
{"error": "仅 YouTube / 抖音→YouTube 类脚本支持 youtube.ini 编辑"}, status_code=400
)
config_file = CONFIG_DIR / "youtube.ini"
if not config_file.exists():
return {"content": "# youtube.ini 文件不存在，将创建空文件\n"}
try:
content = config_file.read_text(encoding="utf-8")
return {"content": content}
except Exception as e:
except OSError as e:
return JSONResponse({"error": f"读取失败: {str(e)}"}, status_code=500)
@app.post("/save_config")
async def save_config(request: Request, process: str = Query(..., description="进程名")):
sc = script_for_pm2(process)
if not sc or not is_youtube_script(sc):
return JSONResponse({"error": "仅 youtube*.py 对应进程支持 youtube.ini 编辑"}, status_code=400)
return JSONResponse(
{"error": "仅 YouTube / 抖音→YouTube 类脚本支持 youtube.ini 编辑"}, status_code=400
)
try:
data = await request.json()
content = data.get("content", "")
except:
except Exception:
return JSONResponse({"error": "无效的 JSON 数据"}, status_code=400)
config_file = CONFIG_DIR / "youtube.ini"
try:
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
config_file.write_text(content, encoding="utf-8")
return {"message": "配置保存成功"}
except Exception as e:
except OSError as e:
return JSONResponse({"error": f"保存失败: {str(e)}"}, status_code=500)
# ---------------- 配置路由（URL_config.ini，youtube 和 tiktok 共享） ----------------
# ---------------- 配置路由（URL_config.ini） ----------------
@app.get("/get_url_config")
async def get_url_config(process: str = Query(..., description="进程名")):
if process not in URL_CONFIG_PM2:
return JSONResponse({"error": "仅 tiktok*.py / youtube*.py 对应进程支持 URL 配置编辑"}, status_code=400)
return JSONResponse(
{"error": "仅 tiktok*.py / youtube*.py / douyin_youtube*.py 对应进程支持 URL 配置编辑"},
status_code=400,
)
config_file = CONFIG_DIR / "URL_config.ini"
if not config_file.exists():
return {"content": "# URL_config.ini 文件不存在，将创建空文件\n"}
try:
content = config_file.read_text(encoding="utf-8")
return {"content": content}
except Exception as e:
except OSError as e:
return JSONResponse({"error": f"读取失败: {str(e)}"}, status_code=500)
@app.post("/save_url_config")
async def save_url_config(request: Request, process: str = Query(..., description="进程名")):
if process not in URL_CONFIG_PM2:
return JSONResponse({"error": "仅 tiktok*.py / youtube*.py 对应进程支持 URL 配置编辑"}, status_code=400)
return JSONResponse(
{
"error": "仅 tiktok*.py / youtube*.py / douyin_youtube*.py 对应进程支持 URL 配置编辑",
},
status_code=400,
)
try:
data = await request.json()
content = data.get("content", "")
except:
except Exception:
return JSONResponse({"error": "无效的 JSON 数据"}, status_code=400)
config_file = CONFIG_DIR / "URL_config.ini"
try:
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
config_file.write_text(content, encoding="utf-8")
return {"message": "配置保存成功"}
except Exception as e:
except OSError as e:
return JSONResponse({"error": f"保存失败: {str(e)}"}, status_code=500)
# ---------------- PM2 控制路由 ----------------
def _not_found_hint(process: str) -> str:
if process_backend.mode == "local":
return (
f"进程「{process}」未在本地注册表中；请点击「启动」由控制台直接拉起脚本，"
"或安装 PM2 后先用 pm2 start 注册同名进程。"
)
return "进程未注册到 PM2（可能被 pm2 delete 删除或从未保存）"
# ---------------- 进程控制 ----------------
@app.get("/start")
async def start(process: str = Query(..., description="进程名")):
if process not in VALID_PROCESSES:
return JSONResponse({"output": f"无效进程名: {process}"}, status_code=400)
output = await run_pm2(f"start {process}")
sc = script_for_pm2(process)
if not sc:
return JSONResponse({"output": "内部错误：找不到脚本映射"}, status_code=500)
output = await process_backend.start(process, sc)
return JSONResponse({"output": output})
@@ -278,31 +320,32 @@ async def stop(process: str = Query(..., description="进程名")):
if process not in VALID_PROCESSES:
return JSONResponse({"output": f"无效进程名: {process}"}, status_code=400)
# 第一步：尝试优雅停止
pm2_output = await run_pm2(f"stop {process}")
pm2_output = await process_backend.stop(process)
# 第二步：等待 PM2 信号传播（通常 1~3 秒）
await asyncio.sleep(2.0)
# 第三步：录制类进程才强杀 ffmpeg（web* 控制台不杀 ffmpeg）
if not process.startswith("web"):
await force_kill_ffmpeg(process)
# 第四步：再等一小会儿，确认
await asyncio.sleep(1.0)
return JSONResponse({
"message": "已执行 stop" + ("" if process.startswith("web") else " + 强杀 ffmpeg"),
"pm2_output": pm2_output,
"note": "如果仍有残留，可多次点击停止或检查 pm2 logs"
})
return JSONResponse(
{
"message": "已执行 stop" + ("" if process.startswith("web") else " + 强杀 ffmpeg"),
"pm2_output": pm2_output,
"note": "如果仍有残留，可多次点击停止或检查日志目录 .pm2/logs",
}
)
@app.get("/restart")
async def restart(process: str = Query(..., description="进程名")):
if process not in VALID_PROCESSES:
return JSONResponse({"output": f"无效进程名: {process}"}, status_code=400)
output = await run_pm2(f"restart {process}")
sc = script_for_pm2(process)
if not sc:
return JSONResponse({"output": "内部错误：找不到脚本映射"}, status_code=500)
output = await process_backend.restart(process, sc)
if not process.startswith("web"):
await force_kill_ffmpeg(process)
return JSONResponse({"output": output})
@@ -311,43 +354,63 @@ async def restart(process: str = Query(..., description="进程名")):
@app.get("/status")
async def status(process: str = Query(..., description="进程名")):
if process not in VALID_PROCESSES:
return JSONResponse({
"raw_status": "",
"process_status": "invalid",
"recent_log": f"无效进程名: {process}",
"recent_error": ""
})
return JSONResponse(
{
"raw_status": "",
"process_status": "invalid",
"recent_log": f"无效进程名: {process}",
"recent_error": "",
}
)
full_status_raw = await run_pm2("status", timeout=8.0)
full_status_raw = await process_backend.full_status_raw()
full_status = strip_ansi_codes(full_status_raw)
info = await get_process_info(process)
recent_log = await read_log_path(info["out_path"], MAX_LOG_LINES)
recent_error = await read_log_path(info["err_path"], MAX_LOG_LINES)
recent_log = await read_log_path(info.get("out_path"), MAX_LOG_LINES)
recent_error = await read_log_path(info.get("err_path"), MAX_LOG_LINES)
if info["process_status"] == "not_found":
recent_log = "进程未注册到 PM2（可能被 pm2 delete 删除或从未保存）\n"
recent_log = _not_found_hint(process) + "\n"
recent_error = ""
return JSONResponse({
"raw_status": full_status,
"process_status": info["process_status"],
"recent_log": recent_log,
"recent_error": recent_error,
})
return JSONResponse(
{
"raw_status": full_status,
"process_status": info["process_status"],
"recent_log": recent_log,
"recent_error": recent_error,
}
)
# ---------------- 进程列表（供前端同步，由 discover_script_entries 自动扫描） ----------------
@app.get("/process_monitor")
async def process_monitor():
await process_backend.ensure_mode()
return {
"entries": [
{"pm2": p["pm2"], "script": p["script"], "label": p["label"]}
for p in PROCESS_MONITOR
]
],
"process_backend": process_backend.mode,
}
# ---------------- 首页 ----------------
@app.get("/favicon.ico", include_in_schema=False)
async def favicon():
ico = CONSOLE_OUT / "favicon.ico"
if ico.is_file():
return FileResponse(ico)
return JSONResponse({}, status_code=404)
@app.get("/")
async def index():
return FileResponse(BASE_DIR / "index.html")
next_index = CONSOLE_OUT / "index.html"
if next_index.is_file():
return FileResponse(next_index)
return FileResponse(BASE_DIR / "index.html")
_next_dir = CONSOLE_OUT / "_next"
if _next_dir.is_dir():
app.mount("/_next", StaticFiles(directory=str(_next_dir)), name="next_static")