zhanghongyu_0626
c9cac1cdf9
- 在 `risk_suitability_log` 表中新增 `check_source` 枚举值 `platform`，支持代销平台的适当性检查。 - 更新相关 SQL 脚本以适应新的数据结构，确保数据一致性。 - 修改文档以反映 API 的最新状态和测试基线，确保文档与实现保持一致。 - 测试基线更新至 530 passed, 0 skipped，确保系统稳定性。 此更新为代销平台提供了更全面的适当性检查能力，提升了系统的功能性与可维护性。
52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
"""Execute a UTF-8 SQL file via mysql CLI (avoids Windows pipe encoding loss)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def _load_env_password() -> tuple[str, str, str]:
|
|
host, user, password = "127.0.0.1", "root", ""
|
|
env_path = ROOT / ".env"
|
|
if env_path.exists():
|
|
for line in env_path.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if line.startswith("MYSQL_HOST="):
|
|
host = line.split("=", 1)[1].strip()
|
|
elif line.startswith("MYSQL_USER="):
|
|
user = line.split("=", 1)[1].strip()
|
|
elif line.startswith("MYSQL_PASSWORD="):
|
|
password = line.split("=", 1)[1].strip()
|
|
return host, user, password
|
|
|
|
|
|
def main() -> None:
|
|
if len(sys.argv) != 2:
|
|
raise SystemExit("usage: run_sql_file.py <path.sql>")
|
|
path = Path(sys.argv[1]).resolve()
|
|
if not path.is_file():
|
|
raise SystemExit(f"file not found: {path}")
|
|
host, user, password = _load_env_password()
|
|
if not password:
|
|
raise SystemExit("MYSQL_PASSWORD missing in .env")
|
|
sql = path.read_text(encoding="utf-8")
|
|
proc = subprocess.run(
|
|
["mysql", "-h", host, "-u", user, f"-p{password}", "--default-character-set=utf8mb4"],
|
|
input=sql.encode("utf-8"),
|
|
capture_output=True,
|
|
)
|
|
if proc.stdout:
|
|
sys.stdout.buffer.write(proc.stdout)
|
|
if proc.returncode != 0:
|
|
if proc.stderr:
|
|
sys.stderr.buffer.write(proc.stderr)
|
|
raise SystemExit(proc.returncode)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|