ツイート・YouTube・note記事の埋め込みをAPIで自動化する

"""
URLがどのサービスの埋め込みに対応しているかを判定する。
"""
import json

from note_auth_client import NoteAuthClient, CookieExpiredError

SAMPLE_URLS = [
    "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
    "https://x.com/note_PR/status/1234567890",
    "https://open.spotify.com/track/xxxxxxxx",
    "https://github.com/python/cpython",
    "https://example.com/plain-page",
]


def check_type(client: NoteAuthClient, url: str) -> dict:
    """URLの埋め込み種別を判定する"""
    data = client.get_json(
        "/api/v2/embed_by_external_api/check_type",
        params={"url": url},
    )
    return data or {}


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

        for url in SAMPLE_URLS:
            result = check_type(client, url)
            service = result.get("type") or result.get("service") or "不明"
            print(f"{service:<16} {url[:52]}")

            # 初回は生のレスポンスも確認しておく
            if url == SAMPLE_URLS[0]:
                print("  --- 生のレスポンス ---")
                print("  " + json.dumps(result, ensure_ascii=False)[:300])

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

"""
外部サービスの埋め込みを扱うクライアント。
"""
import logging

from note_auth_client import NoteAuthClient

logger = logging.getLogger(__name__)


class EmbedClient:

    def __init__(self, client: NoteAuthClient):
        self.client = client

    def check_type(self, url: str) -> str:
        """URLの種別を返す。判定できなければ空文字"""
        data = self.client.get_json(
            "/api/v2/embed_by_external_api/check_type",
            params={"url": url},
        )
        if not data:
            return ""
        return data.get("type") or data.get("service") or ""

    def fetch_embed(self, url: str, note_key: str,
                    service: str = None) -> dict:
        """
        埋め込み情報を取得する。
        note_key は埋め込み先の記事キー。
        """
        service = service or self.check_type(url)

        if not service:
            logger.info("埋め込みに対応していないURLです: %s", url[:60])
            return {}

        data = self.client.get_json(
            "/api/v2/embed_by_external_api",
            params={
                "url": url,
                "service": service,
                "embeddable_key": note_key,
            },
        )

        if not data:
            logger.warning("埋め込み情報を取得できません: %s", url[:60])
            return {}

        return data

    def embed_note(self, note_url: str, target_key: str,
                   height: int = 380) -> dict:
        """
        note記事をネイティブ埋め込みする。
        note_url は埋め込みたい記事のURL。
        target_key は埋め込み先の記事キー。
        """
        payload = {
            "url": note_url,
            "height": height,
            "embeddable_type": "Note",
            "embeddable_key": target_key,
        }

        res = self.client.request("POST", "/api/v1/embed", json=payload)

        if res.status_code not in (200, 201):
            logger.error("note埋め込み失敗 HTTP %s: %s",
                         res.status_code, res.text[:200])
            return {}

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

        embedded = data.get("embedded_content", {})
        return {
            "key": embedded.get("key", ""),
            "html": data.get("html_for_embed", ""),
            "raw": data,
        }


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

    client = NoteAuthClient()
    client.verify()

    embed = EmbedClient(client)

    # 埋め込み先の記事キーを指定する
    target = "nXXXXXXXXXXXX"

    info = embed.fetch_embed(
        "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
        note_key=target,
    )
    print("取得結果:", list(info.keys()) if info else "なし")

"""
note記事を別の記事に埋め込む。
自分の過去記事へ誘導する導線として使える。
"""
import time

from note_auth_client import NoteAuthClient, CookieExpiredError
from embed_client import EmbedClient


def embed_related_notes(client: NoteAuthClient, target_key: str,
                        related_urls: list, height: int = 380) -> list:
    """複数のnote記事を埋め込み、埋め込みキーの一覧を返す"""
    embed = EmbedClient(client)
    results = []

    for url in related_urls:
        print(f"埋め込み中: {url[:60]}")

        result = embed.embed_note(url, target_key, height=height)

        if result.get("key"):
            print(f"  成功: key={result['key']}")
            results.append(result)
        else:
            print("  失敗しました")

        time.sleep(2.0)

    return results


def build_embed_html(embed_key: str) -> str:
    """埋め込みキーから、本文に入れるHTMLを組み立てる"""
    # 実際に返される html_for_embed を使うのが確実だが、
    # 形式が分かっていればこの形でも動く
    return f'<p><embed data-key="{embed_key}"></p>'


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

        target = "nXXXXXXXXXXXX"     # 埋め込み先の記事
        related = [
            "https://note.com/someone/n/nAAAAAAAAAAA",
            "https://note.com/someone/n/nBBBBBBBBBBB",
        ]

        results = embed_related_notes(client, target, related)

        print(f"\n{len(results)} 件を埋め込みました")
        for r in results:
            if r.get("html"):
                print("  HTML:", r["html"][:80])

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

"""
原稿内の単独行URLを、埋め込みに変換する。

    本文の途中に書いたリンクは変換しない。
    行全体がURLだけの場合のみ埋め込み扱いにする。
"""
import re
import time

from note_auth_client import NoteAuthClient
from embed_client import EmbedClient

# 行全体がURLの場合のみマッチする
STANDALONE_URL = re.compile(r"^\s*(https?://\S+)\s*$", re.MULTILINE)

NOTE_URL = re.compile(r"^https?://note\.com/[^/]+/n/n\w+")


def convert_embeds(client: NoteAuthClient, md_text: str,
                   target_key: str, sleep_sec: float = 2.0) -> tuple:
    """
    原稿内の単独行URLを埋め込みHTMLに置き換える。
    戻り値: (置換後のテキスト, 埋め込んだ数)
    """
    embed = EmbedClient(client)
    count = 0

    def replace(match):
        nonlocal count

        url = match.group(1)

        # note記事はネイティブ埋め込みを使う
        if NOTE_URL.match(url):
            result = embed.embed_note(url, target_key)
            time.sleep(sleep_sec)

            if result.get("html"):
                count += 1
                return "\n" + result["html"] + "\n"

            print(f"  note埋め込みに失敗: {url[:60]}")
            return match.group(0)

        # 外部サービス
        info = embed.fetch_embed(url, target_key)
        time.sleep(sleep_sec)

        html = info.get("html") or info.get("html_for_embed") or ""
        if html:
            count += 1
            return "\n" + html + "\n"

        print(f"  埋め込みに対応していません: {url[:60]}")
        return match.group(0)

    result = STANDALONE_URL.sub(replace, md_text)
    return result, count


def list_urls(md_text: str) -> list:
    """原稿内の単独行URLを列挙する（事前確認用）"""
    return STANDALONE_URL.findall(md_text)


if __name__ == "__main__":
    sample = """## 参考動画

https://www.youtube.com/watch?v=dQw4w9WgXcQ

本文の中に書いた https://example.com は変換されません。

## 関連記事

https://note.com/someone/n/nAAAAAAAAAAA
"""

    urls = list_urls(sample)
    print("変換対象のURL:")
    for u in urls:
        print("  -", u)

"""
自分の記事に含まれる埋め込みを一覧し、リンク切れを検出する。
"""
import re
import time

import requests

from note_auth_client import NoteAuthClient, CookieExpiredError

URLNAME = "your_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")

EMBED_PATTERN = re.compile(r"<(embed|iframe)[^>]*>", re.IGNORECASE)
URL_IN_TAG = re.compile(r'(?:src|href|data-url)="([^"]+)"')


def collect_embeds(client: NoteAuthClient, urlname: str,
                   max_pages: int = 20) -> list:
    """全記事から埋め込みを収集する"""
    found = []

    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

        for item in contents:
            key = item.get("key")
            if not key:
                continue

            detail = client.get_json(f"/api/v3/notes/{key}")
            body = detail.get("body") or ""

            tags = EMBED_PATTERN.findall(body)
            urls = URL_IN_TAG.findall(body)

            if tags or urls:
                found.append({
                    "key": key,
                    "title": item.get("name", ""),
                    "embed_count": len(tags),
                    "urls": [u for u in urls
                             if u.startswith("http")],
                })

            time.sleep(1.2)

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

    return found


def check_urls(items: list) -> list:
    """埋め込み先のURLが生きているか確認する"""
    broken = []
    session = requests.Session()
    session.headers.update({"User-Agent": UA})

    for item in items:
        for url in item["urls"]:
            try:
                res = session.head(url, timeout=10,
                                   allow_redirects=True)
                if res.status_code >= 400:
                    broken.append((item["title"], url, res.status_code))
            except requests.RequestException:
                broken.append((item["title"], url, "接続エラー"))

            time.sleep(0.8)

    return broken


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

        items = collect_embeds(client, URLNAME)

        print(f"=== 埋め込みを含む記事: {len(items)} 件 ===")
        for item in items:
            print(f"  {item['embed_count']} 個  {item['title'][:44]}")

        print("\n=== リンク切れの確認 ===")
        broken = check_urls(items)

        if broken:
            for title, url, status in broken:
                print(f"  [{status}] {title[:30]}")
                print(f"        {url[:70]}")
        else:
            print("  問題は見つかりませんでした")

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

"""
原稿内のURLが埋め込みに対応しているかを事前に確認する。

    python precheck_urls.py article.md

- 短縮URLを展開してから判定する
- リンク切れも同時に検出する
"""
import re
import sys
import time

import requests

from note_auth_client import NoteAuthClient, CookieExpiredError
from embed_client import EmbedClient

UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
      "AppleWebKit/537.36 (KHTML, like Gecko) "
      "Chrome/120.0.0.0 Safari/537.36")

URL_PATTERN = re.compile(r"https?://[^\s)\]>\"']+")

# 展開が必要な短縮URLのホスト
SHORTENERS = {
    "bit.ly", "t.co", "goo.gl", "ow.ly", "buff.ly",
    "amzn.to", "youtu.be", "is.gd", "tinyurl.com",
}


def extract_urls(text: str) -> list:
    """重複を除いてURLを抽出する"""
    seen = set()
    result = []

    for url in URL_PATTERN.findall(text):
        url = url.rstrip(".,、。")
        if url not in seen:
            seen.add(url)
            result.append(url)

    return result


def expand(url: str, session: requests.Session) -> tuple:
    """
    短縮URLを展開し、生存確認もする。
    戻り値: (展開後のURL, ステータス)
    """
    host = url.split("/")[2] if "://" in url else ""

    try:
        res = session.head(url, timeout=12, allow_redirects=True)
        final = res.url
        status = res.status_code
    except requests.RequestException as e:
        return url, f"接続エラー（{type(e).__name__}）"

    if host in SHORTENERS and final != url:
        return final, status

    return final, status


def main(path: str):
    with open(path, encoding="utf-8") as f:
        text = f.read()

    urls = extract_urls(text)

    if not urls:
        print("URLが見つかりませんでした")
        return

    print(f"=== {len(urls)} 件のURLを確認します ===\n")

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

    session = requests.Session()
    session.headers.update({"User-Agent": UA})

    embeddable = []
    plain = []
    broken = []

    for url in urls:
        final, status = expand(url, session)

        if isinstance(status, str) or status >= 400:
            broken.append((url, status))
            print(f"[NG   ] {url[:58]}")
            print(f"         {status}")
            time.sleep(0.8)
            continue

        service = embed.check_type(final)

        if service:
            embeddable.append((final, service))
            print(f"[埋込可] {service:<14} {final[:48]}")
        else:
            plain.append(final)
            print(f"[リンク] {'':<14} {final[:48]}")

        if final != url:
            print(f"         ← 展開元 {url[:48]}")

        time.sleep(1.5)

    print("\n=== 集計 ===")
    print(f"  埋め込み可能 : {len(embeddable)} 件")
    print(f"  通常リンク   : {len(plain)} 件")
    print(f"  問題あり     : {len(broken)} 件")

    if broken:
        print("\n  修正が必要なURL:")
        for url, status in broken:
            print(f"    [{status}] {url[:60]}")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("使い方: python precheck_urls.py article.md")
        sys.exit(1)

    main(sys.argv[1])