feat: 同步本地三端改动并清理上传凭证风险
整合小程序、管理端与后端的最新本地改动，补齐用户管理与首页入口相关能力；提交前已完成敏感信息扫描，并移除本地 gitea 远程 URL 中的明文凭证，避免隐私信息进入远程仓库。 Made-with: Cursor
This commit is contained in:
卡若
183
scripts/deploy_kr_btapi_verify.py
Normal file
183
scripts/deploy_kr_btapi_verify.py
Normal file
@@ -0,0 +1,183 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
kr 宝塔：正式环境发布 + 宝塔面板 API 重启 Go 项目 + 线上冒烟验证。
|
||||
|
||||
说明（与现状一致）：
|
||||
- soul-api 的**文件上传与解压**仍通过 SSH/SFTP（稳定）；**进程重启**使用宝塔「Go 项目」插件 API（与 master.py --restart-method btapi 一致）。
|
||||
- soul-admin 静态资源仍通过 soul-admin/master.py（SSH），宝塔无统一「站点文件 API」封装。
|
||||
- 环境变量与 soul-api/master.py 相同：DEPLOY_HOST、DEPLOY_PASSWORD、BT_PANEL_URL、BT_API_KEY、BT_GO_PROJECT_NAME 等。
|
||||
|
||||
用法：
|
||||
python3 scripts/deploy_kr_btapi_verify.py # 部署 API+管理端 + 验证
|
||||
python3 scripts/deploy_kr_btapi_verify.py --verify-only # 仅公网冒烟（不部署）
|
||||
python3 scripts/deploy_kr_btapi_verify.py --skip-bt-ping # 跳过宝塔 API 探活
|
||||
"""
|
||||
|
||||
from __future__ import print_function
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
||||
SOUL_API = os.path.join(ROOT, "soul-api")
|
||||
SOUL_ADMIN = os.path.join(ROOT, "soul-admin")
|
||||
|
||||
|
||||
def _load_api_cfg():
|
||||
import importlib.util
|
||||
|
||||
path = os.path.join(SOUL_API, "master.py")
|
||||
spec = importlib.util.spec_from_file_location("soul_api_master_mod", path)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod.get_cfg()
|
||||
|
||||
|
||||
def bt_panel_ping(cfg):
|
||||
"""调用宝塔 /system?action=GetDiskInfo 校验 API 密钥与面板可达性。"""
|
||||
try:
|
||||
import requests
|
||||
|
||||
try:
|
||||
import urllib3
|
||||
|
||||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||
except Exception:
|
||||
pass
|
||||
except ImportError:
|
||||
print("[失败] 需要 requests：pip install requests")
|
||||
return False
|
||||
|
||||
url = (cfg.get("bt_panel_url") or "").rstrip("/")
|
||||
key = cfg.get("bt_api_key") or ""
|
||||
if not url or not key:
|
||||
print("[失败] 缺少 BT_PANEL_URL 或 BT_API_KEY")
|
||||
return False
|
||||
req_time = int(time.time())
|
||||
sk_md5 = hashlib.md5(key.encode()).hexdigest()
|
||||
req_token = hashlib.md5(("%s%s" % (req_time, sk_md5)).encode()).hexdigest()
|
||||
try:
|
||||
r = requests.post(
|
||||
url + "/system?action=GetDiskInfo",
|
||||
data={"request_time": req_time, "request_token": req_token},
|
||||
timeout=20,
|
||||
verify=False,
|
||||
)
|
||||
if r.status_code != 200:
|
||||
print("[宝塔API 探活] HTTP %s" % r.status_code)
|
||||
return False
|
||||
j = r.json() if "application/json" in (r.headers.get("content-type") or "") else {}
|
||||
# 部分面板返回 data 为列表或非标准结构，仅当明确 status=False 视为失败
|
||||
if isinstance(j, dict) and j.get("status") is False:
|
||||
print("[宝塔API 探活] %s" % (j.get("msg") or j))
|
||||
return False
|
||||
print("[宝塔API 探活] 成功（GetDiskInfo）", flush=True)
|
||||
return True
|
||||
except Exception as e:
|
||||
print("[宝塔API 探活] 异常: %s" % e)
|
||||
return False
|
||||
|
||||
|
||||
def run_cmd(cwd, args, env=None):
|
||||
print("", flush=True)
|
||||
print(">>> %s" % " ".join(args), flush=True)
|
||||
e = os.environ.copy()
|
||||
if env:
|
||||
e.update(env)
|
||||
r = subprocess.run(args, cwd=cwd, env=e)
|
||||
return r.returncode == 0
|
||||
|
||||
|
||||
def verify_public(base_api=None, base_admin=None):
|
||||
"""公网冒烟：健康检查 + 只读业务接口 + 管理端首页。"""
|
||||
try:
|
||||
import requests
|
||||
|
||||
try:
|
||||
import urllib3
|
||||
|
||||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||
except Exception:
|
||||
pass
|
||||
except ImportError:
|
||||
print("[失败] 验证阶段需要 requests")
|
||||
return False
|
||||
|
||||
base_api = (base_api or os.environ.get("VERIFY_API_BASE", "https://soulapi.quwanzhi.com")).rstrip("/")
|
||||
base_admin = (base_admin or os.environ.get("VERIFY_ADMIN_BASE", "https://souladmin.quwanzhi.com")).rstrip("/")
|
||||
|
||||
checks = [
|
||||
("GET %s/health" % base_api, "%s/health" % base_api, lambda r: r.status_code == 200 and '"status"' in r.text),
|
||||
("GET %s/api/book/parts" % base_api, "%s/api/book/parts" % base_api, lambda r: r.status_code == 200),
|
||||
("GET %s/api/config" % base_api, "%s/api/config" % base_api, lambda r: r.status_code == 200),
|
||||
("GET %s/" % base_admin, "%s/" % base_admin, lambda r: r.status_code == 200 and "root" in r.text),
|
||||
]
|
||||
|
||||
print("")
|
||||
print("=" * 60)
|
||||
print(" 线上冒烟验证")
|
||||
print("=" * 60)
|
||||
all_ok = True
|
||||
for name, url, pred in checks:
|
||||
try:
|
||||
r = requests.get(url, timeout=20, verify=True)
|
||||
ok = pred(r)
|
||||
print(" [%s] %s" % ("OK" if ok else "FAIL", name))
|
||||
if not ok:
|
||||
print(" status=%s len=%s" % (r.status_code, len(r.text or "")))
|
||||
all_ok = False
|
||||
except Exception as ex:
|
||||
print(" [FAIL] %s — %s" % (name, ex))
|
||||
all_ok = False
|
||||
return all_ok
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="kr 宝塔：BT API 重启 + 部署 + 验证")
|
||||
parser.add_argument("--verify-only", action="store_true", help="仅执行公网冒烟，不部署")
|
||||
parser.add_argument("--skip-bt-ping", action="store_true", help="跳过宝塔面板 API 探活")
|
||||
parser.add_argument(
|
||||
"--restart-method",
|
||||
choices=("btapi", "auto"),
|
||||
default="auto",
|
||||
help="soul-api master.py 的 --restart-method（默认先宝塔 Go 插件 API，失败再 SSH）",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
cfg = _load_api_cfg()
|
||||
|
||||
if not args.verify_only and not args.skip_bt_ping:
|
||||
if not bt_panel_ping(cfg):
|
||||
print("[失败] 宝塔 API 探活未通过，已中止（可用 --skip-bt-ping 跳过）")
|
||||
return 1
|
||||
|
||||
if args.verify_only:
|
||||
return 0 if verify_public() else 1
|
||||
|
||||
py = sys.executable
|
||||
if not run_cmd(SOUL_API, [py, "master.py", "--restart-method", args.restart_method]):
|
||||
print("[失败] soul-api 部署失败")
|
||||
return 1
|
||||
|
||||
if not run_cmd(SOUL_ADMIN, [py, "master.py"]):
|
||||
print("[失败] soul-admin 部署失败")
|
||||
return 1
|
||||
|
||||
time.sleep(2)
|
||||
if not verify_public():
|
||||
print("")
|
||||
print("[失败] 线上冒烟存在未通过项（部署已执行，请排查 Nginx/证书/业务）")
|
||||
return 1
|
||||
|
||||
print("")
|
||||
print(" 全部完成：已部署且冒烟通过。")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main() or 0)
|
||||
157
scripts/download_chapter_images_from_md.py
Normal file
157
scripts/download_chapter_images_from_md.py
Normal file
@@ -0,0 +1,157 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
从书稿 Markdown 中抓取远程图片（ 与 <img src="...">），下载到本地 images/，
|
||||
并把正文中的 URL 替换为相对路径 images/文件名。
|
||||
|
||||
用法：
|
||||
python3 scripts/download_chapter_images_from_md.py /path/to/第140场.md
|
||||
python3 scripts/download_chapter_images_from_md.py --images-dir /path/to/images a.md b.md
|
||||
python3 scripts/download_chapter_images_from_md.py --dry-run a.md
|
||||
|
||||
依赖：requests（与 content_download.py 一致）
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from urllib.parse import unquote, urlparse
|
||||
|
||||
try:
|
||||
import requests
|
||||
except ImportError:
|
||||
print("请安装: pip install requests", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
MD_IMG = re.compile(r"!\[[^\]]*\]\((https?://[^)\s]+)\)")
|
||||
HTML_IMG = re.compile(r'<img[^>]+src=["\'](https?://[^"\']+)["\']', re.I)
|
||||
|
||||
|
||||
def collect_urls(text: str) -> list[str]:
|
||||
seen: set[str] = set()
|
||||
out: list[str] = []
|
||||
for m in MD_IMG.finditer(text):
|
||||
u = m.group(1).strip()
|
||||
if u not in seen:
|
||||
seen.add(u)
|
||||
out.append(u)
|
||||
for m in HTML_IMG.finditer(text):
|
||||
u = m.group(1).strip()
|
||||
if u not in seen:
|
||||
seen.add(u)
|
||||
out.append(u)
|
||||
return out
|
||||
|
||||
|
||||
def safe_filename(url: str, index: int) -> str:
|
||||
path = unquote(urlparse(url).path)
|
||||
base = Path(path).name
|
||||
if not base or base == "/" or ".." in base:
|
||||
base = f"image_{index:02d}.bin"
|
||||
base = re.sub(r"[^\w.\-一-龥]", "_", base)
|
||||
return base
|
||||
|
||||
|
||||
def download_one(url: str, dest: Path, session: requests.Session) -> None:
|
||||
r = session.get(url, timeout=60, headers={"User-Agent": "Mozilla/5.0"})
|
||||
r.raise_for_status()
|
||||
dest.write_bytes(r.content)
|
||||
|
||||
|
||||
def process_file(
|
||||
md_path: Path,
|
||||
images_dir: Path,
|
||||
dry_run: bool,
|
||||
session: requests.Session,
|
||||
) -> tuple[int, int]:
|
||||
text = md_path.read_text(encoding="utf-8")
|
||||
urls = collect_urls(text)
|
||||
if not urls:
|
||||
print(f"{md_path}: 无远程图片")
|
||||
return 0, 0
|
||||
|
||||
used_names: dict[str, str] = {} # url -> local filename
|
||||
basename_count: dict[str, int] = {}
|
||||
ok = fail = 0
|
||||
|
||||
for i, url in enumerate(urls, 1):
|
||||
name = safe_filename(url, i)
|
||||
stem = Path(name).stem
|
||||
ext = Path(name).suffix
|
||||
if name in basename_count:
|
||||
basename_count[name] += 1
|
||||
name = f"{stem}_{basename_count[name]}{ext}"
|
||||
else:
|
||||
basename_count[name] = 0
|
||||
|
||||
local_rel = f"images/{name}"
|
||||
used_names[url] = name
|
||||
|
||||
dest = images_dir / name
|
||||
print(f" GET {url} -> {local_rel}")
|
||||
if dry_run:
|
||||
ok += 1
|
||||
continue
|
||||
images_dir.mkdir(parents=True, exist_ok=True)
|
||||
try:
|
||||
download_one(url, dest, session)
|
||||
ok += 1
|
||||
except Exception as e:
|
||||
print(f" FAIL {url}: {e}", file=sys.stderr)
|
||||
fail += 1
|
||||
|
||||
if dry_run or fail:
|
||||
return ok, fail
|
||||
|
||||
new_text = text
|
||||
for url, name in used_names.items():
|
||||
local_rel = f"images/{name}"
|
||||
new_text = new_text.replace(f"]({url})", f"]({local_rel})")
|
||||
new_text = new_text.replace(f'src="{url}"', f'src="{local_rel}"')
|
||||
new_text = new_text.replace(f"src='{url}'", f"src='{local_rel}'")
|
||||
|
||||
md_path.write_text(new_text, encoding="utf-8")
|
||||
print(f" 已写回: {md_path}")
|
||||
return ok, fail
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser(description="从 md 下载远程插图并改为相对路径")
|
||||
ap.add_argument("md", nargs="+", type=Path, help="Markdown 文件路径")
|
||||
ap.add_argument(
|
||||
"--images-dir",
|
||||
type=Path,
|
||||
default=None,
|
||||
help="图片目录，默认与第一个 md 同级的 images/",
|
||||
)
|
||||
ap.add_argument("--dry-run", action="store_true", help="只打印不下载、不改文件")
|
||||
args = ap.parse_args()
|
||||
|
||||
first = args.md[0].resolve()
|
||||
images_dir = (
|
||||
args.images_dir.resolve()
|
||||
if args.images_dir
|
||||
else first.parent / "images"
|
||||
)
|
||||
|
||||
session = requests.Session()
|
||||
total_ok = total_fail = 0
|
||||
for p in args.md:
|
||||
p = p.resolve()
|
||||
if not p.is_file():
|
||||
print(f"跳过（非文件）: {p}", file=sys.stderr)
|
||||
continue
|
||||
print(f"=== {p.name}")
|
||||
ok, fail = process_file(p, images_dir, args.dry_run, session)
|
||||
total_ok += ok
|
||||
total_fail += fail
|
||||
|
||||
print(f"完成: 成功 {total_ok}, 失败 {total_fail}")
|
||||
if total_fail:
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
89
scripts/download_mbti_avatars_16p_colored.py
Normal file
89
scripts/download_mbti_avatars_16p_colored.py
Normal file
@@ -0,0 +1,89 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
下载 16Personalities 官网静态彩色小人 SVG（国内最流行的 MBTI 视觉）。
|
||||
|
||||
与 DiceBear 不同，这是站内「紫人/绿人」梗图同源画风。
|
||||
|
||||
⚠️ 版权：素材版权归 NERIS Analytics Limited / 16personalities.com。
|
||||
商用、再分发、放入小程序包体前请自行阅读其条款并取得授权；本脚本仅作本地备份与联调用途，风险自负。
|
||||
|
||||
用法:
|
||||
python3 scripts/download_mbti_avatars_16p_colored.py
|
||||
python3 scripts/download_mbti_avatars_16p_colored.py --gender male
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_QUERY = "?v=3"
|
||||
|
||||
# 与官网路径一致：{type}-{role}-{gender}.svg
|
||||
SIXTEEN: list[tuple[str, str]] = [
|
||||
("INTJ", "architect"),
|
||||
("INTP", "logician"),
|
||||
("ENTJ", "commander"),
|
||||
("ENTP", "debater"),
|
||||
("INFJ", "advocate"),
|
||||
("INFP", "mediator"),
|
||||
("ENFJ", "protagonist"),
|
||||
("ENFP", "campaigner"),
|
||||
("ISTJ", "logistician"),
|
||||
("ISFJ", "defender"),
|
||||
("ESTJ", "executive"),
|
||||
("ESFJ", "consul"),
|
||||
("ISTP", "virtuoso"),
|
||||
("ISFP", "adventurer"),
|
||||
("ESTP", "entrepreneur"),
|
||||
("ESFP", "entertainer"),
|
||||
]
|
||||
|
||||
BASE = "https://www.16personalities.com/static/images/personality-types/avatars"
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument(
|
||||
"--gender",
|
||||
choices=("female", "male"),
|
||||
default="female",
|
||||
help="官网提供 female / male 两版，默认 female（国内梗多对应女版小人）",
|
||||
)
|
||||
args = ap.parse_args()
|
||||
gender = args.gender
|
||||
out = ROOT / "static" / f"mbti-avatars-16personalities-{gender}"
|
||||
out.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
readme = out / "README.txt"
|
||||
warn = """16Personalities 彩色小人 SVG（官方静态站）
|
||||
|
||||
版权归属：16personalities.com / NERIS Analytics Limited。
|
||||
请勿在未取得授权的情况下用于商业产品对外分发；上架前请改用自有素材或书面许可。
|
||||
|
||||
文件名：{TYPE}.svg 便于与后台 MBTI key 对应；内容来自官网角色英文名路径。
|
||||
"""
|
||||
readme.write_text(warn, encoding="utf-8")
|
||||
|
||||
headers = {
|
||||
"User-Agent": "SoulProjectMbtiAssetMirror/1.0 (+local dev backup)",
|
||||
"Accept": "image/svg+xml,*/*",
|
||||
}
|
||||
|
||||
for code, role in SIXTEEN:
|
||||
low = code.lower()
|
||||
url = f"{BASE}/{low}-{role}-{gender}.svg{DEFAULT_QUERY}"
|
||||
dest = out / f"{code}.svg"
|
||||
req = urllib.request.Request(url, headers=headers)
|
||||
with urllib.request.urlopen(req, timeout=60) as resp:
|
||||
data = resp.read()
|
||||
dest.write_bytes(data)
|
||||
print(f"ok {dest.name} ({len(data)} bytes)")
|
||||
|
||||
print(f"done -> {out}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
59
scripts/download_mbti_avatars_dicebear.py
Normal file
59
scripts/download_mbti_avatars_dicebear.py
Normal file
@@ -0,0 +1,59 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
下载 16 型 MBTI 占位头像（DiceBear notionists，见 https://www.dicebear.com/licenses/ ）。
|
||||
|
||||
说明：网传「药水姐 / 紫大姐 / 大宝剑」等对应的 3D 小人多为 16personalities 或同人版权素材，
|
||||
未获授权不宜批量爬取用于商用小程序；本脚本仅拉取开源生成器产物作占位，可后续在管理端替换为自有素材 URL。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
OUT = ROOT / "static" / "mbti-avatars-dicebear"
|
||||
|
||||
# 与社区常见梗名对应，便于文件夹内辨认（文件名仍用四字母类型）
|
||||
MBTI_ORDER = [
|
||||
("INTJ", "紫老头"),
|
||||
("INTP", "药水姐"),
|
||||
("ENTJ", "大姐头"),
|
||||
("ENTP", "骨折眉毛"),
|
||||
("INFJ", "绿老头"),
|
||||
("INFP", "小蝴蝶"),
|
||||
("ENFJ", "大宝剑"),
|
||||
("ENFP", "快乐小狗"),
|
||||
("ISTJ", "蓝老头"),
|
||||
("ISFJ", "小护士"),
|
||||
("ESTJ", "尺子姐"),
|
||||
("ESFJ", "雨伞哥"),
|
||||
("ISTP", "电钻哥"),
|
||||
("ISFP", "小画家"),
|
||||
("ESTP", "墨镜哥"),
|
||||
("ESFP", "沙锤姐"),
|
||||
]
|
||||
|
||||
BASE = "https://api.dicebear.com/7.x/notionists/svg"
|
||||
|
||||
|
||||
def main() -> None:
|
||||
OUT.mkdir(parents=True, exist_ok=True)
|
||||
readme = OUT / "README.txt"
|
||||
lines = [
|
||||
"DiceBear 7.x notionists — 占位头像，可商用（以 dicebear.com 许可证为准）。",
|
||||
"替换为「药水姐」等 3D 风素材时请使用自有版权或已授权图床 URL，在管理端 MBTI 头像库粘贴保存。",
|
||||
"",
|
||||
]
|
||||
for code, nick in MBTI_ORDER:
|
||||
url = f"{BASE}?seed={code}"
|
||||
dest = OUT / f"{code}.svg"
|
||||
urllib.request.urlretrieve(url, dest)
|
||||
lines.append(f"{code}.svg — {nick}")
|
||||
print(f"ok {dest.name}")
|
||||
readme.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
print(f"done -> {OUT}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
124
scripts/fix_2026_daily_part.py
Normal file
124
scripts/fix_2026_daily_part.py
Normal file
@@ -0,0 +1,124 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
将误挂在「第四篇 / 第9章」等下的 2026 派对场次，归位到「2026每日派对干货」篇章。
|
||||
|
||||
规则（与 content_upload.py 一致）：
|
||||
- id 为 10.xx 的节：只修正 part_id / chapter_id / part_title / chapter_title，不改 id。
|
||||
- section_title 含「第102场」及以后、且不在 part-2026-daily：同上修正（含 id 非 10.xx 的遗留）。
|
||||
|
||||
用法:
|
||||
python3 scripts/fix_2026_daily_part.py # 预览
|
||||
python3 scripts/fix_2026_daily_part.py --execute # 执行 UPDATE
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import importlib.util
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
def load_db_config() -> dict:
|
||||
mig = ROOT / "scripts" / "migrate_2026_sections.py"
|
||||
spec = importlib.util.spec_from_file_location("_mig", mig)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
assert spec.loader is not None
|
||||
spec.loader.exec_module(mod)
|
||||
return mod.DB_CONFIG
|
||||
|
||||
|
||||
PART_2026 = "part-2026-daily"
|
||||
CHAPTER_2026 = "chapter-2026-daily"
|
||||
TITLE_2026 = "2026每日派对干货"
|
||||
|
||||
|
||||
def session_num(title: str) -> int | None:
|
||||
m = re.search(r"第(\d+)场", title or "")
|
||||
return int(m.group(1)) if m else None
|
||||
|
||||
|
||||
def main():
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument("--execute", action="store_true")
|
||||
args = p.parse_args()
|
||||
|
||||
try:
|
||||
import pymysql
|
||||
except ImportError:
|
||||
print("需要: pip install pymysql", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
cfg = load_db_config()
|
||||
conn = pymysql.connect(**cfg)
|
||||
cur = conn.cursor()
|
||||
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT id, section_title, part_id, chapter_id, part_title, chapter_title, sort_order
|
||||
FROM chapters
|
||||
ORDER BY sort_order, id
|
||||
"""
|
||||
)
|
||||
rows = cur.fetchall()
|
||||
to_fix: list[tuple] = []
|
||||
|
||||
def is_2026_daily_row(sid: str, title: str) -> bool:
|
||||
"""与上传脚本一致：10.01～10.99（两位小数段）、2026.1；不含 10.1/10.2 单段 id。"""
|
||||
s = str(sid)
|
||||
if s == "2026.1":
|
||||
return True
|
||||
if re.match(r"^10\.\d{2}$", s):
|
||||
return True
|
||||
n = session_num(title or "")
|
||||
if n is not None and n >= 102:
|
||||
return True
|
||||
return False
|
||||
|
||||
for r in rows:
|
||||
sid, title, pid, cid, ptitle, ctitle, so = r
|
||||
wrong_part = pid != PART_2026 or cid != CHAPTER_2026
|
||||
if wrong_part and is_2026_daily_row(sid, title or ""):
|
||||
to_fix.append(r)
|
||||
|
||||
if not to_fix:
|
||||
print("没有需要归位的节（10.xx 或第102场及以后且已在 part-2026-daily）。")
|
||||
conn.close()
|
||||
return
|
||||
|
||||
print(f"待归位到「{TITLE_2026}」: {len(to_fix)} 节\n")
|
||||
for r in to_fix:
|
||||
sid, title, pid, cid, _, _, _ = r
|
||||
print(f" {sid}\tpart={pid}\tch={cid}\t{title[:60] if title else ''}")
|
||||
|
||||
if not args.execute:
|
||||
print("\n[预览] 未写入。确认无误后加 --execute")
|
||||
conn.close()
|
||||
return
|
||||
|
||||
n = 0
|
||||
for r in to_fix:
|
||||
sid = r[0]
|
||||
cur.execute(
|
||||
"""
|
||||
UPDATE chapters SET
|
||||
part_id = %s,
|
||||
part_title = %s,
|
||||
chapter_id = %s,
|
||||
chapter_title = %s,
|
||||
updated_at = NOW()
|
||||
WHERE id = %s
|
||||
""",
|
||||
(PART_2026, TITLE_2026, CHAPTER_2026, TITLE_2026, sid),
|
||||
)
|
||||
n += cur.rowcount
|
||||
conn.commit()
|
||||
conn.close()
|
||||
print(f"\n已更新 {n} 行，part/chapter 已归位 {PART_2026} / {CHAPTER_2026}。")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user