Carl Weread

.venv/
__pycache__/
*.pyc
.pytest_cache/
.env
reports/generated/
*.log
.DS_Store

"""carl-weread: context-driven WeRead action coach."""

__version__ = "0.3.0"

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

from .digest_apply import ReadingInput, build_action_card


@dataclass(frozen=True)
class AfterReadInput:
    book_title: str
    chapter_title: str
    current_problem: str = ""
    highlights: list[str] = field(default_factory=list)


@dataclass(frozen=True)
class AfterReadResult:
    status: str
    markdown: str
    should_writeback: bool


def extract_highlights(payload: Any) -> list[str]:
    """Extract user highlights from common WeRead bookmark/underline response shapes."""
    if isinstance(payload, list):
        items = payload
    elif isinstance(payload, dict):
        items = []
        containers = [payload]
        for key in ("data", "result"):
            value = payload.get(key)
            if isinstance(value, dict):
                containers.append(value)
        for container in containers:
            for key in ("updated", "bookmarks", "underlines", "items", "reviews"):
                value = container.get(key)
                if isinstance(value, list):
                    items = value
                    break
            if items:
                break
    else:
        return []

    highlights: list[str] = []
    for item in items:
        if not isinstance(item, dict):
            continue
        for key in ("markText", "abstract", "content", "text", "review", "summary"):
            value = item.get(key)
            if value is not None and str(value).strip():
                highlights.append(str(value).strip())
                break
    return highlights


def build_no_highlight_protocol(reading: AfterReadInput) -> str:
    problem = reading.current_problem or "你读这一节前想解决的那个问题"
    return "\n".join(
        [
            f"# 无划线读后检查｜《{reading.book_title}》{reading.chapter_title}",
            "",
            "这次先不强行总结。没有划线时，更可能是读得太散，或者问题没有咬住文本。",
            "",
            "## 先回答 3 个问题",
            f"1. 这一节里哪句话最接近「{problem}」？请补一条原文或自己的转述。",
            "2. 如果只能把这一节变成一个动作，你会改哪个项目、文章或判断？",
            "3. 这节有没有让你改变原来的看法？如果没有，下一次应该换书还是换问题？",
            "",
            "## 下一步",
            "补一条划线/转述后，再生成阅读行动卡；否则这次阅读只记录为浏览，不算进入工作。",
        ]
    )


def build_after_read_result(reading: AfterReadInput) -> AfterReadResult:
    highlights = [item.strip() for item in reading.highlights if item.strip()]
    if not highlights:
        return AfterReadResult(
            status="no-highlights",
            markdown=build_no_highlight_protocol(reading),
            should_writeback=False,
        )
    card = build_action_card(
        ReadingInput(
            book_title=reading.book_title,
            chapter_title=reading.chapter_title,
            highlights=highlights,
            current_problem=reading.current_problem,
        )
    )
    return AfterReadResult(status="action-card", markdown=card, should_writeback=True)

from __future__ import annotations

from dataclasses import asdict, dataclass
from typing import Any

from .today_chapter import Chapter, is_substantive_chapter_title


@dataclass(frozen=True)
class BookRef:
    book_id: str
    title: str


def _unwrap_items(payload: Any, keys: tuple[str, ...]) -> list[Any]:
    if isinstance(payload, list):
        return payload
    if not isinstance(payload, dict):
        return []

    containers = [payload]
    for wrapper_key in ("data", "result"):
        wrapped = payload.get(wrapper_key)
        if isinstance(wrapped, dict):
            containers.append(wrapped)

    for container in containers:
        for key in keys:
            value = container.get(key)
            if isinstance(value, list):
                return value
    return []


def _string_value(item: dict[str, Any], keys: tuple[str, ...]) -> str:
    for key in keys:
        value = item.get(key)
        if value is not None:
            text = str(value).strip()
            if text:
                return text
    return ""


def extract_book_refs(payload: Any, limit: int | None = None) -> list[BookRef]:
    refs: list[BookRef] = []
    for raw_item in _unwrap_items(payload, ("books", "bookProgress", "updated", "synckeys")):
        if not isinstance(raw_item, dict):
            continue
        item = raw_item.get("book") if isinstance(raw_item.get("book"), dict) else raw_item
        book_id = _string_value(item, ("book_id", "bookId", "bookid", "id"))
        title = _string_value(item, ("book_title", "title", "bookName", "name"))
        if book_id and title:
            refs.append(BookRef(book_id=book_id, title=title))
        if limit is not None and len(refs) >= limit:
            break
    return refs


def _is_readable_chapter(title: str, item: dict[str, Any]) -> bool:
    normalized = title.strip()
    if not is_substantive_chapter_title(normalized):
        return False
    word_count = item.get("wordCount") or item.get("word_count")
    if isinstance(word_count, int) and word_count <= 20:
        return False
    return True


def chapters_from_chapterinfo(book_id: str, book_title: str, payload: Any) -> list[Chapter]:
    chapters: list[Chapter] = []
    for item in _unwrap_items(payload, ("chapters", "chapterInfos", "chapterInfo")):
        if not isinstance(item, dict):
            continue
        chapter_uid = _string_value(item, ("chapter_uid", "chapterUid", "uid", "chapterId", "id"))
        title = _string_value(item, ("title", "chapterName", "name"))
        if not _is_readable_chapter(title, item):
            continue
        if chapter_uid and title:
            chapters.append(Chapter(book_id=book_id, book_title=book_title, chapter_uid=chapter_uid, title=title))
    return chapters


def merge_book_refs(*groups: list[BookRef], limit: int | None = None) -> list[BookRef]:
    merged: list[BookRef] = []
    seen: set[str] = set()
    for group in groups:
        for ref in group:
            if ref.book_id in seen:
                continue
            seen.add(ref.book_id)
            merged.append(ref)
            if limit is not None and len(merged) >= limit:
                return merged
    return merged


def build_candidate_chapters(
    shelf_payload: Any,
    notebooks_payload: Any,
    chapterinfo_by_book_id: dict[str, Any],
    limit_books: int | None = None,
) -> list[Chapter]:
    book_refs = merge_book_refs(
        extract_book_refs(shelf_payload),
        extract_book_refs(notebooks_payload),
        limit=limit_books,
    )
    candidates: list[Chapter] = []
    for ref in book_refs:
        payload = chapterinfo_by_book_id.get(ref.book_id)
        if payload is None:
            continue
        candidates.extend(chapters_from_chapterinfo(ref.book_id, ref.title, payload))
    return candidates


def chapters_to_jsonable(chapters: list[Chapter]) -> list[dict[str, str]]:
    return [asdict(chapter) for chapter in chapters]

from __future__ import annotations

import argparse
import json
from pathlib import Path
import subprocess
import sys
from typing import Any

from .after_read import AfterReadInput, build_after_read_result, extract_highlights
from .config import ContextConfig, load_config
from .context import collect_context_for_config
from .unread_advisor import format_unread_recommendation, recommend_unread_book
from .weekly_loop import WeeklyReportInput, build_weekly_report, load_cards
from .writeback import write_action_card


def default_config_path() -> Path:
    return Path.home() / ".config" / "carl-weread" / "config.toml"


def _load_config_or_chat(path: Path) -> ContextConfig:
    return load_config(path) if path.exists() else ContextConfig(mode="chat")


def _run_weread(weread_script: Path, args: list[str]) -> Any:
    command = [str(weread_script), *args]
    if weread_script.suffix == ".py":
        command = [sys.executable, str(weread_script), *args]
    try:
        result = subprocess.run(command, text=True, capture_output=True, timeout=90)
    except subprocess.TimeoutExpired:
        print(f"WeRead命令超时：{' '.join(command[:2])} ...", file=sys.stderr)
        raise SystemExit(4)
    if result.returncode != 0:
        sys.stderr.write(result.stderr)
        raise SystemExit(result.returncode)
    return json.loads(result.stdout)


def _cmd_recommend(args: argparse.Namespace) -> int:
    config = _load_config_or_chat(args.config)
    context = collect_context_for_config(config, brief=args.brief)
    shelf_payload = _run_weread(args.weread_script, ["shelf"])
    notebooks_payload = _run_weread(args.weread_script, ["notebooks", f"--count={args.notebooks_count}"])
    recommendation_payloads = [_run_weread(args.weread_script, ["recommend", f"--count={args.count}"])]
    similar_payloads = []
    for book_id in args.similar_book_id or []:
        similar_payloads.append(_run_weread(args.weread_script, ["similar", f"--bookId={book_id}", f"--count={args.count}"]))
    search_payloads = []
    if args.keyword:
        search_payloads.append(_run_weread(args.weread_script, ["search", f"--keyword={args.keyword}"]))
    rec = recommend_unread_book(
        context=context,
        shelf_payload=shelf_payload,
        notebooks_payload=notebooks_payload,
        recommendation_payloads=recommendation_payloads,
        similar_payloads=similar_payloads,
        search_payloads=search_payloads,
        brief=args.brief or args.keyword or "",
    )
    if args.json:
        print(json.dumps(rec.__dict__, ensure_ascii=False, indent=2))
    else:
        print(format_unread_recommendation(rec))
    return 0


def _cmd_after_read(args: argparse.Namespace) -> int:
    highlights = list(args.highlight or [])
    if args.highlights_file:
        highlights.extend(
            line.strip()
            for line in args.highlights_file.read_text(encoding="utf-8").splitlines()
            if line.strip()
        )
    if args.auto_fetch:
        if not args.book_id or not args.chapter_uid:
            print("--auto-fetch 需要同时提供 --book-id 和 --chapter-uid", file=sys.stderr)
            return 2
        payload = _run_weread(
            args.weread_script,
            ["underlines", f"--bookId={args.book_id}", f"--chapterUid={args.chapter_uid}"],
        )
        highlights.extend(extract_highlights(payload))
    if not args.book_title or not args.chapter_title:
        print("after-read 需要 --book-title 和 --chapter-title", file=sys.stderr)
        return 2
    result = build_after_read_result(
        AfterReadInput(
            book_title=args.book_title,
            chapter_title=args.chapter_title,
            current_problem=args.current_problem or "",
            highlights=highlights,
        )
    )
    if args.writeback and result.should_writeback:
        config = _load_config_or_chat(args.config)
        write_result = write_action_card(
            result.markdown,
            config,
            f"{args.book_title}-{args.chapter_title}",
            output_dir=args.writeback_dir,
        )
        print(write_result.message)
        if write_result.path is None:
            print(result.markdown)
        return 0
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(result.markdown.rstrip() + "\n", encoding="utf-8")
        print(f"已写入：{args.output}")
        return 0
    print(result.markdown)
    return 0


def _cmd_weekly(args: argparse.Namespace) -> int:
    report = build_weekly_report(WeeklyReportInput(cards=load_cards(args.cards), context=args.context or ""))
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(report + "\n", encoding="utf-8")
        print(f"已写入周复盘：{args.output}")
    else:
        print(report)
    return 0


def _cmd_today(args: argparse.Namespace) -> int:
    script = Path(__file__).resolve().parents[1] / "scripts" / "today_live.py"
    command = [
        sys.executable,
        str(script),
        "--config",
        str(args.config),
        "--weread-script",
        str(args.weread_script),
        "--limit-books",
        str(args.limit_books),
        "--notebooks-count",
        str(args.notebooks_count),
    ]
    if args.brief:
        command.extend(["--brief", args.brief])
    result = subprocess.run(command, text=True)
    return result.returncode


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="carl-weread V0.3 unified reading coach CLI")
    subparsers = parser.add_subparsers(dest="command", required=True)

    recommend = subparsers.add_parser("recommend", help="推荐一本没读透但现在该读的书")
    recommend.add_argument("--config", type=Path, default=default_config_path())
    recommend.add_argument("--weread-script", type=Path, default=Path("scripts/weread.sh"))
    recommend.add_argument("--brief", help="当前问题/项目上下文")
    recommend.add_argument("--keyword", help="额外搜索关键词")
    recommend.add_argument("--similar-book-id", action="append", help="用相似书扩展候选，可传多次")
    recommend.add_argument("--count", type=int, default=12)
    recommend.add_argument("--notebooks-count", type=int, default=200)
    recommend.add_argument("--json", action="store_true")
    recommend.set_defaults(func=_cmd_recommend)

    today = subparsers.add_parser("today", help="推荐今天只读哪一小节")
    today.add_argument("--config", type=Path, default=default_config_path())
    today.add_argument("--weread-script", type=Path, default=Path("scripts/weread.sh"))
    today.add_argument("--brief", help="当前问题/项目上下文")
    today.add_argument("--limit-books", type=int, default=5)
    today.add_argument("--notebooks-count", type=int, default=200)
    today.set_defaults(func=_cmd_today)

    after_read = subparsers.add_parser("after-read", help="读后检查划线；有划线生成行动卡，无划线走追问协议")
    after_read.add_argument("--config", type=Path, default=default_config_path())
    after_read.add_argument("--weread-script", type=Path, default=Path("scripts/weread.sh"))
    after_read.add_argument("--book-id")
    after_read.add_argument("--chapter-uid")
    after_read.add_argument("--book-title")
    after_read.add_argument("--chapter-title")
    after_read.add_argument("--current-problem")
    after_read.add_argument("--highlight", action="append")
    after_read.add_argument("--highlights-file", type=Path)
    after_read.add_argument("--auto-fetch", action="store_true")
    after_read.add_argument("--writeback", action="store_true")
    after_read.add_argument("--writeback-dir", type=Path)
    after_read.add_argument("--output", type=Path)
    after_read.set_defaults(func=_cmd_after_read)

    weekly = subparsers.add_parser("weekly", help="从阅读行动卡生成周复盘")
    weekly.add_argument("--cards", nargs="*", type=Path, default=[])
    weekly.add_argument("--context", default="")
    weekly.add_argument("--output", type=Path)
    weekly.set_defaults(func=_cmd_weekly)

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    return args.func(args)


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

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path


VALID_MODES = {"obsidian", "folder", "chat", "weread-only"}
PATH_REQUIRED_MODES = {"obsidian", "folder"}


@dataclass(frozen=True)
class ContextConfig:
    mode: str
    path: Path | None = None

    def validate(self) -> None:
        if self.mode not in VALID_MODES:
            raise ValueError(f"unsupported mode: {self.mode}")
        if self.mode in PATH_REQUIRED_MODES and self.path is None:
            raise ValueError(f"path is required for {self.mode} mode")


def _escape_toml_string(value: str) -> str:
    return value.replace("\\", "\\\\").replace('"', '\\"')


def write_config(config: ContextConfig, output_path: str | Path) -> None:
    config.validate()
    output = Path(output_path).expanduser()
    output.parent.mkdir(parents=True, exist_ok=True)

    lines = ["[context]", f'mode = "{_escape_toml_string(config.mode)}"']
    if config.path is not None:
        lines.append(f'path = "{_escape_toml_string(str(Path(config.path).expanduser()))}"')
    lines.append("")
    output.write_text("\n".join(lines), encoding="utf-8")


def _parse_simple_toml_context(text: str) -> dict[str, str]:
    in_context = False
    values: dict[str, str] = {}
    for raw_line in text.splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        if line.startswith("[") and line.endswith("]"):
            in_context = line == "[context]"
            continue
        if not in_context or "=" not in line:
            continue
        key, raw_value = line.split("=", 1)
        value = raw_value.strip()
        if value.startswith('"') and value.endswith('"'):
            value = value[1:-1].replace('\\"', '"').replace("\\\\", "\\")
        values[key.strip()] = value
    return values


def load_config(config_path: str | Path) -> ContextConfig:
    path = Path(config_path).expanduser()
    values = _parse_simple_toml_context(path.read_text(encoding="utf-8"))
    mode = values.get("mode")
    if mode is None:
        raise ValueError("config missing context.mode")
    context_path = Path(values["path"]).expanduser() if values.get("path") else None
    config = ContextConfig(mode=mode, path=context_path)
    config.validate()
    return config

from __future__ import annotations

from dataclasses import dataclass
from datetime import date, timedelta
from pathlib import Path

from .config import ContextConfig


@dataclass(frozen=True)
class ContextItem:
    kind: str
    source: str
    text: str


def _read_snippet(path: Path, max_chars: int) -> str:
    return path.read_text(encoding="utf-8", errors="ignore")[:max_chars].strip()


def _recent_daily_paths(vault: Path, today: date, days: int) -> list[Path]:
    daily_dir = vault / "10_日记"
    return [daily_dir / f"{today - timedelta(days=offset):%Y-%m-%d}.md" for offset in range(days)]


def _first_markdown_files(root: Path, limit: int = 8) -> list[Path]:
    if not root.exists():
        return []
    return sorted((p for p in root.rglob("*.md") if p.is_file()), key=lambda p: p.stat().st_mtime, reverse=True)[:limit]


def _first_context_files(root: Path, limit: int = 8) -> list[Path]:
    if not root.exists():
        return []
    patterns = ("*.md", "*.txt")
    candidates: list[Path] = []
    for pattern in patterns:
        candidates.extend(p for p in root.rglob(pattern) if p.is_file())
    return sorted(candidates, key=lambda p: p.stat().st_mtime, reverse=True)[:limit]


def collect_recent_context(
    vault_path: str | Path,
    today: date | None = None,
    days: int = 3,
    max_chars_per_file: int = 1200,
) -> list[ContextItem]:
    """Collect a compact context bundle from Obsidian daily/project/topic notes."""
    vault = Path(vault_path).expanduser()
    today = today or date.today()
    items: list[ContextItem] = []

    structured_dirs = [vault / "10_日记", vault / "20_项目", vault / "15_自媒体" / "选题库"]
    has_obsidian_structure = any(path.exists() for path in structured_dirs)

    if not has_obsidian_structure:
        for path in _first_context_files(vault, limit=8):
            items.append(ContextItem("context", str(path.relative_to(vault)), _read_snippet(path, max_chars_per_file)))
        return [item for item in items if item.text]

    for path in _recent_daily_paths(vault, today, days):
        if path.exists():
            items.append(ContextItem("daily", str(path.relative_to(vault)), _read_snippet(path, max_chars_per_file)))

    for path in _first_markdown_files(vault / "20_项目", limit=6):
        items.append(ContextItem("project", str(path.relative_to(vault)), _read_snippet(path, max_chars_per_file)))

    for path in _first_markdown_files(vault / "15_自媒体" / "选题库", limit=6):
        items.append(ContextItem("topic", str(path.relative_to(vault)), _read_snippet(path, max_chars_per_file)))

    return [item for item in items if item.text]


def collect_context_for_config(
    config: ContextConfig,
    brief: str | None = None,
    today: date | None = None,
    days: int = 3,
    max_chars_per_file: int = 1200,
) -> list[ContextItem]:
    """Collect context using the configured mode."""
    config.validate()
    if config.mode in {"obsidian", "folder"}:
        return collect_recent_context(config.path, today=today, days=days, max_chars_per_file=max_chars_per_file)
    if config.mode == "chat":
        text = (brief or "").strip()
        return [ContextItem("brief", "current-input", text)] if text else []
    return []

from __future__ import annotations

from dataclasses import dataclass, field


@dataclass(frozen=True)
class ReadingInput:
    book_title: str
    chapter_title: str
    highlights: list[str] = field(default_factory=list)
    current_problem: str = ""


def _first_highlight(reading: ReadingInput) -> str:
    return reading.highlights[0].strip() if reading.highlights else "这次阅读需要补一条具体划线或读后想法。"


def _concept_name(reading: ReadingInput) -> str:
    if "验证" in _first_highlight(reading) or "验证" in reading.current_problem:
        return "阅读进入验证链路"
    if "行动" in reading.current_problem:
        return "行动型阅读"
    return "上下文驱动阅读"


def build_action_card(reading: ReadingInput) -> str:
    """Build an Obsidian-ready action card from a reading note."""
    highlight = _first_highlight(reading)
    concept = _concept_name(reading)
    problem = reading.current_problem or "这次阅读还没有绑定到具体问题，需要补充它解释了哪个真实卡点。"

    return f"""# 阅读行动卡｜《{reading.book_title}》{reading.chapter_title}

## 一句话收获
{highlight}

## 它解释了我最近哪个问题
{problem}

## 可以马上改的一个动作
把这次阅读里的一个判断，改写成当前项目里可检查的验证标准，并在今天推进一次。

## 可沉淀的原子笔记
[[{concept}]]：阅读不是在脑子里多存一条知识，而是要进入一个可验证的行动链路。只有能改变下一步动作的内容，才真正进入了工作系统。

## 可发展的内容选题
标题方向：AI 时代读书，不该从书单开始，而该从你今天卡住的问题开始。
切入角度：用《{reading.book_title}》这一节作为例子，展示如何把一段阅读变成一个项目动作。
""".strip()

from __future__ import annotations

from datetime import datetime
from zoneinfo import ZoneInfo
from urllib.parse import urlencode


def seconds_to_readable(seconds: int) -> str:
    """Convert WeRead duration seconds to a Chinese human-readable string."""
    minutes = max(0, int(seconds)) // 60
    hours, mins = divmod(minutes, 60)
    if hours and mins:
        return f"{hours} 小时 {mins} 分钟"
    if hours:
        return f"{hours} 小时"
    return f"{mins} 分钟"


def format_date(timestamp: int, tz: str = "Asia/Shanghai") -> str:
    """Format a Unix timestamp as YYYY-MM-DD in the requested timezone."""
    return datetime.fromtimestamp(int(timestamp), ZoneInfo(tz)).strftime("%Y-%m-%d")


def weread_link(book_id: str, chapter_uid: str | None = None) -> str:
    """Build a WeRead deep link for a book or a specific chapter."""
    params = {"bId": book_id}
    if chapter_uid:
        params["chapterUid"] = str(chapter_uid)
    return "weread://reading?" + urlencode(params)

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

from .candidates import _unwrap_items


@dataclass(frozen=True)
class BookProfile:
    book_id: str
    title: str
    author: str = ""
    category: str = ""
    intro: str = ""
    source: str = "unknown"
    note_count: int = 0
    progress_percent: float | None = None


def _string_value(item: dict[str, Any], keys: tuple[str, ...]) -> str:
    for key in keys:
        value = item.get(key)
        if value is None:
            continue
        text = str(value).strip()
        if text:
            return text
    return ""


def _number_value(item: dict[str, Any], keys: tuple[str, ...]) -> float | None:
    for key in keys:
        value = item.get(key)
        if value in (None, ""):
            continue
        try:
            return float(value)
        except (TypeError, ValueError):
            continue
    return None


def _book_dict(raw_item: Any) -> dict[str, Any] | None:
    if not isinstance(raw_item, dict):
        return None
    for key in ("book", "bookInfo", "bookinfo", "info"):
        value = raw_item.get(key)
        if isinstance(value, dict):
            merged = dict(value)
            for outer_key in ("noteCount", "note_count", "readingTime", "progress", "progressPercent", "source"):
                if outer_key in raw_item and outer_key not in merged:
                    merged[outer_key] = raw_item[outer_key]
            return merged
    return raw_item


def extract_book_profiles(payload: Any, source: str = "unknown") -> list[BookProfile]:
    """Extract book profiles from common WeRead response shapes.

    The WeRead gateway returns slightly different envelopes across endpoints. This
    helper stays deliberately permissive so workflow code can reason about books
    without depending on one exact JSON shape.
    """
    raw_items = _unwrap_items(
        payload,
        (
            "books",
            "bookInfos",
            "bookInfo",
            "items",
            "records",
            "recommendBooks",
            "data",
            "bookProgress",
            "updated",
            "synckeys",
        ),
    )
    profiles: list[BookProfile] = []
    for raw_item in raw_items:
        item = _book_dict(raw_item)
        if item is None:
            continue
        book_id = _string_value(item, ("book_id", "bookId", "bookid", "id"))
        title = _string_value(item, ("book_title", "title", "bookName", "name"))
        if not book_id or not title:
            continue
        note_count = int(_number_value(item, ("noteCount", "note_count", "bookmarkCount", "reviewCount")) or 0)
        progress = _number_value(item, ("progressPercent", "progress", "percent", "readProgress"))
        profiles.append(
            BookProfile(
                book_id=book_id,
                title=title,
                author=_string_value(item, ("author", "authorName", "writer")),
                category=_string_value(item, ("category", "categoryName", "newCategory", "type")),
                intro=_string_value(item, ("intro", "brief", "description", "shortIntro")),
                source=source,
                note_count=note_count,
                progress_percent=progress,
            )
        )
    return profiles


def merge_profiles(*groups: list[BookProfile]) -> list[BookProfile]:
    merged: dict[str, BookProfile] = {}
    for group in groups:
        for profile in group:
            existing = merged.get(profile.book_id)
            if existing is None:
                merged[profile.book_id] = profile
                continue
            merged[profile.book_id] = BookProfile(
                book_id=existing.book_id,
                title=existing.title or profile.title,
                author=existing.author or profile.author,
                category=existing.category or profile.category,
                intro=existing.intro or profile.intro,
                source="/".join(dict.fromkeys([*existing.source.split("/"), *profile.source.split("/")])),
                note_count=max(existing.note_count, profile.note_count),
                progress_percent=existing.progress_percent if existing.progress_percent is not None else profile.progress_percent,
            )
    return list(merged.values())


def reading_evidence_ids(notebooks_payload: Any, progress_payloads: dict[str, Any] | None = None) -> set[str]:
    ids = {profile.book_id for profile in extract_book_profiles(notebooks_payload, source="notebooks") if profile.note_count > 0}
    for book_id, payload in (progress_payloads or {}).items():
        profiles = extract_book_profiles(payload, source="progress")
        if any((profile.progress_percent or 0) >= 20 for profile in profiles):
            ids.add(book_id)
    return ids

from __future__ import annotations

from dataclasses import dataclass
import re

from .formatting import weread_link


@dataclass(frozen=True)
class ContextItem:
    kind: str
    source: str
    text: str


@dataclass(frozen=True)
class Chapter:
    book_id: str
    book_title: str
    chapter_uid: str
    title: str


@dataclass(frozen=True)
class TodayChapterRecommendation:
    book_title: str
    chapter_title: str
    why: str
    reading_question: str
    apply_action: str
    deep_link: str


_KEYWORD_ALIASES = {
    "agent": ["agent", "skill", "subagent", "codex", "hermes", "ai software", "harness"],
    "product": ["产品", "pm", "用户", "增长", "需求"],
    "writing": ["写作", "选题", "公众号", "脚本", "内容"],
    "design": ["设计", "审美", "ui", "ux", "视觉"],
}

_NON_CONTENT_CHAPTER_TITLES = {
    "封面",
    "目录",
    "版权",
    "版权页",
    "版权信息",
    "书名页",
    "扉页",
    "献词",
    "插图",
    "插图目录",
    "内容简介",
    "作者简介",
    "前言",
    "序",
    "推荐序",
    "作者序",
    "再版序",
}

_BRIEF_TRIGGER_PHRASES = {
    "今天读什么",
    "今天读哪一小节",
    "推荐一章",
    "推荐一小节",
}


def _tokens(text: str) -> set[str]:
    return {t.lower() for t in re.findall(r"[A-Za-z][A-Za-z0-9_-]*|[\u4e00-\u9fff]{2,}", text)}


def _expanded_tokens(text: str) -> set[str]:
    tokens = _tokens(text)
    expanded = set(tokens)
    joined = " ".join(tokens)
    for key, aliases in _KEYWORD_ALIASES.items():
        if any(alias.lower() in joined for alias in aliases):
            expanded.update(alias.lower() for alias in aliases)
            expanded.add(key)
    return expanded


def is_substantive_chapter_title(title: str) -> bool:
    normalized = re.sub(r"\s+", "", title).strip("：:,.，。！？!?")
    if not normalized:
        return False
    return normalized not in _NON_CONTENT_CHAPTER_TITLES


def _has_specific_context(text: str) -> bool:
    normalized = re.sub(r"\s+", "", text).strip("：:,.，。！？!?")
    if not normalized:
        return False
    return normalized not in _BRIEF_TRIGGER_PHRASES


def choose_today_chapter(context: list[ContextItem], chapters: list[Chapter]) -> TodayChapterRecommendation:
    """Pick the chapter most connected to the current context."""
    chapters = [chapter for chapter in chapters if is_substantive_chapter_title(chapter.title)]
    if not chapters:
        raise ValueError("chapters must not be empty")
    context_text = "\n".join(item.text for item in context)
    context_tokens = _expanded_tokens(context_text)

    def score(chapter: Chapter) -> int:
        chapter_tokens = _expanded_tokens(f"{chapter.book_title} {chapter.title}")
        return len(context_tokens & chapter_tokens)

    chosen = max(chapters, key=score)
    strongest_context = context[0].text if context else "最近上下文不足"
    short_context = strongest_context.replace("\n", " ")[:80]
    if _has_specific_context(context_text):
        why = f"你最近的上下文里反复出现「{short_context}」，这一节最接近当前问题。"
    else:
        why = "当前没有可用的项目/笔记上下文；我先按微信读书最近书架和笔记里的可读章节推荐这一节。"

    return TodayChapterRecommendation(
        book_title=chosen.book_title,
        chapter_title=chosen.title,
        why=why,
        reading_question="读这一节时只问一个问题：它能解释我最近哪个真实卡点？",
        apply_action="读完后写下一条可以在今天项目里立刻改掉的动作。",
        deep_link=weread_link(chosen.book_id, chosen.chapter_uid),
    )

from __future__ import annotations

from dataclasses import dataclass
import re
from typing import Any

from .reading_profile import BookProfile, extract_book_profiles, merge_profiles
from .today_chapter import ContextItem, _expanded_tokens


@dataclass(frozen=True)
class UnreadRecommendation:
    book_id: str
    book_title: str
    author: str
    source: str
    why: str
    evidence: list[str]
    first_reading_question: str
    next_action: str


def _context_text(context: list[ContextItem], brief: str = "") -> str:
    return "\n".join([brief, *(item.text for item in context)]).strip()


def _known_ids(*groups: list[BookProfile]) -> set[str]:
    return {profile.book_id for group in groups for profile in group}


def _tokens_for_profile(profile: BookProfile) -> set[str]:
    return _expanded_tokens(" ".join([profile.title, profile.author, profile.category, profile.intro]))


def _score_profile(profile: BookProfile, context_tokens: set[str], shelf_ids: set[str], notebook_ids: set[str]) -> tuple[int, str]:
    profile_tokens = _tokens_for_profile(profile)
    overlap = context_tokens & profile_tokens
    score = len(overlap) * 10
    reasons: list[str] = []
    if overlap:
        reasons.append("命中当前问题关键词：" + "、".join(sorted(overlap)[:5]))
    if profile.book_id not in shelf_ids:
        score += 8
        reasons.append("不在当前书架里，更像真正的新候选")
    elif profile.book_id not in notebook_ids:
        score += 4
        reasons.append("在书架里但还没有明显笔记痕迹")
    if "recommend" in profile.source:
        score += 3
        reasons.append("来自微信读书个性化推荐")
    if "similar" in profile.source:
        score += 2
        reasons.append("来自相似书扩展")
    if "search" in profile.source:
        score += 1
        reasons.append("来自当前问题搜索")
    return score, "；".join(reasons) or "候选来源完整，但和当前问题的文本重合较少"


def recommend_unread_book(
    context: list[ContextItem],
    shelf_payload: Any,
    notebooks_payload: Any,
    recommendation_payloads: list[Any] | None = None,
    similar_payloads: list[Any] | None = None,
    search_payloads: list[Any] | None = None,
    brief: str = "",
) -> UnreadRecommendation:
    """Recommend one not-yet-digested book using shelf, notes and candidate sources."""
    shelf = extract_book_profiles(shelf_payload, source="shelf")
    notebooks = extract_book_profiles(notebooks_payload, source="notebooks")
    recommended = [
        profile
        for payload in (recommendation_payloads or [])
        for profile in extract_book_profiles(payload, source="recommend")
    ]
    similar = [
        profile
        for payload in (similar_payloads or [])
        for profile in extract_book_profiles(payload, source="similar")
    ]
    search = [
        profile
        for payload in (search_payloads or [])
        for profile in extract_book_profiles(payload, source="search")
    ]
    candidates = merge_profiles(recommended, similar, search, shelf)
    if not candidates:
        raise ValueError("no book candidates")

    shelf_ids = _known_ids(shelf)
    notebook_ids = _known_ids(notebooks)
    context_tokens = _expanded_tokens(_context_text(context, brief))

    ranked = []
    for profile in candidates:
        if profile.book_id in notebook_ids and profile.note_count > 0:
            continue
        score, reason = _score_profile(profile, context_tokens, shelf_ids, notebook_ids)
        ranked.append((score, profile, reason))
    if not ranked:
        raise ValueError("no unread candidates after filtering notebook evidence")

    ranked.sort(key=lambda item: (item[0], item[1].title), reverse=True)
    score, chosen, reason = ranked[0]
    evidence = [reason]
    if chosen.book_id in shelf_ids:
        evidence.append("书架里能看到这本书，但当前笔记列表没有显示已消化证据。")
    else:
        evidence.append("书架和笔记中暂未看到这本书，适合作为新输入。")
    if chosen.category:
        evidence.append(f"分类线索：{chosen.category}")
    if chosen.intro:
        clean_intro = re.sub(r"\s+", " ", chosen.intro).strip()[:120]
        evidence.append(f"简介线索：{clean_intro}")

    return UnreadRecommendation(
        book_id=chosen.book_id,
        book_title=chosen.title,
        author=chosen.author,
        source=chosen.source,
        why=f"这本书和当前问题最接近，同时没有明显读后笔记证据。评分口径是：当前上下文匹配、是否已在书架、是否已有笔记、候选来源可信度。",
        evidence=evidence,
        first_reading_question="翻开后先看目录，只问：哪一章能解释我现在最卡的那个问题？",
        next_action="不要先通读。先选一个章节，读完只写一条划线和一张行动卡。",
    )


def format_unread_recommendation(rec: UnreadRecommendation) -> str:
    author = f"｜{rec.author}" if rec.author else ""
    lines = [
        "今天推荐一本没读透但现在该读的书：",
        f"《{rec.book_title}》{author}",
        "",
        "为什么是它：",
        rec.why,
        "",
        "证据：",
        *[f"- {item}" for item in rec.evidence],
        "",
        "读前问题：",
        rec.first_reading_question,
        "",
        "读完只做一个动作：",
        rec.next_action,
    ]
    return "\n".join(lines)

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
import re


@dataclass(frozen=True)
class WeeklyReportInput:
    cards: list[str]
    context: str = ""


def _first_heading(markdown: str) -> str:
    for line in markdown.splitlines():
        line = line.strip()
        if line.startswith("# "):
            return line[2:].strip()
    return "未命名阅读卡"


def _section(markdown: str, heading: str) -> str:
    pattern = re.compile(rf"^## {re.escape(heading)}\s*$", re.MULTILINE)
    match = pattern.search(markdown)
    if not match:
        return ""
    start = match.end()
    next_match = re.search(r"^## ", markdown[start:], re.MULTILINE)
    end = start + next_match.start() if next_match else len(markdown)
    return markdown[start:end].strip()


def load_cards(paths: list[str | Path]) -> list[str]:
    cards: list[str] = []
    for raw_path in paths:
        path = Path(raw_path).expanduser()
        if path.is_dir():
            files = sorted(path.glob("*.md"), key=lambda p: p.stat().st_mtime, reverse=True)
            cards.extend(p.read_text(encoding="utf-8", errors="ignore") for p in files)
        elif path.exists():
            cards.append(path.read_text(encoding="utf-8", errors="ignore"))
    return [card for card in cards if card.strip()]


def build_weekly_report(report_input: WeeklyReportInput) -> str:
    """Build a concise weekly reading-action report from action cards and context."""
    cards = report_input.cards
    headings = [_first_heading(card) for card in cards]
    actions = [_section(card, "可以马上改的一个动作") for card in cards]
    topics = [_section(card, "可发展的内容选题") for card in cards]
    problems = [_section(card, "它解释了我最近哪个问题") for card in cards]
    evidence = [action for action in actions if action]

    if not cards:
        main_reading = "本周还没有可复盘的阅读行动卡。"
        evidence_block = "没有证据。先完成一次 today → after-read，再复盘。"
        impact_block = "阅读还没有进入文章、项目或判断。"
        escape_block = "如果只是收藏书和章节，但没有行动卡，就算作逃避式收藏。"
        next_line = "下周只做一件事：完成一张阅读行动卡。"
        stop_line = "先停止扩书单，直到第一张行动卡出现。"
        candidates = ["从当前最卡的问题开始，而不是从书单开始。"]
    else:
        main_reading = "\n".join(f"- {heading}" for heading in headings[:5])
        evidence_block = "\n".join(f"- {item.splitlines()[0]}" for item in evidence[:5]) or "还缺少行动证据。"
        impact_block = _derive_impact(report_input.context, problems, topics)
        escape_block = "没有行动卡支撑的章节先不算读进去；下周减少只收藏不应用的阅读。"
        next_line = _derive_next_line(report_input.context, headings, topics)
        stop_line = _derive_stop_line(headings, evidence)
        candidates = _derive_candidates(headings, topics)

    return "\n".join([
        "# 本周阅读行动闭环",
        "",
        "## 本周真正读进去的东西",
        main_reading,
        "",
        "## 已经进入工作的证据",
        evidence_block,
        "",
        "## 阅读影响",
        impact_block,
        "",
        "## 逃避式收藏",
        escape_block,
        "",
        "## 下周停止读什么",
        stop_line,
        "",
        "## 下周只保留一条阅读主线",
        next_line,
        "",
        "## 下周 3 个 Today Chapter 候选",
        *[f"{idx}. {candidate}" for idx, candidate in enumerate(candidates[:3], start=1)],
    ]).strip()


def _derive_next_line(context: str, headings: list[str], topics: list[str]) -> str:
    text = " ".join([context, *headings, *topics])
    if any(word in text for word in ("文章", "选题", "公众号", "内容", "脚本")):
        return "把阅读服务于一篇可发布内容：每次读完只沉淀一个能进入文章的判断。"
    if any(word.lower() in text.lower() for word in ("agent", "skill", "workflow", "hermes")):
        return "把阅读服务于一个可演示工作流：每次读完只改进一个可验证动作。"
    return "把阅读服务于当前最卡的项目：每次读完只推进一个今天能完成的小动作。"


def _derive_impact(context: str, problems: list[str], topics: list[str]) -> str:
    text = " ".join([context, *problems, *topics])
    if any(word in text for word in ("文章", "选题", "公众号", "脚本", "内容")):
        return "本周阅读已经进入内容生产：优先把行动卡里的判断改写成文章段落或视频脚本测试点。"
    if any(word.lower() in text.lower() for word in ("agent", "skill", "workflow", "cli", "hermes")):
        return "本周阅读已经进入产品/工程判断：优先把行动卡里的概念变成可运行命令、测试或README说明。"
    if any(item.strip() for item in problems):
        return "本周阅读至少解释了一个真实问题；下周要看它有没有改变具体行动。"
    return "行动卡里还缺少清晰问题，影响证据偏弱。下周先补「它解释了我最近哪个问题」。"


def _derive_stop_line(headings: list[str], evidence: list[str]) -> str:
    if not evidence:
        return "停止继续收藏新书；先把本周已读章节补成一张能行动的卡。"
    if len(headings) >= 4:
        return "停止横向开新主题；下周只沿着最有行动证据的一条线往下读。"
    return "停止为了安心而补书单；只保留能推动当前动作的章节。"


def _derive_candidates(headings: list[str], topics: list[str]) -> list[str]:
    candidates: list[str] = []
    for heading in headings[:3]:
        candidates.append(f"延续 {heading}：找下一节能直接改动作的内容")
    for topic in topics:
        lines = topic.splitlines()
        if not lines:
            continue
        first = lines[0].replace("标题方向：", "").strip()
        if first:
            candidates.append(f"围绕「{first}」补一节案例型阅读")
    candidates.append("从本周最卡的项目问题倒推一节书，而不是顺着目录读。")
    return candidates[:3]

from __future__ import annotations

from dataclasses import dataclass
from datetime import date
from pathlib import Path
import re

from .config import ContextConfig


@dataclass(frozen=True)
class WritebackResult:
    path: Path | None
    mode: str
    message: str


def slugify_title(title: str, max_length: int = 48) -> str:
    """Build a filesystem-safe, readable slug for Chinese/English titles."""
    cleaned = re.sub(r'[\\/:*?"<>|#\n\r\t]+', "-", title).strip(" -")
    cleaned = re.sub(r"\s+", "-", cleaned)
    cleaned = re.sub(r"-+", "-", cleaned)
    return (cleaned or "reading-card")[:max_length].strip(" -") or "reading-card"


def default_writeback_dir(config: ContextConfig) -> Path | None:
    """Return the default directory for persisted action cards, if this mode writes files."""
    if config.mode == "obsidian" and config.path is not None:
        vault = Path(config.path).expanduser()
        preferred = vault / "15_自媒体" / "阅读行动卡"
        fallback = vault / "阅读行动卡"
        return preferred if preferred.parent.exists() else fallback
    if config.mode == "folder" and config.path is not None:
        return Path(config.path).expanduser() / "carl-weread" / "reading-cards"
    return None


def write_action_card(
    card_markdown: str,
    config: ContextConfig,
    title: str,
    output_dir: str | Path | None = None,
    today: date | None = None,
) -> WritebackResult:
    """Persist an action card when the configured mode has a real filesystem target.

    Chat and WeRead-only modes intentionally do not write files; the caller should return
    the markdown in conversation instead.
    """
    target_dir = Path(output_dir).expanduser() if output_dir is not None else default_writeback_dir(config)
    if target_dir is None:
        return WritebackResult(path=None, mode=config.mode, message="当前模式不写文件，已在对话中返回行动卡。")

    day = today or date.today()
    target_dir.mkdir(parents=True, exist_ok=True)
    path = target_dir / f"{day:%Y-%m-%d}-{slugify_title(title)}.md"
    path.write_text(card_markdown.rstrip() + "\n", encoding="utf-8")
    return WritebackResult(path=path, mode=config.mode, message=f"已写入阅读行动卡：{path}")

[
  {
    "book_id": "demo-agent-book",
    "book_title": "AI Harness Engineering",
    "chapter_uid": "chapter-7",
    "title": "Agent Skill 交付验证"
  },
  {
    "book_id": "demo-writing-book",
    "book_title": "写作是门手艺",
    "chapter_uid": "chapter-2",
    "title": "从真实问题开始写"
  }
]

[project]
name = "carl-weread"
version = "0.3.0"
description = "Context-driven WeRead action coach skill."
requires-python = ">=3.9"
dependencies = []

[project.scripts]
carl-weread = "carl_weread.cli:main"

[tool.pytest.ini_options]
pythonpath = ["."]
testpaths = ["tests"]

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

import argparse
import json
from pathlib import Path
import sys

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from carl_weread.candidates import build_candidate_chapters, chapters_to_jsonable


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


def _load_chapterinfo_by_book_id(chapter_dir: Path) -> dict[str, object]:
    payloads: dict[str, object] = {}
    for path in sorted(chapter_dir.glob("*.json")):
        payloads[path.stem] = _read_json(path)
    return payloads


def main() -> int:
    parser = argparse.ArgumentParser(description="Build today-compatible chapter candidates from WeRead API JSON files.")
    parser.add_argument("--shelf", required=True, type=Path, help="JSON output from scripts/weread.sh shelf")
    parser.add_argument("--notebooks", required=True, type=Path, help="JSON output from scripts/weread.sh notebooks")
    parser.add_argument("--chapter-dir", required=True, type=Path, help="Directory containing <bookId>.json chapterinfo files")
    parser.add_argument("--output", required=True, type=Path, help="Output candidate chapters JSON")
    parser.add_argument("--limit-books", type=int, help="Maximum books to include after merging shelf and notebooks")
    args = parser.parse_args()

    candidates = build_candidate_chapters(
        _read_json(args.shelf),
        _read_json(args.notebooks),
        _load_chapterinfo_by_book_id(args.chapter_dir),
        limit_books=args.limit_books,
    )
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(chapters_to_jsonable(candidates), ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"已写入候选章节：{args.output}（{len(candidates)} 条）")
    return 0


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

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

from pathlib import Path
import sys

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from carl_weread.cli import main


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