bwstudio
2fe59e15ef
scripts/import_nav_urls.py：从 C://Users//PF//Desktop//nav_urls.csv 导入 业务规则（与用户确认）： - 「热门网址」分类不导入（仅是使用频度的复述，与内网应用重叠 7 条） - 新建 4 个 NavCategory：内网应用(sort=9) / 外网应用(sort=8) / 其他应用(sort=7) / 常用工具(sort=6) - 删除 2 个 E2E 测试残留 NavCategory（icon=layui-icon-test 且 nav_count=0） - Nav 增量导入：URL 已存在（按 category+url_normalized 去重）则跳过 - 同 URL 在不同分类可双写（按用户语义分类的本质） - 同分类内 URL 末尾 / 差异视作重复 支持 --dry-run 干跑模式。 本次实跑结果： - NavCategory created: 4 - Nav total inserted: 67 - 内网应用: 35 - 外网应用: 18 - 其他应用: 12 - 常用工具: 2 - Skipped (热门网址): 10 - Skipped (duplicate url in same cat): 1（液位仪末尾 /） - E2E trash NavCategory deleted: 2 Nav 总数从 54 → 121；NavCategory 从 7 → 9。
185 lines
7.6 KiB
Python
185 lines
7.6 KiB
Python
"""
|
||
从 CSV 增量导入 Nav + 同步 NavCategory。
|
||
|
||
用法：
|
||
./venv/Scripts/python.exe scripts/import_nav_urls.py # 真实导入
|
||
./venv/Scripts/python.exe scripts/import_nav_urls.py --dry-run # 只打印，不写库
|
||
|
||
CSV 格式（首行为表头）：
|
||
category,name,url
|
||
|
||
业务规则（与用户确认）：
|
||
- 「热门网址」分类不导入（只是常用度的复述，与内网应用重叠）
|
||
- 4 个新分类建 NavCategory：
|
||
内网应用 (sort=9) 外网应用 (sort=8) 其他应用 (sort=7) 常用工具 (sort=6)
|
||
- 删除 2 个 E2E 测试残留的 NavCategory（enable=0 且 nav_count=0）
|
||
- Nav：URL 已存在则跳过；同分类同名则跳过；其他新增
|
||
- 同 URL 在不同分类仍可双写（这是「按用户语义分类」的本质）
|
||
- 同分类内 URL 完全相同视为同一记录
|
||
"""
|
||
import argparse
|
||
import csv
|
||
import os
|
||
import sys
|
||
from collections import OrderedDict
|
||
|
||
# 让脚本能找到项目根目录的 applications 包
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from app import app # noqa: E402
|
||
from applications.extensions import db # noqa: E402
|
||
from applications.models import Nav, NavCategory # noqa: E402
|
||
|
||
|
||
# ---- 配置文件 ----
|
||
CSV_PATH = os.path.expanduser(
|
||
r"C:\Users\PF\Desktop\nav_urls.csv"
|
||
if os.name == "nt"
|
||
else "~/Desktop/nav_urls.csv"
|
||
)
|
||
|
||
# 不导入的分类（用户确认：热门网址是"使用最多的十个"的复述，与内网应用重叠）
|
||
SKIP_CATEGORIES = {"热门网址"}
|
||
|
||
# 新建 NavCategory 的配置（按用户确认的映射）
|
||
NEW_CATEGORIES = OrderedDict([
|
||
("内网应用", {"sort": 9, "icon": "layui-icon-cellphone", "description": "石化内网办公 / 业务系统"}),
|
||
("外网应用", {"sort": 8, "icon": "layui-icon-website", "description": "外网办公 / 政府 / 公共服务系统"}),
|
||
("其他应用", {"sort": 7, "icon": "layui-icon-template-1", "description": "其他常用工具与服务"}),
|
||
("常用工具", {"sort": 6, "icon": "layui-icon-tools", "description": "日常工具与下载"}),
|
||
])
|
||
|
||
# E2E 测试残留分类（icon=layui-icon-test 且 enable=1 但 nav_count=0 时删除）
|
||
E2E_TRASH_INDICATOR = "layui-icon-test"
|
||
|
||
|
||
def normalize_url(u: str) -> str:
|
||
"""URL 归一化：去末尾 /，去空白，小写。便于跨分类 / 拼写差异去重。"""
|
||
return (u or "").strip().rstrip("/").lower()
|
||
|
||
|
||
def main(dry_run: bool = False):
|
||
with app.app_context():
|
||
# ---- Step 1：清理 E2E 残留分类 ----
|
||
trash_cats = [
|
||
c for c in NavCategory.query.filter(NavCategory.enable == 1).all()
|
||
if (c.icon or "").startswith(E2E_TRASH_INDICATOR)
|
||
and Nav.query.filter(Nav.category == c.name).count() == 0
|
||
]
|
||
print(f"[Step 1] {'WOULD delete' if dry_run else 'delete'} "
|
||
f"E2E trash NavCategory: {[c.name for c in trash_cats]}")
|
||
for c in trash_cats:
|
||
print(f" - id={c.id} name={c.name!r} icon={c.icon!r}")
|
||
if not dry_run:
|
||
db.session.delete(c)
|
||
|
||
# ---- Step 2：新建 4 个分类（已存在则跳过） ----
|
||
existing_cat_names = {c.name for c in NavCategory.query.all()}
|
||
created_cats = []
|
||
for name, cfg in NEW_CATEGORIES.items():
|
||
if name in existing_cat_names:
|
||
print(f"[Step 2] SKIP NavCategory (exists): {name}")
|
||
continue
|
||
print(f"[Step 2] {'WOULD create' if dry_run else 'create'} "
|
||
f"NavCategory: {name} sort={cfg['sort']} icon={cfg['icon']}")
|
||
cat = NavCategory(
|
||
name=name,
|
||
icon=cfg["icon"],
|
||
description=cfg["description"],
|
||
sort=cfg["sort"],
|
||
enable=1,
|
||
)
|
||
if not dry_run:
|
||
db.session.add(cat)
|
||
created_cats.append(cat)
|
||
if not dry_run:
|
||
db.session.flush() # 取到 cat.id 给后面的 INSERT 用
|
||
|
||
# ---- Step 3：导入 Nav 行 ----
|
||
# 预加载现有 Nav 的 (url_normalized, category) 集合，用于 O(1) 去重判断
|
||
existing_pairs = {
|
||
(normalize_url(n.url), n.category)
|
||
for n in Nav.query.filter(Nav.enable == 1).all()
|
||
}
|
||
# 重新拉一次刷新缓存（包括刚被删除/已存在的）
|
||
existing_pairs = {
|
||
(normalize_url(n.url), n.category)
|
||
for n in Nav.query.filter(Nav.enable == 1).all()
|
||
}
|
||
|
||
inserted, skipped_skip_cat, skipped_dup = 0, 0, 0
|
||
per_cat_added = OrderedDict((n, 0) for n in NEW_CATEGORIES.keys())
|
||
|
||
with open(CSV_PATH, encoding="utf-8-sig", newline="") as fh:
|
||
reader = csv.DictReader(fh)
|
||
for row_no, row in enumerate(reader, start=2): # 数据从第 2 行开始
|
||
cat_name = (row.get("category") or "").strip()
|
||
title = (row.get("name") or "").strip()
|
||
url = (row.get("url") or "").strip()
|
||
if not (cat_name and title and url):
|
||
print(f" WARN row {row_no}: 字段为空，跳过 -> {row}")
|
||
continue
|
||
|
||
# 跳过的分类
|
||
if cat_name in SKIP_CATEGORIES:
|
||
skipped_skip_cat += 1
|
||
continue
|
||
|
||
# 该分类必须是本次脚本要导入的 4 个之一
|
||
if cat_name not in NEW_CATEGORIES:
|
||
print(f" WARN row {row_no}: 未知分类 {cat_name!r}（不在 4 个目标分类内），跳过")
|
||
continue
|
||
|
||
# URL 归一化去重
|
||
key = (normalize_url(url), cat_name)
|
||
if key in existing_pairs:
|
||
skipped_dup += 1
|
||
continue
|
||
|
||
print(f" + {cat_name} | {title} | {url}")
|
||
if not dry_run:
|
||
db.session.add(Nav(
|
||
category=cat_name,
|
||
title=title,
|
||
url=url,
|
||
description="",
|
||
icon="layui-icon-link",
|
||
sort=0,
|
||
enable=1,
|
||
is_external=1, # CSV 全是外链
|
||
create_by="csv-import",
|
||
))
|
||
existing_pairs.add(key)
|
||
inserted += 1
|
||
per_cat_added[cat_name] += 1
|
||
|
||
# ---- Step 4：commit + 统计 ----
|
||
if not dry_run:
|
||
db.session.commit()
|
||
|
||
print()
|
||
print("=" * 60)
|
||
print(f"Dry-run: {dry_run}")
|
||
print(f"NavCategory created: {len(created_cats)}")
|
||
for n, k in per_cat_added.items():
|
||
print(f" {n}: +{k} nav")
|
||
print(f"Nav total inserted: {inserted}")
|
||
print(f"Skipped (热门网址): {skipped_skip_cat}")
|
||
print(f"Skipped (duplicate url in same cat): {skipped_dup}")
|
||
print(f"E2E trash NavCategory deleted: {len(trash_cats)}")
|
||
print("=" * 60)
|
||
|
||
# ---- Step 5：现状汇报 ----
|
||
if not dry_run:
|
||
print("\n--- after import ---")
|
||
cats = NavCategory.query.filter(NavCategory.enable == 1).order_by(NavCategory.sort.desc()).all()
|
||
for c in cats:
|
||
n = Nav.query.filter(Nav.category == c.name, Nav.enable == 1).count()
|
||
print(f" NavCategory {c.name!r} (sort={c.sort}): {n} nav")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser(description="CSV → Nav/NavCategory 增量导入")
|
||
parser.add_argument("--dry-run", action="store_true", help="只打印要做的变更，不写库")
|
||
args = parser.parse_args()
|
||
main(dry_run=args.dry_run) |