80b96d236f
把原本散落在 /root/zikai 的 FastAPI 服务整理到 server/ 子目录， 开启独立 git 与 venv。运行时数据（config.yaml / keys / uploads） 按 .gitignore 留在工作目录但不入仓。
86 lines
2.8 KiB
Python
86 lines
2.8 KiB
Python
"""一次性建库脚本。
|
||
|
||
通过本机 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()
|