后端第一步开发完成，第一版回归测试完成
This commit is contained in:
Your Name
@@ -0,0 +1,79 @@
|
||||
# 脚本说明
|
||||
|
||||
## seed_test_users.py：插入测试用户（供测试工程师使用）
|
||||
|
||||
向 `users` 表插入 6 个测试账号（管理员、市场部、工程部、技经部、财务部、物贸部各一），**统一密码：123456**。
|
||||
|
||||
**用法**
|
||||
|
||||
```bash
|
||||
cd backend
|
||||
source .venv/bin/activate
|
||||
PYTHONPATH=src python scripts/seed_test_users.py
|
||||
```
|
||||
|
||||
**测试账号**
|
||||
|
||||
| 账号 (username) | 角色 | 密码 |
|
||||
|-----------------|----------|--------|
|
||||
| admin | 管理员 | 123456 |
|
||||
| market | 市场部 | 123456 |
|
||||
| engineer | 工程部 | 123456 |
|
||||
| tech | 技经部 | 123456 |
|
||||
| finance | 财务部 | 123456 |
|
||||
| material | 物贸部 | 123456 |
|
||||
|
||||
（新建项目仅市场部可调，其他角色可查看/修改。）
|
||||
|
||||
---
|
||||
|
||||
## clean_db.py：清空项目与操作日志表
|
||||
|
||||
清空 `projects`、`operation_logs` 表（保留 `users`）。用于重新导入前清理数据。
|
||||
|
||||
**用法**
|
||||
|
||||
```bash
|
||||
cd backend
|
||||
source .venv/bin/activate
|
||||
PYTHONPATH=src python scripts/clean_db.py
|
||||
```
|
||||
|
||||
**清理并重新导入（一条命令）**
|
||||
|
||||
```bash
|
||||
cd backend && source .venv/bin/activate && PYTHONPATH=src python scripts/clean_db.py && PYTHONPATH=src python scripts/import_xls.py ../docs/example.xls
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## import_xls.py：导入 example.xls 到 projects 表
|
||||
|
||||
将 `docs/example.xls` 中的项目数据导入 ocean 数据库的 `projects` 表。
|
||||
|
||||
**前置条件**
|
||||
|
||||
1. 已执行 `schema.sql` 创建数据库和表。
|
||||
2. 已配置 `.env`（MySQL 连接信息）。
|
||||
3. 已安装依赖：`pip install -r requirements.txt`（含 mysql-connector-python、xlrd）。
|
||||
|
||||
**用法**
|
||||
|
||||
```bash
|
||||
cd backend
|
||||
source .venv/bin/activate
|
||||
PYTHONPATH=src python scripts/import_xls.py
|
||||
```
|
||||
|
||||
默认读取 `../docs/example.xls`（相对 backend）。也可指定路径：
|
||||
|
||||
```bash
|
||||
PYTHONPATH=src python scripts/import_xls.py /path/to/example.xls
|
||||
```
|
||||
|
||||
**说明**
|
||||
|
||||
- 使用第一个 Sheet；表头为第 3 行（row 2），数据从第 6 行（row 5）开始。
|
||||
- 仅导入「合同编号」非空的行；若「项目名称」为空则用合同编号或「未命名」。
|
||||
- 表头与 `schema.sql`、`初始需求` 对齐（含「其中：」等前缀的列已映射）。
|
||||
- 导入成功后输出：成功条数、跳过条数。
|
||||
@@ -0,0 +1,26 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
清空 projects 与 operation_logs 表（保留 users）。
|
||||
用法：cd backend && PYTHONPATH=src python3 scripts/clean_db.py
|
||||
"""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_backend = Path(__file__).resolve().parent.parent
|
||||
if str(_backend) not in sys.path:
|
||||
sys.path.insert(0, str(_backend))
|
||||
|
||||
from src.database import get_connection
|
||||
|
||||
def main():
|
||||
with get_connection() as conn:
|
||||
cur = conn.cursor()
|
||||
cur.execute("SET FOREIGN_KEY_CHECKS = 0")
|
||||
cur.execute("TRUNCATE TABLE operation_logs")
|
||||
cur.execute("TRUNCATE TABLE projects")
|
||||
cur.execute("SET FOREIGN_KEY_CHECKS = 1")
|
||||
cur.close()
|
||||
print("已清空 operation_logs、projects 表。")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,7 @@
|
||||
-- 创建 ocean 专用数据库用户（解决 root@localhost 的 Access denied）
|
||||
-- 执行：sudo mysql < backend/scripts/create_db_user.sql
|
||||
-- 然后将 backend/.env 中 MYSQL_USER=ocean, MYSQL_PASSWORD=你设置的密码
|
||||
|
||||
CREATE USER IF NOT EXISTS 'ocean'@'localhost' IDENTIFIED BY 'ocean123';
|
||||
GRANT ALL PRIVILEGES ON ocean.* TO 'ocean'@'localhost';
|
||||
FLUSH PRIVILEGES;
|
||||
@@ -0,0 +1,314 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
将 docs/example.xls 中的项目数据导入 ocean 数据库 projects 表。
|
||||
用法（在项目根目录或 backend 目录执行）：
|
||||
cd backend && PYTHONPATH=src python scripts/import_xls.py [path_to_example.xls]
|
||||
默认路径：../docs/example.xls（相对 backend）或 docs/example.xls（相对项目根）。
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import xlrd
|
||||
|
||||
# 数据库中的日期列（schema DATE 类型）；Excel 里可能填「在建中」「未开工」等文字
|
||||
DATE_COLUMNS = {
|
||||
"sign_date",
|
||||
"start_date",
|
||||
"planned_completion_date",
|
||||
"actual_completion_date",
|
||||
"actual_warranty_refund_date",
|
||||
}
|
||||
# 质保期截止日、中标合同金额、工程款拨付方式：按字符串存储，原样导入
|
||||
STRING_STORAGE_COLUMNS = {"warranty_end_date", "bid_contract_amount", "payment_method"}
|
||||
# 签订/开工/计划竣工/实际竣工：有 _str 字段，日期正常写 DATE+「日期正常」，非日期写 _str
|
||||
DATE_COLUMNS_WITH_STR = {
|
||||
"sign_date",
|
||||
"start_date",
|
||||
"planned_completion_date",
|
||||
"actual_completion_date",
|
||||
}
|
||||
DATE_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}$")
|
||||
|
||||
# 确保可导入 src
|
||||
_backend = Path(__file__).resolve().parent.parent
|
||||
if str(_backend) not in sys.path:
|
||||
sys.path.insert(0, str(_backend))
|
||||
os.chdir(_backend)
|
||||
|
||||
from src.database import get_connection, new_id
|
||||
|
||||
# Excel 表头（中文，strip 后）-> 数据库列名（snake_case）
|
||||
# 依据 schema.sql、初始需求；表头可能含换行或「其中：」等前缀
|
||||
EXCEL_HEADER_TO_DB = {
|
||||
"序号": "serial_no",
|
||||
"合同编号": "contract_code",
|
||||
"供电局项目合同编号": "power_bureau_contract_code",
|
||||
"项目名称": "project_name",
|
||||
"子项个数": "sub_item_count",
|
||||
"子项编码": "sub_item_code",
|
||||
"项目总投资（万元）": "total_investment",
|
||||
"中标合同金额（万元）": "bid_contract_amount",
|
||||
"质保金比例": "warranty_ratio",
|
||||
"结算金额（万元）": "settlement_amount",
|
||||
"总成本测算": "total_cost_estimate",
|
||||
"工程电压等级": "voltage_level",
|
||||
"工程类别": "project_category",
|
||||
"业主单位": "owner_unit",
|
||||
"业主联系人及电话": "owner_contact",
|
||||
"中标形式": "bid_type",
|
||||
"签订日期": "sign_date",
|
||||
"开工日期": "start_date",
|
||||
"计划竣工日期": "planned_completion_date",
|
||||
"实际竣工日期": "actual_completion_date",
|
||||
"质保金（万元）": "warranty_amount",
|
||||
"质保期截止日": "warranty_end_date",
|
||||
"实际退质保金日期": "actual_warranty_refund_date",
|
||||
"所属项目部": "project_department",
|
||||
"项目负责人及电话": "project_leader_contact",
|
||||
"工程款拨付方式": "payment_method",
|
||||
"总体成本（控制）": "total_cost",
|
||||
"是否调整（是/否）": "is_adjusted",
|
||||
"农民工工资（按进度计划）": "migrant_worker_plan",
|
||||
"农民工工资（实付）": "migrant_worker_actual",
|
||||
"其中：乙供材料费（控制）": "self_supply_material_control",
|
||||
"应付材料费（按收款比例）": "material_payable_by_ratio",
|
||||
"实际发生材料费": "material_actual_occurred",
|
||||
"实际支付材料费": "material_actual_paid",
|
||||
"其中：其他费用（控制）": "other_cost_control",
|
||||
"应付其他费": "other_payable",
|
||||
"实际其他费用": "other_actual",
|
||||
"税金": "tax",
|
||||
"利润（万元）": "profit",
|
||||
"实际利润（万元）": "actual_profit",
|
||||
"成本结算金额（万元）": "cost_settlement_amount",
|
||||
"累计进度": "cumulative_progress",
|
||||
"应收款（完成进度款）": "receivable_progress",
|
||||
"开票金额（万元）": "invoice_amount",
|
||||
"实际收款金额（万元）": "actual_receipt_amount",
|
||||
"实际收款完成率": "actual_receipt_rate",
|
||||
"应付款金额（万元）": "payable_amount",
|
||||
"实际付款金额（万元）": "actual_payment_amount",
|
||||
"未收款（万元）": "unreceived_amount",
|
||||
"实际付款完成率": "actual_payment_rate",
|
||||
"民工工资清欠金额（万元）": "migrant_worker_arrears",
|
||||
"结算后成本测算金额": "settlement_cost_estimate",
|
||||
"结算人工费": "settlement_labor_cost",
|
||||
"结算材料费": "settlement_material_cost",
|
||||
"结算其他费": "settlement_other_cost",
|
||||
"到期应结算项目个数": "due_settlement_count",
|
||||
"到期未完成结算个数": "overdue_settlement_count",
|
||||
"存在的问题": "existing_problems",
|
||||
"建议措施": "suggestions",
|
||||
"备注": "remark",
|
||||
}
|
||||
|
||||
|
||||
def _normalize_header(h: str) -> str:
|
||||
return h.strip().replace("\n", "").replace("\r", "")
|
||||
|
||||
|
||||
def _cell_value(sh: xlrd.sheet.Sheet, row: int, col: int):
|
||||
"""取单元格值，日期转 YYYY-MM-DD，空转 None。"""
|
||||
try:
|
||||
cell = sh.cell(row, col)
|
||||
except IndexError:
|
||||
return None
|
||||
if cell.ctype == xlrd.XL_CELL_EMPTY:
|
||||
return None
|
||||
if cell.ctype == xlrd.XL_CELL_DATE:
|
||||
try:
|
||||
d = xlrd.xldate.xldate_as_datetime(cell.value, 0)
|
||||
return d.strftime("%Y-%m-%d")
|
||||
except Exception:
|
||||
return None
|
||||
v = cell.value
|
||||
if isinstance(v, float):
|
||||
if v != v: # NaN
|
||||
return None
|
||||
if v == int(v):
|
||||
v = int(v)
|
||||
if v == "" or (isinstance(v, str) and not v.strip()):
|
||||
return None
|
||||
if isinstance(v, str):
|
||||
v = v.strip()
|
||||
return v
|
||||
|
||||
|
||||
def _cell_value_date_and_str(sh: xlrd.sheet.Sheet, row: int, col: int) -> tuple:
|
||||
"""专用于带 _str 的日期列。返回 (date_val, str_val)：date_val 为 YYYY-MM-DD 或 None，str_val 为「日期正常」或原文或 None。"""
|
||||
try:
|
||||
cell = sh.cell(row, col)
|
||||
except IndexError:
|
||||
return None, None
|
||||
if cell.ctype == xlrd.XL_CELL_EMPTY:
|
||||
return None, None
|
||||
# 日期型或数字型（Excel 序列日期）
|
||||
if cell.ctype == xlrd.XL_CELL_DATE:
|
||||
try:
|
||||
d = xlrd.xldate.xldate_as_datetime(cell.value, 0)
|
||||
return d.strftime("%Y-%m-%d"), "日期正常"
|
||||
except Exception:
|
||||
return None, None
|
||||
if cell.ctype == xlrd.XL_CELL_NUMBER and isinstance(cell.value, (int, float)):
|
||||
v = cell.value
|
||||
if 20000 <= v <= 50000 or (0 < v < 1): # 常见 Excel 日期范围或时间部分
|
||||
try:
|
||||
d = xlrd.xldate.xldate_as_datetime(v, 0)
|
||||
return d.strftime("%Y-%m-%d"), "日期正常"
|
||||
except Exception:
|
||||
pass
|
||||
return None, str(int(v)) if v == int(v) else str(v)
|
||||
# 文本
|
||||
v = cell.value
|
||||
if v is None or (isinstance(v, str) and not v.strip()):
|
||||
return None, None
|
||||
s = (v if isinstance(v, str) else str(v)).strip()
|
||||
if DATE_PATTERN.match(s):
|
||||
return s, "日期正常"
|
||||
return None, s
|
||||
|
||||
|
||||
def _cell_value_raw_str(sh: xlrd.sheet.Sheet, row: int, col: int):
|
||||
"""取单元格原值并转为字符串（用于 warranty_end_date、bid_contract_amount 等字符串存储列）。"""
|
||||
try:
|
||||
cell = sh.cell(row, col)
|
||||
except IndexError:
|
||||
return None
|
||||
if cell.ctype == xlrd.XL_CELL_EMPTY:
|
||||
return None
|
||||
v = cell.value
|
||||
if v is None:
|
||||
return None
|
||||
s = (str(v) if not isinstance(v, str) else v).strip()
|
||||
return s if s else None
|
||||
|
||||
|
||||
def _row_to_db_dict(sh: xlrd.sheet.Sheet, header_to_col: dict, row_idx: int) -> dict:
|
||||
"""将一行转为 DB 列名 -> 值的字典。header_to_col: 表头(已 normalize) -> 列索引。"""
|
||||
row_dict = {}
|
||||
for header, col in header_to_col.items():
|
||||
db_col = EXCEL_HEADER_TO_DB.get(header)
|
||||
if not db_col:
|
||||
continue
|
||||
if db_col in DATE_COLUMNS_WITH_STR:
|
||||
date_val, str_val = _cell_value_date_and_str(sh, row_idx, col)
|
||||
if date_val is not None:
|
||||
row_dict[db_col] = date_val
|
||||
row_dict[db_col + "_str"] = "日期正常"
|
||||
elif str_val is not None and str_val.strip():
|
||||
row_dict[db_col + "_str"] = str_val
|
||||
continue
|
||||
if db_col in STRING_STORAGE_COLUMNS:
|
||||
val = _cell_value_raw_str(sh, row_idx, col)
|
||||
if val is not None:
|
||||
row_dict[db_col] = val
|
||||
continue
|
||||
if db_col == "total_cost":
|
||||
val = _cell_value(sh, row_idx, col)
|
||||
if val is not None:
|
||||
if isinstance(val, (int, float)) and (not isinstance(val, float) or not (val != val)):
|
||||
row_dict[db_col] = float(val)
|
||||
elif isinstance(val, str):
|
||||
s = val.strip()
|
||||
try:
|
||||
row_dict[db_col] = float(s)
|
||||
except (ValueError, TypeError):
|
||||
row_dict[db_col] = 0
|
||||
row_dict["_total_cost_text"] = s
|
||||
else:
|
||||
row_dict[db_col] = 0
|
||||
continue
|
||||
val = _cell_value(sh, row_idx, col)
|
||||
if val is None:
|
||||
continue
|
||||
if db_col in DATE_COLUMNS and isinstance(val, str) and not DATE_PATTERN.match(val.strip()):
|
||||
continue
|
||||
row_dict[db_col] = val
|
||||
if "_total_cost_text" in row_dict:
|
||||
prefix = "总体成本（文字）：" + row_dict["_total_cost_text"] + "\n"
|
||||
row_dict["remark"] = prefix + (row_dict.get("remark") or "")
|
||||
del row_dict["_total_cost_text"]
|
||||
return row_dict
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) >= 2:
|
||||
xls_path = Path(sys.argv[1]).resolve()
|
||||
else:
|
||||
for p in [_backend.parent / "docs" / "example.xls", _backend / "docs" / "example.xls"]:
|
||||
if p.exists():
|
||||
xls_path = p
|
||||
break
|
||||
else:
|
||||
print("未找到 example.xls，请指定路径：python scripts/import_xls.py <path>")
|
||||
sys.exit(1)
|
||||
|
||||
if not xls_path.exists():
|
||||
print(f"文件不存在：{xls_path}")
|
||||
sys.exit(1)
|
||||
|
||||
wb = xlrd.open_workbook(str(xls_path))
|
||||
sh = wb.sheet_by_index(0)
|
||||
header_row_idx = 2
|
||||
data_start_idx = 5
|
||||
|
||||
raw_headers = [str(sh.cell_value(header_row_idx, j)) for j in range(sh.ncols)]
|
||||
# header_to_col: 归一化表头 -> 列索引（取首次出现）
|
||||
header_to_col = {}
|
||||
for j, h in enumerate(raw_headers):
|
||||
nh = _normalize_header(h)
|
||||
if nh and nh not in header_to_col:
|
||||
header_to_col[nh] = j
|
||||
|
||||
contract_code_col = header_to_col.get("合同编号")
|
||||
project_name_col = header_to_col.get("项目名称")
|
||||
if contract_code_col is None:
|
||||
print("未找到列「合同编号」，请检查 Excel 表头。")
|
||||
sys.exit(1)
|
||||
|
||||
inserted = 0
|
||||
skipped = 0
|
||||
with get_connection() as conn:
|
||||
cur = conn.cursor()
|
||||
for r in range(data_start_idx, sh.nrows):
|
||||
row_dict = _row_to_db_dict(sh, header_to_col, r)
|
||||
contract_code = _cell_value(sh, r, contract_code_col) if contract_code_col is not None else row_dict.get("contract_code")
|
||||
project_name = _cell_value(sh, r, project_name_col) if project_name_col is not None else row_dict.get("project_name")
|
||||
|
||||
if contract_code is None or (isinstance(contract_code, str) and not contract_code.strip()):
|
||||
skipped += 1
|
||||
continue
|
||||
if project_name is None or (isinstance(project_name, str) and not project_name.strip()):
|
||||
project_name = str(contract_code) if contract_code else "未命名"
|
||||
row_dict["contract_code"] = contract_code
|
||||
row_dict["project_name"] = project_name
|
||||
|
||||
pid = new_id()
|
||||
now = datetime.utcnow()
|
||||
row_dict["id"] = pid
|
||||
row_dict["progress"] = (str(row_dict.get("cumulative_progress") or "")[:32]) if row_dict.get("cumulative_progress") is not None else ""
|
||||
row_dict["cost"] = ""
|
||||
row_dict["created_at"] = now
|
||||
row_dict["updated_at"] = now
|
||||
|
||||
cols = list(row_dict.keys())
|
||||
vals = [row_dict[k] for k in cols]
|
||||
placeholders = ", ".join(["%s"] * len(cols))
|
||||
try:
|
||||
cur.execute(
|
||||
"INSERT INTO projects (" + ", ".join(cols) + ") VALUES (" + placeholders + ")",
|
||||
vals,
|
||||
)
|
||||
inserted += 1
|
||||
except Exception as e:
|
||||
print(f"Row {r + 1} 跳过（{e}）")
|
||||
skipped += 1
|
||||
cur.close()
|
||||
print(f"导入完成：成功 {inserted} 条，跳过 {skipped} 条。")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,39 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
为 projects 表增加四个日期原文列（若不存在）。
|
||||
用法：cd backend && PYTHONPATH=src python3 scripts/migrate_add_date_str_columns.py
|
||||
"""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_backend = Path(__file__).resolve().parent.parent
|
||||
if str(_backend) not in sys.path:
|
||||
sys.path.insert(0, str(_backend))
|
||||
|
||||
from src.database import get_connection
|
||||
|
||||
ALTER_SQL = """
|
||||
ALTER TABLE projects
|
||||
ADD COLUMN sign_date_str varchar(128) DEFAULT NULL COMMENT '签订日期原文：日期正常或导入字符串' AFTER sign_date,
|
||||
ADD COLUMN start_date_str varchar(128) DEFAULT NULL COMMENT '开工日期原文' AFTER start_date,
|
||||
ADD COLUMN planned_completion_date_str varchar(128) DEFAULT NULL COMMENT '计划竣工日期原文' AFTER planned_completion_date,
|
||||
ADD COLUMN actual_completion_date_str varchar(128) DEFAULT NULL COMMENT '实际竣工日期原文' AFTER actual_completion_date;
|
||||
"""
|
||||
|
||||
def main():
|
||||
with get_connection() as conn:
|
||||
cur = conn.cursor()
|
||||
try:
|
||||
cur.execute(ALTER_SQL)
|
||||
conn.commit()
|
||||
print("已为 projects 表添加 sign_date_str、start_date_str、planned_completion_date_str、actual_completion_date_str 列。")
|
||||
except Exception as e:
|
||||
if "Duplicate column name" in str(e):
|
||||
print("列已存在，无需迁移。")
|
||||
else:
|
||||
raise
|
||||
finally:
|
||||
cur.close()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,35 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
将 projects 表中「质保期截止日」「中标合同金额」「工程款拨付方式」改为字符串类型。
|
||||
用法：cd backend && PYTHONPATH=src python3 scripts/migrate_string_columns.py
|
||||
"""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_backend = Path(__file__).resolve().parent.parent
|
||||
if str(_backend) not in sys.path:
|
||||
sys.path.insert(0, str(_backend))
|
||||
|
||||
from src.database import get_connection
|
||||
|
||||
def main():
|
||||
with get_connection() as conn:
|
||||
cur = conn.cursor()
|
||||
try:
|
||||
cur.execute(
|
||||
"ALTER TABLE projects "
|
||||
"MODIFY COLUMN bid_contract_amount varchar(512) DEFAULT NULL COMMENT '中标合同金额（万元，字符串存储）', "
|
||||
"MODIFY COLUMN warranty_end_date varchar(256) DEFAULT NULL COMMENT '质保期截止日（字符串存储）', "
|
||||
"MODIFY COLUMN payment_method varchar(2048) DEFAULT NULL COMMENT '工程款拨付方式', "
|
||||
"MODIFY COLUMN is_adjusted varchar(512) DEFAULT NULL COMMENT '是否调整：是/否'"
|
||||
)
|
||||
conn.commit()
|
||||
print("已将 bid_contract_amount、warranty_end_date、payment_method、is_adjusted 改为目标类型/长度。")
|
||||
except Exception as e:
|
||||
print(f"迁移失败: {e}")
|
||||
raise
|
||||
finally:
|
||||
cur.close()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,59 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
插入测试用户，供测试工程师使用。统一密码：123456
|
||||
用法：cd backend && PYTHONPATH=src python3 scripts/seed_test_users.py
|
||||
"""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
_backend = Path(__file__).resolve().parent.parent
|
||||
if str(_backend) not in sys.path:
|
||||
sys.path.insert(0, str(_backend))
|
||||
|
||||
import bcrypt
|
||||
|
||||
from src.database import get_connection, new_id
|
||||
|
||||
|
||||
def _hash(pwd: str) -> str:
|
||||
"""bcrypt 哈希，与 auth 中 passlib 校验兼容"""
|
||||
return bcrypt.hashpw(pwd.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
|
||||
|
||||
# 角色枚举与测试账号： (username, role, display_name)
|
||||
TEST_USERS = [
|
||||
("admin", "管理员", "管理员"),
|
||||
("market", "市场部", "市场部"),
|
||||
("engineer", "工程部", "工程部"),
|
||||
("tech", "技经部", "技经部"),
|
||||
("finance", "财务部", "财务部"),
|
||||
("material", "物贸部", "物贸部"),
|
||||
]
|
||||
|
||||
TEST_PASSWORD = "123456"
|
||||
|
||||
|
||||
def main():
|
||||
pwd_hash = _hash(TEST_PASSWORD)
|
||||
with get_connection() as conn:
|
||||
cur = conn.cursor()
|
||||
for username, role, display_name in TEST_USERS:
|
||||
uid = new_id()
|
||||
try:
|
||||
cur.execute(
|
||||
"INSERT INTO users (id, username, password_hash, role, display_name) VALUES (%s, %s, %s, %s, %s)",
|
||||
(uid, username, pwd_hash, role, display_name),
|
||||
)
|
||||
conn.commit()
|
||||
print(f"已插入: {username} ({role})")
|
||||
except Exception as e:
|
||||
if "Duplicate entry" in str(e) or "1062" in str(e):
|
||||
print(f"已存在，跳过: {username}")
|
||||
else:
|
||||
raise
|
||||
cur.close()
|
||||
print(f"\n测试账号统一密码: {TEST_PASSWORD}")
|
||||
print("账号列表: admin, market, engineer, tech, finance, material")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user