209 lines
9.0 KiB
Python
209 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
||
"""scripts/import_journals.py —— 期刊目录导入 CLI（source_v4.xlsx → SQLite journal_directory）。
|
||
|
||
核心解析/upsert/schema 在 app/services/journal_seed.py（与启动自动 seed 共享）；
|
||
本 CLI 仅 import 该模块（sys.path bootstrap），零网络零 BFF，仍不碰 db.py。
|
||
职责：口径校验 → 幂等 upsert → 对账报告 data/journals/import_report.txt。
|
||
|
||
数据流：
|
||
openpyxl 非 read_only 加载（需要读 cell.hyperlink.target 来源链接）
|
||
→ 「国际期刊」「中文期刊」两 sheet 按【列名】映射（不按索引，防列序漂移）
|
||
→ 行级清洗（空→None 绝不填 0；分区数字→"N区"；JCR学科→JSON 数组）
|
||
→ 按 id 幂等 upsert（ON CONFLICT DO UPDATE，重跑 count 不变）
|
||
→ 对账报告 data/journals/import_report.txt
|
||
|
||
v4 源（2026-10-10）：JIF=2025 统计年/2026 JCR 发布；中文 IF=2024 统计年/
|
||
年报 2025 版；新增官网三列 + JCR发布年；JCR学科列改名「JCR学科（2024源表）」。
|
||
旧文件 source.xlsx（v2）原样保留为回退点，勿覆盖。
|
||
|
||
schema 迁移（幂等 + 备份）：既有表缺列 → PRAGMA table_info 探测后 ALTER TABLE
|
||
ADD COLUMN，缺哪加哪、重复跑不报错；跑前先 cp review.db review.db.bak-<日期>。
|
||
|
||
id 策略：有 ISSN → "{lang}:{ISSN原样(strip)}"；无 ISSN → "{lang}:{slug(name)}"；
|
||
同 lang 内 id 冲突 → 报告冲突列表并【中止导入】（非零退出，不写库）。
|
||
|
||
用法：python scripts/import_journals.py （cwd 任意；路径按本文件位置推导）
|
||
退出码：0 成功；1 id 冲突中止；2 源数据行数与口径不符；3 结构/文件错误。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import time
|
||
from collections import Counter
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
# ---- 路径（review_service/scripts/ → parents[1] = review_service）----
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
if str(ROOT) not in sys.path:
|
||
sys.path.insert(0, str(ROOT)) # 供 app.services.journal_seed import
|
||
|
||
from app.services.journal_seed import ( # noqa: E402
|
||
UPSERT_SQL,
|
||
EXPECT_INTL,
|
||
EXPECT_ZH,
|
||
SHEET_INTL,
|
||
SHEET_ZH,
|
||
JournalSeedError,
|
||
extract_rows_from_xlsx,
|
||
new_stats,
|
||
upsert_to_db,
|
||
)
|
||
|
||
SOURCE_XLSX = ROOT / "data" / "journals" / "source_v4.xlsx" # v4；v2=source.xlsx 保留回退
|
||
REPORT_TXT = ROOT / "data" / "journals" / "import_report.txt"
|
||
|
||
|
||
def build_report(
|
||
stage: str,
|
||
stats: dict,
|
||
counts: dict[str, int] | None,
|
||
conflicts: list[str],
|
||
idempotent_ok: bool | None,
|
||
exit_note: str = "",
|
||
) -> str:
|
||
L: list[str] = []
|
||
L.append("journals directory 导入对账报告")
|
||
L.append(f"生成时间: {datetime.now().isoformat(timespec='seconds')}")
|
||
L.append(f"源文件: {SOURCE_XLSX}（{SOURCE_XLSX.stat().st_size if SOURCE_XLSX.exists() else 0} bytes）")
|
||
L.append(f"阶段: {stage}")
|
||
L.append("")
|
||
L.append("=== 计数 ===")
|
||
L.append(f"读取有效行: 国际 {stats['read_intl']} / 中文 {stats['read_zh']} / 合计 {stats['read_intl'] + stats['read_zh']}")
|
||
L.append(f"成功: {stats['success']} 跳过(名称空): {stats['skipped']} 失败: {stats['failed']}")
|
||
if counts is not None:
|
||
L.append(f"写入后 count(*): en={counts.get('en', 0)} zh={counts.get('zh', 0)} total={counts.get('total', 0)}")
|
||
L.append(f"期望: en={EXPECT_INTL} zh={EXPECT_ZH} total={EXPECT_INTL + EXPECT_ZH} → {'✓ 相符' if counts.get('total') == EXPECT_INTL + EXPECT_ZH else '✗ 不符'}")
|
||
L.append("")
|
||
L.append("=== 分区分布（国际）===")
|
||
L.append(" / ".join(f"{k} {v}" for k, v in sorted(stats["part_dist"].items())))
|
||
L.append(f"分区空白(→null 无note): {len(stats['part_blank'])}")
|
||
L.append(f"分区未检得(→null + note): {len(stats['part_notfound'])}")
|
||
L.append("")
|
||
L.append(f"=== JIF 缺失 {len(stats['jif_missing'])} 条（jif=null + note=核对状态）===")
|
||
for n, s in stats["jif_missing"]:
|
||
L.append(f" - {n} [{s}]")
|
||
L.append("")
|
||
L.append(f"=== 中文复合IF 缺失 {len(stats['cif_missing'])} 条 ===")
|
||
for n, s in stats["cif_missing"]:
|
||
L.append(f" - {n} [{s}]")
|
||
L.append("")
|
||
L.append("=== 分区未检得 期刊名 ===")
|
||
if stats["part_notfound"]:
|
||
for n in stats["part_notfound"]:
|
||
L.append(f" - {n}")
|
||
else:
|
||
L.append(" （无——源表以空白表示未检得，见上方'分区空白'）")
|
||
L.append("")
|
||
L.append("=== 重复 id 冲突（应为空）===")
|
||
if conflicts:
|
||
for c in conflicts:
|
||
L.append(f" !! {c}")
|
||
else:
|
||
L.append(" （无）")
|
||
L.append("")
|
||
L.append("=== id 策略 ===")
|
||
L.append(f"ISSN 型: {stats['id_issn']} 条 slug 型: {stats['id_slug']} 条")
|
||
L.append("")
|
||
L.append("=== 来源超链接 ===")
|
||
L.append(f"国际 JIF来源: {stats['hl_jif']}/{stats['read_intl']} 分区来源: {stats['hl_part']}/{stats['read_intl']} 中文数据来源: {stats['hl_zh']}/{stats['read_zh']}")
|
||
L.append("")
|
||
L.append("=== 结构迁移（ALTER TABLE，幂等）===")
|
||
added = stats.get("migrated_cols") or []
|
||
skipped = stats.get("migrated_skipped") or []
|
||
L.append(f" 本次加列: {', '.join(added) if added else '无（列已齐）'}")
|
||
L.append(f" 已存在跳过: {', '.join(skipped) if skipped else '无'}")
|
||
L.append("")
|
||
L.append("=== 官网映射（决策：official_source 不入库）===")
|
||
L.append(f" 识别列: {stats.get('official_url_headers', '未识别')}")
|
||
tot = stats.get("official_total", 0)
|
||
L.append(f" official_url 填充: {stats.get('official_filled', 0)}/{tot}"
|
||
f" official_check_status 填充: {stats.get('official_check_filled', 0)}/{tot}")
|
||
L.append(f" jcr_release_year 填充: {stats.get('jcr_filled', 0)}（仅国际=2026；中文表无此列→null）")
|
||
missing = stats.get("official_missing") or []
|
||
if missing:
|
||
L.append(f" 官网缺失 {len(missing)} 条（official_url=null，前端显示「—」不生成链接）:")
|
||
for lg, n in missing:
|
||
L.append(f" - [{lg}] {n}")
|
||
else:
|
||
L.append(" 官网缺失: （无）")
|
||
L.append("")
|
||
L.append("=== 幂等性验证 ===")
|
||
if idempotent_ok is None:
|
||
L.append(" 未执行")
|
||
else:
|
||
L.append(f" 第二次 upsert 后 count 不变: {'✓ 通过' if idempotent_ok else '✗ 失败'}")
|
||
L.append("")
|
||
if exit_note:
|
||
L.append(f"!!! {exit_note}")
|
||
return "\n".join(L) + "\n"
|
||
|
||
|
||
def main() -> int:
|
||
if not SOURCE_XLSX.exists():
|
||
print(f"[错误] 源文件不存在: {SOURCE_XLSX}", file=sys.stderr)
|
||
return 3
|
||
|
||
stats: dict[str, Any] = new_stats()
|
||
imported_at = datetime.now().isoformat(timespec="seconds")
|
||
t0 = time.time()
|
||
|
||
# 1) 读 Excel + 解析 + 口径/id 校验（journal_seed 共享；失败不写库）
|
||
try:
|
||
rows_en, rows_zh, stats = extract_rows_from_xlsx(
|
||
SOURCE_XLSX, imported_at=imported_at, stats=stats)
|
||
except JournalSeedError as exc:
|
||
REPORT_TXT.write_text(
|
||
build_report(exc.stage, stats, None, exc.conflicts, None, str(exc)),
|
||
encoding="utf-8")
|
||
if exc.exit_code == 1:
|
||
print(f"[中止] {exc}", file=sys.stderr)
|
||
for c in exc.conflicts:
|
||
print(f" !! {c}", file=sys.stderr)
|
||
elif exc.exit_code == 2:
|
||
print(f"[中止] {exc}", file=sys.stderr)
|
||
else:
|
||
print(f"[错误] {exc}", file=sys.stderr)
|
||
return exc.exit_code
|
||
|
||
all_rows = rows_en + rows_zh
|
||
|
||
# 2) 写 SQLite（独立 stdlib 连接；表由 journal_seed 创建，db.py 零触碰）
|
||
db_path = Path(os.environ.get("REVIEW_DB_PATH") or (ROOT / "review.db"))
|
||
if not db_path.is_absolute():
|
||
db_path = ROOT / db_path
|
||
counts = upsert_to_db(db_path, rows_en, rows_zh)
|
||
stats["success"] = len(all_rows)
|
||
|
||
# 3) 幂等验证：同数据第二次 upsert → count 不变
|
||
conn = sqlite3.connect(str(db_path))
|
||
try:
|
||
conn.executemany(UPSERT_SQL, all_rows)
|
||
conn.commit()
|
||
cur = conn.execute(
|
||
"SELECT lang, COUNT(*) FROM journal_directory GROUP BY lang")
|
||
d = {lang: n for lang, n in cur.fetchall()}
|
||
idempotent_ok = {"en": d.get("en", 0), "zh": d.get("zh", 0),
|
||
"total": sum(d.values())} == counts
|
||
|
||
# 分区分布来自实际库（防统计与写入脱节；未检得 note 单列）
|
||
stats["part_dist"] = Counter(dict(conn.execute(
|
||
"SELECT COALESCE(cas_partition, "
|
||
"CASE WHEN cas_partition_note='未检得' THEN '未检得' ELSE '空白' END), COUNT(*) "
|
||
"FROM journal_directory WHERE lang='en' GROUP BY 1").fetchall()))
|
||
finally:
|
||
conn.close()
|
||
|
||
report = build_report("完成", stats, counts, [], idempotent_ok)
|
||
REPORT_TXT.write_text(report, encoding="utf-8")
|
||
print(report)
|
||
print(f"[完成] 耗时 {time.time() - t0:.2f}s db={db_path} 报告={REPORT_TXT}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|