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-29 01:34:23 -05:00
parent 8bce529255
commit 784a4468b8
8 changed files with 308 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
web.py
View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
import re
import sys
from contextlib import asynccontextmanager
from pathlib import Path
@@ -159,13 +160,57 @@ def _build_process_monitor():
PROCESS_MONITOR = _build_process_monitor()
VALID_PROCESSES = frozenset(p["pm2"] for p in PROCESS_MONITOR)
URL_CONFIG_PM2 = frozenset(
pm2_name_from_script(e["script"])
for e in SCRIPT_ENTRIES
if is_tiktok_script(e["script"]) or is_youtube_script(e["script"])
)
def _pm2_safe_segment(pm2: str) -> str:
return re.sub(r"[^a-zA-Z0-9_.-]", "_", pm2)
def _pro_script_head(pm2: str) -> Optional[str]:
if "__" not in pm2:
return None
head = pm2.split("__", 1)[0].strip()
return head.lower() if head else None
def script_for_pm2_or_pro(pm2: str) -> Optional[str]:
s = script_for_pm2(pm2)
if s:
return s
head_l = _pro_script_head(pm2)
if not head_l:
return None
for p in PROCESS_MONITOR:
if Path(p["script"]).stem.lower() == head_l:
return p["script"]
return None
def is_valid_control_process(pm2: str) -> bool:
return script_for_pm2_or_pro(pm2) is not None
def allows_url_config(pm2: str) -> bool:
sc = script_for_pm2_or_pro(pm2)
return sc is not None and (is_youtube_script(sc) or is_tiktok_script(sc))
def managed_url_config_relpath(pm2: str) -> str:
sc = script_for_pm2_or_pro(pm2)
if not sc:
return "URL_config.ini"
if "__" in pm2 and (is_youtube_script(sc) or is_tiktok_script(sc)):
return f"URL_config.{_pm2_safe_segment(pm2)}.ini"
return "URL_config.ini"
def managed_youtube_ini_relpath(pm2: str) -> Optional[str]:
sc = script_for_pm2_or_pro(pm2)
if not sc or not is_youtube_script(sc):
return None
if "__" in pm2:
return f"youtube.{_pm2_safe_segment(pm2)}.ini"
return "youtube.ini"
def script_for_pm2(pm2: str) -> Optional[str]:
@@ -502,15 +547,15 @@ async def get_system_snapshot():
# ---------------- 配置路由（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):
rel = managed_youtube_ini_relpath(process)
if not rel:
return JSONResponse(
{"error": "仅 YouTube / 抖音→YouTube 类脚本支持 youtube.ini 编辑"}, status_code=400
)
config_file = CONFIG_DIR / "youtube.ini"
config_file = CONFIG_DIR / rel
if not config_file.exists():
return {"content": "# youtube.ini 文件不存在，将创建空文件\n"}
return {"content": f"# {rel} 尚未创建，保存后将写入磁盘\n"}
try:
content = config_file.read_text(encoding="utf-8")
return {"content": content}
@@ -520,8 +565,8 @@ async def get_config(process: str = Query(..., description="进程名")):
@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):
rel = managed_youtube_ini_relpath(process)
if not rel:
return JSONResponse(
{"error": "仅 YouTube / 抖音→YouTube 类脚本支持 youtube.ini 编辑"}, status_code=400
)
@@ -533,7 +578,7 @@ async def save_config(request: Request, process: str = Query(..., description="
return JSONResponse({"error": "无效的 JSON 数据"}, status_code=400)
try:
governed_save_managed_file("app-config", "youtube.ini", content)
governed_save_managed_file("app-config", rel, content)
return {"message": "配置保存成功"}
except OSError as e:
return JSONResponse({"error": f"保存失败: {str(e)}"}, status_code=500)
@@ -542,15 +587,16 @@ async def save_config(request: Request, process: str = Query(..., description="
# ---------------- 配置路由（URL_config.ini） ----------------
@app.get("/get_url_config")
async def get_url_config(process: str = Query(..., description="进程名")):
if process not in URL_CONFIG_PM2:
if not allows_url_config(process):
return JSONResponse(
{"error": "仅 tiktok*.py / youtube*.py / douyin_youtube*.py 对应进程支持 URL 配置编辑"},
status_code=400,
)
config_file = CONFIG_DIR / "URL_config.ini"
rel = managed_url_config_relpath(process)
config_file = CONFIG_DIR / rel
if not config_file.exists():
return {"content": "# URL_config.ini 文件不存在，将创建空文件\n"}
return {"content": f"# {rel} 尚未创建，保存后将写入磁盘\n"}
try:
content = config_file.read_text(encoding="utf-8")
return {"content": content}
@@ -560,7 +606,7 @@ async def get_url_config(process: str = Query(..., description="进程名")):
@app.post("/save_url_config")
async def save_url_config(request: Request, process: str = Query(..., description="进程名")):
if process not in URL_CONFIG_PM2:
if not allows_url_config(process):
return JSONResponse(
{
"error": "仅 tiktok*.py / youtube*.py / douyin_youtube*.py 对应进程支持 URL 配置编辑",
@@ -575,7 +621,8 @@ async def save_url_config(request: Request, process: str = Query(..., descriptio
return JSONResponse({"error": "无效的 JSON 数据"}, status_code=400)
try:
governed_save_managed_file("app-config", "URL_config.ini", content)
rel = managed_url_config_relpath(process)
governed_save_managed_file("app-config", rel, content)
return {"message": "配置保存成功"}
except OSError as e:
return JSONResponse({"error": f"保存失败: {str(e)}"}, status_code=500)
@@ -593,9 +640,9 @@ def _not_found_hint(process: str) -> str:
# ---------------- 进程控制 ----------------
@app.get("/start")
async def start(process: str = Query(..., description="进程名")):
if process not in VALID_PROCESSES:
if not is_valid_control_process(process):
return JSONResponse({"output": f"无效进程名: {process}"}, status_code=400)
sc = script_for_pm2(process)
sc = script_for_pm2_or_pro(process)
if not sc:
return JSONResponse({"output": "内部错误：找不到脚本映射"}, status_code=500)
output = await process_backend.start(process, sc)
@@ -609,7 +656,7 @@ async def start_post(body: ProcessNameBody):
@app.get("/stop")
async def stop(process: str = Query(..., description="进程名")):
if process not in VALID_PROCESSES:
if not is_valid_control_process(process):
return JSONResponse({"output": f"无效进程名: {process}"}, status_code=400)
pm2_output = await process_backend.stop(process)
@@ -637,9 +684,9 @@ async def stop_post(body: ProcessNameBody):
@app.get("/restart")
async def restart(process: str = Query(..., description="进程名")):
if process not in VALID_PROCESSES:
if not is_valid_control_process(process):
return JSONResponse({"output": f"无效进程名: {process}"}, status_code=400)
sc = script_for_pm2(process)
sc = script_for_pm2_or_pro(process)
if not sc:
return JSONResponse({"output": "内部错误：找不到脚本映射"}, status_code=500)
output = await process_backend.restart(process, sc)
@@ -655,7 +702,7 @@ async def restart_post(body: ProcessNameBody):
@app.get("/status")
async def status(process: str = Query(..., description="进程名")):
if process not in VALID_PROCESSES:
if not is_valid_control_process(process):
return JSONResponse(
{
"raw_status": "",