85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
创建项目所需 MySQL 数据库（liaotian_db）。
|
||
从环境变量 DATABASE_URL 或下方默认值读取连接信息，连接后执行 CREATE DATABASE。
|
||
用法:
|
||
cd backend && python scripts/create_database.py
|
||
或指定: export DATABASE_URL='mysql+pymysql://user:pass@host:port/liaotian_db?charset=utf8mb4'
|
||
python scripts/create_database.py
|
||
"""
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
# 与 app/config.py 中默认一致，不导入 app 以免依赖 Flask 等
|
||
DEFAULT_DATABASE_URL = (
|
||
"mysql+pymysql://root:!Rjb12191@"
|
||
"gz-cynosdbmysql-grp-d26pzce5.sql.tencentcdb.com:24936/"
|
||
"liaotian_db?charset=utf8mb4"
|
||
)
|
||
|
||
|
||
def parse_database_url(url):
|
||
"""从 DATABASE_URL 解析 host, port, user, password, database。"""
|
||
# mysql+pymysql://user:password@host:port/database?query
|
||
m = re.match(
|
||
r"mysql\+pymysql://([^:]+):([^@]+)@([^/:]+):(\d+)/([^?]+)",
|
||
url.strip(),
|
||
)
|
||
if not m:
|
||
raise ValueError("DATABASE_URL 格式应为: mysql+pymysql://user:pass@host:port/dbname?charset=utf8mb4")
|
||
user, password, host, port, database = m.groups()
|
||
import urllib.parse
|
||
password = urllib.parse.unquote(password)
|
||
return {
|
||
"host": host,
|
||
"port": int(port),
|
||
"user": user,
|
||
"password": password,
|
||
"database": database.strip("/"),
|
||
}
|
||
|
||
|
||
def main():
|
||
url = os.getenv("DATABASE_URL", DEFAULT_DATABASE_URL)
|
||
try:
|
||
params = parse_database_url(url)
|
||
except Exception as e:
|
||
print("解析 DATABASE_URL 失败:", e)
|
||
sys.exit(1)
|
||
|
||
db_name = params["database"]
|
||
print(f"连接 {params['host']}:{params['port']} (用户 {params['user']})，创建数据库: {db_name}")
|
||
|
||
try:
|
||
import pymysql
|
||
except ImportError:
|
||
print("请先安装: pip install pymysql")
|
||
sys.exit(1)
|
||
|
||
# 连接时不指定 database，才能执行 CREATE DATABASE
|
||
conn = pymysql.connect(
|
||
host=params["host"],
|
||
port=params["port"],
|
||
user=params["user"],
|
||
password=params["password"],
|
||
charset="utf8mb4",
|
||
)
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"CREATE DATABASE IF NOT EXISTS `%s` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
|
||
% db_name
|
||
)
|
||
conn.commit()
|
||
print(f"数据库 {db_name} 已存在或已创建成功。")
|
||
except Exception as e:
|
||
print("创建失败:", e)
|
||
sys.exit(1)
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|