noteのコメントをPythonで取得・投稿する｜AST形式の書き方

{
  "type": "root",
  "children": [
    {
      "type": "element",
      "tag_name": "p",
      "children": [
        {
          "type": "text",
          "value": "記事を読みました。参考になりました。"
        }
      ]
    }
  ]
}

"""
noteのコメントAST（抽象構文木）とテキストを相互変換する。
"""


def ast_to_text(node) -> str:
    """
    ASTからプレーンテキストを取り出す。
    段落ごとに改行を入れる。
    """
    if node is None:
        return ""

    # 文字列がそのまま入っている場合もある
    if isinstance(node, str):
        return node

    if not isinstance(node, dict):
        return ""

    node_type = node.get("type")

    if node_type == "text":
        return node.get("value", "")

    children = node.get("children") or []
    parts = [ast_to_text(child) for child in children]

    if node_type == "element":
        tag = node.get("tag_name", "")
        if tag == "br":
            return "\n"
        if tag in ("p", "div"):
            return "".join(parts) + "\n"

    return "".join(parts)


def text_to_ast(text: str) -> dict:
    """
    プレーンテキストからASTを組み立てる。
    改行で段落を分ける。
    """
    if not text or not text.strip():
        raise ValueError("本文が空です")

    lines = [l.strip() for l in text.split("\n")]

    children = []
    for line in lines:
        if not line:
            continue

        children.append({
            "type": "element",
            "tag_name": "p",
            "children": [
                {"type": "text", "value": line}
            ],
        })

    if not children:
        raise ValueError("有効な行がありません")

    return {"type": "root", "children": children}


def describe_ast(node, depth: int = 0):
    """ASTの構造を確認する（デバッグ用）"""
    if not isinstance(node, dict):
        print("  " * depth + repr(node)[:60])
        return

    node_type = node.get("type", "?")

    if node_type == "text":
        value = node.get("value", "")
        print("  " * depth + f"text: {value[:40]}")
        return

    tag = node.get("tag_name", "")
    label = f"{node_type}" + (f"({tag})" if tag else "")
    print("  " * depth + label)

    for child in node.get("children") or []:
        describe_ast(child, depth + 1)


if __name__ == "__main__":
    sample_text = "記事を読みました。\n参考になりました。"

    ast = text_to_ast(sample_text)
    print("=== 組み立てたAST ===")
    describe_ast(ast)

    print("\n=== 戻したテキスト ===")
    print(repr(ast_to_text(ast)))

"""
X-Note-Client-Code を扱う。

このヘッダーは、ブラウザがコメント投稿時に送っている識別子。
DevToolsのネットワークタブで、実際のコメント投稿リクエストを
確認すると値が分かる。
"""
import os

from dotenv import load_dotenv

load_dotenv()


def get_client_code() -> str:
    """
    .env から X-Note-Client-Code の値を読む。

    .env の例:
        NOTE_CLIENT_CODE=64文字の16進数
    """
    code = os.getenv("NOTE_CLIENT_CODE", "").strip()

    if not code:
        raise ValueError(
            "NOTE_CLIENT_CODE が設定されていません。\n"
            "ブラウザでコメントを投稿し、DevToolsのネットワークタブから\n"
            "リクエストヘッダーの X-Note-Client-Code を確認して\n"
            ".env に設定してください"
        )

    if len(code) != 64:
        print(f"警告: 想定と長さが違います（{len(code)} 文字）")

    return code


def comment_headers() -> dict:
    """コメント投稿用のヘッダーを組み立てる"""
    return {
        "X-Note-Client-Code": get_client_code(),
        "Content-Type": "application/json",
    }


if __name__ == "__main__":
    try:
        code = get_client_code()
        print(f"取得しました（先頭8文字: {code[:8]}...）")
    except ValueError as e:
        print(e)

"""
noteのコメント操作をまとめたクライアント。
"""
import logging

from note_auth_client import NoteAuthClient
from ast_utils import ast_to_text, text_to_ast
from client_code import get_client_code

logger = logging.getLogger(__name__)


class CommentClient:

    def __init__(self, client: NoteAuthClient,
                 client_code: str = None):
        self.client = client
        self._client_code = client_code

    def _headers(self) -> dict:
        code = self._client_code or get_client_code()
        return {"X-Note-Client-Code": code}

    # ---- 取得 ----

    def list_comments(self, note_key: str, order: str = "newest",
                      per_page: int = 20,
                      max_pages: int = 10) -> list:
        """記事のコメント一覧を取得する"""
        comments = []

        for page in range(1, max_pages + 1):
            data = self.client.get_json(
                f"/api/v3/notes/{note_key}/note_comments",
                params={
                    "page": page,
                    "per_page": per_page,
                    "order": order,
                },
            )

            items = data if isinstance(data, list) else data.get("data", [])
            if not items:
                break

            comments.extend(items)

            if len(items) < per_page:
                break

        return comments

    def list_replies(self, note_key: str, parent_key: str) -> list:
        """特定コメントへの返信を取得する"""
        data = self.client.get_json(
            f"/api/v3/notes/{note_key}/note_comments",
            params={"parent_key": parent_key, "per_page": 50},
        )
        return data if isinstance(data, list) else data.get("data", [])

    @staticmethod
    def to_readable(comment: dict) -> dict:
        """コメントを扱いやすい形に変換する"""
        user = comment.get("user") or {}

        return {
            "key": comment.get("key", ""),
            "text": ast_to_text(comment.get("comment")).strip(),
            "user_key": user.get("key", ""),
            "nickname": user.get("nickname", ""),
            "urlname": user.get("urlname", ""),
            "is_root": comment.get("is_root", True),
            "created_at": comment.get("created_at",
                                      comment.get("createdAt", "")),
        }

    # ---- 投稿 ----

    def post_comment(self, note_key: str, text: str,
                     parent_key: str = None,
                     dry_run: bool = True) -> dict:
        """
        コメントを投稿する。
        parent_key を指定すると返信になる。
        """
        payload = {
            "comment": text_to_ast(text),
            "acknowledgement": False,
        }

        if parent_key:
            payload["parent_key"] = parent_key

        print("=== 投稿内容 ===")
        print("  記事:", note_key)
        print("  種別:", "返信" if parent_key else "新規コメント")
        print("  本文:", text[:80])

        if dry_run:
            print("\n[確認モード] 実際には投稿していません")
            return {}

        res = self.client.request(
            "POST",
            f"/api/v3/notes/{note_key}/note_comments",
            json=payload,
            headers=self._headers(),
        )

        if res.status_code not in (200, 201):
            logger.error("投稿失敗 HTTP %s: %s",
                         res.status_code, res.text[:200])
            return {}

        try:
            data = res.json().get("data", {})
        except ValueError:
            data = {}

        logger.info("コメントを投稿しました")
        return data

    def edit_comment(self, note_key: str, comment_key: str,
                     text: str, dry_run: bool = True) -> bool:
        """自分のコメントを編集する"""
        payload = {"comment": text_to_ast(text)}

        print(f"=== 編集: {comment_key} ===")
        print("  新しい本文:", text[:80])

        if dry_run:
            print("\n[確認モード] 実際には送信していません")
            return False

        res = self.client.request(
            "PUT",
            f"/api/v3/notes/{note_key}/note_comments/{comment_key}",
            json=payload,
            headers=self._headers(),
        )

        if res.status_code not in (200, 201):
            logger.error("編集失敗 HTTP %s", res.status_code)
            return False

        return True

    def delete_comment(self, note_key: str,
                       comment_key: str) -> bool:
        """コメントを削除する（comment_keyはnc形式）"""
        res = self.client.request(
            "DELETE",
            f"/api/v3/notes/{note_key}/note_comments/{comment_key}",
            headers=self._headers(),
        )

        if res.status_code not in (200, 204):
            logger.error("削除失敗 HTTP %s", res.status_code)
            return False

        logger.info("コメントを削除しました: %s", comment_key)
        return True


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)

    client = NoteAuthClient()
    client.verify()

    cc = CommentClient(client)

    note_key = "nXXXXXXXXXXXX"
    comments = cc.list_comments(note_key)

    print(f"=== コメント {len(comments)} 件 ===")
    for c in comments:
        r = cc.to_readable(c)
        print(f"\n  {r['nickname']}（{r['created_at'][:10]}）")
        print(f"  {r['text'][:100]}")

"""
全記事を横断して、未返信のコメントを抽出する。
返信忘れを防ぐための確認スクリプト。
"""
import time

from note_auth_client import NoteAuthClient, CookieExpiredError
from comment_client import CommentClient

URLNAME = "your_note_id"
SLEEP_SEC = 1.5
MAX_ARTICLES = 100


def fetch_my_notes(client: NoteAuthClient, urlname: str,
                   max_pages: int = 50) -> list:
    items = []

    for page in range(1, max_pages + 1):
        data = client.get_json(
            f"/api/v2/creators/{urlname}/contents",
            params={"kind": "note", "page": page},
        )
        contents = data.get("contents", [])
        if not contents:
            break

        items.extend(contents)

        if data.get("isLastPage") or data.get("is_last_page"):
            break

        time.sleep(1.2)

    return items


def has_comments(item: dict) -> bool:
    """コメント数のフィールドで事前に絞り込む"""
    for key in ("commentCount", "comment_count"):
        if item.get(key):
            return True
    return False


def find_unreplied(client: NoteAuthClient, my_user_key: str,
                   notes: list) -> list:
    """自分が返信していないコメントを抽出する"""
    cc = CommentClient(client)
    unreplied = []

    targets = [n for n in notes if has_comments(n)][:MAX_ARTICLES]
    print(f"コメントがある記事: {len(targets)} 本\n")

    for i, note in enumerate(targets, start=1):
        note_key = note.get("key")
        title = note.get("name", "")

        if not note_key:
            continue

        comments = cc.list_comments(note_key)
        readable = [cc.to_readable(c) for c in comments]

        # 自分以外のルートコメントを抽出
        others = [
            r for r in readable
            if r["is_root"] and r["user_key"] != my_user_key
        ]

        for parent in others:
            replies = cc.list_replies(note_key, parent["key"])
            reply_users = {
                cc.to_readable(r)["user_key"] for r in replies
            }

            if my_user_key not in reply_users:
                unreplied.append({
                    "note_key": note_key,
                    "title": title,
                    "comment": parent,
                })

            time.sleep(SLEEP_SEC)

        print(f"[{i}/{len(targets)}] {title[:40]}"
              f"（コメント {len(others)} 件）")

        time.sleep(SLEEP_SEC)

    return unreplied


def main():
    try:
        client = NoteAuthClient()
        user = client.verify()
    except CookieExpiredError as e:
        print("認証エラー:", e)
        return

    my_user_key = user.get("key") or str(user.get("id", ""))

    notes = fetch_my_notes(client, URLNAME)
    print(f"記事 {len(notes)} 本を確認します\n")

    unreplied = find_unreplied(client, my_user_key, notes)

    print(f"\n=== 未返信のコメント: {len(unreplied)} 件 ===")

    for item in unreplied:
        c = item["comment"]
        print(f"\n[{item['title'][:40]}]")
        print(f"  {c['nickname']}（{c['created_at'][:10]}）")
        print(f"  {c['text'][:120]}")
        print(f"  https://note.com/notes/{item['note_key']}")


if __name__ == "__main__":
    main()

"""
コメントを集計して、傾向を把握する。
- どの記事にコメントが集まるか
- 誰が繰り返しコメントしてくれているか
- コメントの長さの分布
"""
import csv
import time
from collections import Counter
from datetime import datetime

from note_auth_client import NoteAuthClient, CookieExpiredError
from comment_client import CommentClient
from find_unreplied import fetch_my_notes, has_comments

URLNAME = "your_note_id"
OUTPUT = "comments.csv"


def collect_all(client: NoteAuthClient, notes: list,
                limit: int = 80) -> list:
    cc = CommentClient(client)
    rows = []

    targets = [n for n in notes if has_comments(n)][:limit]

    for i, note in enumerate(targets, start=1):
        note_key = note.get("key")
        title = note.get("name", "")

        if not note_key:
            continue

        for c in cc.list_comments(note_key):
            r = cc.to_readable(c)
            rows.append({
                "note_key": note_key,
                "title": title,
                "comment_key": r["key"],
                "nickname": r["nickname"],
                "urlname": r["urlname"],
                "text": r["text"].replace("\n", " "),
                "length": len(r["text"]),
                "created_at": r["created_at"],
            })

        print(f"[{i}/{len(targets)}] {title[:40]}")
        time.sleep(1.5)

    return rows


def analyze(rows: list):
    if not rows:
        print("コメントがありません")
        return

    print(f"\n=== コメント {len(rows)} 件 ===")

    by_article = Counter(r["title"] for r in rows)
    print("\n--- コメントが多い記事 ---")
    for title, count in by_article.most_common(10):
        print(f"  {count:>3} 件  {title[:44]}")

    by_user = Counter(
        r["nickname"] for r in rows if r["nickname"]
    )
    print("\n--- よくコメントをくれる方 ---")
    for name, count in by_user.most_common(10):
        print(f"  {count:>3} 件  {name[:24]}")

    lengths = [r["length"] for r in rows]
    avg = sum(lengths) / len(lengths)
    print(f"\n--- コメントの長さ ---")
    print(f"  平均: {avg:.0f} 文字")
    print(f"  最長: {max(lengths)} 文字")

    long_comments = [r for r in rows if r["length"] > 100]
    print(f"  100文字超: {len(long_comments)} 件")

    if long_comments:
        print("\n  --- 長文コメントの例 ---")
        for r in long_comments[:3]:
            print(f"  [{r['title'][:30]}]")
            print(f"  {r['text'][:100]}...")


def save_csv(rows: list, path: str = OUTPUT):
    if not rows:
        return

    with open(path, "w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
        writer.writeheader()
        writer.writerows(rows)

    print(f"\n{path} に保存しました")


def main():
    try:
        client = NoteAuthClient()
        client.verify()
    except CookieExpiredError as e:
        print("認証エラー:", e)
        return

    notes = fetch_my_notes(client, URLNAME)
    rows = collect_all(client, notes)

    analyze(rows)
    save_csv(rows)


if __name__ == "__main__":
    main()