Paper Select Journal

skill_info:
  name: "paper-select-journal"
  version: "0.3.1"
  description: "根据 manuscript 与用户需求筛选合适的 SCI 投稿期刊，并输出带证据的排序报告。"
  author: "Bensz Conan"
  category: "academic-writing"

workspace:
  hidden_dir: ".bensz-api/skills/paper-select-journal"
  run_prefix: ""
  timestamp_format: "%Y-%m-%d-%H-%M"
  latest_run_pointer: "latest-run.txt"
  keep_intermediates_inside_hidden_dir: true
  subdirs:
    - "input"
    - "output"
    - "log"
    - "inputs"
    - "analysis"
    - "candidates"
    - "pubmed"
    - "reports"
    - "logs"

directories:
  plans: "plans"
  tests: "tests"
  default_test_dir: "tests/paper-select-journal"

assets:
  catalog_xlsx: "assets/journal_catalog/2023IF.xlsx"
  source_catalog_xlsx: "/Volumes/2T01/winE/RCloud/RFactory/easySCI/raw/2023IF.xlsx"

screening:
  min_impact_factor: 3.0
  default_set1_limit: 40
  default_set2_limit: 20
  default_set3_limit: 10
  manual_exception_key: "manual_exceptions"
  require_quality_screening_after_set1: true
  prefer_verified_cas_data_in_final: true
  open_access_gold_threshold: 40.0

pubmed:
  lookback_days: 90
  max_results_per_journal: 12
  top_articles_per_journal: 5
  retmax: 20
  timeout_seconds: 30
  delay_seconds_between_requests: 0.34
  esearch_url: "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi"
  efetch_url: "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/efetch.fcgi"
  tool_name: "paper-select-journal"

reports:
  run_manifest: "run-manifest.json"
  manuscript_profile: "analysis/manuscript_profile.json"
  set1_json: "candidates/set1_candidates.json"
  set1_markdown: "candidates/set1_candidates.md"
  scope_review: "analysis/set2_scope_review.json"
  pubmed_json: "pubmed/recent_articles.json"
  pubmed_markdown: "pubmed/recent_articles.md"
  set3_similarity_review: "analysis/set3_similarity_review.json"
  final_json: "analysis/final_recommendations.json"
  final_report: "reports/paper-select-journal-report.md"

scripts:
  init_workspace: "scripts/init_workspace.py"
  shortlist_journals: "scripts/shortlist_journals.py"
  fetch_pubmed_recent: "scripts/fetch_pubmed_recent.py"
  render_report: "scripts/render_report.py"

from __future__ import annotations

import datetime as dt
import json
import re
from pathlib import Path
from typing import Any

import yaml


def skill_root() -> Path:
    return Path(__file__).resolve().parents[1]


def load_config(root: Path | None = None) -> dict[str, Any]:
    config_path = (root or skill_root()) / "config.yaml"
    return yaml.safe_load(config_path.read_text(encoding="utf-8"))


def ensure_dir(path: Path) -> Path:
    path.mkdir(parents=True, exist_ok=True)
    return path


def write_json(path: Path, payload: Any) -> None:
    ensure_dir(path.parent)
    path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=False) + "\n",
        encoding="utf-8",
    )


def read_json(path: Path) -> Any:
    return json.loads(path.read_text(encoding="utf-8"))


def write_text(path: Path, text: str) -> None:
    ensure_dir(path.parent)
    path.write_text(text, encoding="utf-8")


def resolve_path(path_str: str, base: Path | None = None) -> Path:
    path = Path(path_str).expanduser()
    if not path.is_absolute():
        path = (base or Path.cwd()) / path
    return path.resolve()


def is_within(parent: Path, child: Path) -> bool:
    parent_resolved = parent.resolve()
    child_resolved = child.resolve()
    return child_resolved == parent_resolved or parent_resolved in child_resolved.parents


def require_within(parent: Path, child: Path, label: str) -> Path:
    if not is_within(parent, child):
        raise ValueError(f"{label} 必须位于工作区内: {child}")
    return child


def resolve_path_within(path_str: str, *, parent: Path, label: str, base: Path | None = None) -> Path:
    return require_within(parent.resolve(), resolve_path(path_str, base=base), label)


def workspace_output_path(workspace: Path, relative_path: str) -> Path:
    return require_within(workspace.resolve(), (workspace / relative_path).resolve(), "输出路径")


def generate_run_id(config: dict[str, Any], now: dt.datetime | None = None) -> str:
    workspace_cfg = config["workspace"]
    stamp = (now or dt.datetime.now()).strftime(workspace_cfg["timestamp_format"])
    return f"{workspace_cfg['run_prefix']}{stamp}"


def allocate_unique_run_id(workspace_base: Path, run_id: str) -> str:
    if not (workspace_base / run_id).exists():
        return run_id
    for idx in range(2, 100):
        candidate = f"{run_id}-{idx:02d}"
        if not (workspace_base / candidate).exists():
            return candidate
    raise ValueError(f"无法在 {workspace_base} 下分配唯一工作目录: {run_id}")


def normalize_text(text: str | None) -> str:
    if not text:
        return ""
    collapsed = re.sub(r"[^a-z0-9]+", " ", text.lower())
    return re.sub(r"\s+", " ", collapsed).strip()


def float_or_none(value: Any) -> float | None:
    if value in (None, ""):
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def quartile_from_category(category: str | None) -> str:
    text = (category or "").upper()
    for quartile in ("Q1", "Q2", "Q3", "Q4"):
        if quartile in text:
            return quartile
    return "unknown"


def slugify(text: str, max_length: int = 80) -> str:
    slug = normalize_text(text).replace(" ", "-")
    slug = re.sub(r"-{2,}", "-", slug).strip("-")
    return slug[:max_length] or "item"


def compact_text(text: str | None, max_chars: int = 220) -> str:
    cleaned = re.sub(r"\s+", " ", (text or "")).strip()
    if len(cleaned) <= max_chars:
        return cleaned
    return cleaned[: max_chars - 1].rstrip() + "…"

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import datetime as dt
import json
import time
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from typing import Any

from common import (
    compact_text,
    load_config,
    read_json,
    resolve_path,
    resolve_path_within,
    workspace_output_path,
    write_json,
    write_text,
)


MONTH_MAP = {
    "1": "01",
    "01": "01",
    "jan": "01",
    "january": "01",
    "2": "02",
    "02": "02",
    "feb": "02",
    "february": "02",
    "3": "03",
    "03": "03",
    "mar": "03",
    "march": "03",
    "4": "04",
    "04": "04",
    "apr": "04",
    "april": "04",
    "5": "05",
    "05": "05",
    "may": "05",
    "6": "06",
    "06": "06",
    "jun": "06",
    "june": "06",
    "7": "07",
    "07": "07",
    "jul": "07",
    "july": "07",
    "8": "08",
    "08": "08",
    "aug": "08",
    "august": "08",
    "9": "09",
    "09": "09",
    "sep": "09",
    "sept": "09",
    "september": "09",
    "10": "10",
    "oct": "10",
    "october": "10",
    "11": "11",
    "nov": "11",
    "november": "11",
    "12": "12",
    "dec": "12",
    "december": "12",
}


def build_esearch_url(config: dict[str, Any], journal_name: str, start_date: dt.date, end_date: dt.date) -> str:
    params = {
        "db": "pubmed",
        "retmode": "json",
        "retmax": str(config["pubmed"]["retmax"]),
        "sort": "pub_date",
        "tool": config["pubmed"]["tool_name"],
        "term": f'"{journal_name}"[Journal] AND ("{start_date:%Y/%m/%d}"[Date - Publication] : "{end_date:%Y/%m/%d}"[Date - Publication])',
    }
    return config["pubmed"]["esearch_url"] + "?" + urllib.parse.urlencode(params)


def build_efetch_url(config: dict[str, Any], ids: list[str]) -> str:
    params = {
        "db": "pubmed",
        "retmode": "xml",
        "tool": config["pubmed"]["tool_name"],
        "id": ",".join(ids),
    }
    return config["pubmed"]["efetch_url"] + "?" + urllib.parse.urlencode(params)


def http_get(url: str, timeout: int) -> str:
    with urllib.request.urlopen(url, timeout=timeout) as response:
        return response.read().decode("utf-8")


def search_pubmed_ids(config: dict[str, Any], journal_name: str, start_date: dt.date, end_date: dt.date) -> list[str]:
    payload = json.loads(http_get(build_esearch_url(config, journal_name, start_date, end_date), int(config["pubmed"]["timeout_seconds"])))
    return payload.get("esearchresult", {}).get("idlist", [])


def _normalize_month(value: str | None) -> str:
    text = str(value or "").strip().lower()
    if not text:
        return "01"
    return MONTH_MAP.get(text, "01")


def _extract_pub_date(article: ET.Element) -> str:
    article_date = article.find(".//Article/ArticleDate")
    if article_date is not None:
        year = article_date.findtext("Year", default="1900")
        month = _normalize_month(article_date.findtext("Month", default="01"))
        day = article_date.findtext("Day", default="01")
        return f"{year}-{month.zfill(2)}-{day.zfill(2)}"

    pub_date = article.find(".//JournalIssue/PubDate")
    if pub_date is None:
        return ""
    year = pub_date.findtext("Year", default="1900")
    month = _normalize_month(pub_date.findtext("Month", default="01"))
    day = pub_date.findtext("Day", default="01")
    return f"{year}-{month.zfill(2)}-{day.zfill(2)}"


def parse_pubmed_articles(xml_payload: str) -> list[dict[str, Any]]:
    root = ET.fromstring(xml_payload)
    articles: list[dict[str, Any]] = []
    for article in root.findall(".//PubmedArticle"):
        title_node = article.find(".//ArticleTitle")
        title = "".join(title_node.itertext()).strip() if title_node is not None else ""
        pmid = article.findtext(".//PMID", default="").strip()
        abstract_parts: list[str] = []
        for node in article.findall(".//Abstract/AbstractText"):
            label = node.attrib.get("Label", "").strip()
            text = "".join(node.itertext()).strip()
            if not text:
                continue
            abstract_parts.append(f"{label}: {text}" if label else text)
        abstract = " ".join(abstract_parts).strip()
        articles.append(
            {
                "pmid": pmid,
                "title": title,
                "abstract": abstract,
                "journal": article.findtext(".//Journal/Title", default="").strip(),
                "publication_date": _extract_pub_date(article),
                "url": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/" if pmid else "",
            }
        )
    return articles


def fetch_articles_for_journal(
    *,
    config: dict[str, Any],
    journal_name: str,
    start_date: dt.date,
    end_date: dt.date,
) -> list[dict[str, Any]]:
    ids = search_pubmed_ids(config, journal_name, start_date, end_date)
    if not ids:
        return []
    xml_payload = http_get(build_efetch_url(config, ids), int(config["pubmed"]["timeout_seconds"]))
    articles = parse_pubmed_articles(xml_payload)
    articles.sort(key=lambda item: item.get("publication_date") or "", reverse=True)
    return articles[: int(config["pubmed"]["top_articles_per_journal"])]


def normalize_scope_review(payload: Any) -> list[dict[str, Any]]:
    if isinstance(payload, dict):
        journals = payload.get("journals", [])
    else:
        journals = payload
    return [journal for journal in journals if journal.get("include_in_set2", True)]


def build_pubmed_summary(
    *,
    config: dict[str, Any],
    scope_review: list[dict[str, Any]],
    end_date: dt.date,
) -> dict[str, Any]:
    start_date = end_date - dt.timedelta(days=int(config["pubmed"]["lookback_days"]))
    journals_output: list[dict[str, Any]] = []
    for index, journal in enumerate(scope_review):
        journal_name = str(journal.get("journal_name") or "").strip()
        entry = {
            "journal_name": journal_name,
            "scope_fit_summary": journal.get("scope_fit_summary", ""),
            "official_website": journal.get("official_website", ""),
            "articles": [],
        }
        if not journal_name:
            entry["error"] = "缺少 journal_name，已跳过"
            journals_output.append(entry)
            continue
        try:
            entry["articles"] = fetch_articles_for_journal(
                config=config,
                journal_name=journal_name,
                start_date=start_date,
                end_date=end_date,
            )
        except Exception as exc:
            entry["error"] = f"{type(exc).__name__}: {exc}"
        journals_output.append(entry)
        if index < len(scope_review) - 1:
            time.sleep(float(config["pubmed"]["delay_seconds_between_requests"]))
    return {
        "generated_at": dt.datetime.now().isoformat(timespec="seconds"),
        "start_date": start_date.isoformat(),
        "end_date": end_date.isoformat(),
        "journals": journals_output,
    }


def render_markdown(summary: dict[str, Any]) -> str:
    lines = [
        "# Set2 PubMed 近 3 个月论文证据",
        "",
        f"- 生成时间：{summary['generated_at']}",
        f"- 检索时间窗：{summary['start_date']} 至 {summary['end_date']}",
        "",
    ]
    for journal in summary["journals"]:
        lines.append(f"## {journal['journal_name']}")
        lines.append("")
        if journal.get("error"):
            lines.append(f"- 检索失败：{journal['error']}")
            lines.append("")
            continue
        if not journal["articles"]:
            lines.append("- PubMed 最近 3 个月未检索到可用论文，或该刊在 PubMed 收录不足。")
            lines.append("")
            continue
        lines.append("| 日期 | 标题 | 摘要 | PMID |")
        lines.append("| --- | --- | --- | --- |")
        for article in journal["articles"]:
            lines.append(
                "| {date} | {title} | {abstract} | [PMID {pmid}]({url}) |".format(
                    date=article["publication_date"] or "未知",
                    title=compact_text(article["title"], 100).replace("|", "/"),
                    abstract=compact_text(article.get("abstract", ""), 140).replace("|", "/"),
                    pmid=article["pmid"] or "N/A",
                    url=article["url"] or "https://pubmed.ncbi.nlm.nih.gov/",
                )
            )
        lines.append("")
    return "\n".join(lines) + "\n"


def main() -> None:
    parser = argparse.ArgumentParser(description="抓取 Set2 期刊最近 3 个月 PubMed 论文")
    parser.add_argument("--workspace", required=True, help="隐藏工作区 run 根目录")
    parser.add_argument("--profile", required=True, help="manuscript_profile.json 路径")
    parser.add_argument("--scope-review", required=True, help="set2_scope_review.json 路径")
    parser.add_argument("--today", default="", help="覆盖今天日期，格式 YYYY-MM-DD")
    args = parser.parse_args()

    config = load_config()
    workspace = resolve_path(args.workspace)
    _ = read_json(resolve_path_within(args.profile, parent=workspace, label="profile"))
    scope_review = normalize_scope_review(
        read_json(resolve_path_within(args.scope_review, parent=workspace, label="scope-review"))
    )
    end_date = dt.date.fromisoformat(args.today) if args.today else dt.date.today()
    payload = build_pubmed_summary(
        config=config,
        scope_review=scope_review,
        end_date=end_date,
    )
    json_path = workspace_output_path(workspace, config["reports"]["pubmed_json"])
    markdown_path = workspace_output_path(workspace, config["reports"]["pubmed_markdown"])
    write_json(json_path, payload)
    write_text(markdown_path, render_markdown(payload))
    print(f"pubmed_json={json_path}")
    print(f"pubmed_markdown={markdown_path}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import datetime as dt
import json
from pathlib import Path

from common import allocate_unique_run_id, ensure_dir, generate_run_id, is_within, load_config, resolve_path, skill_root


def build_manifest(
    *,
    config: dict,
    project_root: Path,
    workspace_base: Path,
    workspace_root: Path,
    run_id: str,
) -> dict:
    reports_cfg = config["reports"]
    return {
        "created_at": dt.datetime.now().isoformat(timespec="seconds"),
        "skill_root": str(skill_root()),
        "project_root": str(project_root),
        "workspace_base": str(workspace_base),
        "workspace_root": str(workspace_root),
        "run_id": run_id,
        "workspace_inside_project_root": is_within(project_root, workspace_base),
        "catalog_xlsx": str((skill_root() / config["assets"]["catalog_xlsx"]).resolve()),
        "final_report": str((workspace_root / reports_cfg["final_report"]).resolve()),
    }


def main() -> None:
    parser = argparse.ArgumentParser(description="初始化 paper-select-journal 隐藏工作区")
    parser.add_argument("--project-root", default=".", help="用户当前项目根目录")
    parser.add_argument("--workspace-base", default="", help="自定义隐藏工作区根目录")
    parser.add_argument("--run-id", default="", help="显式指定 run 目录名")
    args = parser.parse_args()

    config = load_config()
    project_root = resolve_path(args.project_root)
    default_hidden_dir = str(config["workspace"]["hidden_dir"])
    workspace_base = (
        resolve_path(args.workspace_base, base=project_root)
        if args.workspace_base
        else (project_root / default_hidden_dir).resolve()
    )
    run_id = args.run_id or allocate_unique_run_id(workspace_base, generate_run_id(config))
    workspace_root = ensure_dir(workspace_base / run_id)

    for subdir in config["workspace"]["subdirs"]:
        ensure_dir(workspace_root / str(subdir))

    latest_run = workspace_base / config["workspace"]["latest_run_pointer"]
    ensure_dir(workspace_base)
    latest_run.write_text(run_id + "\n", encoding="utf-8")

    manifest = build_manifest(
        config=config,
        project_root=project_root,
        workspace_base=workspace_base,
        workspace_root=workspace_root,
        run_id=run_id,
    )
    manifest_path = workspace_root / config["reports"]["run_manifest"]
    manifest_path.write_text(
        json.dumps(manifest, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )

    print(f"project_root={project_root}")
    print(f"workspace_base={workspace_base}")
    print(f"workspace_root={workspace_root}")
    print(f"run_id={run_id}")
    print(f"manifest={manifest_path}")
    if not manifest["workspace_inside_project_root"]:
        print("warning=workspace_base 位于 project_root 之外；只有用户明确指定时才应这样做")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations

import argparse
from typing import Any

from common import (
    compact_text,
    load_config,
    read_json,
    resolve_path,
    resolve_path_within,
    workspace_output_path,
    write_text,
)


def normalize_final_payload(payload: Any, limit: int = 10) -> dict[str, Any]:
    if isinstance(payload, dict):
        journals = payload.get("journals", [])
        title = payload.get("manuscript_title", "")
        notes = payload.get("summary", "")
    else:
        journals = payload
        title = ""
        notes = ""
    journals = [item for item in journals if str(item.get("journal_name") or "").strip()]
    journals = sorted(
        journals,
        key=lambda item: (
            -(item.get("recommendation_score") or item.get("rank_score") or 0.0),
            item.get("rank", 999),
        ),
    )[:limit]
    return {
        "manuscript_title": title,
        "summary": notes,
        "journals": journals,
    }


def render_markdown(payload: dict[str, Any]) -> str:
    lines = ["# 期刊投稿推荐报告", ""]
    if payload["manuscript_title"]:
        lines.append(f"- 稿件题目：{payload['manuscript_title']}")
    if payload["summary"]:
        lines.append(f"- 总结：{payload['summary']}")
    if len(lines) > 2:
        lines.append("")

    for index, journal in enumerate(payload["journals"], start=1):
        lines.append(f"# {index}. {journal.get('journal_name', '未命名期刊')}")
        lines.append("")
        lines.append("## 核心信息")
        lines.append("")
        lines.append(f"- 影响因子：{journal.get('impact_factor', journal.get('jif', '待补充'))}")
        lines.append(
            "- 中科院小类及其分区：{value}".format(
                value=journal.get("cas_small_category_quartile", journal.get("cas_quartile", "待联网核验"))
            )
        )
        lines.append(f"- 业内认可度：{journal.get('recognition', '待联网核验')}")
        official_website = journal.get("official_website", "")
        if official_website:
            lines.append(f"- 官方网站：[{official_website}]({official_website})")
        else:
            lines.append("- 官方网站：待联网核验")
        lines.append("")
        lines.append("## 为什么推荐这个杂志")
        lines.append("")
        reasons = journal.get("recommendation_reasons", [])
        if reasons:
            for reason in reasons:
                lines.append(f"- {reason}")
        else:
            lines.append("- 待补充推荐理由")
        scope_fit = journal.get("scope_fit_summary", "")
        if scope_fit:
            lines.append(f"- Scope 匹配摘要：{scope_fit}")
        lines.append("")
        lines.append("## 最近 3 个月类似主题论文")
        lines.append("")
        lines.append("| 发表日期 | 标题 | 摘要 | 相关性说明 | PMID / 链接 |")
        lines.append("| --- | --- | --- | --- | --- |")
        articles = journal.get("recent_articles", [])
        if not articles:
            lines.append("| 暂无 | 暂无 | 暂无 | 暂无 | 暂无 |")
        for article in articles:
            url = article.get("url") or ""
            pmid = article.get("pmid") or "无"
            if url:
                link = f"[PMID {pmid}]({url})"
            else:
                link = pmid
            lines.append(
                "| {date} | {title} | {abstract} | {relevance} | {link} |".format(
                    date=article.get("publication_date", "未知"),
                    title=compact_text(article.get("title", ""), 90).replace("|", "/"),
                    abstract=compact_text(article.get("abstract", ""), 140).replace("|", "/"),
                    relevance=compact_text(article.get("relevance", "待 AI 补充"), 120).replace("|", "/"),
                    link=link,
                )
            )
        lines.append("")
    return "\n".join(lines) + "\n"


def main() -> None:
    parser = argparse.ArgumentParser(description="把最终推荐 JSON 渲染成 Markdown 报告")
    parser.add_argument("--workspace", required=True, help="隐藏工作区 run 根目录")
    parser.add_argument("--final-json", required=True, help="final_recommendations.json 路径")
    parser.add_argument("--output", default="", help="覆盖默认输出路径")
    args = parser.parse_args()

    config = load_config()
    workspace = resolve_path(args.workspace)
    payload = normalize_final_payload(
        read_json(resolve_path_within(args.final_json, parent=workspace, label="final-json")),
        limit=int(config["screening"]["default_set3_limit"]),
    )
    output_path = (
        resolve_path_within(args.output, parent=workspace, label="output", base=workspace)
        if args.output
        else workspace_output_path(workspace, config["reports"]["final_report"])
    )
    write_text(output_path, render_markdown(payload))
    print(f"report_markdown={output_path}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations

import argparse
from pathlib import Path
from typing import Any

from openpyxl import load_workbook

from common import (
    compact_text,
    float_or_none,
    load_config,
    quartile_from_category,
    read_json,
    resolve_path,
    resolve_path_within,
    skill_root,
    workspace_output_path,
    write_json,
    write_text,
)


REQUIRED_HEADERS = {
    "journal_name",
    "issn",
    "eissn",
    "category",
    "citations",
    "jif",
    "jci",
    "percentageOAGold",
}


def load_catalog(catalog_path: Path) -> list[dict[str, Any]]:
    workbook = load_workbook(catalog_path, read_only=True, data_only=True)
    sheet = workbook[workbook.sheetnames[0]]
    rows = sheet.iter_rows(values_only=True)
    headers = [str(item).strip() if item is not None else "" for item in next(rows)]
    missing = REQUIRED_HEADERS.difference(headers)
    if missing:
        raise ValueError(f"catalog 缺少列: {sorted(missing)}")

    catalog: list[dict[str, Any]] = []
    for row_values in rows:
        row = dict(zip(headers, row_values))
        catalog.append(
            {
                "journal_name": str(row.get("journal_name") or "").strip(),
                "issn": str(row.get("issn") or "").strip(),
                "eissn": str(row.get("eissn") or "").strip(),
                "category": str(row.get("category") or "").strip(),
                "citations": row.get("citations"),
                "jif": float_or_none(row.get("jif")),
                "jci": float_or_none(row.get("jci")),
                "percentage_oa_gold": float_or_none(row.get("percentageOAGold")),
            }
        )
    return catalog


def quartile_rank(quartile: str) -> int:
    return {"Q1": 0, "Q2": 1, "Q3": 2, "Q4": 3}.get(quartile, 4)


def build_selection_notes(row: dict[str, Any], profile: dict[str, Any], config: dict[str, Any]) -> list[str]:
    quartile = quartile_from_category(row["category"])
    notes = [f"JIF {row['jif']:.1f} 达到阈值", f"分区信号：{quartile}"]
    if profile.get("prefer_open_access") and (row["percentage_oa_gold"] or 0.0) >= float(
        config["screening"]["open_access_gold_threshold"]
    ):
        notes.append(f"OA Gold 占比 {row['percentage_oa_gold']:.1f}%")
    if row.get("citations"):
        notes.append(f"引用量 {row['citations']}")
    return notes


def summarize_candidate(row: dict[str, Any], profile: dict[str, Any], config: dict[str, Any]) -> dict[str, Any]:
    quartile = quartile_from_category(row["category"])
    return {
        **row,
        "quartile": quartile,
        "selection_notes": build_selection_notes(row, profile, config),
        "manual_exception": False,
    }


def sort_candidates(candidates: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return sorted(
        candidates,
        key=lambda item: (
            0 if item.get("manual_exception") else 1,
            quartile_rank(str(item.get("quartile") or "unknown")),
            -(item.get("jif") or 0.0),
            -(float_or_none(item.get("citations")) or 0.0),
            str(item.get("journal_name") or "").lower(),
        ),
    )


def apply_manual_exceptions(
    *,
    catalog: list[dict[str, Any]],
    shortlisted: list[dict[str, Any]],
    exceptions: list[dict[str, Any]] | None,
    profile: dict[str, Any],
    config: dict[str, Any],
) -> list[dict[str, Any]]:
    if not exceptions:
        return shortlisted
    journal_index = {item["journal_name"].lower(): item for item in catalog}
    deduped: dict[str, dict[str, Any]] = {
        item["journal_name"].lower(): item for item in shortlisted
    }
    for exception in exceptions:
        journal_name = str(exception.get("journal_name") or "").strip()
        if not journal_name:
            continue
        row = journal_index.get(journal_name.lower())
        if not row:
            continue
        candidate = {
            **summarize_candidate(row, profile, config),
            "manual_exception": True,
            "manual_exception_reason": str(exception.get("reason") or "").strip(),
        }
        deduped[journal_name.lower()] = candidate
    return sort_candidates(list(deduped.values()))


def build_shortlist(
    *,
    catalog: list[dict[str, Any]],
    profile: dict[str, Any],
    config: dict[str, Any],
    limit: int,
) -> list[dict[str, Any]]:
    if limit <= 0:
        raise ValueError(f"limit 必须是正整数，当前值为 {limit}")
    min_jif = max(float(config["screening"]["min_impact_factor"]), float(profile.get("target_if_min", 0.0) or 0.0))
    excluded = {
        str(name).strip().lower()
        for name in profile.get("excluded_journals", [])
        if str(name).strip()
    }
    candidates: list[dict[str, Any]] = []
    for row in catalog:
        if not row["journal_name"]:
            continue
        if row["journal_name"].lower() in excluded:
            continue
        if row["jif"] is None or row["jif"] < min_jif:
            continue
        candidates.append(summarize_candidate(row, profile, config))
    return sort_candidates(candidates)[:limit]


def render_markdown(payload: dict[str, Any]) -> str:
    lines = [
        "# Set1 候选池",
        "",
        f"- 生成时间：{payload['generated_at']}",
        f"- 候选数量：{payload['candidate_count']}",
        f"- 最低影响因子阈值：{payload['min_impact_factor']}",
        "- 本步骤只做最小硬过滤，不代表已完成语义匹配排序。",
        "",
        "| 排名 | 期刊 | 类别 | JIF | 分区 | 入池理由 |",
        "| --- | --- | --- | --- | --- | --- |",
    ]
    for index, item in enumerate(payload["candidates"], start=1):
        clue_text = "；".join(item.get("selection_notes") or []) or "待 AI 进一步判断"
        if item.get("manual_exception") and item.get("manual_exception_reason"):
            clue_text = f"{clue_text}；人工例外：{item['manual_exception_reason']}"
        lines.append(
            "| {idx} | {name} | {category} | {jif:.1f} | {quartile} | {clues} |".format(
                idx=index,
                name=item["journal_name"],
                category=item["category"].replace("|", "/"),
                jif=item["jif"] or 0.0,
                quartile=item.get("quartile", "unknown"),
                clues=clue_text.replace("|", "/"),
            )
        )
    lines.extend(
        [
            "",
            "## 下一步",
            "",
            "- 先通读 manuscript 画像与当前候选池，自主决定优先核验顺序。",
            "- 对真正值得继续看的候选联网核验：官方 scope、官网链接、业内认可度、预警/垃圾期刊风险。",
            "- 只把 scope 真匹配且质量信号干净的期刊写入 `analysis/set2_scope_review.json`。",
            "- 不要把当前 Set1 候选池直接当最终推荐清单。",
        ]
    )
    return "\n".join(lines) + "\n"


def main() -> None:
    parser = argparse.ArgumentParser(description="基于 2023IF.xlsx 生成 Set1 候选池")
    parser.add_argument("--profile", required=True, help="manuscript_profile.json 路径")
    parser.add_argument("--workspace", required=True, help="隐藏工作区 run 根目录")
    parser.add_argument("--catalog", default="", help="覆盖默认内置 2023IF.xlsx")
    parser.add_argument("--limit", type=int, default=0, help="Set1 数量上限")
    parser.add_argument("--manual-include-json", default="", help="低 IF 例外补录 JSON")
    args = parser.parse_args()

    config = load_config()
    workspace = resolve_path(args.workspace)
    profile = read_json(resolve_path_within(args.profile, parent=workspace, label="profile"))
    catalog_path = (
        resolve_path(args.catalog)
        if args.catalog
        else (skill_root() / config["assets"]["catalog_xlsx"]).resolve()
    )
    catalog = load_catalog(catalog_path)
    limit = args.limit or int(config["screening"]["default_set1_limit"])
    shortlisted = build_shortlist(
        catalog=catalog,
        profile=profile,
        config=config,
        limit=limit,
    )
    if args.manual_include_json:
        shortlisted = apply_manual_exceptions(
            catalog=catalog,
            shortlisted=shortlisted,
            exceptions=read_json(resolve_path(args.manual_include_json)),
            profile=profile,
            config=config,
        )[:limit]

    payload = {
        "generated_at": __import__("datetime").datetime.now().isoformat(timespec="seconds"),
        "catalog_path": str(catalog_path),
        "profile_title": profile.get("title", ""),
        "candidate_count": len(shortlisted),
        "min_impact_factor": max(
            float(config["screening"]["min_impact_factor"]),
            float(profile.get("target_if_min", 0.0) or 0.0),
        ),
        "candidates": shortlisted,
        "notes": compact_text(profile.get("notes", "")),
        "selection_strategy": "minimal-hard-filtering",
    }
    json_path = workspace_output_path(workspace, config["reports"]["set1_json"])
    markdown_path = workspace_output_path(workspace, config["reports"]["set1_markdown"])
    write_json(json_path, payload)
    write_text(markdown_path, render_markdown(payload))
    print(f"set1_json={json_path}")
    print(f"set1_markdown={markdown_path}")


if __name__ == "__main__":
    main()

{
  "manuscript_title": "请填写稿件标题",
  "summary": "1-3 句总述，说明为什么这些期刊最值得优先考虑。",
  "journals": [
    {
      "rank": 1,
      "recommendation_score": 9.4,
      "journal_name": "示例期刊",
      "impact_factor": 8.7,
      "cas_small_category_quartile": "材料科学 2 区",
      "recognition": "业内认可度高，主流实验室常投",
      "official_website": "https://example.org/journal",
      "scope_fit_summary": "论文主题与期刊 scope 匹配点。",
      "recommendation_reasons": [
        "期刊 scope 与论文高度匹配",
        "近期发表过相似主题文章",
        "不是预警或垃圾期刊"
      ],
      "recent_articles": [
        {
          "publication_date": "2026-03-01",
          "title": "Recent article title",
          "abstract": "Recent article abstract",
          "relevance": "AI 说明这篇论文为何能支持该期刊进入最终推荐。",
          "pmid": "12345678",
          "url": "https://pubmed.ncbi.nlm.nih.gov/12345678/"
        }
      ]
    }
  ]
}

{
  "title": "请填写论文题目",
  "abstract": "请填写论文摘要或 AI 对全文的精炼摘要",
  "keywords": [
    "关键词1",
    "关键词2"
  ],
  "manuscript_summary": "用 2-4 句总结稿件的核心问题、主要方法和目标读者。",
  "target_journal_brief": "如果用户有明确偏好，可简述期刊方向、风格、读者群或投稿策略；没有可留空。",
  "excluded_journals": [
    "明确不想投的期刊"
  ],
  "prefer_open_access": false,
  "target_if_min": 3.0,
  "notes": "用户的额外偏好，例如审稿速度、是否接受 OA、是否回避版面费过高期刊，以及 AI 对选刊策略的自由规划说明。"
}

{
  "journals": [
    {
      "journal_name": "示例期刊",
      "include_in_set2": true,
      "official_website": "https://example.org/journal",
      "scope_fit_summary": "说明该刊 scope 为什么与论文匹配。",
      "recognition": "说明业内认可度，例如主流领域认可、编辑部稳定、非预警。",
      "cas_small_category_quartile": "例如：医学 2 区 / 肿瘤学 2 区",
      "quality_check": {
        "warning_screening_result": "已查，无预警信号",
        "predatory_screening_result": "已查，无垃圾期刊信号"
      },
      "sources": [
        {
          "label": "官方 scope",
          "url": "https://example.org/aims-and-scope"
        }
      ]
    }
  ]
}

{
  "journals": [
    {
      "journal_name": "示例期刊",
      "include_in_set3": true,
      "similarity_assessment": "该刊近 3 个月发表了 2 篇与稿件主题和方法都较接近的论文，可作为进入最终推荐的重要证据。",
      "relevant_articles": [
        {
          "pmid": "12345678",
          "relevance": "研究问题与稿件接近，且使用了相似的实验策略。"
        }
      ],
      "irrelevant_articles_count": 3,
      "overall_relevance_level": "high"
    }
  ]
}