noteの予約投稿をPythonで自作する｜スケジューラ設計

note_scheduler/
├── .env                     # 認証情報（gitignore対象）
├── queue.json               # 投稿キュー
├── history.json             # 実行履歴
├── scheduler.lock           # 二重実行防止用（自動生成）
├── note_auth_client.py      # 認証クライアント
├── note_publisher.py        # 公開処理
├── queue_manager.py         # キューの操作
├── scheduler.py             # 本体（定期実行される）
├── cli.py                   # キューの登録・確認用
└── logs/
    └── scheduler.log

"""
投稿キューの管理。
queue.json に「いつ・どの記事を公開するか」を保持する。
"""
import json
import os
from datetime import datetime, timedelta

QUEUE_FILE = "queue.json"
HISTORY_FILE = "history.json"


def load_queue(path: str = QUEUE_FILE) -> list:
    if not os.path.exists(path):
        return []
    with open(path, encoding="utf-8") as f:
        return json.load(f)


def save_queue(items: list, path: str = QUEUE_FILE):
    with open(path, "w", encoding="utf-8") as f:
        json.dump(items, f, ensure_ascii=False, indent=2)


def add(note_key: str, publish_at: str, title: str = "",
        tags: list = None, path: str = QUEUE_FILE) -> dict:
    """
    キューに追加する。
    publish_at は "2026-08-05 07:30" の形式。
    """
    try:
        scheduled = datetime.strptime(publish_at, "%Y-%m-%d %H:%M")
    except ValueError:
        raise ValueError("日時は '2026-08-05 07:30' の形式で指定してください")

    items = load_queue(path)

    # 同じ記事が既に登録されていないか確認する
    for item in items:
        if item["note_key"] == note_key and item["status"] == "pending":
            raise ValueError(f"この記事は既に登録されています: {note_key}")

    entry = {
        "note_key": note_key,
        "title": title,
        "publish_at": scheduled.isoformat(),
        "tags": tags or [],
        "status": "pending",
        "created_at": datetime.now().isoformat(),
        "attempts": 0,
    }

    items.append(entry)
    items.sort(key=lambda x: x["publish_at"])
    save_queue(items, path)

    return entry


def due_items(now: datetime = None, path: str = QUEUE_FILE) -> list:
    """公開時刻を過ぎた pending の項目を返す"""
    now = now or datetime.now()
    items = load_queue(path)

    due = []
    for item in items:
        if item["status"] != "pending":
            continue
        scheduled = datetime.fromisoformat(item["publish_at"])
        if scheduled <= now:
            due.append(item)

    return due


def mark(note_key: str, status: str, message: str = "",
         path: str = QUEUE_FILE):
    """項目の状態を更新する"""
    items = load_queue(path)

    for item in items:
        if item["note_key"] == note_key and item["status"] == "pending":
            item["status"] = status
            item["message"] = message
            item["processed_at"] = datetime.now().isoformat()
            item["attempts"] = item.get("attempts", 0) + 1
            break

    save_queue(items, path)


def cleanup(days: int = 30, path: str = QUEUE_FILE):
    """完了から一定期間経った項目を履歴へ移す"""
    items = load_queue(path)
    threshold = datetime.now() - timedelta(days=days)

    keep = []
    archived = []

    for item in items:
        processed = item.get("processed_at")
        if item["status"] == "done" and processed:
            if datetime.fromisoformat(processed) < threshold:
                archived.append(item)
                continue
        keep.append(item)

    if archived:
        history = []
        if os.path.exists(HISTORY_FILE):
            with open(HISTORY_FILE, encoding="utf-8") as f:
                history = json.load(f)
        history.extend(archived)

        with open(HISTORY_FILE, "w", encoding="utf-8") as f:
            json.dump(history, f, ensure_ascii=False, indent=2)

        save_queue(keep, path)
        print(f"{len(archived)} 件を履歴へ移しました")


def summary(path: str = QUEUE_FILE):
    """キューの状態を表示する"""
    items = load_queue(path)
    if not items:
        print("キューは空です")
        return

    counts = {}
    for item in items:
        counts[item["status"]] = counts.get(item["status"], 0) + 1

    print("=== キューの状態 ===")
    for status, count in counts.items():
        print(f"  {status}: {count} 件")

    pending = [i for i in items if i["status"] == "pending"]
    if pending:
        print("\n=== 公開待ち ===")
        for item in pending[:15]:
            when = datetime.fromisoformat(item["publish_at"])
            title = item.get("title") or item["note_key"]
            print(f"  {when.strftime('%m/%d %H:%M')}  {title[:40]}")


if __name__ == "__main__":
    summary()

"""
定期実行され、公開時刻を過ぎた記事を公開する。

    python scheduler.py           # 確認のみ
    python scheduler.py --run     # 実際に公開する
"""
import logging
import os
import sys
import time
from datetime import datetime

from note_auth_client import NoteAuthClient, CookieExpiredError
from note_publisher import NotePublisher
import queue_manager as qm

LOCK_FILE = "scheduler.lock"
LOCK_TIMEOUT_SEC = 1800     # ロックが残り続けた場合の解除時間
SLEEP_BETWEEN = 5.0
MAX_ATTEMPTS = 3

os.makedirs("logs", exist_ok=True)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("logs/scheduler.log", encoding="utf-8"),
        logging.StreamHandler(),
    ],
)
logger = logging.getLogger(__name__)


# ---- 二重実行の防止 ----

def acquire_lock() -> bool:
    """ロックを取得する。既にロックがあればFalse"""
    if os.path.exists(LOCK_FILE):
        age = time.time() - os.path.getmtime(LOCK_FILE)
        if age < LOCK_TIMEOUT_SEC:
            logger.warning("別のプロセスが実行中です（%.0f秒前から）", age)
            return False
        logger.warning("古いロックを解除します（%.0f秒経過）", age)
        release_lock()

    with open(LOCK_FILE, "w", encoding="utf-8") as f:
        f.write(f"{os.getpid()}\n{datetime.now().isoformat()}")
    return True


def release_lock():
    if os.path.exists(LOCK_FILE):
        os.remove(LOCK_FILE)


# ---- 公開処理 ----

def publish_one(pub: NotePublisher, item: dict) -> bool:
    """1件を公開する"""
    note_key = item["note_key"]
    title = item.get("title") or note_key

    logger.info("公開処理: %s", title)

    try:
        current = pub.fetch(note_key)
    except Exception as e:
        logger.error("記事を取得できません（%s）: %s", note_key, e)
        qm.mark(note_key, "error", str(e))
        return False

    # すでに公開済みならスキップ
    if current.get("status") == "published":
        logger.info("すでに公開されています: %s", title)
        qm.mark(note_key, "done", "already published")
        return True

    changes = {}
    if item.get("tags"):
        changes["hashtags"] = item["tags"]

    result = pub.publish(note_key, dry_run=False, **changes)

    if result.get("ok"):
        logger.info("公開しました: %s", title)
        qm.mark(note_key, "done", "published")
        return True

    attempts = item.get("attempts", 0) + 1
    if attempts >= MAX_ATTEMPTS:
        logger.error("試行上限に達しました: %s", title)
        qm.mark(note_key, "failed", "max attempts")
    else:
        logger.warning("失敗（%s回目）。次回に再試行します: %s",
                       attempts, title)
        # pending のまま attempts だけ増やす
        items = qm.load_queue()
        for i in items:
            if i["note_key"] == note_key and i["status"] == "pending":
                i["attempts"] = attempts
        qm.save_queue(items)

    return False


def main(dry_run: bool = True):
    due = qm.due_items()

    if not due:
        logger.info("公開対象はありません")
        return 0

    logger.info("公開対象: %s 件", len(due))
    for item in due:
        when = datetime.fromisoformat(item["publish_at"])
        print(f"  {when.strftime('%m/%d %H:%M')}  "
              f"{item.get('title') or item['note_key']}")

    if dry_run:
        print("\n[確認モード] --run を付けると実際に公開します")
        return 0

    if not acquire_lock():
        return 1

    try:
        client = NoteAuthClient()
        client.verify()
        pub = NotePublisher(client)

        success = 0
        for item in due:
            if publish_one(pub, item):
                success += 1
            time.sleep(SLEEP_BETWEEN)

        logger.info("完了: %s/%s 件を公開しました", success, len(due))
        qm.cleanup()
        return 0

    except CookieExpiredError as e:
        logger.error("認証エラー: %s", e)
        return 2
    except Exception as e:
        logger.exception("想定外のエラー: %s", e)
        return 3
    finally:
        release_lock()


if __name__ == "__main__":
    sys.exit(main(dry_run="--run" not in sys.argv))

"""
キューを操作するコマンドラインツール。

    python cli.py list
    python cli.py add nXXXXXXXX "2026-08-05 07:30" --title "記事タイトル"
    python cli.py remove nXXXXXXXX
    python cli.py drafts
"""
import sys
from datetime import datetime

import queue_manager as qm


def cmd_list():
    qm.summary()


def cmd_add(args: list):
    if len(args) < 2:
        print('使い方: python cli.py add <note_key> "YYYY-MM-DD HH:MM"')
        return

    note_key = args[0]
    publish_at = args[1]

    title = ""
    if "--title" in args:
        idx = args.index("--title")
        if idx + 1 < len(args):
            title = args[idx + 1]

    tags = []
    if "--tags" in args:
        idx = args.index("--tags")
        if idx + 1 < len(args):
            tags = [t.strip() for t in args[idx + 1].split(",")]

    try:
        entry = qm.add(note_key, publish_at, title=title, tags=tags)
    except ValueError as e:
        print("エラー:", e)
        return

    when = datetime.fromisoformat(entry["publish_at"])
    print(f"登録しました: {when.strftime('%Y-%m-%d %H:%M')} "
          f"{title or note_key}")


def cmd_remove(args: list):
    if not args:
        print("使い方: python cli.py remove <note_key>")
        return

    note_key = args[0]
    items = qm.load_queue()
    before = len(items)
    items = [i for i in items if i["note_key"] != note_key]

    if len(items) == before:
        print("該当する項目がありません")
        return

    qm.save_queue(items)
    print(f"削除しました: {note_key}")


def cmd_drafts():
    """下書き一覧を表示して、キューに登録しやすくする"""
    from note_auth_client import NoteAuthClient
    from note_draft import NoteDraft

    client = NoteAuthClient()
    client.verify()

    drafts = NoteDraft(client).list_drafts()

    if not drafts:
        print("下書きはありません")
        return

    print("=== 下書き一覧 ===")
    for d in drafts:
        key = d.get("key", "")
        title = (d.get("name") or "（無題）")[:44]
        print(f"  {key}  {title}")

    print("\n登録するには:")
    print('  python cli.py add <key> "2026-08-05 07:30" --title "..."')


COMMANDS = {
    "list": lambda args: cmd_list(),
    "add": cmd_add,
    "remove": cmd_remove,
    "drafts": lambda args: cmd_drafts(),
}


if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS:
        print("使えるコマンド:", ", ".join(COMMANDS))
        sys.exit(1)

    COMMANDS[sys.argv[1]](sys.argv[2:])

@echo off
rem タスクスケジューラから15分おきに実行する
rem 「タスクの作成」→「トリガー」→「繰り返し間隔: 15分」

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

python scheduler.py --run

deactivate

# crontab -e で編集する
# 15分ごとに実行する

*/15 * * * * cd /home/user/note_scheduler && venv/bin/python scheduler.py --run

# 毎朝8時にキューの状態を確認する
0 8 * * * cd /home/user/note_scheduler && venv/bin/python cli.py list

"""
下書きを、指定したルールで日付に振り分けてキューに登録する。

    python distribute.py --interval 2 --time 07:30 --start 2026-08-01
"""
import sys
from datetime import datetime, timedelta

from note_auth_client import NoteAuthClient, CookieExpiredError
from note_draft import NoteDraft
import queue_manager as qm

# 投稿しない曜日（0=月曜 ... 6=日曜）
SKIP_WEEKDAYS = set()


def parse_args(argv: list) -> dict:
    opts = {
        "interval": 2,
        "time": "07:30",
        "start": None,
        "limit": 20,
    }

    for key in opts:
        flag = f"--{key}"
        if flag in argv:
            idx = argv.index(flag)
            if idx + 1 < len(argv):
                value = argv[idx + 1]
                opts[key] = int(value) if key in ("interval", "limit") \
                    else value

    return opts


def next_slot(current: datetime, interval_days: int) -> datetime:
    """次の投稿日を求める（スキップ曜日を避ける）"""
    candidate = current + timedelta(days=interval_days)

    while candidate.weekday() in SKIP_WEEKDAYS:
        candidate += timedelta(days=1)

    return candidate


def main(argv: list):
    opts = parse_args(argv)

    hour, minute = (int(x) for x in opts["time"].split(":"))

    if opts["start"]:
        start = datetime.strptime(opts["start"], "%Y-%m-%d")
    else:
        start = datetime.now() + timedelta(days=1)

    start = start.replace(hour=hour, minute=minute,
                          second=0, microsecond=0)

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

    drafts = NoteDraft(client).list_drafts()

    # 本文がある程度あるものだけを対象にする
    targets = [
        d for d in drafts
        if len(d.get("body") or "") > 500 and (d.get("name") or "").strip()
    ]

    if not targets:
        print("登録できる下書きがありません")
        return

    targets = targets[:opts["limit"]]

    print(f"=== {len(targets)} 件を振り分けます ===")
    print(f"  開始    : {start.strftime('%Y-%m-%d %H:%M')}")
    print(f"  間隔    : {opts['interval']} 日おき")

    current = start
    registered = 0

    for d in targets:
        key = d.get("key")
        title = (d.get("name") or "")[:44]

        if not key:
            continue

        try:
            qm.add(
                key,
                current.strftime("%Y-%m-%d %H:%M"),
                title=title,
            )
            print(f"  {current.strftime('%m/%d %H:%M')}  {title}")
            registered += 1
            current = next_slot(current, opts["interval"])
        except ValueError as e:
            print(f"  スキップ: {title}（{e}）")

    print(f"\n{registered} 件をキューに登録しました")
    print("確認: python cli.py list")


if __name__ == "__main__":
    main(sys.argv[1:])

"""
スケジューラの状態を確認し、問題があれば通知する。
1日1回実行する想定。
"""
import os
from datetime import datetime, timedelta

import requests

import queue_manager as qm

# Discord や Slack の Webhook URL（環境変数から読む）
WEBHOOK_URL = os.getenv("NOTIFY_WEBHOOK", "")
LOG_FILE = "logs/scheduler.log"
STALE_HOURS = 6


def send(message: str):
    print(message)
    if not WEBHOOK_URL:
        return
    try:
        requests.post(WEBHOOK_URL, json={"content": message}, timeout=10)
    except requests.RequestException as e:
        print("通知の送信に失敗:", e)


def check() -> list:
    """異常を検出して、メッセージの配列を返す"""
    problems = []
    items = qm.load_queue()
    now = datetime.now()

    # 1. 公開時刻を大きく過ぎているのに残っている
    overdue = []
    for item in items:
        if item["status"] != "pending":
            continue
        scheduled = datetime.fromisoformat(item["publish_at"])
        if scheduled < now - timedelta(hours=1):
            overdue.append(item)

    if overdue:
        titles = ", ".join(
            (i.get("title") or i["note_key"])[:20] for i in overdue[:5]
        )
        problems.append(
            f"公開時刻を過ぎた記事が {len(overdue)} 件あります: {titles}"
        )

    # 2. 失敗した項目
    failed = [i for i in items if i["status"] in ("failed", "error")]
    if failed:
        problems.append(f"失敗した項目が {len(failed)} 件あります")

    # 3. ログが更新されていない（スケジューラが動いていない）
    if os.path.exists(LOG_FILE):
        age_hours = (
            datetime.now().timestamp() - os.path.getmtime(LOG_FILE)
        ) / 3600
        if age_hours > STALE_HOURS:
            problems.append(
                f"スケジューラが {age_hours:.0f} 時間動いていません"
            )
    else:
        problems.append("ログファイルがありません")

    return problems


def main():
    problems = check()

    if not problems:
        print("問題ありません")
        return 0

    message = "【note スケジューラ】\n" + "\n".join(
        f"- {p}" for p in problems
    )
    send(message)
    return 1


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