RNAQwen / scripts /14_quality_report.py

add scripts

acd280c verified 22 days ago

20.2 kB

"""P0 数据质量报告：对 10_collect.py 产出的 JSONL 做全面体检。
只读，不改数据。用法：
    python scripts/14_quality_report.py \
        --in data/raw/rna_literature_1k.jsonl \
        --out reports/collect_1k_report.json \
        --stats '{"nt_scanned":...}'      # 可选：把 10_collect.py 打印的 [stats] 原样传进来
检查项
------
1. 总条数 + [stats] 全计数
2. 序列长度分布（min/p25/median/p75/p90/max）+ >MAX_NT 的条数（必须 0）
3. 字母表：去掉 ACGU 之后剩余字符集合（必须为空集或极少数简并碱基）
4. 物种分布 top10 / RNA 类型分布（DEFINITION 关键词粗分）
5. 摘要长度分布
6. 去重：uid 唯一、accession 唯一、序列本身 sha1 唯一
7. 每条必须有非空 pmid 和 abstract
8. 抽样 10 条
任何违规项会进入 `critical_issues`（顶层），方便上层直接判红。
"""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import statistics
import sys
import time
from collections import Counter
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
MIN_NT, MAX_NT = 30, 1022  # 与 10_collect.py 保持一致（1022 = RNA-FM 实测上限）
MIN_ABSTRACT = 200

# IUPAC 简并碱基（去掉 ACGU 之后"合法"的残留）
IUPAC = set("NRYWSKMBDHV")
VALID = set("ACGU")

# DEFINITION 关键词 → RNA 类型。**顺序即优先级**（先命中先归类）。
TYPE_RULES: list[tuple[str, str]] = [
    ("circRNA", r"circular rna|\bcircrna\b|circ_?rna"),
    ("miRNA", r"\bmirna\b|microrna|\bmiR-|\blet-7\b|\bmir-\d"),
    ("snoRNA", r"\bsnorna\b|small nucleolar"),
    ("snRNA", r"\bsnrna\b|small nuclear rna"),
    ("tRNA", r"\btrna\b|transfer rna|\btrna-|\bmt[-_]t"),
    ("rRNA", r"\brrna\b|ribosomal rna|\brrna[-_]|\b16s\b|\b18s\b|\b28s\b|\b5\.8s\b|\b5s\b"),
    ("lncRNA", r"lncrna|long non-?coding|lincrna|antisense rna|non-?coding rna|\bncrna\b"),
    ("ribozyme", r"ribozyme|riboswitch|\baptamer\b|\bires\b|hammerhead|pseudoknot"),
    ("SRP RNA", r"\bsrp rna\b|signal recognition particle"),
    ("vaultRNA/YRNA", r"vault rna|\byrna\b|\by rna\b"),
]
TYPE_RES = [(name, re.compile(pat, re.I)) for name, pat in TYPE_RULES]


def classify(definition: str) -> str:
    d = definition or ""
    for name, rx in TYPE_RES:
        if rx.search(d):
            return name
    return "其它"


def pct(sorted_vals: list[int], q: float) -> int:
    """线性插值分位数（与 numpy percentile 默认一致）。"""
    if not sorted_vals:
        return 0
    if len(sorted_vals) == 1:
        return sorted_vals[0]
    pos = (len(sorted_vals) - 1) * q
    lo, hi = int(pos), min(int(pos) + 1, len(sorted_vals) - 1)
    frac = pos - lo
    return int(round(sorted_vals[lo] + (sorted_vals[hi] - sorted_vals[lo]) * frac))


def dist(vals: list[int]) -> dict:
    s = sorted(vals)
    if not s:
        return {"n": 0, "min": 0, "p25": 0, "median": 0, "p75": 0, "p90": 0, "max": 0, "mean": 0}
    return {
        "n": len(s),
        "min": s[0],
        "p25": pct(s, 0.25),
        "median": pct(s, 0.50),
        "p75": pct(s, 0.75),
        "p90": pct(s, 0.90),
        "max": s[-1],
        "mean": round(statistics.fmean(s), 2),
    }


def load(path: Path) -> tuple[list[dict], int]:
    recs: list[dict] = []
    bad_lines = 0
    with path.open("r", encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            try:
                recs.append(json.loads(line))
            except json.JSONDecodeError:
                bad_lines += 1
    return recs, bad_lines


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--in", dest="inp", type=Path, required=True)
    ap.add_argument("--out", dest="outp", type=Path, required=True)
    ap.add_argument("--stats", default=None, help="10_collect.py 打印的 [stats] JSON 字符串")
    ap.add_argument("--stats-file", default=None,
                    help="同上，但从文件读（Windows 下传内联 JSON 会被 PowerShell 吞引号）")
    ap.add_argument("--n-sample", type=int, default=10)
    args = ap.parse_args()

    recs, bad_lines = load(args.inp)
    n = len(recs)
    critical: list[str] = []   # 硬失败：长度越界 / 缺 pmid / 缺 abstract / 重复
    warnings: list[str] = []   # 可用性告警：多样性、文献复用（不阻断管线，但影响训练）

    # ---------------- 1. 总条数 ----------------
    raw_stats = args.stats
    if not raw_stats and args.stats_file:
        # utf-8-sig：PowerShell 的 Set-Content -Encoding utf8 会写 BOM
        raw_stats = Path(args.stats_file).read_text(encoding="utf-8-sig").strip()
        # 允许直接粘 "[stats] {...}" 这一整行
        if raw_stats.startswith("[stats]"):
            raw_stats = raw_stats[len("[stats]"):].strip()
    stats = json.loads(raw_stats) if raw_stats else None
    if stats and stats.get("kept") is not None and stats["kept"] != n:
        critical.append(f"[stats].kept={stats['kept']} 与文件实际条数 {n} 不一致")

    # ---------------- 2. 长度 ----------------
    seqs = [r.get("sequence") or "" for r in recs]
    lens = [len(s) for s in seqs]
    lens_d = dist(lens)
    over = sorted(
        [{"uid": r.get("uid"), "nuc_accession": r.get("nuc_accession"), "len": len(r.get("sequence") or "")}
         for r in recs if len(r.get("sequence") or "") > MAX_NT],
        key=lambda d: -d["len"],
    )
    under = [r.get("uid") for r in recs if len(r.get("sequence") or "") < MIN_NT]
    if over:
        critical.append(f"有 {len(over)} 条序列长度 > {MAX_NT} nt（RNA-FM 会截断/报错）")
    if under:
        critical.append(f"有 {len(under)} 条序列长度 < {MIN_NT} nt")

    brackets = {
        "30-99": sum(1 for x in lens if 30 <= x <= 99),
        "100-249": sum(1 for x in lens if 100 <= x <= 249),
        "250-499": sum(1 for x in lens if 250 <= x <= 499),
        "500-749": sum(1 for x in lens if 500 <= x <= 749),
        "750-1022": sum(1 for x in lens if 750 <= x <= MAX_NT),
        ">1022": len(over),
    }

    # ---------------- 3. 字母表 ----------------
    residue_counter: Counter[str] = Counter()
    n_with_residue = 0
    residue_examples: list[dict] = []
    for r in recs:
        s = (r.get("sequence") or "").upper()
        residue = "".join(sorted({c for c in s if c not in VALID}))
        if residue:
            n_with_residue += 1
            for c in residue:
                residue_counter[c] += s.count(c)
            if len(residue_examples) < 15:
                residue_examples.append({"uid": r.get("uid"), "residue_chars": residue,
                                         "len": len(s)})
            unknown = sorted(set(residue) - IUPAC)
            if unknown:
                critical.append(f"{r.get('uid')} 含非 IUPAC 字符 {unknown}")
    alphabet_ok = n_with_residue == 0 or all(set(r["residue_chars"]) <= IUPAC for r in residue_examples)

    # ---------------- 4. 物种 / 类型 ----------------
    orgs = Counter((r.get("organism") or "<空>") for r in recs)
    types = Counter(classify(r.get("definition") or "") for r in recs)
    no_org = sum(1 for r in recs if not (r.get("organism") or "").strip())
    no_def = sum(1 for r in recs if not (r.get("definition") or "").strip())
    if no_org:
        critical.append(f"有 {no_org} 条缺 organism")
    if no_def:
        critical.append(f"有 {no_def} 条缺 definition")

    # ---------------- 5. 摘要长度 ----------------
    abs_lens = [len(r.get("literature") or "") for r in recs]
    abs_d = dist(abs_lens)
    short_abs = [r.get("uid") for r in recs if len(r.get("literature") or "") < MIN_ABSTRACT]

    # ---------------- 5b. 来源分布（后面 7c 要用） ----------------
    qname = Counter((r.get("provenance") or {}).get("query_name", "?") for r in recs)

    # 与其它 JSONL 的交叉核对（例如 50 条基线、20 条 probe）
    cross_checks = []
    for other in sorted((ROOT / "data" / "raw").glob("*.jsonl")):
        if other.resolve() == args.inp.resolve() or other.stat().st_size == 0:
            continue
        o_recs, _ = load(other)
        o_acc = {r.get("nuc_accession") for r in o_recs}
        cross_checks.append({
            "file": other.name,
            "n": len(o_recs),
            "shared_accessions": sum(1 for r in recs if r.get("nuc_accession") in o_acc),
            "shared_sequences": len({r.get("sequence") for r in recs}
                                    & {r.get("sequence") for r in o_recs}),
        })

    # ---------------- 6. 去重 ----------------
    def dups(key: str) -> dict:
        c = Counter((r.get(key) or "") for r in recs)
        d = {k: v for k, v in c.items() if v > 1}
        return {
            "unique": len(c),
            "n_dup_keys": len(d),
            "n_extra_rows": sum(v - 1 for v in d.values()),
            "top": sorted(({"key": k, "count": v} for k, v in d.items()), key=lambda x: -x["count"])[:10],
        }

    uid_d = dups("uid")
    acc_d = dups("nuc_accession")

    sha: dict[str, list[str]] = {}
    for r in recs:
        h = hashlib.sha1((r.get("sequence") or "").encode()).hexdigest()[:16]
        sha.setdefault(h, []).append(r.get("uid") or "")
    sha_dups = {h: u for h, u in sha.items() if len(u) > 1}
    if uid_d["n_dup_keys"]:
        critical.append(f"uid 有 {uid_d['n_dup_keys']} 个重复键（多出 {uid_d['n_extra_rows']} 行）")
    if acc_d["n_dup_keys"]:
        critical.append(f"nuc_accession 有 {acc_d['n_dup_keys']} 个重复（多出 {acc_d['n_extra_rows']} 行）")
    if sha_dups:
        critical.append(f"序列本身有 {sum(len(v) - 1 for v in sha_dups.values())} 条重复"
                        f"（{len(sha_dups)} 组相同序列）")

    # ---------------- 7. 必填字段 ----------------
    miss_pmid = [r.get("uid") for r in recs if not (r.get("pmid") or "").strip()]
    miss_abs = [r.get("uid") for r in recs if not (r.get("literature") or "").strip()]
    if miss_pmid:
        critical.append(f"有 {len(miss_pmid)} 条缺 pmid: {miss_pmid[:10]}")
    if miss_abs:
        critical.append(f"有 {len(miss_abs)} 条缺 abstract: {miss_abs[:10]}")
    # uid 里嵌的 pmid 也要和 pmid 字段一致
    uid_mismatch = [r.get("uid") for r in recs
                    if (r.get("uid") or "").split("|pmid:")[-1] != (r.get("pmid") or "")]
    if uid_mismatch:
        critical.append(f"有 {len(uid_mismatch)} 条 uid 内嵌 pmid 与 pmid 字段不一致")

    # ---------------- 7b. 文献复用（序列唯一 ≠ 文献唯一） ----------------
    pmid_c = Counter((r.get("pmid") or "") for r in recs)
    abs_c = Counter((r.get("literature") or "") for r in recs)
    lit_dup_rows = sum(v - 1 for v in abs_c.values() if v > 1)
    pm_dup_rows = sum(v - 1 for v in pmid_c.values() if v > 1)
    if n and lit_dup_rows / n > 0.20:
        warnings.append(
            f"文献复用严重：只有 {len(pmid_c)} 个不同 pmid / {len(abs_c)} 个不同摘要，"
            f"却有 {n} 条记录（{lit_dup_rows} 行为重复摘要，占 {lit_dup_rows / n:.0%}）——"
            "同一条 abstract 被配到大量不同序列上，AIS/摘要类监督信号会被稀释"
        )
    lit_titles = {}
    for r in recs:
        p = r.get("pmid") or ""
        if p not in lit_titles:
            pr = r.get("provenance") or {}
            lit_titles[p] = {"pmid": p, "count": pmid_c[p], "year": pr.get("year", ""),
                             "journal": pr.get("journal", ""), "title": (pr.get("title") or "")[:140]}
    lit_reuse_top = sorted(lit_titles.values(), key=lambda d: -d["count"])[:10]

    # ---------------- 7c. 多样性 ----------------
    acc_pref = Counter(re.match(r"[A-Za-z_]+", (r.get("nuc_accession") or "")).group(0)
                       for r in recs if r.get("nuc_accession"))
    n_orgs = len(orgs)
    top_org_share = (orgs.most_common(1)[0][1] / n) if n else 0
    if n and n_orgs <= 10:
        warnings.append(f"物种集中度高：仅 {n_orgs} 个物种，"
                        f"'{orgs.most_common(1)[0][0]}' 占 {top_org_share:.0%}")
    n_types = len(types)
    if n and types:
        top_type, top_type_n = types.most_common(1)[0]
        if top_type_n / n > 0.90:
            warnings.append(f"RNA 类型单一：'{top_type}' 占 {top_type_n / n:.0%}"
                            f"（全部只有 {n_types} 类）")
    if n and len(qname) < 3:
        warnings.append(f"只命中了 {len(qname)} 条 query（{dict(qname)}），"
                        "多路查询的多样性设计没有生效——数据被单一的 "
                        f"{qname.most_common(1)[0][0]} 查询塞满")
    for cc in cross_checks:
        if cc["shared_accessions"] == cc["n"] and cc["n"]:
            warnings.append(
                f"与 {cc['file']} 完全重叠：它的 {cc['n']} 条 accession 全部出现在本文件中"
                "（esearch 按 relevance 排序是确定性的，所以每次跑都从同一片记录开始）——"
                "如果后续要把几个 jsonl 合并训练，必须先按 accession 去重，否则这些行会被重复计入"
            )

    # ---------------- 8. 抽样 ----------------
    k = min(args.n_sample, n)
    step = max(n // k, 1)
    picks = recs[::step][:k]
    samples = [
        {
            "uid": r.get("uid"),
            "seq_len": len(r.get("sequence") or ""),
            "organism": r.get("organism"),
            "rna_type": classify(r.get("definition") or ""),
            "definition": (r.get("definition") or "")[:200],
            "abstract_head150": re.sub(r"\s+", " ", (r.get("literature") or ""))[:150],
            "pmid": r.get("pmid"),
            "journal": (r.get("provenance") or {}).get("journal", ""),
            "year": (r.get("provenance") or {}).get("year", ""),
        }
        for r in picks
    ]

    qname_report = qname
    file_bytes = args.inp.stat().st_size

    # 与其它 JSONL 的交叉核对（例如 50 条基线、20 条 probe）
    cross_checks = []
    for other in (ROOT / "data" / "raw").glob("*.jsonl"):
        if other.resolve() == args.inp.resolve() or other.stat().st_size == 0:
            continue
        o_recs, _ = load(other)
        o_acc = {r.get("nuc_accession") for r in o_recs}
        overlap = sum(1 for r in recs if r.get("nuc_accession") in o_acc)
        dup_seqs = len({r.get("sequence") for r in recs} & {r.get("sequence") for r in o_recs})
        cross_checks.append({
            "file": other.name, "n": len(o_recs),
            "shared_accessions": overlap, "shared_sequences": dup_seqs,
        })

    report_notes = [
        "10_collect.py 的逻辑**未改动**；本次只新增了只读的报告脚本 scripts/14_quality_report.py。",
        "注意 10_collect.py 顶部注释写的是 30–1024 nt，而实际常量 MIN_NT, MAX_NT = 30, 1022 —— "
        "注释与代码不一致（1022 = 探针实测的 RNA-FM 上限）。本次按任务要求保持 1022，未改注释。",
        "MAX_NT 之外还有一层 esearch 服务端闸门：query 里写的是 60:1000[slen]，"
        "所以实际落盘长度天然 <= 1000，1022 闸门是双保险而非唯一约束。",
        "所有查询都命中了 srcdb_refseq + biomol_rna，因此 1000 条全部是 RefSeq NR_* 记录。",
        "10_collect.py 把记录攒在内存里、结束时一次性写盘，所以采集过程中 jsonl 文件不存在 —— "
        "想看进度只能读进程 stdout（本次是后台任务日志）。",
    ]

    report = {
        "generated": time.strftime("%Y-%m-%d %H:%M:%S"),
        "source_file": str(args.inp),
        "source_bytes": file_bytes,
        "source_mb": round(file_bytes / 1024 / 1024, 2),
        "pipeline": "esearch nuccore -> efetch nuccore rettype=gb -> efetch pubmed",
        "gates": {"MIN_NT": MIN_NT, "MAX_NT": MAX_NT, "MIN_ABSTRACT": MIN_ABSTRACT},

        "verdict": "PASS" if not critical else "FAIL",
        "critical_issues": critical,
        "warnings": warnings,
        "hard_gates_ok": not critical,
        "downstream_usability": "OK" if not warnings else "DEGRADED",
        "notes": report_notes,
        "cross_checks": cross_checks,

        "counts": {
            "total_records": n,
            "bad_jsonl_lines": bad_lines,
            "collect_stats": stats,
            "by_query": dict(qname_report),
        },

        "length_distribution": {
            **lens_d,
            "n_over_max_nt": len(over),
            "max_nt_gate": MAX_NT,
            "over_max_examples": over[:10],
            "under_min_examples": under[:10],
            "buckets": brackets,
        },

        "alphabet": {
            "allowed_core": sorted(VALID),
            "allowed_iupac": sorted(IUPAC),
            "records_with_residue": n_with_residue,
            "residue_char_counts": dict(residue_counter),
            "residue_is_empty_or_degenerate_only": alphabet_ok,
            "examples": residue_examples,
        },

        "organisms_top10": [{"organism": k, "count": v} for k, v in orgs.most_common(10)],
        "n_distinct_organisms": len(orgs),
        "rna_type_distribution": dict(types.most_common()),
        "n_distinct_types": len(types),
        "accession_prefixes": dict(acc_pref.most_common()),

        "literature_reuse": {
            "n_distinct_pmids": len(pmid_c),
            "n_distinct_abstracts": len(abs_c),
            "n_rows_with_reused_abstract": lit_dup_rows,
            "reuse_ratio": round(lit_dup_rows / n, 4) if n else 0,
            "top_reused": lit_reuse_top,
        },

        "abstract_length_distribution": abs_d,
        "n_abstract_under_min": len(short_abs),

        "duplicates": {
            "uid": uid_d,
            "nuc_accession": acc_d,
            "sequence_sha1": {
                "unique_sequences": len(sha),
                "n_dup_groups": len(sha_dups),
                "n_extra_duplicate_rows": sum(len(v) - 1 for v in sha_dups.values()),
                "groups": [{"sha1_16": h, "count": len(u), "uids": u[:6]} for h, u in list(sha_dups.items())[:10]],
            },
        },

        "required_fields": {
            "n_missing_pmid": len(miss_pmid),
            "n_missing_abstract": len(miss_abs),
            "n_uid_pmid_mismatch": len(uid_mismatch),
            "examples_missing": (miss_pmid + miss_abs)[:10],
            "ok": not (miss_pmid or miss_abs or uid_mismatch),
        },

        "samples": samples,
    }

    args.outp.parent.mkdir(parents=True, exist_ok=True)
    args.outp.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")

    # 控制台摘要
    print(f"[report] {args.outp}  verdict={report['verdict']}")
    print(f"  total={n}  bytes={file_bytes}  bad_lines={bad_lines}")
    print(f"  len: min={lens_d['min']} p25={lens_d['p25']} med={lens_d['median']} "
          f"p75={lens_d['p75']} p90={lens_d['p90']} max={lens_d['max']}  >{MAX_NT}={len(over)}")
    print(f"  alphabet residue records={n_with_residue} chars={dict(residue_counter)}")
    print(f"  organisms={len(orgs)}  types={dict(types.most_common())}")
    print(f"  abstract: min={abs_d['min']} med={abs_d['median']} max={abs_d['max']}")
    print(f"  dup: uid={uid_d['n_dup_keys']} acc={acc_d['n_dup_keys']} seq={len(sha_dups)}")
    print(f"  missing pmid={len(miss_pmid)} abstract={len(miss_abs)}")
    print(f"  literature: distinct pmid={len(pmid_c)} distinct abstract={len(abs_c)} "
          f"reused_rows={lit_dup_rows} ({lit_dup_rows / n:.0%})")
    print(f"  diversity: organisms={len(orgs)} types={len(types)} queries={len(qname)}")
    print("  --- 抽样 ---")
    for s in samples:
        print(f"   {s['uid']}  len={s['seq_len']:4d}  {s['rna_type']}")
        print(f"     DEF: {s['definition'][:100]}")
        print(f"     ABS: {s['abstract_head150'][:110]}")
    if critical:
        print("  !! CRITICAL:")
        for c in critical:
            print("   -", c)
    if warnings:
        print("  ~ warnings:")
        for w in warnings:
            print("   -", w)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

	"""P0 数据质量报告：对 10_collect.py 产出的 JSONL 做全面体检。

	只读，不改数据。用法：

	python scripts/14_quality_report.py \
	--in data/raw/rna_literature_1k.jsonl \
	--out reports/collect_1k_report.json \
	--stats '{"nt_scanned":...}' # 可选：把 10_collect.py 打印的 [stats] 原样传进来

	检查项
	------
	1. 总条数 + [stats] 全计数
	2. 序列长度分布（min/p25/median/p75/p90/max）+ >MAX_NT 的条数（必须 0）
	3. 字母表：去掉 ACGU 之后剩余字符集合（必须为空集或极少数简并碱基）
	4. 物种分布 top10 / RNA 类型分布（DEFINITION 关键词粗分）
	5. 摘要长度分布
	6. 去重：uid 唯一、accession 唯一、序列本身 sha1 唯一
	7. 每条必须有非空 pmid 和 abstract
	8. 抽样 10 条

	任何违规项会进入 `critical_issues`（顶层），方便上层直接判红。
	"""

	from __future__ import annotations

	import argparse
	import hashlib
	import json
	import re
	import statistics
	import sys
	import time
	from collections import Counter
	from pathlib import Path

	ROOT = Path(__file__).resolve().parents[1]
	MIN_NT, MAX_NT = 30, 1022 # 与 10_collect.py 保持一致（1022 = RNA-FM 实测上限）
	MIN_ABSTRACT = 200

	# IUPAC 简并碱基（去掉 ACGU 之后"合法"的残留）
	IUPAC = set("NRYWSKMBDHV")
	VALID = set("ACGU")

	# DEFINITION 关键词 → RNA 类型。顺序即优先级（先命中先归类）。
	TYPE_RULES: list[tuple[str, str]] = [
	("circRNA", r"circular rna\|\bcircrna\b\|circ_?rna"),
	("miRNA", r"\bmirna\b\|microrna\|\bmiR-\|\blet-7\b\|\bmir-\d"),
	("snoRNA", r"\bsnorna\b\|small nucleolar"),
	("snRNA", r"\bsnrna\b\|small nuclear rna"),
	("tRNA", r"\btrna\b\|transfer rna\|\btrna-\|\bmt[-_]t"),
	("rRNA", r"\brrna\b\|ribosomal rna\|\brrna[-_]\|\b16s\b\|\b18s\b\|\b28s\b\|\b5\.8s\b\|\b5s\b"),
	("lncRNA", r"lncrna\|long non-?coding\|lincrna\|antisense rna\|non-?coding rna\|\bncrna\b"),
	("ribozyme", r"ribozyme\|riboswitch\|\baptamer\b\|\bires\b\|hammerhead\|pseudoknot"),
	("SRP RNA", r"\bsrp rna\b\|signal recognition particle"),
	("vaultRNA/YRNA", r"vault rna\|\byrna\b\|\by rna\b"),
	]
	TYPE_RES = [(name, re.compile(pat, re.I)) for name, pat in TYPE_RULES]


	def classify(definition: str) -> str:
	d = definition or ""
	for name, rx in TYPE_RES:
	if rx.search(d):
	return name
	return "其它"


	def pct(sorted_vals: list[int], q: float) -> int:
	"""线性插值分位数（与 numpy percentile 默认一致）。"""
	if not sorted_vals:
	return 0
	if len(sorted_vals) == 1:
	return sorted_vals[0]
	pos = (len(sorted_vals) - 1) * q
	lo, hi = int(pos), min(int(pos) + 1, len(sorted_vals) - 1)
	frac = pos - lo
	return int(round(sorted_vals[lo] + (sorted_vals[hi] - sorted_vals[lo]) * frac))


	def dist(vals: list[int]) -> dict:
	s = sorted(vals)
	if not s:
	return {"n": 0, "min": 0, "p25": 0, "median": 0, "p75": 0, "p90": 0, "max": 0, "mean": 0}
	return {
	"n": len(s),
	"min": s[0],
	"p25": pct(s, 0.25),
	"median": pct(s, 0.50),
	"p75": pct(s, 0.75),
	"p90": pct(s, 0.90),
	"max": s[-1],
	"mean": round(statistics.fmean(s), 2),
	}


	def load(path: Path) -> tuple[list[dict], int]:
	recs: list[dict] = []
	bad_lines = 0
	with path.open("r", encoding="utf-8") as fh:
	for line in fh:
	line = line.strip()
	if not line:
	continue
	try:
	recs.append(json.loads(line))
	except json.JSONDecodeError:
	bad_lines += 1
	return recs, bad_lines


	def main() -> int:
	ap = argparse.ArgumentParser()
	ap.add_argument("--in", dest="inp", type=Path, required=True)
	ap.add_argument("--out", dest="outp", type=Path, required=True)
	ap.add_argument("--stats", default=None, help="10_collect.py 打印的 [stats] JSON 字符串")
	ap.add_argument("--stats-file", default=None,
	help="同上，但从文件读（Windows 下传内联 JSON 会被 PowerShell 吞引号）")
	ap.add_argument("--n-sample", type=int, default=10)
	args = ap.parse_args()

	recs, bad_lines = load(args.inp)
	n = len(recs)
	critical: list[str] = [] # 硬失败：长度越界 / 缺 pmid / 缺 abstract / 重复
	warnings: list[str] = [] # 可用性告警：多样性、文献复用（不阻断管线，但影响训练）

	# ---------------- 1. 总条数 ----------------
	raw_stats = args.stats
	if not raw_stats and args.stats_file:
	# utf-8-sig：PowerShell 的 Set-Content -Encoding utf8 会写 BOM
	raw_stats = Path(args.stats_file).read_text(encoding="utf-8-sig").strip()
	# 允许直接粘 "[stats] {...}" 这一整行
	if raw_stats.startswith("[stats]"):
	raw_stats = raw_stats[len("[stats]"):].strip()
	stats = json.loads(raw_stats) if raw_stats else None
	if stats and stats.get("kept") is not None and stats["kept"] != n:
	critical.append(f"[stats].kept={stats['kept']} 与文件实际条数 {n} 不一致")

	# ---------------- 2. 长度 ----------------
	seqs = [r.get("sequence") or "" for r in recs]
	lens = [len(s) for s in seqs]
	lens_d = dist(lens)
	over = sorted(
	[{"uid": r.get("uid"), "nuc_accession": r.get("nuc_accession"), "len": len(r.get("sequence") or "")}
	for r in recs if len(r.get("sequence") or "") > MAX_NT],
	key=lambda d: -d["len"],
	)
	under = [r.get("uid") for r in recs if len(r.get("sequence") or "") < MIN_NT]
	if over:
	critical.append(f"有 {len(over)} 条序列长度 > {MAX_NT} nt（RNA-FM 会截断/报错）")
	if under:
	critical.append(f"有 {len(under)} 条序列长度 < {MIN_NT} nt")

	brackets = {
	"30-99": sum(1 for x in lens if 30 <= x <= 99),
	"100-249": sum(1 for x in lens if 100 <= x <= 249),
	"250-499": sum(1 for x in lens if 250 <= x <= 499),
	"500-749": sum(1 for x in lens if 500 <= x <= 749),
	"750-1022": sum(1 for x in lens if 750 <= x <= MAX_NT),
	">1022": len(over),
	}

	# ---------------- 3. 字母表 ----------------
	residue_counter: Counter[str] = Counter()
	n_with_residue = 0
	residue_examples: list[dict] = []
	for r in recs:
	s = (r.get("sequence") or "").upper()
	residue = "".join(sorted({c for c in s if c not in VALID}))
	if residue:
	n_with_residue += 1
	for c in residue:
	residue_counter[c] += s.count(c)
	if len(residue_examples) < 15:
	residue_examples.append({"uid": r.get("uid"), "residue_chars": residue,
	"len": len(s)})
	unknown = sorted(set(residue) - IUPAC)
	if unknown:
	critical.append(f"{r.get('uid')} 含非 IUPAC 字符 {unknown}")
	alphabet_ok = n_with_residue == 0 or all(set(r["residue_chars"]) <= IUPAC for r in residue_examples)

	# ---------------- 4. 物种 / 类型 ----------------
	orgs = Counter((r.get("organism") or "<空>") for r in recs)
	types = Counter(classify(r.get("definition") or "") for r in recs)
	no_org = sum(1 for r in recs if not (r.get("organism") or "").strip())
	no_def = sum(1 for r in recs if not (r.get("definition") or "").strip())
	if no_org:
	critical.append(f"有 {no_org} 条缺 organism")
	if no_def:
	critical.append(f"有 {no_def} 条缺 definition")

	# ---------------- 5. 摘要长度 ----------------
	abs_lens = [len(r.get("literature") or "") for r in recs]
	abs_d = dist(abs_lens)
	short_abs = [r.get("uid") for r in recs if len(r.get("literature") or "") < MIN_ABSTRACT]

	# ---------------- 5b. 来源分布（后面 7c 要用） ----------------
	qname = Counter((r.get("provenance") or {}).get("query_name", "?") for r in recs)

	# 与其它 JSONL 的交叉核对（例如 50 条基线、20 条 probe）
	cross_checks = []
	for other in sorted((ROOT / "data" / "raw").glob("*.jsonl")):
	if other.resolve() == args.inp.resolve() or other.stat().st_size == 0:
	continue
	o_recs, _ = load(other)
	o_acc = {r.get("nuc_accession") for r in o_recs}
	cross_checks.append({
	"file": other.name,
	"n": len(o_recs),
	"shared_accessions": sum(1 for r in recs if r.get("nuc_accession") in o_acc),
	"shared_sequences": len({r.get("sequence") for r in recs}
	& {r.get("sequence") for r in o_recs}),
	})

	# ---------------- 6. 去重 ----------------
	def dups(key: str) -> dict:
	c = Counter((r.get(key) or "") for r in recs)
	d = {k: v for k, v in c.items() if v > 1}
	return {
	"unique": len(c),
	"n_dup_keys": len(d),
	"n_extra_rows": sum(v - 1 for v in d.values()),
	"top": sorted(({"key": k, "count": v} for k, v in d.items()), key=lambda x: -x["count"])[:10],
	}

	uid_d = dups("uid")
	acc_d = dups("nuc_accession")

	sha: dict[str, list[str]] = {}
	for r in recs:
	h = hashlib.sha1((r.get("sequence") or "").encode()).hexdigest()[:16]
	sha.setdefault(h, []).append(r.get("uid") or "")
	sha_dups = {h: u for h, u in sha.items() if len(u) > 1}
	if uid_d["n_dup_keys"]:
	critical.append(f"uid 有 {uid_d['n_dup_keys']} 个重复键（多出 {uid_d['n_extra_rows']} 行）")
	if acc_d["n_dup_keys"]:
	critical.append(f"nuc_accession 有 {acc_d['n_dup_keys']} 个重复（多出 {acc_d['n_extra_rows']} 行）")
	if sha_dups:
	critical.append(f"序列本身有 {sum(len(v) - 1 for v in sha_dups.values())} 条重复"
	f"（{len(sha_dups)} 组相同序列）")

	# ---------------- 7. 必填字段 ----------------
	miss_pmid = [r.get("uid") for r in recs if not (r.get("pmid") or "").strip()]
	miss_abs = [r.get("uid") for r in recs if not (r.get("literature") or "").strip()]
	if miss_pmid:
	critical.append(f"有 {len(miss_pmid)} 条缺 pmid: {miss_pmid[:10]}")
	if miss_abs:
	critical.append(f"有 {len(miss_abs)} 条缺 abstract: {miss_abs[:10]}")
	# uid 里嵌的 pmid 也要和 pmid 字段一致
	uid_mismatch = [r.get("uid") for r in recs
	if (r.get("uid") or "").split("\|pmid:")[-1] != (r.get("pmid") or "")]
	if uid_mismatch:
	critical.append(f"有 {len(uid_mismatch)} 条 uid 内嵌 pmid 与 pmid 字段不一致")

	# ---------------- 7b. 文献复用（序列唯一 ≠ 文献唯一） ----------------
	pmid_c = Counter((r.get("pmid") or "") for r in recs)
	abs_c = Counter((r.get("literature") or "") for r in recs)
	lit_dup_rows = sum(v - 1 for v in abs_c.values() if v > 1)
	pm_dup_rows = sum(v - 1 for v in pmid_c.values() if v > 1)
	if n and lit_dup_rows / n > 0.20:
	warnings.append(
	f"文献复用严重：只有 {len(pmid_c)} 个不同 pmid / {len(abs_c)} 个不同摘要，"
	f"却有 {n} 条记录（{lit_dup_rows} 行为重复摘要，占 {lit_dup_rows / n:.0%}）——"
	"同一条 abstract 被配到大量不同序列上，AIS/摘要类监督信号会被稀释"
	)
	lit_titles = {}
	for r in recs:
	p = r.get("pmid") or ""
	if p not in lit_titles:
	pr = r.get("provenance") or {}
	lit_titles[p] = {"pmid": p, "count": pmid_c[p], "year": pr.get("year", ""),
	"journal": pr.get("journal", ""), "title": (pr.get("title") or "")[:140]}
	lit_reuse_top = sorted(lit_titles.values(), key=lambda d: -d["count"])[:10]

	# ---------------- 7c. 多样性 ----------------
	acc_pref = Counter(re.match(r"[A-Za-z_]+", (r.get("nuc_accession") or "")).group(0)
	for r in recs if r.get("nuc_accession"))
	n_orgs = len(orgs)
	top_org_share = (orgs.most_common(1)[0][1] / n) if n else 0
	if n and n_orgs <= 10:
	warnings.append(f"物种集中度高：仅 {n_orgs} 个物种，"
	f"'{orgs.most_common(1)[0][0]}' 占 {top_org_share:.0%}")
	n_types = len(types)
	if n and types:
	top_type, top_type_n = types.most_common(1)[0]
	if top_type_n / n > 0.90:
	warnings.append(f"RNA 类型单一：'{top_type}' 占 {top_type_n / n:.0%}"
	f"（全部只有 {n_types} 类）")
	if n and len(qname) < 3:
	warnings.append(f"只命中了 {len(qname)} 条 query（{dict(qname)}），"
	"多路查询的多样性设计没有生效——数据被单一的 "
	f"{qname.most_common(1)[0][0]} 查询塞满")
	for cc in cross_checks:
	if cc["shared_accessions"] == cc["n"] and cc["n"]:
	warnings.append(
	f"与 {cc['file']} 完全重叠：它的 {cc['n']} 条 accession 全部出现在本文件中"
	"（esearch 按 relevance 排序是确定性的，所以每次跑都从同一片记录开始）——"
	"如果后续要把几个 jsonl 合并训练，必须先按 accession 去重，否则这些行会被重复计入"
	)

	# ---------------- 8. 抽样 ----------------
	k = min(args.n_sample, n)
	step = max(n // k, 1)
	picks = recs[::step][:k]
	samples = [
	{
	"uid": r.get("uid"),
	"seq_len": len(r.get("sequence") or ""),
	"organism": r.get("organism"),
	"rna_type": classify(r.get("definition") or ""),
	"definition": (r.get("definition") or "")[:200],
	"abstract_head150": re.sub(r"\s+", " ", (r.get("literature") or ""))[:150],
	"pmid": r.get("pmid"),
	"journal": (r.get("provenance") or {}).get("journal", ""),
	"year": (r.get("provenance") or {}).get("year", ""),
	}
	for r in picks
	]

	qname_report = qname
	file_bytes = args.inp.stat().st_size

	# 与其它 JSONL 的交叉核对（例如 50 条基线、20 条 probe）
	cross_checks = []
	for other in (ROOT / "data" / "raw").glob("*.jsonl"):
	if other.resolve() == args.inp.resolve() or other.stat().st_size == 0:
	continue
	o_recs, _ = load(other)
	o_acc = {r.get("nuc_accession") for r in o_recs}
	overlap = sum(1 for r in recs if r.get("nuc_accession") in o_acc)
	dup_seqs = len({r.get("sequence") for r in recs} & {r.get("sequence") for r in o_recs})
	cross_checks.append({
	"file": other.name, "n": len(o_recs),
	"shared_accessions": overlap, "shared_sequences": dup_seqs,
	})

	report_notes = [
	"10_collect.py 的逻辑未改动；本次只新增了只读的报告脚本 scripts/14_quality_report.py。",
	"注意 10_collect.py 顶部注释写的是 30–1024 nt，而实际常量 MIN_NT, MAX_NT = 30, 1022 —— "
	"注释与代码不一致（1022 = 探针实测的 RNA-FM 上限）。本次按任务要求保持 1022，未改注释。",
	"MAX_NT 之外还有一层 esearch 服务端闸门：query 里写的是 60:1000[slen]，"
	"所以实际落盘长度天然 <= 1000，1022 闸门是双保险而非唯一约束。",
	"所有查询都命中了 srcdb_refseq + biomol_rna，因此 1000 条全部是 RefSeq NR_* 记录。",
	"10_collect.py 把记录攒在内存里、结束时一次性写盘，所以采集过程中 jsonl 文件不存在 —— "
	"想看进度只能读进程 stdout（本次是后台任务日志）。",
	]

	report = {
	"generated": time.strftime("%Y-%m-%d %H:%M:%S"),
	"source_file": str(args.inp),
	"source_bytes": file_bytes,
	"source_mb": round(file_bytes / 1024 / 1024, 2),
	"pipeline": "esearch nuccore -> efetch nuccore rettype=gb -> efetch pubmed",
	"gates": {"MIN_NT": MIN_NT, "MAX_NT": MAX_NT, "MIN_ABSTRACT": MIN_ABSTRACT},

	"verdict": "PASS" if not critical else "FAIL",
	"critical_issues": critical,
	"warnings": warnings,
	"hard_gates_ok": not critical,
	"downstream_usability": "OK" if not warnings else "DEGRADED",
	"notes": report_notes,
	"cross_checks": cross_checks,

	"counts": {
	"total_records": n,
	"bad_jsonl_lines": bad_lines,
	"collect_stats": stats,
	"by_query": dict(qname_report),
	},

	"length_distribution": {
	**lens_d,
	"n_over_max_nt": len(over),
	"max_nt_gate": MAX_NT,
	"over_max_examples": over[:10],
	"under_min_examples": under[:10],
	"buckets": brackets,
	},

	"alphabet": {
	"allowed_core": sorted(VALID),
	"allowed_iupac": sorted(IUPAC),
	"records_with_residue": n_with_residue,
	"residue_char_counts": dict(residue_counter),
	"residue_is_empty_or_degenerate_only": alphabet_ok,
	"examples": residue_examples,
	},

	"organisms_top10": [{"organism": k, "count": v} for k, v in orgs.most_common(10)],
	"n_distinct_organisms": len(orgs),
	"rna_type_distribution": dict(types.most_common()),
	"n_distinct_types": len(types),
	"accession_prefixes": dict(acc_pref.most_common()),

	"literature_reuse": {
	"n_distinct_pmids": len(pmid_c),
	"n_distinct_abstracts": len(abs_c),
	"n_rows_with_reused_abstract": lit_dup_rows,
	"reuse_ratio": round(lit_dup_rows / n, 4) if n else 0,
	"top_reused": lit_reuse_top,
	},

	"abstract_length_distribution": abs_d,
	"n_abstract_under_min": len(short_abs),

	"duplicates": {
	"uid": uid_d,
	"nuc_accession": acc_d,
	"sequence_sha1": {
	"unique_sequences": len(sha),
	"n_dup_groups": len(sha_dups),
	"n_extra_duplicate_rows": sum(len(v) - 1 for v in sha_dups.values()),
	"groups": [{"sha1_16": h, "count": len(u), "uids": u[:6]} for h, u in list(sha_dups.items())[:10]],
	},
	},

	"required_fields": {
	"n_missing_pmid": len(miss_pmid),
	"n_missing_abstract": len(miss_abs),
	"n_uid_pmid_mismatch": len(uid_mismatch),
	"examples_missing": (miss_pmid + miss_abs)[:10],
	"ok": not (miss_pmid or miss_abs or uid_mismatch),
	},

	"samples": samples,
	}

	args.outp.parent.mkdir(parents=True, exist_ok=True)
	args.outp.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")

	# 控制台摘要
	print(f"[report] {args.outp} verdict={report['verdict']}")
	print(f" total={n} bytes={file_bytes} bad_lines={bad_lines}")
	print(f" len: min={lens_d['min']} p25={lens_d['p25']} med={lens_d['median']} "
	f"p75={lens_d['p75']} p90={lens_d['p90']} max={lens_d['max']} >{MAX_NT}={len(over)}")
	print(f" alphabet residue records={n_with_residue} chars={dict(residue_counter)}")
	print(f" organisms={len(orgs)} types={dict(types.most_common())}")
	print(f" abstract: min={abs_d['min']} med={abs_d['median']} max={abs_d['max']}")
	print(f" dup: uid={uid_d['n_dup_keys']} acc={acc_d['n_dup_keys']} seq={len(sha_dups)}")
	print(f" missing pmid={len(miss_pmid)} abstract={len(miss_abs)}")
	print(f" literature: distinct pmid={len(pmid_c)} distinct abstract={len(abs_c)} "
	f"reused_rows={lit_dup_rows} ({lit_dup_rows / n:.0%})")
	print(f" diversity: organisms={len(orgs)} types={len(types)} queries={len(qname)}")
	print(" --- 抽样 ---")
	for s in samples:
	print(f" {s['uid']} len={s['seq_len']:4d} {s['rna_type']}")
	print(f" DEF: {s['definition'][:100]}")
	print(f" ABS: {s['abstract_head150'][:110]}")
	if critical:
	print(" !! CRITICAL:")
	for c in critical:
	print(" -", c)
	if warnings:
	print(" ~ warnings:")
	for w in warnings:
	print(" -", w)
	return 0


	if __name__ == "__main__":
	raise SystemExit(main())