init: 从 /root/zikai 根目录迁入
把原本散落在 /root/zikai 的 FastAPI 服务整理到 server/ 子目录, 开启独立 git 与 venv。运行时数据(config.yaml / keys / uploads) 按 .gitignore 留在工作目录但不入仓。
This commit is contained in:
1
app/scripts/__init__.py
Normal file
1
app/scripts/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""一次性脚本。"""
|
||||
85
app/scripts/init_db.py
Normal file
85
app/scripts/init_db.py
Normal file
@@ -0,0 +1,85 @@
|
||||
"""一次性建库脚本。
|
||||
|
||||
通过本机 socket 以 root 调用 mysql:
|
||||
1. 创建独立数据库;
|
||||
2. 创建独立账户,生成随机密码并授权;
|
||||
3. 把新密码写回 config.yaml;
|
||||
4. 导入 sql/schema.sql。
|
||||
|
||||
幂等。设置环境变量 KEEP_DB_PASSWORD=1 可复用 config.yaml 中的现有密码。
|
||||
|
||||
python -m app.scripts.init_db
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import secrets
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import yaml
|
||||
|
||||
from ..config import PROJECT_ROOT, reload_settings
|
||||
|
||||
SCHEMA_PATH = PROJECT_ROOT / "sql" / "schema.sql"
|
||||
CFG_PATH = PROJECT_ROOT / "config.yaml"
|
||||
|
||||
|
||||
def _mysql_root(args: list[str], input_str: str | None = None) -> str:
|
||||
cmd = ["sudo", "-n", "mysql", "--protocol=socket", "-uroot", *args]
|
||||
proc = subprocess.run(cmd, input=input_str, text=True, capture_output=True)
|
||||
if proc.returncode != 0:
|
||||
sys.stderr.write(proc.stderr)
|
||||
raise SystemExit(f"mysql 执行失败 (exit {proc.returncode})")
|
||||
return proc.stdout
|
||||
|
||||
|
||||
def _save_password(db: str, user: str, password: str) -> None:
|
||||
if not CFG_PATH.exists():
|
||||
raise SystemExit("config.yaml 不存在;请先从 config.example.yaml 复制。")
|
||||
data = yaml.safe_load(CFG_PATH.read_text(encoding="utf-8")) or {}
|
||||
data.setdefault("database", {})
|
||||
data["database"].update(database=db, user=user, password=password)
|
||||
CFG_PATH.write_text(
|
||||
yaml.safe_dump(data, sort_keys=False, default_flow_style=False),
|
||||
encoding="utf-8",
|
||||
)
|
||||
print(f"[init_db] 凭据已写入 {CFG_PATH}")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
settings = reload_settings()
|
||||
db = settings.database.database
|
||||
user = settings.database.user
|
||||
|
||||
keep = os.getenv("KEEP_DB_PASSWORD") == "1"
|
||||
if keep and settings.database.password and settings.database.password != "CHANGE_ME":
|
||||
password = settings.database.password
|
||||
print(f"[init_db] KEEP_DB_PASSWORD=1,沿用 '{user}' 的现有密码")
|
||||
else:
|
||||
password = secrets.token_urlsafe(24)
|
||||
print(f"[init_db] 为 '{user}'@'localhost' 生成新密码")
|
||||
|
||||
sql = (
|
||||
f"CREATE DATABASE IF NOT EXISTS `{db}` "
|
||||
f" DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;"
|
||||
f"CREATE USER IF NOT EXISTS '{user}'@'localhost' IDENTIFIED BY '{password}';"
|
||||
f"ALTER USER '{user}'@'localhost' IDENTIFIED BY '{password}';"
|
||||
f"GRANT ALL PRIVILEGES ON `{db}`.* TO '{user}'@'localhost';"
|
||||
f"FLUSH PRIVILEGES;"
|
||||
)
|
||||
_mysql_root(["-e", sql])
|
||||
print(f"[init_db] 数据库 '{db}' 与账户 '{user}'@'localhost' 已就绪")
|
||||
|
||||
_mysql_root([db], input_str=SCHEMA_PATH.read_text(encoding="utf-8"))
|
||||
print(f"[init_db] 已导入 {SCHEMA_PATH}")
|
||||
|
||||
if not keep:
|
||||
_save_password(db, user, password)
|
||||
|
||||
print("[init_db] 完成。")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user