スキ・フォローをPythonで操作する前に読む記事｜自動化の境界線

"""
スキ・フォローの操作をまとめたクライアント。

【重要】
このクラスは仕様の理解と、自分のデータ整理のために提供する。
他人への機械的な大量実行には使わないこと。
間隔を意図的に長く設定しているのはそのため。
"""
import logging
import time

from note_auth_client import NoteAuthClient

logger = logging.getLogger(__name__)

# 意図的に長めの間隔にしている
MIN_INTERVAL = 5.0
# 1回の実行での上限も低く抑える
MAX_ACTIONS_PER_RUN = 20


class SocialClient:

    def __init__(self, client: NoteAuthClient,
                 interval: float = MIN_INTERVAL):
        self.client = client
        self.interval = max(interval, MIN_INTERVAL)
        self._actions = 0

    def _guard(self):
        """実行回数の上限を守る"""
        if self._actions >= MAX_ACTIONS_PER_RUN:
            raise RuntimeError(
                f"1回の実行での上限（{MAX_ACTIONS_PER_RUN}件）に達しました。"
                "時間をおいて実行してください"
            )
        self._actions += 1
        time.sleep(self.interval)

    # ---- ユーザーIDの解決 ----

    def resolve_user_id(self, urlname: str) -> int:
        """urlname から数値のユーザーIDを取得する"""
        data = self.client.get_json(f"/api/v2/creators/{urlname}")

        user_id = data.get("id")
        if not user_id:
            raise ValueError(f"ユーザーIDを取得できません: {urlname}")

        return int(user_id)

    # ---- スキ ----

    def get_likers(self, note_key: str) -> list:
        """記事にスキを付けた人の一覧（読み取りのみ・安全）"""
        data = self.client.get_json(
            f"/api/v3/notes/{note_key}/likes"
        )

        if isinstance(data, list):
            return data

        return data.get("likes", data.get("contents", []))

    def like(self, note_key: str) -> bool:
        """記事にスキを付ける"""
        self._guard()

        res = self.client.request(
            "POST", f"/api/v3/notes/{note_key}/likes"
        )

        if res.status_code not in (200, 201):
            logger.warning("スキに失敗 HTTP %s（%s）",
                           res.status_code, note_key)
            return False

        return True

    def unlike(self, note_key: str) -> bool:
        """スキを取り消す"""
        self._guard()

        res = self.client.request(
            "DELETE", f"/api/v3/notes/{note_key}/likes"
        )

        return res.status_code in (200, 204)

    # ---- フォロー ----

    def follow(self, user_id: int) -> bool:
        """フォローする（数値IDを渡す）"""
        self._guard()

        res = self.client.request(
            "POST", f"/api/v3/users/{user_id}/following"
        )

        if res.status_code not in (200, 201):
            logger.warning("フォローに失敗 HTTP %s（id=%s）",
                           res.status_code, user_id)
            return False

        return True

    def unfollow(self, user_id: int) -> bool:
        """フォローを外す"""
        self._guard()

        res = self.client.request(
            "DELETE", f"/api/v3/users/{user_id}/following"
        )

        return res.status_code in (200, 204)


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

    client = NoteAuthClient()
    client.verify()

    social = SocialClient(client)

    # 読み取りのみの例（安全）
    likers = social.get_likers("nXXXXXXXXXXXX")
    print(f"スキを付けた人: {len(likers)} 人")

    for u in likers[:10]:
        user = u.get("user", u)
        print(f"  {user.get('nickname', '')}"
              f"（@{user.get('urlname', '')}）")

"""
自分の記事にスキを付けてくれた人を分析する。
読み取りのみなので安全。

- 繰り返しスキをくれる人が誰か
- どの記事に反応が集まるか
"""
import time
from collections import Counter

from note_auth_client import NoteAuthClient, CookieExpiredError
from social_client import SocialClient

URLNAME = "your_note_id"
SLEEP_SEC = 1.5
MAX_ARTICLES = 60


def fetch_my_notes(client: NoteAuthClient, urlname: str,
                   max_pages: int = 30) -> 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 analyze(client: NoteAuthClient, notes: list):
    social = SocialClient(client)

    liker_counter = Counter()
    liker_names = {}
    by_article = []

    targets = notes[:MAX_ARTICLES]

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

        if not note_key:
            continue

        likers = social.get_likers(note_key)

        for entry in likers:
            user = entry.get("user", entry)
            urlname = user.get("urlname", "")
            if not urlname:
                continue

            liker_counter[urlname] += 1
            liker_names[urlname] = user.get("nickname", urlname)

        by_article.append((len(likers), title))
        print(f"[{i}/{len(targets)}] {title[:40]}"
              f"（{len(likers)} 人）")

        time.sleep(SLEEP_SEC)

    print("\n=== 繰り返しスキをくれる方 ===")
    for urlname, count in liker_counter.most_common(20):
        name = liker_names.get(urlname, urlname)
        print(f"  {count:>3} 回  {name[:24]}（@{urlname}）")

    print("\n=== スキが多い記事 ===")
    for count, title in sorted(by_article, reverse=True)[:10]:
        print(f"  {count:>4} 人  {title[:44]}")

    print(f"\nスキをくれた人の総数（重複なし）: {len(liker_counter)} 人")


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

    notes = fetch_my_notes(client, URLNAME)
    analyze(client, notes)


if __name__ == "__main__":
    main()

"""
自分のフォロー一覧を確認し、整理の候補を提示する。

判定するのはスクリプト、決めるのは人間。
実際の解除は確認したうえで手動または少数ずつ行う。
"""
import time
from datetime import datetime, timezone

from note_auth_client import NoteAuthClient, CookieExpiredError

SLEEP_SEC = 1.5
STALE_DAYS = 365


def fetch_following(client: NoteAuthClient, urlname: str,
                    max_pages: int = 40) -> list:
    """自分がフォローしている人の一覧"""
    users = []

    for page in range(1, max_pages + 1):
        data = client.get_json(
            f"/api/v2/creators/{urlname}/followings",
            params={"page": page},
        )

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

        users.extend(items)
        print(f"  page {page}: 累計 {len(users)} 人")

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

        time.sleep(SLEEP_SEC)

    return users


def parse_date(value: str):
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None


def review(client: NoteAuthClient, users: list):
    """整理の候補を分類する"""
    now = datetime.now(timezone.utc)

    inactive = []
    no_notes = []
    active = []

    for u in users:
        urlname = u.get("urlname", "")
        nickname = u.get("nickname", urlname)
        note_count = u.get("noteCount", u.get("note_count", 0)) or 0
        last = parse_date(
            u.get("lastNoteAt") or u.get("last_note_at") or ""
        )

        if note_count == 0:
            no_notes.append((nickname, urlname))
            continue

        if last and (now - last).days > STALE_DAYS:
            days = (now - last).days
            inactive.append((nickname, urlname, days))
            continue

        active.append((nickname, urlname))

    print(f"\n=== フォロー {len(users)} 人の内訳 ===")
    print(f"  活動中           : {len(active)} 人")
    print(f"  1年以上更新なし  : {len(inactive)} 人")
    print(f"  記事が0本        : {len(no_notes)} 人")

    if inactive:
        print(f"\n--- 1年以上更新がない方（整理の候補）---")
        for nickname, urlname, days in inactive[:25]:
            print(f"  {days:>5}日前  {nickname[:24]}（@{urlname}）")

    if no_notes:
        print(f"\n--- 記事が0本の方 ---")
        for nickname, urlname in no_notes[:15]:
            print(f"  {nickname[:24]}（@{urlname}）")

    print("\n※ 解除するかどうかは、ご自身で判断してください")
    print("※ タイムラインに出てきたとき読みたいかが基準です")


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

    urlname = user.get("urlname", "")
    if not urlname:
        print("ユーザー情報を取得できません")
        return

    print(f"@{urlname} のフォロー一覧を取得します")
    users = fetch_following(client, urlname)

    review(client, users)


if __name__ == "__main__":
    main()

"""
自分がスキした記事を記録し、あとから検索できるようにする。
読み取りのみなので安全。

noteの画面でもスキした記事は一覧できるが、
手元にCSVで持っておくと検索・分類がしやすい。
"""
import csv
import os
import time

from note_auth_client import NoteAuthClient, CookieExpiredError

OUTPUT = "my_likes.csv"
SLEEP_SEC = 1.5
FIELDS = ["key", "title", "author", "author_urlname",
          "price", "url", "recorded_at"]


def fetch_my_likes(client: NoteAuthClient, urlname: str,
                   max_pages: int = 40) -> list:
    """自分がスキした記事の一覧を取得する"""
    items = []

    for page in range(1, max_pages + 1):
        data = client.get_json(
            f"/api/v2/creators/{urlname}/likes",
            params={"page": page},
        )

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

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

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

        time.sleep(SLEEP_SEC)

    return items


def normalize(items: list) -> list:
    from datetime import datetime

    stamp = datetime.now().strftime("%Y-%m-%d")
    rows = []

    for entry in items:
        # note が入れ子になっている場合とそうでない場合に対応
        note = entry.get("note", entry)
        user = note.get("user") or {}
        key = note.get("key", "")
        author_url = user.get("urlname", "")

        rows.append({
            "key": key,
            "title": note.get("name", ""),
            "author": user.get("nickname", ""),
            "author_urlname": author_url,
            "price": note.get("price", 0) or 0,
            "url": f"https://note.com/{author_url}/n/{key}",
            "recorded_at": stamp,
        })

    return rows


def save(rows: list, path: str = OUTPUT):
    if not rows:
        print("記録対象がありません")
        return

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

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


def summarize(rows: list):
    from collections import Counter

    authors = Counter(r["author"] for r in rows if r["author"])

    print("\n=== よくスキしている書き手 ===")
    for name, count in authors.most_common(15):
        print(f"  {count:>3} 件  {name[:24]}")

    paid = [r for r in rows if r["price"]]
    print(f"\n有料記事へのスキ: {len(paid)} 件")


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

        urlname = user.get("urlname", "")
        if not urlname:
            raise SystemExit("ユーザー情報を取得できません")

        print(f"@{urlname} がスキした記事を取得します")
        rows = normalize(fetch_my_likes(client, urlname))

        save(rows)
        summarize(rows)

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

"""
フォローとフォロワーを比較して、関係の内訳を確認する。

【注意】
この結果をもとに「フォローバックしてくれない人を外す」という
運用は推奨しない。フォローは一方向でも成立する関係である。
数字を把握するための情報として使うこと。
"""
import time

from note_auth_client import NoteAuthClient, CookieExpiredError

SLEEP_SEC = 1.5


def fetch_users(client: NoteAuthClient, urlname: str,
                kind: str, max_pages: int = 40) -> list:
    """
    kind: "followings"（フォロー中） / "followers"（フォロワー）
    """
    users = []

    for page in range(1, max_pages + 1):
        data = client.get_json(
            f"/api/v2/creators/{urlname}/{kind}",
            params={"page": page},
        )

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

        users.extend(items)

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

        time.sleep(SLEEP_SEC)

    return users


def to_map(users: list) -> dict:
    """urlname をキーにした辞書にする"""
    result = {}
    for u in users:
        urlname = u.get("urlname")
        if urlname:
            result[urlname] = u.get("nickname", urlname)
    return result


def report(following: dict, followers: dict):
    mutual = set(following) & set(followers)
    only_following = set(following) - set(followers)
    only_followers = set(followers) - set(following)

    print("\n=== 関係の内訳 ===")
    print(f"  フォロー中     : {len(following)} 人")
    print(f"  フォロワー     : {len(followers)} 人")
    print(f"  相互           : {len(mutual)} 人")
    print(f"  自分から片方向 : {len(only_following)} 人")
    print(f"  相手から片方向 : {len(only_followers)} 人")

    if following:
        rate = len(mutual) / len(following) * 100
        print(f"\n  フォロー中のうち相互の割合: {rate:.1f}%")

    if only_followers:
        print("\n=== フォローしてくれている方（自分は未フォロー）===")
        for urlname in list(only_followers)[:20]:
            print(f"  {followers[urlname][:24]}（@{urlname}）")
        print("  → 記事を見て、読みたいと思う人だけフォローすれば十分です")


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

        urlname = user.get("urlname", "")
        if not urlname:
            raise SystemExit("ユーザー情報を取得できません")

        print("フォロー中を取得します")
        following = to_map(fetch_users(client, urlname, "followings"))

        print("フォロワーを取得します")
        followers = to_map(fetch_users(client, urlname, "followers"))

        report(following, followers)

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

"""
スキ・フォローなどの操作を記録するラッパー。

- 1日の実行上限を管理する
- 何をいつ実行したかCSVに残す
- 上限に達したら例外を投げて止まる
"""
import csv
import os
from datetime import date, datetime

LOG_FILE = "action_log.csv"
DAILY_LIMIT = 30
FIELDS = ["date", "time", "action", "target", "result"]


class DailyLimitExceeded(Exception):
    """1日の上限に達した"""


def count_today(path: str = LOG_FILE) -> int:
    """今日すでに何件実行したかを数える"""
    if not os.path.exists(path):
        return 0

    today = date.today().isoformat()
    count = 0

    with open(path, encoding="utf-8-sig", newline="") as f:
        for row in csv.DictReader(f):
            if row.get("date") == today:
                count += 1

    return count


def record(action: str, target: str, result: bool,
           path: str = LOG_FILE):
    """操作を記録する"""
    now = datetime.now()
    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()

        writer.writerow({
            "date": now.date().isoformat(),
            "time": now.strftime("%H:%M:%S"),
            "action": action,
            "target": target,
            "result": "ok" if result else "failed",
        })


def guard(path: str = LOG_FILE, limit: int = DAILY_LIMIT):
    """上限に達していたら例外を投げる"""
    count = count_today(path)

    if count >= limit:
        raise DailyLimitExceeded(
            f"1日の上限（{limit}件）に達しています。"
            f"今日はすでに {count} 件実行しました"
        )

    return count


def summary(path: str = LOG_FILE):
    """記録の集計を表示する"""
    if not os.path.exists(path):
        print("記録がありません")
        return

    from collections import Counter

    with open(path, encoding="utf-8-sig", newline="") as f:
        rows = list(csv.DictReader(f))

    by_date = Counter(r["date"] for r in rows)
    by_action = Counter(r["action"] for r in rows)
    failed = sum(1 for r in rows if r["result"] == "failed")

    print(f"=== 記録 {len(rows)} 件 ===")
    print(f"  失敗: {failed} 件")

    print("\n--- 操作の種類 ---")
    for action, count in by_action.most_common():
        print(f"  {action}: {count} 件")

    print("\n--- 日別の実行数（直近14日）---")
    for d in sorted(by_date)[-14:]:
        count = by_date[d]
        bar = "|" * count
        warn = "  ← 多い" if count > DAILY_LIMIT else ""
        print(f"  {d}  {count:>3} 件  {bar}{warn}")


if __name__ == "__main__":
    print(f"今日の実行数: {count_today()} / {DAILY_LIMIT}")
    summary()