Pythonで自分のnote全記事を一括取得してCSV保存する

import time

import requests

URLNAME = "your_note_id"   # ← 自分のnote IDに変更
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
      "AppleWebKit/537.36 (KHTML, like Gecko) "
      "Chrome/120.0.0.0 Safari/537.36")
SLEEP_SEC = 1.2
MAX_PAGES = 100   # 安全のため上限を置く


def fetch_contents(urlname: str) -> list:
    """クリエイターの記事一覧を全ページ取得する"""
    session = requests.Session()
    session.headers.update({"User-Agent": UA, "Accept": "application/json"})

    all_items = []

    for page in range(1, MAX_PAGES + 1):
        res = session.get(
            f"https://note.com/api/v2/creators/{urlname}/contents",
            params={"kind": "note", "page": page},
            timeout=20,
        )

        if res.status_code != 200:
            print(f"page {page}: HTTP {res.status_code} で中断")
            break

        data = res.json().get("data", {})
        contents = data.get("contents", [])

        if not contents:
            print(f"page {page}: 記事がないため終了")
            break

        all_items.extend(contents)
        print(f"page {page}: {len(contents)} 件（累計 {len(all_items)} 件）")

        if data.get("isLastPage") or data.get("is_last_page"):
            print("最終ページに到達しました")
            break

        time.sleep(SLEEP_SEC)

    return all_items


if __name__ == "__main__":
    items = fetch_contents(URLNAME)
    print(f"\n合計 {len(items)} 件を取得しました")

    for item in items[:10]:
        likes = item.get("likeCount", item.get("like_count", 0))
        print(f"  スキ{likes:>4} | {item.get('name', '')}")

import csv
from datetime import datetime

from fetch_all import fetch_contents, URLNAME


def pick(d: dict, *keys, default=None):
    """camelCase / snake_case の表記ゆれを吸収する"""
    for k in keys:
        if k in d and d[k] is not None:
            return d[k]
    return default


def to_rows(items: list, urlname: str) -> list:
    """APIのレスポンスをCSV用の行に変換する"""
    rows = []
    for item in items:
        key = item.get("key", "")
        price = pick(item, "price", default=0) or 0

        rows.append({
            "key": key,
            "id": item.get("id", ""),
            "title": item.get("name", ""),
            "published_at": pick(item, "publishAt", "publish_at", default=""),
            "likes": pick(item, "likeCount", "like_count", default=0),
            "comments": pick(item, "commentCount", "comment_count", default=0),
            "price": price,
            "is_paid": "有料" if price else "無料",
            "type": item.get("type", ""),
            "has_eyecatch": "あり" if item.get("eyecatch") else "なし",
            "url": f"https://note.com/{urlname}/n/{key}",
        })
    return rows


def save_csv(rows: list, filename: str = None) -> str:
    if not rows:
        print("保存対象がありません")
        return ""

    if filename is None:
        stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"note_articles_{stamp}.csv"

    # utf-8-sig にするとExcelで開いても文字化けしない
    with open(filename, "w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
        writer.writeheader()
        writer.writerows(rows)

    print(f"{filename} に {len(rows)} 件を保存しました")
    return filename


if __name__ == "__main__":
    items = fetch_contents(URLNAME)
    rows = to_rows(items, URLNAME)
    save_csv(rows)

import pandas as pd


def load(filename: str) -> pd.DataFrame:
    df = pd.read_csv(filename, encoding="utf-8-sig")

    # 日付型に変換（タイムゾーン付き文字列にも対応）
    df["published_at"] = pd.to_datetime(
        df["published_at"], errors="coerce", utc=True
    ).dt.tz_convert("Asia/Tokyo")

    df["year_month"] = df["published_at"].dt.strftime("%Y-%m")
    df["weekday"] = df["published_at"].dt.day_name()
    df["hour"] = df["published_at"].dt.hour
    df["title_len"] = df["title"].astype(str).str.len()

    return df


def summarize(df: pd.DataFrame):
    print("=== 全体 ===")
    print(f"  記事数     : {len(df)} 本")
    print(f"  合計スキ   : {df['likes'].sum():,}")
    print(f"  平均スキ   : {df['likes'].mean():.1f}")
    print(f"  中央値スキ : {df['likes'].median():.0f}")
    print(f"  有料記事   : {(df['price'] > 0).sum()} 本")

    print("\n=== 月別の投稿数と平均スキ ===")
    monthly = df.groupby("year_month").agg(
        posts=("key", "count"),
        avg_likes=("likes", "mean"),
    ).round(1)
    print(monthly.tail(12))

    print("\n=== スキが多い記事 TOP10 ===")
    top = df.nlargest(10, "likes")[["likes", "title"]]
    for _, row in top.iterrows():
        print(f"  {row['likes']:>5} | {row['title'][:45]}")

    print("\n=== タイトル文字数とスキ数の相関 ===")
    corr = df[["title_len", "likes"]].corr().iloc[0, 1]
    print(f"  相関係数: {corr:.3f}")
    print("  （0.3以上なら弱い関連、0.5以上でやや強い関連）")


if __name__ == "__main__":
    import sys

    filename = sys.argv[1] if len(sys.argv) > 1 else "note_articles.csv"
    df = load(filename)
    summarize(df)

"""
インストール:
    pip install pandas openpyxl
"""
from datetime import datetime

import pandas as pd

from analyze import load


def export_excel(df: pd.DataFrame, filename: str = None) -> str:
    if filename is None:
        stamp = datetime.now().strftime("%Y%m%d")
        filename = f"note_report_{stamp}.xlsx"

    monthly = df.groupby("year_month").agg(
        posts=("key", "count"),
        total_likes=("likes", "sum"),
        avg_likes=("likes", "mean"),
    ).round(1).reset_index()

    paid = df[df["price"] > 0][
        ["title", "price", "likes", "published_at", "url"]
    ].sort_values("price", ascending=False)

    top = df.nlargest(30, "likes")[["title", "likes", "comments", "url"]]

    with pd.ExcelWriter(filename, engine="openpyxl") as writer:
        df.to_excel(writer, sheet_name="全記事", index=False)
        monthly.to_excel(writer, sheet_name="月別集計", index=False)
        top.to_excel(writer, sheet_name="スキ上位", index=False)
        if not paid.empty:
            paid.to_excel(writer, sheet_name="有料記事", index=False)

        # 列幅を自動調整する
        for sheet in writer.sheets.values():
            for column in sheet.columns:
                length = max(
                    (len(str(cell.value)) for cell in column if cell.value),
                    default=10,
                )
                letter = column[0].column_letter
                sheet.column_dimensions[letter].width = min(length + 2, 60)

    print(f"{filename} を出力しました")
    return filename


if __name__ == "__main__":
    df = load("note_articles.csv")
    export_excel(df)

"""
既存CSVに無い記事だけを取得して追記する。
毎日実行しても負荷が小さい。
"""
import csv
import os
import time

import requests

URLNAME = "your_note_id"
CSV_FILE = "note_articles.csv"
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
      "AppleWebKit/537.36 (KHTML, like Gecko) "
      "Chrome/120.0.0.0 Safari/537.36")

FIELDS = ["key", "id", "title", "published_at", "likes",
          "comments", "price", "is_paid", "type", "url"]


def load_existing_keys(path: str) -> set:
    if not os.path.exists(path):
        return set()
    with open(path, encoding="utf-8-sig", newline="") as f:
        return {row["key"] for row in csv.DictReader(f)}


def fetch_new(urlname: str, known: set, max_pages: int = 100) -> list:
    """既知のキーに当たるまで新しい順に取得する"""
    session = requests.Session()
    session.headers.update({"User-Agent": UA})

    new_items = []

    for page in range(1, max_pages + 1):
        res = session.get(
            f"https://note.com/api/v2/creators/{urlname}/contents",
            params={"kind": "note", "page": page},
            timeout=20,
        )
        if res.status_code != 200:
            break

        data = res.json().get("data", {})
        contents = data.get("contents", [])
        if not contents:
            break

        hit_known = False
        for item in contents:
            if item.get("key") in known:
                hit_known = True
                continue
            new_items.append(item)

        # 既知の記事に到達したら、それ以降は取得済みとみなす
        if hit_known:
            print(f"page {page} で既知の記事に到達。取得を終了します")
            break

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

        time.sleep(1.2)

    return new_items


def append_rows(items: list, path: str, urlname: str):
    if not items:
        print("新しい記事はありません")
        return

    exists = os.path.exists(path)
    with open(path, "a", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        if not exists:
            writer.writeheader()

        for item in items:
            key = item.get("key", "")
            price = item.get("price", 0) or 0
            writer.writerow({
                "key": key,
                "id": item.get("id", ""),
                "title": item.get("name", ""),
                "published_at": item.get("publishAt",
                                         item.get("publish_at", "")),
                "likes": item.get("likeCount", item.get("like_count", 0)),
                "comments": item.get("commentCount",
                                     item.get("comment_count", 0)),
                "price": price,
                "is_paid": "有料" if price else "無料",
                "type": item.get("type", ""),
                "url": f"https://note.com/{urlname}/n/{key}",
            })

    print(f"{len(items)} 件を追記しました")


if __name__ == "__main__":
    known = load_existing_keys(CSV_FILE)
    print(f"既存: {len(known)} 件")

    new_items = fetch_new(URLNAME, known)
    append_rows(new_items, CSV_FILE, URLNAME)

"""
毎日実行して、その日時点のスキ数を履歴として残す。
後から「どの記事がいつ伸びたか」を追える。
"""
import csv
import os
from datetime import date

from fetch_all import fetch_contents, URLNAME

HISTORY_FILE = "note_likes_history.csv"


def snapshot(urlname: str, path: str = HISTORY_FILE):
    items = fetch_contents(urlname)
    today = date.today().isoformat()

    exists = os.path.exists(path)
    with open(path, "a", encoding="utf-8-sig", newline="") as f:
        writer = csv.writer(f)
        if not exists:
            writer.writerow(["date", "key", "title", "likes", "comments"])

        for item in items:
            writer.writerow([
                today,
                item.get("key", ""),
                item.get("name", ""),
                item.get("likeCount", item.get("like_count", 0)),
                item.get("commentCount", item.get("comment_count", 0)),
            ])

    print(f"{today} 時点のスナップショットを {len(items)} 件記録しました")


def show_growth(path: str = HISTORY_FILE, days: int = 7):
    """直近の伸び幅を表示する"""
    import pandas as pd

    df = pd.read_csv(path, encoding="utf-8-sig")
    df["date"] = pd.to_datetime(df["date"])

    latest = df["date"].max()
    past = latest - pd.Timedelta(days=days)

    now = df[df["date"] == latest].set_index("key")
    before = df[df["date"] <= past]

    if before.empty:
        print("比較できる過去データがまだありません")
        return

    before = before[before["date"] == before["date"].max()].set_index("key")

    joined = now.join(before["likes"], rsuffix="_before").dropna()
    joined["growth"] = joined["likes"] - joined["likes_before"]

    print(f"=== 直近{days}日で伸びた記事 ===")
    for _, row in joined.nlargest(10, "growth").iterrows():
        if row["growth"] > 0:
            print(f"  +{int(row['growth']):>3} | {row['title'][:45]}")


if __name__ == "__main__":
    snapshot(URLNAME)
    show_growth()

"""
認証ありで、下書きを含む自分の記事を取得する。
note_auth_client.py（前回の記事で作成）を使う。
"""
from note_auth_client import NoteAuthClient, CookieExpiredError


def fetch_my_notes(client: NoteAuthClient, status: str = "all",
                   max_pages: int = 50) -> list:
    """
    自分の記事を取得する。
    status: all / published / draft（環境により対応状況が異なる）
    """
    items = []

    for page in range(1, max_pages + 1):
        data = client.get_json(
            "/api/v3/notes",
            params={"kind": "note", "status": status, "page": page},
        )

        contents = data.get("notes", data.get("contents", []))
        if not contents:
            break

        items.extend(contents)
        print(f"page {page}: {len(contents)} 件（累計 {len(items)} 件）")

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

    return items


def summarize(items: list):
    published = [i for i in items if i.get("status") == "published"]
    drafts = [i for i in items if i.get("status") == "draft"]

    print("\n=== 集計 ===")
    print(f"  合計   : {len(items)} 本")
    print(f"  公開済 : {len(published)} 本")
    print(f"  下書き : {len(drafts)} 本")

    if drafts:
        print("\n=== 眠っている下書き ===")
        for d in drafts[:15]:
            title = d.get("name") or "（無題）"
            print(f"  - {title[:50]}")


if __name__ == "__main__":
    try:
        client = NoteAuthClient()
        client.verify()

        notes = fetch_my_notes(client)
        summarize(notes)

    except CookieExpiredError as e:
        print("認証エラー:", e)

@echo off
rem タスクスケジューラから呼び出すバッチ
rem 「基本タスクの作成」→ 毎日 → このファイルを指定

cd /d C:\note_api
call venv\Scripts\activate.bat

python 05_incremental.py >> logs\incremental.log 2>&1
python 06_snapshot.py >> logs\snapshot.log 2>&1

deactivate

# crontab -e で編集する
# 毎日 朝6時に実行（分 時 日 月 曜日）

0 6 * * * cd /home/user/note_api && venv/bin/python 05_incremental.py >> logs/incremental.log 2>&1
5 6 * * * cd /home/user/note_api && venv/bin/python 06_snapshot.py >> logs/snapshot.log 2>&1