diff --git a/scripts/import_nav_urls.py b/scripts/import_nav_urls.py new file mode 100644 index 0000000..82ac5e8 --- /dev/null +++ b/scripts/import_nav_urls.py @@ -0,0 +1,185 @@ +""" +从 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) \ No newline at end of file