{
  "name": "insane-search-templates",
  "version": "0.1.0",
  "private": true,
  "description": "Script-only dependencies for local Chrome templates driven from js eval; Chrome must already be installed.",
  "dependencies": {
    "playwright-core": "^1.63.0",
    "playwright-extra": "^4.3.6",
    "puppeteer-extra-plugin-stealth": "^2.11.2"
  }
}

#!/usr/bin/env node
/**
 * Generic Playwright mobile fetcher — real Chrome + device emulation.
 *
 * Usage:
 *   echo '{"url":"...", "device":"iPhone 13 Pro"}' | node playwright_mobile_chrome.js
 *
 * Device name must match playwright `devices[...]` keys (Pixel 7, iPhone 13 Pro,
 * iPad Pro 11, etc.). When in doubt, omit `device` — default is iPhone 13 Pro.
 *
 * NO-SITE-NAME RULE: same as playwright_real_chrome.js — no hostname branches.
 */

async function readStdinJson() {
  return await new Promise((resolve, reject) => {
    let data = '';
    process.stdin.on('data', (c) => (data += c));
    process.stdin.on('end', () => {
      try { resolve(JSON.parse(data || '{}')); }
      catch (e) { reject(e); }
    });
    process.stdin.on('error', reject);
  });
}

function describeError(error) {
  if (error instanceof Error) {
    return `${error.name}: ${error.message}`;
  }
  return String(error);
}

function warnBestEffort(action, error) {
  process.stderr.write(`best-effort ${action} failed: ${describeError(error)}\n`);
}

function isMissingTopLevelModule(error, moduleName) {
  return (
    error instanceof Error &&
    error.code === 'MODULE_NOT_FOUND' &&
    typeof error.message === 'string' &&
    error.message.includes(`Cannot find module '${moduleName}'`)
  );
}

function requireOptionalModule(moduleName) {
  let resolvedModule;
  try {
    resolvedModule = require.resolve(moduleName);
  } catch (e) {
    if (isMissingTopLevelModule(e, moduleName)) {
      warnBestEffort(`optional module ${moduleName}`, e);
      return null;
    }
    throw e;
  }
  return require(resolvedModule);
}

async function main() {
  const args = await readStdinJson();
  const url = args.url;
  if (!url) {
    process.stderr.write('missing url\n');
    process.exitCode = 2;
    return;
  }

  const profileDir = args.profileDir || '/tmp/.insane_pw_mobile_profile';
  const deviceName = args.device || 'iPhone 13 Pro';
  const waitSelector = args.waitSelector || null;
  const timeoutMs = args.timeout || 60000;
  const headless = args.headless ?? false;

  let chromium, devices;
  const playwrightExtra = requireOptionalModule('playwright-extra');
  const stealthPlugin = playwrightExtra ? requireOptionalModule('puppeteer-extra-plugin-stealth') : null;
  if (playwrightExtra && stealthPlugin) {
    ({ devices } = require('playwright-core'));
    chromium = playwrightExtra.addExtra(require('playwright-core').chromium);
    const stealth = stealthPlugin();
    chromium.use(stealth);
  } else {
    ({ chromium, devices } = require('playwright-core'));
  }

  const dev = devices[deviceName];
  if (!dev) {
    process.stderr.write(`unknown device: ${deviceName}\n`);
    process.exitCode = 2;
    return;
  }

  let ctx;
  try {
    ctx = await chromium.launchPersistentContext(profileDir, {
      channel: 'chrome',
      args: ['--disable-blink-features=AutomationControlled'],
      ignoreDefaultArgs: ['--enable-automation'],
      headless,
      ...dev,
    });
    const page = await ctx.newPage();
    const navTimeout = Math.min(timeoutMs, 90000);
    await page.goto(url, { waitUntil: 'domcontentloaded', timeout: navTimeout });

    if (waitSelector) {
      try {
        await page.waitForSelector(waitSelector, { timeout: Math.min(timeoutMs, 20000) });
      } catch (e) {
        warnBestEffort('waitSelector', e);
      }
    }

    const html = await page.content();
    process.stdout.write(html);
    process.exitCode = 0;
    return;
  } catch (e) {
    process.stderr.write(`${describeError(e)}\n`);
    process.exitCode = 1;
    return;
  } finally {
    try {
      if (ctx) await ctx.close();
    } catch (e) {
      warnBestEffort('browser context close', e);
    }
  }
}

main();

#!/usr/bin/env node
/**
 * Generic Playwright fetcher — real Chrome channel (not bundled Chromium).
 *
 * Usage (driven by engine/executor.py):
 *   echo '{"url":"...", "profileDir":"/tmp/.p", "waitSelector":"article"}' | node playwright_real_chrome.js
 *
 * Outputs page HTML to stdout on success; errors to stderr with non-zero exit.
 *
 * NO-SITE-NAME RULE: this file must never branch on specific hostnames.
 * All site specifics come from the JSON input (url, waitSelector).
 *
 * User setup: ../../references/chrome-stealth.md (engine-local dependencies).
 * Chrome must already be installed; optional stealth plugins run only in this script.
 * Execute this script from js eval; profileDir must be task-owned or a CLONED profile.
 */

const fs = require('fs');

async function readStdinJson() {
  return await new Promise((resolve, reject) => {
    let data = '';
    process.stdin.on('data', (c) => (data += c));
    process.stdin.on('end', () => {
      try { resolve(JSON.parse(data || '{}')); }
      catch (e) { reject(e); }
    });
    process.stdin.on('error', reject);
  });
}

function describeError(error) {
  if (error instanceof Error) {
    return `${error.name}: ${error.message}`;
  }
  return String(error);
}

function warnBestEffort(action, error) {
  process.stderr.write(`best-effort ${action} failed: ${describeError(error)}\n`);
}

function isMissingTopLevelModule(error, moduleName) {
  return (
    error instanceof Error &&
    error.code === 'MODULE_NOT_FOUND' &&
    typeof error.message === 'string' &&
    error.message.includes(`Cannot find module '${moduleName}'`)
  );
}

function requireOptionalModule(moduleName) {
  let resolvedModule;
  try {
    resolvedModule = require.resolve(moduleName);
  } catch (e) {
    if (isMissingTopLevelModule(e, moduleName)) {
      warnBestEffort(`optional module ${moduleName}`, e);
      return null;
    }
    throw e;
  }
  return require(resolvedModule);
}

async function main() {
  const args = await readStdinJson();
  const url = args.url;
  if (!url) {
    process.stderr.write('missing url\n');
    process.exitCode = 2;
    return;
  }

  const profileDir = args.profileDir || '/tmp/.insane_pw_profile';
  const waitSelector = args.waitSelector || null;
  const timeoutMs = args.timeout || 60000;
  const headless = args.headless ?? false;     // Akamai/etc detect headless
  const viewport = args.viewport || { width: 1366, height: 900 };

  let chromium;
  const playwrightExtra = requireOptionalModule('playwright-extra');
  const stealthPlugin = playwrightExtra ? requireOptionalModule('puppeteer-extra-plugin-stealth') : null;
  if (playwrightExtra && stealthPlugin) {
    chromium = playwrightExtra.addExtra(require('playwright-core').chromium);
    const stealth = stealthPlugin();
    chromium.use(stealth);
  } else {
    // Core controls installed Chrome without downloading a managed browser.
    ({ chromium } = require('playwright-core'));
  }

  let ctx;
  try {
    ctx = await chromium.launchPersistentContext(profileDir, {
      channel: 'chrome',          // real Chrome, not bundled Chromium
      args: ['--disable-blink-features=AutomationControlled'],
      ignoreDefaultArgs: ['--enable-automation'],
      headless,
      viewport,
    });
    const page = await ctx.newPage();
    const navTimeout = Math.min(timeoutMs, 90000);

    // Warmup hop: visit the site root first so Akamai-style bot managers
    // can run their JS sensor and set a resolved session cookie. Direct
    // landing on a search/deep URL is the classic first-hit rejection pattern.
    // Use domcontentloaded (not networkidle) — many SPAs keep analytics/xhr
    // open indefinitely and would hit the 90s timeout.
    try {
      const urlObj = new URL(url);
      const rootUrl = `${urlObj.protocol}//${urlObj.host}/`;
      if (rootUrl !== url) {
        await page.goto(rootUrl, { waitUntil: 'domcontentloaded', timeout: navTimeout });
        await page.waitForTimeout(3500);   // let sensor JS finish
      }
    } catch (e) {
      warnBestEffort('warmup navigation', e);
      // warmup is best-effort; continue even if it hiccups
    }

    // Main page — DOM loaded then give the sensor a moment.
    await page.goto(url, { waitUntil: 'domcontentloaded', timeout: navTimeout });
    await page.waitForTimeout(2500);

    if (waitSelector) {
      try {
        await page.waitForSelector(waitSelector, { timeout: Math.min(timeoutMs, 20000) });
      } catch (e) {
        warnBestEffort('waitSelector', e);
        // Selector still missing — try one hard reload in case the first hit
        // landed on a challenge page and the sensor has just cleared.
        try {
          await page.reload({ waitUntil: 'domcontentloaded', timeout: navTimeout });
          await page.waitForTimeout(2000);
          try {
            await page.waitForSelector(waitSelector, { timeout: 10000 });
          } catch (e2) {
            warnBestEffort('retry waitSelector', e2);
            // Still no luck — caller validates HTML anyway.
          }
        } catch (e3) {
          warnBestEffort('selector recovery reload', e3);
          // reload failed — proceed with whatever we have
        }
      }
    } else {
      // Without a positive-proof selector, give the sensor a couple more seconds.
      await page.waitForTimeout(2000);
    }

    const html = await page.content();
    process.stdout.write(html);
    process.exitCode = 0;
    return;
  } catch (e) {
    process.stderr.write(`${describeError(e)}\n`);
    process.exitCode = 1;
    return;
  } finally {
    try {
      if (ctx) await ctx.close();
    } catch (e) {
      warnBestEffort('browser context close', e);
    }
  }
}

main();

from __future__ import annotations

import sys
import unittest
from pathlib import Path
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from engine.fetch_chain import fetch  # noqa: E402
from engine.result_schema import Attempt  # noqa: E402
from engine.validators import Verdict  # noqa: E402


class _Resp:
    def __init__(self, text: str = "<article>ok</article>", url: str = "https://example.com/"):
        self.text = text
        self.url = url


class _Hit:
    profile_id = "cloudflare_turnstile"
    confidence = 1.0
    signals = ["test"]


class FetchChain(unittest.TestCase):
    def test_probe_success_returns_without_grid(self) -> None:
        attempt = Attempt(
            phase="probe",
            executor="curl_cffi",
            url="https://example.com",
            url_transform="original",
            impersonate="safari",
            referer="self_root",
            verdict=Verdict.WEAK_OK.value,
        )

        with patch("engine.fetch_chain._load_profiles", return_value={}), \
                patch("engine.fetch_chain.last_load_error", return_value=None), \
                patch("engine.fetch_chain.run_attempt", return_value=(attempt, _Resp())) as run_attempt:
            result = fetch("https://example.com", enable_playwright=False)

        self.assertTrue(result.ok)
        self.assertEqual(result.verdict, Verdict.WEAK_OK.value)
        self.assertEqual(len(result.trace), 1)
        self.assertEqual(run_attempt.call_count, 1)

    def test_max_attempts_stops_after_probe_before_grid(self) -> None:
        attempt = Attempt(
            phase="probe",
            executor="curl_cffi",
            url="https://example.com",
            url_transform="original",
            impersonate="safari",
            referer="self_root",
            verdict=Verdict.CHALLENGE.value,
        )

        with patch("engine.fetch_chain._load_profiles", return_value={}), \
                patch("engine.fetch_chain.last_load_error", return_value=None), \
                patch("engine.fetch_chain.run_attempt", return_value=(attempt, _Resp("blocked"))), \
                patch("engine.fetch_chain.detect", return_value=[_Hit()]), \
                patch("engine.fetch_chain.load_profile", return_value={
                    "tls_impersonate_candidates": [["chrome"]],
                    "referer_strategies": ["self_root"],
                    "url_transform_order": ["original"],
                }):
            result = fetch("https://example.com", max_attempts=1, enable_playwright=False)

        self.assertFalse(result.ok)
        self.assertEqual(len(result.trace), 1)
        self.assertEqual(result.trace[0].phase, "probe")


if __name__ == "__main__":
    unittest.main()

from __future__ import annotations

import json
import os
import subprocess
import tempfile
import unittest
from pathlib import Path

from test_playwright_templates import (
    TEMPLATE_NAMES,
    TEMPLATES_DIR,
    _install_fake_playwright,
    _write_file,
)


class PlaywrightTemplateStealth(unittest.TestCase):
    def test_extracts_html_and_closes_wrapped_chrome_when_stealth_is_installed(self) -> None:
        for template_name in TEMPLATE_NAMES:
            with self.subTest(template_name=template_name), tempfile.TemporaryDirectory() as tmp:
                # Given: core has no plugin API; only addExtra supplies a registered wrapper.
                root = Path(tmp)
                modules = root / "node_modules"
                profile = root / "profile"
                receipt = root / "launch.json"
                _install_fake_playwright(modules)
                _write_file(modules / "puppeteer-extra-plugin-stealth/index.js", """
                    module.exports = () => ({ name: 'stealth' });
                """)
                _write_file(modules / "playwright-extra/index.js", """
                    const assert = require('node:assert/strict');
                    const fs = require('node:fs');
                    exports.addExtra = (core) => {
                      assert.equal(core, require('playwright-core').chromium);
                      assert.equal(Object.hasOwn(core, 'use'), false);
                      const events = ['wrap'];
                      let registered = false;
                      return {
                        use(plugin) {
                          assert.equal(plugin.name, 'stealth');
                          registered = true;
                          events.push(plugin.name);
                        },
                        async launchPersistentContext(profileDir, options) {
                          assert.equal(registered, true);
                          events.push({ profileDir, channel: options.channel, viewport: options.viewport });
                          const context = await core.launchPersistentContext(profileDir, options);
                          return {
                            ...context,
                            async close() {
                              await context.close();
                              events.push('close');
                              fs.writeFileSync(process.env.STEALTH_RECEIPT, JSON.stringify(events));
                            },
                          };
                        },
                      };
                    };
                """)
                script = root / template_name
                script.write_bytes((TEMPLATES_DIR / template_name).read_bytes())
                env = {key: value for key, value in os.environ.items() if key != "NODE_PATH"}
                env["STEALTH_RECEIPT"] = str(receipt)

                # When: the actual bundled entry point consumes its normal stdin protocol.
                result = subprocess.run(
                    ["node", str(script)],
                    input=json.dumps({"url": "https://example.com/", "profileDir": str(profile), "headless": True}),
                    text=True, capture_output=True, timeout=5, env=env, check=False,
                )

                # Then: HTML arrives through the registered wrapper and the context is closed.
                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertEqual(result.stdout, "<html><article>ok</article></html>")
                self.assertEqual(result.stderr, "")
                viewport = {"width": 390, "height": 844} if "mobile" in template_name else {"width": 1366, "height": 900}
                self.assertEqual(json.loads(receipt.read_text(encoding="utf-8")), [
                    "wrap", "stealth",
                    {"profileDir": str(profile), "channel": "chrome", "viewport": viewport},
                    "close",
                ])


if __name__ == "__main__":
    unittest.main()

from __future__ import annotations

import json
import os
import shutil
import subprocess
import tempfile
import textwrap
import unittest
from pathlib import Path

TEMPLATES_DIR = Path(__file__).resolve().parents[1] / "templates"
TEMPLATE_NAMES = ("playwright_real_chrome.js", "playwright_mobile_chrome.js")
JsonValue = None | bool | int | float | str | list["JsonValue"] | dict[str, "JsonValue"]


def _write_file(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(textwrap.dedent(content).lstrip(), encoding="utf-8")


def _install_fake_playwright(node_modules: Path) -> None:
    _write_file(
        node_modules / "playwright-core" / "index.js",
        """
        const page = {
          async goto() {},
          async waitForTimeout() {},
          async waitForSelector(selector) {
            if (process.env.PW_FAKE_SELECTOR_FAIL === '1') {
              throw new Error(`missing selector ${selector}`);
            }
          },
          async reload() {},
          async content() {
            return '<html><article>ok</article></html>';
          },
        };

        const context = {
          async newPage() { return page; },
          async close() {
            if (process.env.PW_FAKE_CLOSE_FAIL === '1') {
              throw new Error('close failed');
            }
          },
        };

        exports.chromium = {
          async launchPersistentContext() { return context; },
        };
        exports.devices = {
          'iPhone 13 Pro': { viewport: { width: 390, height: 844 }, userAgent: 'fake-mobile' },
        };
        """,
    )


def _install_broken_playwright_extra(node_modules: Path) -> None:
    _write_file(
        node_modules / "playwright-extra" / "index.js",
        """
        const error = new Error('stealth init failed');
        error.code = 'EACCES';
        throw error;
        """,
    )
    _write_file(
        node_modules / "puppeteer-extra-plugin-stealth" / "index.js",
        """
        module.exports = function stealth() { return {}; };
        """,
    )


def _install_working_playwright_extra(node_modules: Path) -> None:
    _write_file(
        node_modules / "playwright-extra" / "index.js",
        """
        exports.addExtra = (browserType) => browserType;
        """,
    )


def _install_playwright_extra_with_internal_missing_dependency(node_modules: Path) -> None:
    _write_file(
        node_modules / "playwright-extra" / "index.js",
        """
        require('transitive-stealth-runtime');
        """,
    )


def _install_self_missing_optional_module(node_modules: Path, module_name: str) -> None:
    _write_file(
        node_modules / module_name / "index.js",
        f"""
        const error = new Error("Cannot find module '{module_name}'");
        error.code = 'MODULE_NOT_FOUND';
        throw error;
        """,
    )


def _run_template(
    template_name: str,
    payload: dict[str, JsonValue],
    *,
    include_broken_extra: bool = False,
    include_working_extra: bool = False,
    include_internal_missing_extra: bool = False,
    include_self_missing_module: str | None = None,
    env_overrides: dict[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
    with tempfile.TemporaryDirectory(prefix="ultimate-browsing-template-test-") as tmp:
        tmp_path = Path(tmp)
        node_modules = tmp_path / "node_modules"
        _install_fake_playwright(node_modules)
        if include_working_extra:
            _install_working_playwright_extra(node_modules)
        if include_broken_extra:
            _install_broken_playwright_extra(node_modules)
        if include_internal_missing_extra:
            _install_playwright_extra_with_internal_missing_dependency(node_modules)
        if include_self_missing_module:
            _install_self_missing_optional_module(node_modules, include_self_missing_module)

        script_path = tmp_path / template_name
        script_path.write_text((TEMPLATES_DIR / template_name).read_text(encoding="utf-8"), encoding="utf-8")

        env = os.environ.copy()
        env.pop("NODE_PATH", None)
        if env_overrides:
            env.update(env_overrides)

        return subprocess.run(
            ["node", str(script_path)],
            input=json.dumps(payload),
            text=True,
            capture_output=True,
            timeout=5,
            env=env,
            check=False,
        )


@unittest.skipUnless(shutil.which("node"), "node is required for Playwright template tests")
class PlaywrightTemplateErrorHandling(unittest.TestCase):
    def test_missing_playwright_extra_warns_and_falls_back_to_playwright_core(self) -> None:
        for template_name in TEMPLATE_NAMES:
            with self.subTest(template_name=template_name):
                result = _run_template(
                    template_name,
                    {
                        "url": "https://example.com/article",
                        "profileDir": "/tmp/ultimate-browsing-test-profile",
                        "headless": True,
                    },
                )

                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertIn("<article>ok</article>", result.stdout)
                self.assertIn("best-effort optional module playwright-extra failed:", result.stderr)
                self.assertIn("Cannot find module 'playwright-extra'", result.stderr)

    def test_missing_stealth_plugin_warns_and_falls_back_to_playwright_core(self) -> None:
        for template_name in TEMPLATE_NAMES:
            with self.subTest(template_name=template_name):
                result = _run_template(
                    template_name,
                    {
                        "url": "https://example.com/article",
                        "profileDir": "/tmp/ultimate-browsing-test-profile",
                        "headless": True,
                    },
                    include_working_extra=True,
                )

                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertIn("<article>ok</article>", result.stdout)
                self.assertIn("best-effort optional module puppeteer-extra-plugin-stealth failed:", result.stderr)
                self.assertIn("Cannot find module 'puppeteer-extra-plugin-stealth'", result.stderr)

    def test_non_missing_stealth_dependency_error_is_not_swallowed(self) -> None:
        for template_name in TEMPLATE_NAMES:
            with self.subTest(template_name=template_name):
                result = _run_template(
                    template_name,
                    {
                        "url": "https://example.com/article",
                        "profileDir": "/tmp/ultimate-browsing-test-profile",
                        "headless": True,
                    },
                    include_broken_extra=True,
                )

                self.assertNotEqual(result.returncode, 0)
                self.assertIn("stealth init failed", result.stderr)
                self.assertNotIn("best-effort optional module", result.stderr)
                self.assertEqual(result.stdout, "")

    def test_internal_module_resolution_error_is_not_treated_as_optional_missing_dependency(self) -> None:
        for template_name in TEMPLATE_NAMES:
            with self.subTest(template_name=template_name):
                result = _run_template(
                    template_name,
                    {
                        "url": "https://example.com/article",
                        "profileDir": "/tmp/ultimate-browsing-test-profile",
                        "headless": True,
                    },
                    include_internal_missing_extra=True,
                )

                self.assertNotEqual(result.returncode, 0)
                self.assertIn("Cannot find module 'transitive-stealth-runtime'", result.stderr)
                self.assertNotIn("best-effort optional module", result.stderr)
                self.assertEqual(result.stdout, "")

    def test_present_optional_module_self_missing_error_is_not_swallowed(self) -> None:
        cases = (("playwright-extra", False), ("puppeteer-extra-plugin-stealth", True))
        for template_name in TEMPLATE_NAMES:
            for module_name, include_working_extra in cases:
                with self.subTest(template_name=template_name, module_name=module_name):
                    result = _run_template(
                        template_name,
                        {
                            "url": "https://example.com/article",
                            "profileDir": "/tmp/ultimate-browsing-test-profile",
                            "headless": True,
                        },
                        include_working_extra=include_working_extra,
                        include_self_missing_module=module_name,
                    )

                    self.assertNotEqual(result.returncode, 0)
                    self.assertIn(f"Cannot find module '{module_name}'", result.stderr)
                    self.assertNotIn("best-effort optional module", result.stderr)
                    self.assertEqual(result.stdout, "")

    def test_selector_failures_are_reported_as_best_effort_warnings(self) -> None:
        for template_name in TEMPLATE_NAMES:
            with self.subTest(template_name=template_name):
                result = _run_template(
                    template_name,
                    {
                        "url": "https://example.com/article",
                        "profileDir": "/tmp/ultimate-browsing-test-profile",
                        "headless": True,
                        "waitSelector": "article.ready",
                    },
                    env_overrides={"PW_FAKE_SELECTOR_FAIL": "1"},
                )

                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertIn("<article>ok</article>", result.stdout)
                self.assertIn("best-effort waitSelector failed:", result.stderr)
                self.assertNotIn("waitSelector article.ready", result.stderr)

    def test_context_close_failures_are_reported_after_successful_html_output(self) -> None:
        for template_name in TEMPLATE_NAMES:
            with self.subTest(template_name=template_name):
                result = _run_template(
                    template_name,
                    {
                        "url": "https://example.com/article",
                        "profileDir": "/tmp/ultimate-browsing-test-profile",
                        "headless": True,
                    },
                    env_overrides={"PW_FAKE_CLOSE_FAIL": "1"},
                )

                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertIn("<article>ok</article>", result.stdout)
                self.assertIn("best-effort browser context close failed:", result.stderr)


if __name__ == "__main__":
    unittest.main()

"""Surrogate registry/fetch/ordering tests; registry entries and responses are faked, never live network."""
from __future__ import annotations

import json
import os
import sys
import unittest
from pathlib import Path
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from engine.result_schema import Attempt  # noqa: E402
from engine.validators import Verdict  # noqa: E402

FIXTURES = Path(__file__).resolve().parent / "fixtures"
TARGET = "https://en.wikipedia.org/wiki/Web_archiving"  # NOTE-BIAS-OK — fixture URL from a third-party surrogate service, not a target site


def _fixture(name: str) -> str:
    return (FIXTURES / name).read_text(encoding="utf-8", errors="replace")


class _Resp:
    def __init__(self, text: str = "", status: int = 200, url: str = "", payload: dict | None = None):
        self.text = text
        self.status_code = status
        self.url = url
        self.cookies = {}
        self.headers = {}
        self._payload = payload

    def json(self):
        if self._payload is None:
            raise ValueError("no json payload")
        return self._payload


def _fail_attempt(phase: str) -> Attempt:
    return Attempt(
        phase=phase,
        executor="curl_cffi",
        url=TARGET,
        url_transform="original",
        impersonate="safari",
        referer="self_root",
        verdict=Verdict.CHALLENGE.value,
    )


class SurrogateRegistry(unittest.TestCase):
    def test_default_registry_contains_wayback(self) -> None:
        from engine import surrogate

        reg = surrogate.load_surrogates()
        self.assertIn("wayback", reg)
        self.assertEqual(reg["wayback"]["kind"], "archive")
        self.assertEqual(reg["wayback"]["trust"], "archive")

    def test_stale_entries_detected(self) -> None:
        from engine import surrogate

        self.assertTrue(surrogate.is_stale({"last_verified": "2025-01-01"}, max_age_days=90))
        self.assertFalse(surrogate.is_stale({"last_verified": "2999-01-01"}, max_age_days=90))


class SurrogateFetch(unittest.TestCase):
    def test_wayback_success_sets_snapshot_provenance(self) -> None:
        from engine import surrogate

        discovery = json.loads(_fixture("wayback_available.json"))

        def fake_http(u: str, **kw):
            if "wayback/available" in u:
                return _Resp(json.dumps(discovery), 200, u, payload=discovery)
            return _Resp(_fixture("wayback_snapshot.html"), 200, u)

        with patch.object(surrogate, "_http_get", side_effect=fake_http):
            atts, meta, content = surrogate.run_surrogate(TARGET, registry=surrogate.load_surrogates(), timeout=5)

        att = atts[-1]
        self.assertEqual(att.phase, "surrogate")
        self.assertEqual(att.executor, "surrogate_wayback")
        self.assertEqual(att.verdict, Verdict.WEAK_OK.value)
        self.assertEqual(meta["provenance"], "snapshot")
        self.assertTrue(meta["snapshot_timestamp"])
        self.assertEqual(meta["trust"], "archive")
        self.assertGreater(len(content), 3000)

    def test_stub_snapshot_is_not_accepted(self) -> None:
        from engine import surrogate

        disc = {"archived_snapshots": {"closest": {"available": True, "url": "https://web.archive.org/web/2024/x", "timestamp": "20240101000000"}}}  # NOTE-BIAS-OK — fixture URL from a third-party surrogate service, not a target site

        def fake_http(u: str, **kw):
            if "wayback/available" in u:
                return _Resp(json.dumps(disc), 200, u, payload=disc)
            return _Resp(_fixture("amp_redirect_stub.html"), 200, u)

        with patch.object(surrogate, "_http_get", side_effect=fake_http):
            atts, _, _ = surrogate.run_surrogate(TARGET, registry={"wayback": surrogate.DEFAULT_SURROGATES["wayback"]}, timeout=5)
        att = atts[-1]
        self.assertNotIn(att.verdict, (Verdict.STRONG_OK.value, Verdict.WEAK_OK.value))

    def test_proxy_skips_without_allow_flag(self) -> None:
        from engine import surrogate

        reg = {"anon_relay": {"kind": "proxy", "trust": "untrusted", "enabled": False,
                              "last_verified": "2999-01-01", "fetch": "https://relay.invalid/{target}"}}  # NOTE-BIAS-OK — fixture URL from a third-party surrogate service, not a target site
        called: list[str] = []

        with patch.object(surrogate, "_http_get", side_effect=lambda u, **kw: (called.append(u), _Resp("<html>" + "y" * 4000 + "</html>", 200, u))[1]):
            atts, _, _ = surrogate.run_surrogate(TARGET, registry=reg, allow_proxy=False, timeout=5)
        self.assertEqual(len(called), 0)

        with patch.object(surrogate, "_http_get", side_effect=lambda u, **kw: (called.append(u), _Resp("<html>" + "y" * 4000 + "</html>", 200, u))[1]):
            atts, meta, _ = surrogate.run_surrogate(TARGET, registry=reg, allow_proxy=True, timeout=5)
        self.assertEqual(len(called), 1)
        att = atts[-1]
        self.assertEqual(att.executor, "surrogate_anon_relay")
        self.assertEqual(meta["trust"], "untrusted")

    def test_proxy_never_carries_credentials(self) -> None:
        from engine import surrogate

        seen: list[dict] = []

        def fake_http(u: str, **kw):
            seen.append(kw.get("headers", {}))
            return _Resp("<html>" + "z" * 4000 + "</html>", 200, u)

        reg = {"anon_relay": {"kind": "proxy", "trust": "untrusted", "enabled": False,
                              "last_verified": "2999-01-01", "fetch": "https://relay.invalid/{target}"}}  # NOTE-BIAS-OK — fixture URL from a third-party surrogate service, not a target site
        os.environ["OMOB_TEST_TOKEN"] = "secret"
        try:
            with patch.object(surrogate, "_http_get", side_effect=fake_http):
                surrogate.run_surrogate(TARGET, registry=reg, allow_proxy=True, timeout=5)
        finally:
            os.environ.pop("OMOB_TEST_TOKEN", None)

        self.assertEqual(len(seen), 1)
        joined = json.dumps(seen[0]).lower()
        self.assertNotIn("secret", joined)
        self.assertNotIn("authorization", joined)
        self.assertNotIn("cookie", joined)


class ChainOrdering(unittest.TestCase):
    def _profile(self) -> dict:
        return {
            "tls_impersonate_candidates": [[]],
            "referer_strategies": [],
            "url_transform_order": ["original"],
            "fallback_when_challenge": ["surrogate_wayback", "playwright_real_chrome"],
        }

    def test_surrogate_success_short_circuits_playwright(self) -> None:
        from engine.fetch_chain import fetch
        from engine import surrogate
        from engine import executor as ex

        def fake_surrogate(*a, **kw):
            att = Attempt(
                phase="surrogate",
                executor="surrogate_wayback",
                url=TARGET,
                url_transform="original",
                impersonate=None,
                referer="",
                verdict=Verdict.WEAK_OK.value,
                body_size=5000,
            )
            return [att], {"provenance": "snapshot", "snapshot_timestamp": "20240101000000", "trust": "archive"}, "<html>content</html>"

        def fail_playwright(*a, **kw):
            raise AssertionError("playwright must not run after surrogate success")

        with patch("engine.fetch_chain._load_profiles", return_value={}), \
                patch("engine.fetch_chain.last_load_error", return_value=None), \
                patch("engine.fetch_chain.run_attempt", side_effect=lambda *a, **kw: (_fail_attempt(str(kw.get("phase", "grid"))), None)), \
                patch("engine.fetch_chain.detect", return_value=[]), \
                patch("engine.fetch_chain.load_profile", side_effect=lambda *a, **kw: self._profile()), \
                patch.object(surrogate, "run_surrogate", side_effect=fake_surrogate), \
                patch.object(ex, "run_playwright_fallback", side_effect=fail_playwright):
            result = fetch(TARGET, enable_playwright=True, max_attempts=1)

        self.assertTrue(result.ok)
        self.assertEqual(result.provenance, "snapshot")
        self.assertEqual(result.snapshot_timestamp, "20240101000000")

    def test_surrogate_still_runs_when_playwright_disabled(self) -> None:
        from engine.fetch_chain import fetch
        from engine import surrogate

        calls: list[str] = []

        def fake_surrogate(*a, **kw):
            calls.append("surrogate")
            att = Attempt(phase="surrogate", executor="surrogate_wayback", url=TARGET,
                          url_transform="original", impersonate=None, referer="",
                          verdict=Verdict.WEAK_OK.value, body_size=5000)
            return [att], {"provenance": "snapshot", "snapshot_timestamp": "20240101000000", "trust": "archive"}, "<html>snap</html>"

        with patch("engine.fetch_chain._load_profiles", return_value={}), \
                patch("engine.fetch_chain.last_load_error", return_value=None), \
                patch("engine.fetch_chain.run_attempt", side_effect=lambda *a, **kw: (_fail_attempt(str(kw.get("phase", "grid"))), None)), \
                patch("engine.fetch_chain.detect", return_value=[]), \
                patch("engine.fetch_chain.load_profile", side_effect=lambda *a, **kw: self._profile()), \
                patch.object(surrogate, "run_surrogate", side_effect=fake_surrogate):
            result = fetch(TARGET, enable_playwright=False, max_attempts=1)

        self.assertEqual(calls, ["surrogate"])
        self.assertTrue(result.ok)
        self.assertEqual(result.provenance, "snapshot")

    def test_playwright_runs_when_surrogate_fails(self) -> None:
        from engine.fetch_chain import fetch
        from engine import surrogate
        from engine import executor as ex

        calls: list[str] = []

        def fake_surrogate(*a, **kw):
            calls.append("surrogate")
            att = Attempt(phase="surrogate", executor="surrogate_wayback", url=TARGET,
                          url_transform="original", impersonate=None, referer="",
                          verdict=Verdict.UNKNOWN.value)
            return [att], {"provenance": "live", "snapshot_timestamp": None, "trust": "origin"}, ""

        def fake_playwright(*a, **kw):
            calls.append("playwright")
            att = Attempt(phase="fallback", executor="playwright_real_chrome", url=TARGET,
                          url_transform="original", impersonate=None, referer="",
                          verdict=Verdict.WEAK_OK.value, body_size=5000)
            return att, "<html>chrome content</html>"

        with patch("engine.fetch_chain._load_profiles", return_value={}), \
                patch("engine.fetch_chain.last_load_error", return_value=None), \
                patch("engine.fetch_chain.run_attempt", side_effect=lambda *a, **kw: (_fail_attempt(str(kw.get("phase", "grid"))), None)), \
                patch("engine.fetch_chain.detect", return_value=[]), \
                patch("engine.fetch_chain.load_profile", side_effect=lambda *a, **kw: self._profile()), \
                patch.object(surrogate, "run_surrogate", side_effect=fake_surrogate), \
                patch.object(ex, "run_playwright_fallback", side_effect=fake_playwright):
            result = fetch(TARGET, enable_playwright=True, max_attempts=1)

        self.assertTrue(result.ok)
        self.assertEqual(result.provenance, "live")
        self.assertEqual(calls, ["surrogate", "playwright"])


if __name__ == "__main__":
    unittest.main()

"""Surrogate-tier validation tests; fixtures only, never live network, so the suite cannot flake."""
from __future__ import annotations

import sys
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from engine.result_schema import FetchResult  # noqa: E402
from engine.validators import Verdict, validate  # noqa: E402

FIXTURES = Path(__file__).resolve().parent / "fixtures"


def _fixture(name: str) -> str:
    return (FIXTURES / name).read_text(encoding="utf-8", errors="replace")


class _Resp:
    def __init__(self, text: str = "", status: int = 200, url: str = ""):
        self.text = text
        self.status_code = status
        self.url = url
        self.cookies = {}
        self.headers = {}


class RedirectStubRejection(unittest.TestCase):
    def test_amp_redirect_stub_is_not_ok(self) -> None:
        body = _fixture("amp_redirect_stub.html")
        self.assertLess(len(body), 1000)
        resp = _Resp(body, 200, "https://en-wikipedia-org.cdn.ampproject.org/c/s/x")  # NOTE-BIAS-OK — fixture URL from a third-party surrogate service, not a target site
        vr = validate(resp, target_url="https://en.wikipedia.org/wiki/Web_archiving")  # NOTE-BIAS-OK — fixture URL from a third-party surrogate service, not a target site
        self.assertIn(vr.verdict, (Verdict.CHALLENGE, Verdict.UNKNOWN))
        self.assertFalse(vr.ok)

    def test_real_snapshot_still_passes(self) -> None:
        resp = _Resp(_fixture("wayback_snapshot.html"), 200, "https://web.archive.org/web/2024/x")  # NOTE-BIAS-OK — fixture URL from a third-party surrogate service, not a target site
        vr = validate(resp, target_url="https://en.wikipedia.org/wiki/Web_archiving")  # NOTE-BIAS-OK — fixture URL from a third-party surrogate service, not a target site
        self.assertTrue(vr.ok)

    def test_plain_page_still_weak_ok(self) -> None:
        resp = _Resp("<html><body>" + ("x" * 4000) + "</body></html>", 200, "https://example.com/")
        vr = validate(resp, target_url="https://example.com/")
        self.assertTrue(vr.ok)


class InterstitialRejection(unittest.TestCase):
    def test_search_engine_interstitial_is_not_ok(self) -> None:
        body = _fixture("search_interstitial.html")
        resp = _Resp(body, 200, "https://webcache.googleusercontent.com/search?q=cache:x")  # NOTE-BIAS-OK — fixture URL from a third-party surrogate service, not a target site
        vr = validate(resp, target_url="https://en.wikipedia.org/wiki/Web_archiving")  # NOTE-BIAS-OK — fixture URL from a third-party surrogate service, not a target site
        self.assertIn(vr.verdict, (Verdict.CHALLENGE, Verdict.UNKNOWN))
        self.assertFalse(vr.ok)


class ResultSchemaProvenance(unittest.TestCase):
    def test_defaults_are_live_origin(self) -> None:
        r = FetchResult(ok=True)
        self.assertEqual(r.provenance, "live")
        self.assertEqual(r.trust, "origin")
        self.assertIsNone(r.snapshot_timestamp)
        d = r.to_dict()
        self.assertEqual(d["provenance"], "live")
        self.assertEqual(d["trust"], "origin")
        self.assertIn("snapshot_timestamp", d)

    def test_snapshot_fields_roundtrip(self) -> None:
        r = FetchResult(ok=True, provenance="snapshot", snapshot_timestamp="20241227132135", trust="archive")
        d = r.to_dict()
        self.assertEqual(d["provenance"], "snapshot")
        self.assertEqual(d["snapshot_timestamp"], "20241227132135")
        self.assertEqual(d["trust"], "archive")


if __name__ == "__main__":
    unittest.main()

# 职场招聘

LinkedIn。

## LinkedIn

```bash
# 获取个人资料
mcporter call 'linkedin-scraper.get_person_profile(linkedin_url: "https://linkedin.com/in/username")'

# 搜索人才
mcporter call 'linkedin-scraper.search_people(keyword: "AI engineer", limit: 10)'

# 获取公司资料
mcporter call 'linkedin-scraper.get_company_profile(linkedin_url: "https://linkedin.com/company/xxx")'

# 搜索职位
mcporter call 'linkedin-scraper.search_jobs(keyword: "software engineer", limit: 10)'
```

> **需要登录**: LinkedIn scraper 需要有效的登录态。

### Fallback 方案

如果 MCP 不可用，可以用 Jina Reader：

```bash
curl -s "https://r.jina.ai/https://linkedin.com/in/username"
```

# 开发工具

GitHub CLI

## GitHub (gh CLI)

GitHub 官方命令行工具，用于仓库、Issue、PR、Actions、Release 以及 API 访问。

```bash
# 认证
gh auth login
gh auth status

# 搜索
gh search repos "query" --sort stars --limit 10
gh search code "query" --language python

# 仓库
gh repo view owner/repo
gh repo clone owner/repo
gh repo create my-repo --private
gh repo fork owner/repo
gh repo fork owner/repo --clone
gh repo sync owner/repo

# Issues
gh issue list -R owner/repo --state open
gh issue view 123 -R owner/repo
gh issue create -R owner/repo --title "Title" --body "Body"

# Pull Requests
gh pr list -R owner/repo --state open
gh pr view 123 -R owner/repo
gh pr create -R owner/repo --title "Title" --body "Body"
gh pr checks 123 --repo owner/repo

# Actions / CI
gh run list --repo owner/repo --limit 10
gh run view <run-id> --repo owner/repo
gh run view <run-id> --repo owner/repo --log-failed
gh workflow list --repo owner/repo

# Releases
gh release list -R owner/repo
gh release create v1.0.0

# API
gh api /user
gh api repos/owner/repo

# JSON 输出
gh issue list --repo owner/repo --json number,title --jq '.[] | "\(.number): \(.title)"'
```


## 选择指南

| 工具 | 来源 | 用途 |
|-----|------|------|
| gh CLI | agent-reach | Git 操作 |
| zread | my-mcp-tools | 读仓库内容 |
| context7 | my-mcp-tools | 查技术文档 |

# Agent Reach — 路由器

> Part of the **ultimate-browsing** skill (Tier 1.5). Routing and tier-escalation live in [../../SKILL.md](../../SKILL.md).
> Per-category guides in this folder: [search.md](search.md) · [social.md](social.md) · [career.md](career.md) · [dev.md](dev.md) · [web.md](web.md) · [video.md](video.md).


17 平台工具集合。根据用户意图选择对应分类。

## 路由表

| 用户意图 | 分类 | 详细文档 |
|---------|------|---------|
| 网页搜索/代码搜索 | search | [search.md](search.md) |
| 小红书/抖音/微博/推特/B站/V2EX/Reddit | social | [social.md](social.md) |
| 招聘/职位/LinkedIn | career | [career.md](career.md) |
| GitHub/代码 | dev | [dev.md](dev.md) |
| 网页/文章/公众号/RSS | web | [web.md](web.md) |
| YouTube/B站/播客字幕 | video | [video.md](video.md) |

## 零配置快速命令

```bash
# Exa 网页搜索
mcporter call 'exa.web_search_exa(query: "query", numResults: 5)'

# 通用网页阅读
curl -s "https://r.jina.ai/URL"

# GitHub 搜索
gh search repos "query" --sort stars --limit 10

# Twitter 搜索
twitter search "query" --limit 10

# YouTube/B站字幕
yt-dlp --write-sub --skip-download -o "/tmp/%(id)s" "URL"

# Reddit 搜索
rdt search "query" --limit 10

# Reddit 读帖 + 评论
rdt read POST_ID

# V2EX 热门
curl -s "https://www.v2ex.com/api/topics/hot.json" -H "User-Agent: agent-reach/1.0"
```

## 环境检查

```bash
# 检查可用 channel
agent-reach doctor

# 查看所有 MCP 服务
mcporter_list_servers()
```

## 工作区规则

**不要在 agent workspace 创建文件。** 使用 `/tmp/` 存放临时输出。

## 详细文档

根据用户需求，阅读对应的详细文档：

- [搜索工具](search.md) — Exa AI 搜索
- [社交媒体](social.md) — 小红书, 抖音, Twitter, B站, V2EX, Reddit
- [职场招聘](career.md) — LinkedIn
- [开发工具](dev.md) — GitHub CLI
- [网页阅读](web.md) — Jina Reader, 微信公众号, RSS
- [视频播客](video.md) — YouTube, B站, 小宇宙

## 配置渠道

如果某个 channel 需要配置，获取安装指南：
https://raw.githubusercontent.com/Panniantong/agent-reach/main/docs/install.md

用户只需提供 cookies，其他配置由 agent 完成。

# 搜索工具

Exa AI 搜索引擎。

## Exa AI 搜索

高质量 AI 搜索引擎，擅长技术和代码搜索。

```bash
mcporter call 'exa.web_search_exa(query: "query", numResults: 5)'
mcporter call 'exa.get_code_context_exa(query: "code question", tokensNum: 3000)'
```

### 使用场景

| 场景 | 参数 |
|-----|------|
| 网页搜索 | `web_search_exa(query: "...", numResults: 5)` |
| 代码搜索 | `get_code_context_exa(query: "...", tokensNum: 3000)` |

### 特点

- 擅长英文内容和技术文档
- 支持代码上下文搜索
- 结果质量高

## 与其他搜索工具对比

| 工具 | 来源 | 适用场景 |
|-----|------|---------|
| Exa | agent-reach | 英文/技术/代码搜索 |
| 智谱搜索 | my-mcp-tools | 中文搜索 |
| GitHub 搜索 | agent-reach (dev.md) | 仓库/代码搜索 |

# 社交媒体 & 社区

小红书、抖音、Twitter/X、微博、B站、V2EX、Reddit。

## 小红书 / XiaoHongShu (xhs-cli)

### 稳定可用的命令

```bash
# 搜索笔记（推荐入口）
xhs search "query"

# 阅读笔记详情（必须用搜索结果中的 URL 或 ID，不能裸 note_id）
xhs read NOTE_ID_OR_URL

# 查看评论
xhs comments NOTE_ID_OR_URL

# 浏览热门
xhs hot

# 推荐 feed
xhs feed
```

### 已知不稳定的命令（v0.6.4）

```bash
# 以下命令当前可能返回 API error，谨慎使用：
xhs user USER_ID          # 可能返回 {code: -1}
xhs user-posts USER_ID    # 可能返回 {code: -1}
xhs favorites              # 可能返回 API error
```

### 重要注意事项

> **安装**: `pipx install xiaohongshu-cli`，然后 `xhs login`（自动从浏览器提取 Cookie）。
>
> **签名令牌限制**: 小红书强制每个 note 携带一个签名令牌，**不能直接用裸 note_id 去读**。正确流程是：先 `xhs search` 或 `xhs feed` 获取结果，再用结果中的 URL/ID 去 `xhs read`。直接构造 note_id 会被拦截。
>
> **频率控制**: 高频请求（批量搜索、深翻评论）会触发验证码，这是平台限制无法绕过。建议每次操作间隔 2-3 秒。
>
> **POST 操作**: 发帖(post)、评论(comment)、点赞(like) 等写操作在 v0.6.4 已修复签名问题 (PR [#19](https://github.com/jackwener/xiaohongshu-cli/pull/19))，可正常使用。

## 抖音 / Douyin

```bash
# 解析视频信息
mcporter call 'douyin.parse_douyin_video_info(share_link: "https://v.douyin.com/xxx/")'

# 获取无水印下载链接
mcporter call 'douyin.get_douyin_download_link(share_link: "https://v.douyin.com/xxx/")'

# 提取视频文案
mcporter call 'douyin.extract_douyin_text(share_link: "https://v.douyin.com/xxx/")'
```

> **无需登录**

## Twitter/X (twitter-cli)

### 稳定命令

```bash
# 首页时间线（最稳定）
twitter feed -n 20

# 读取单条推文（含回复）
twitter tweet URL_OR_ID

# 读取长文 / X Article
twitter article URL_OR_ID

# 用户时间线
twitter user-posts @username -n 20

# 用户资料
twitter user @username
```

### 可能不稳定的命令

```bash
# 搜索推文（Twitter 频繁改 GraphQL 端点，可能 404）
twitter search "query" -n 10
# 如果 search 返回 404，升级 twitter-cli：pipx upgrade twitter-cli

# likes（2024 年后只能看自己的，平台限制）
twitter likes
```

### 重要注意事项

> **安装**: `pipx install twitter-cli`（确保 v0.8.5+）
>
> **认证**: 如果你有访问权限，导出 Twitter 会话 Cookie 后，把 auth-token 与 ct0 两个值设置为 twitter-cli 文档所要求的认证环境变量。自动提取在 SSH/Docker/无头环境不可用。
>
> **IP 风控**: 不要在 VPS/数据中心 IP 上频繁调用，尤其是 followers/following，有封号风险。使用住宅代理或本地环境。
>
> **search 可能失效**: Twitter 频繁修改 GraphQL API，search 命令可能随时返回 404。如遇到，先 `pipx upgrade twitter-cli`。如果最新版仍不行，说明上游还没跟上 Twitter 的改动，用 `twitter feed` 替代。
>
> **输出格式**: 建议用 `--yaml` 或 `--json` 获得结构化输出，对 AI agent 更友好。

## 微博 / Weibo

```bash
# 使用 Jina Reader 读取
curl -s "https://r.jina.ai/https://weibo.com/USER_ID/POST_ID"
```

> 微博主要通过网页抓取，推荐使用通用网页读取方式。

## B站 / Bilibili

```bash
# 获取视频元数据
yt-dlp --dump-json "https://www.bilibili.com/video/BVxxx"

# 下载字幕
yt-dlp --write-sub --write-auto-sub --sub-lang "zh-Hans,zh,en" --convert-subs vtt --skip-download -o "/tmp/%(id)s" "URL"
```

> **注意**: 服务器 IP 可能遇到 412 错误。使用 `--cookies-from-browser chrome` 或配置代理。

## V2EX (公开 API)

无需认证，直接调用公开 API。

### 热门主题

```bash
curl -s "https://www.v2ex.com/api/topics/hot.json" -H "User-Agent: agent-reach/1.0"
```

### 节点主题

```bash
# node_name 如: python, tech, jobs, qna, programmers
curl -s "https://www.v2ex.com/api/topics/show.json?node_name=python&page=1" -H "User-Agent: agent-reach/1.0"
```

### 主题详情

```bash
# topic_id 从 URL 获取，如 https://www.v2ex.com/t/1234567
curl -s "https://www.v2ex.com/api/topics/show.json?id=TOPIC_ID" -H "User-Agent: agent-reach/1.0"
```

### 主题回复

```bash
curl -s "https://www.v2ex.com/api/replies/show.json?topic_id=TOPIC_ID&page=1" -H "User-Agent: agent-reach/1.0"
```

### 用户信息

```bash
curl -s "https://www.v2ex.com/api/members/show.json?username=USERNAME" -H "User-Agent: agent-reach/1.0"
```

### Python 调用示例

```python
from agent_reach.channels.v2ex import V2EXChannel

ch = V2EXChannel()

# 获取热门帖子
topics = ch.get_hot_topics(limit=10)
for t in topics:
    print(f"[{t['node_title']}] {t['title']} ({t['replies']} 回复)")

# 获取节点帖子
node_topics = ch.get_node_topics("python", limit=5)

# 获取帖子详情 + 回复
topic = ch.get_topic(1234567)
print(topic["title"], "—", topic["author"])

# 获取用户信息
user = ch.get_user("Livid")
```

> **节点列表**: https://www.v2ex.com/planes

## Reddit (rdt-cli)

```bash
# 搜索帖子
rdt search "query" --limit 10

# 读帖子全文 + 评论
rdt read POST_ID

# 浏览 subreddit
rdt sub python --limit 20

# 浏览热门
rdt popular --limit 10

# 浏览 /r/all
rdt all --limit 10
```

> **安装**: `pipx install rdt-cli`（确保 v0.4.2+）。无需登录即可搜索和阅读。
> 需要登录的功能：`rdt feed --subs-only`（订阅列表）、`rdt saved`（收藏）。
> 建议使用 `--yaml` 输出，对 AI agent 更友好。

# 视频/播客

YouTube、B站、小宇宙播客的字幕和转录。

## YouTube (yt-dlp)

### 获取视频元数据

```bash
yt-dlp --dump-json "URL"
```

### 下载字幕

```bash
# 下载字幕 (不下载视频)
yt-dlp --write-sub --write-auto-sub --sub-lang "zh-Hans,zh,en" --skip-download -o "/tmp/%(id)s" "URL"

# 然后读取 .vtt 文件
cat /tmp/VIDEO_ID.*.vtt
```

### 获取评论

```bash
# 提取评论（best-effort，不保证完整）
yt-dlp --write-comments --skip-download --write-info-json \
  --extractor-args "youtube:max_comments=20" \
  -o "/tmp/%(id)s" "URL"
# 评论在 .info.json 的 comments 字段中
```

### 搜索视频

```bash
yt-dlp --dump-json "ytsearch5:query"
```

> **字幕注意**: 手动上传的字幕提取可靠；自动生成字幕可能存在行间重复，需后处理。
> **评论注意**: `--write-comments` 基于网页抓取（非 YouTube Data API），部分评论可能丢失。

## B站 / Bilibili (yt-dlp + bili-cli)

### 视频元数据 (yt-dlp)

```bash
yt-dlp --dump-json "https://www.bilibili.com/video/BVxxx"
```

### 字幕 (yt-dlp)

```bash
yt-dlp --write-sub --write-auto-sub --sub-lang "zh-Hans,zh,en" --convert-subs vtt --skip-download -o "/tmp/%(id)s" "URL"
```

### 搜索/热门/排行 (bili-cli)

```bash
# 搜索视频
bili search "query" --type video -n 5

# 热门视频
bili hot -n 10

# 排行榜
bili rank -n 10
```

> **412 风控**: 海外 IP 必须提供 Cookie（`--cookies-from-browser chrome` 或 `--cookies /path/to/cookies.txt`），国内 IP 一般不受影响。
> **安装 bili-cli**: `pipx install bilibili-cli`，然后 `bili login` 扫码登录。

## 小宇宙播客 / Xiaoyuzhou Podcast

### 转录单集播客

```bash
# 输出 Markdown 文件到 /tmp/ (transcribe via the agent-reach xiaoyuzhou tool)
agent-reach run xiaoyuzhou transcribe "https://www.xiaoyuzhoufm.com/episode/EPISODE_ID"
```

### 前置要求

1. **ffmpeg**: `brew install ffmpeg`
2. **Groq API Key** (免费): https://console.groq.com/keys
3. **配置 Key**: `agent-reach configure groq-key YOUR_KEY`
4. **首次运行**: `agent-reach install --env=auto` 安装工具

### 检查状态

```bash
agent-reach doctor
```

> 输出 Markdown 文件默认保存到 `/tmp/`。

## 抖音视频解析

```bash
# 解析视频信息
mcporter call 'douyin.parse_douyin_video_info(share_link: "https://v.douyin.com/xxx/")'

# 获取无水印下载链接
mcporter call 'douyin.get_douyin_download_link(share_link: "https://v.douyin.com/xxx/")'
```

> 详见 [social.md](social.md#抖音--douyin)

## 选择指南

| 场景 | 推荐工具 |
|-----|---------|
| YouTube 字幕 | yt-dlp |
| B站字幕 | yt-dlp |
| 播客转录 | 小宇宙 transcribe.sh |
| 抖音视频解析 | douyin MCP |

# 网页阅读

通用网页、微信公众号、RSS。

## 通用网页 (Jina Reader)

```bash
# 读取任意网页内容
curl -s "https://r.jina.ai/URL"

# 示例
curl -s "https://r.jina.ai/https://example.com/article"
```

**适用场景**: 大多数网页可以直接用 Jina Reader 读取。

## Web Reader (MCP)

```bash
# 读取网页内容 (Markdown 格式)
mcporter call 'web-reader.webReader(url: "https://example.com")'

# 保留图片
mcporter call 'web-reader.webReader(url: "https://example.com", retain_images: true)'

# 纯文本格式
mcporter call 'web-reader.webReader(url: "https://example.com", return_format: "text")'
```

**适用场景**: 需要更精确控制输出格式时使用。

## 微信公众号 / WeChat Articles

### 搜索公众号文章（通过 Exa）

```bash
# 搜索微信公众号文章
mcporter call 'exa.web_search_exa(query: "搜索关键词", numResults: 5, includeDomains: ["mp.weixin.qq.com"])'
```

### 阅读公众号文章全文（通过 Exa）

```bash
# 抓取文章全文
mcporter call 'exa.crawling_exa(urls: ["https://mp.weixin.qq.com/s/ARTICLE_ID"], maxCharacters: 10000)'
```

### 可选：Camoufox 阅读（反爬更强）

```bash
agent-reach run wechat-article "https://mp.weixin.qq.com/s/ARTICLE_ID"   # Camoufox-backed WeChat reader
```

> **注意**: Jina Reader 无法读取微信文章（被 CAPTCHA 拦截），推荐用 Exa。

## RSS (feedparser)

```python
python3 -c "
import feedparser
for e in feedparser.parse('FEED_URL').entries[:5]:
    print(f'{e.title} — {e.link}')
"
```

**适用场景**: 订阅博客、新闻源、播客等 RSS feed。

## 选择指南

| 场景 | 推荐工具 |
|-----|---------|
| 通用网页 | Jina Reader (`curl r.jina.ai`) |
| 需要图片/格式控制 | web-reader MCP |
| 微信公众号 | Exa (搜索+阅读) / Camoufox (可选阅读) |
| RSS 订阅 | feedparser |
| 微博/知乎等 | Jina Reader |

# 캐시 & 아카이브 (surrogate 경로)

> 원본 사이트가 차단되었을 때 캐시/아카이브된 **사본**으로 접근.
> 2026-08-09 실측 probe 기준으로 정렬. 각 경로는 생명 주기가 짧다 — 이 파일도
> 90일마다 재검증 대상. (당일 probe: 기존 기대 경로 6개 중 4개 사망 또는 스텁 반환.)

## 의존성

없음 (curl만 사용). 수동 경로이며, 자동화는 엔진 Phase 2.5(`engine/surrogates.yaml`)가 담당한다.

## 엔진 자동 폴백 (Phase 2.5)

`waf_profiles.yaml`의 `fallback_when_challenge`가 `surrogate_wayback`을 앞에 두므로,
그리드 실패 후 브라우저 실행 전에 아카이브 경로를 먼저 시도한다.
성공 시 `FetchResult.provenance = "snapshot"`, `snapshot_timestamp` = 아카이브의 자체 타임스탬프,
`trust = "archive"`가 채워진다. **사본이므로 반드시 날짜와 함께 인용할 것.**
`--allow-proxy` 없이는 `kind: proxy` 엔트리는 절대 실행되지 않으며, 프록시에는
Cookie/Authorization 헤더를 보내지 않는다 (중계자 = 구조적 MITM).

## 1. Wayback Machine (Internet Archive) — 1순위

**2026-08-09 probe: 정상 동작.** `available` API가 200 JSON으로 스냅샷 URL과
타임스탬프를 돌려준다 — 출처(provenance) 확보에 가장 좋은 primitive.

```bash
# 스냅샷 존재 여부 + 최신 스냅샷 URL/타임스탬프 (진입점으로 이것을 쓸 것)
curl -sL "https://archive.org/wayback/available?url={URL}"

# 반환 JSON의 archived_snapshots.closest.url 로 접근
curl -sL "https://web.archive.org/web/{timestamp}/{URL}"
```

> **CDX API 주의**: 이전 버전이 권장하던 `web.archive.org/cdx/search/cdx`는
> 2026-08 probe에서 503 반환. 스냅샷 열거가 필요 없으면 `available` API만 사용.

**성공 조건**: 크롤링 대상이었던 공개 URL
**실패 조건**: robots.txt로 차단된 사이트, 스냅샷이 없는 URL, SPA 스냅샷 (렌더링 안 됨)

## 2. archive.today — 2순위

사용자 제출 아카이브. 페이월 기사, 삭제된 콘텐츠에 특히 유용.
**2026-08 probe: 429 rate-limit이 잦고 도메인이 수시로 회전** (archive.ph → archive.md 관찰).
하나가 차단되면 다른 도메인을 순회한다 (엔진 `host_rotation`과 동일 패턴).

```bash
# 최신 스냅샷 조회 — 도메인 회전은 필수 경로, 예외 처리 아님
for domain in archive.ph archive.md archive.li archive.is; do
  resp=$(curl -sL -o /dev/null -w "%{http_code}" "https://$domain/newest/{URL}")
  if [ "$resp" = "200" ] || [ "$resp" = "302" ]; then
    echo "성공: https://$domain/newest/{URL}"
    curl -sL "https://$domain/newest/{URL}"
    break
  fi
done
```

**주의**: 429 응답에도 수 KB 본문이 딸려 오므로 상태코드 대신 본문 검증이 필요하다.

## 3. AMP 캐시 — 강등 (사실상 무용)

과거 1순위였으나 **2026-08 probe에서 사실상 무력화**:
`{host}.cdn.ampproject.org/c/s/...`가 HTTP 200을 돌려주지만, 실제 본문은
**322바이트짜리 `<TITLE>Redirecting</TITLE>` meta-refresh** — 대상은 다시 **원본(차단된) 페이지**다.
이걸 성공으로 착각하면 에이전트가 차단 페이지로 되돌아가는 루프가 생긴다.

엔진은 `engine/validators.py:is_redirect_stub`으로 이 패턴을 CHALLENGE 판정한다
(3KB 미만 + meta-refresh/JS redirect + 대상 호스트 재등장 조합).
수동 사용도 권장하지 않는다.

## 4. Google Cache — 사망 확정

**2024년 7월 종료** 후로도 `webcache.googleusercontent.com`이 HTTP 200 + 수십 KB의
본문을 반환하지만, 실제로는 `<title>Google Search</title>` 인터스티셜 + JS 리다이렉트다
(2026-08 probe 재확인). **캐시가 아니라 검색 홈이다.**
엔진은 `INTERSTITIAL_TITLE_MARKERS`로 판정해 성공 집계에서 배제한다.

## 시도 순서 (probe 근거)

```
1. Wayback available API → 스냅샷 URL + 타임스탬프 (provenance까지 확보)
2. archive.today 도메인 회전 (429 대비, 본문 검증 필수)
3. AMP 캐시: 시도하지 않음 (redirect stub → 원본으로 회귀)
4. Google Cache: 시도하지 않음 (사망, 검색 인터스티셜 반환)
```

# 접근 실패 시 — 적응형 스케줄러

> 인덱스 방법이 실패하거나 인덱스에 없는 사이트일 때 실행.
> Phase 0 → 1 → 2 → 3 순서로 에스컬레이션. 각 Phase에서 성공하면 즉시 종료.

## 원칙

1. **어떤 방법도 미리 제외하지 않는다** — 되는지는 시도해봐야 안다
2. **의존성이 없으면 설치하고 시도한다** — 미설치를 이유로 건너뛰지 않는다
3. **Phase 간 전환은 신호 기반** — 실패 유형에 따라 에스컬레이션
4. **결과 채택 기준**: 정확성/신뢰도 > 신선도 > 완전성 > 구조화 > 비용

---

## Phase 0: 특수 엔드포인트 (인덱스 매칭)

인덱스에 사이트가 있으면 해당 전용 방법을 **먼저** 시도.
정확성과 비용이 가장 좋으므로 generic Phase 1보다 우선.

성공 → 종료 / 실패 → Phase 1

---

## Phase 1: 경량 프로브 (병렬)

**먼저 시도** (동시):
- WebFetch (Claude 내장)
- Jina Reader (기본 / JSON / SPA 모드)
- curl Chrome Desktop UA

**아직 성공 없으면 추가 시도**:
- curl 모바일 UA + 모바일 URL (`m.{domain}`)
- curl Googlebot UA
- URL 변형 시도: `.json`, `/rss`, `/feed`

**사이드카** (1차와 동시, low-trust):
- Google AMP 캐시
- archive.today
- Wayback Machine
→ **원본이 하나라도 성공하면 사이드카는 참고만.** 전부 실패 시에만 사이드카 채택 (provenance 태깅 필수)

**모든 응답에서 메타데이터도 추출**: OGP, JSON-LD — [metadata.md](metadata.md) 참조

상세: [jina.md](jina.md), [cache-archive.md](cache-archive.md), [rss.md](rss.md)

---

## 에스컬레이션 신호

Phase 1 → Phase 2 전환 조건:

| 신호 | 감지 방법 | 의미 |
|------|-----------|------|
| HTTP 403/430 | 상태 코드 | WAF/봇 차단 |
| HTTP 429/503 | 상태 코드 | Rate limit (짧은 jitter retry 먼저, 실패 시 에스컬레이션) |
| WAF 헤더 | `cf-ray`, `server: cloudflare`, `x-datadome` | Cloudflare/Akamai/DataDome |
| WAF 쿠키 | `__cf_bm`, `_abck`, `datadome` | WAF 세션 |
| 챌린지 본문 | `captcha`, `verify`, `enable javascript`, `check your browser` | JS 챌린지 |
| 빈 SPA | `<div id="root"></div>` 외 콘텐츠 없음, 200자 미만 | JS 렌더링 필요 |
| Redirect loop | 3회 이상 302/307 | 챌린지 리다이렉트 |

**login/paywall 감지 시**: `login`, `sign in`, `로그인`, `subscribe`, `구독` 집중 → Phase 2/3으로 올려도 해결 안 됨. **"인증 필요"로 종료.**

---

## Phase 2: TLS 임퍼소네이션 (curl_cffi)

**조건**: Phase 1에서 WAF/봇 차단 신호 감지

**의존성 확보**:
```bash
python3 -c "import curl_cffi" 2>/dev/null || pip install curl_cffi -q
```
설치 실패 시 → 즉시 Phase 3으로.

**다중 타겟 순차 시도**: safari → chrome → firefox

```python
from curl_cffi import requests

TARGETS = ["safari", "chrome", "firefox"]
HEADERS = {
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "ko-KR,ko;q=0.9,en-US;q=0.8",
    "Referer": "https://www.google.com/",
}

for target in TARGETS:
    try:
        session = requests.Session(impersonate=target)
        session.headers.update(HEADERS)
        resp = session.get("{URL}", timeout=20)
        if resp.status_code == 200 and len(resp.text) > 300:
            # 성공 — JSON-LD도 같이 추출
            break
    except:
        continue
```

성공 → 종료 / 실패 또는 JS 챌린지 → Phase 3

상세: [tls-impersonate.md](tls-impersonate.md)

---

## Phase 3: Playwright MCP (브라우저)

**조건**: Phase 2도 실패 또는 JS 챌린지/CAPTCHA 감지

```
browser_navigate → {URL}
browser_wait_for → "body" (3초)
browser_evaluate → () => document.body.innerText  (Light Mode — 먼저)
필요 시 browser_snapshot → 접근성 트리 전체
```

**API 발견**: `browser_network_requests`로 숨은 JSON API를 찾으면 이후 curl_cffi로 재사용 가능.

상세: [playwright.md](playwright.md)

---

## 응답 검증

| 판정 | 조건 | 결과 |
|------|------|------|
| **성공** | 콘텐츠 타입에 맞는 분량 + 주제 관련 키워드 | 채택 |
| **부분 성공** | OG 메타/JSON-LD만 (본문 없음) | 보조 소스 |
| **실패 — 인증** | login/paywall 감지 | "인증 필요"로 종료 |
| **실패 — 챌린지** | CAPTCHA/JS challenge | 다음 Phase로 |
| **실패 — 에러** | 4xx/5xx | 다음 Phase로 |
| **실패 — 빈 SPA** | 콘텐츠 없음 | 다음 Phase로 |

**콘텐츠 분량 기준** (유연하게):
- 기사/블로그: 500자 이상
- 상품 페이지: JSON-LD 있으면 성공
- 트윗/짧은 글: 100자 이상
- 프로필: JSON-LD Person 있으면 성공

## False-Positive 마커 (HTTP 200이지만 실패)

| 패턴 | 감지 방법 | 처리 |
|------|----------|------|
| X SPA 셸 (247KB) | 200 OK + `Sign in to X` 또는 `hasResults: false` | 실패 — 웹 검색 도구+oEmbed 폴백 |
| CAPTCHA 페이지 | 200 OK + `captcha\|recaptcha\|hcaptcha\|cf-turnstile` | 실패 — 다음 Phase |
| 소프트 페이월 | 200 OK + `member-only\|subscribe to read\|구독하세요` | 부분 성공 — 메타만 채택 |
| DDG 소프트 리밋 | 202 Accepted + body 15KB 미만 | 실패 — 다른 엔진 폴백 |
| 빈 JSON | 200 OK + `hasResults.*false\|"entries":\s*\[\]` | 실패 — 다른 방법 시도 |
| 지역 차단 | 200 OK + `not available in your region\|geo-restricted` | 실패 — "지역 차단" 알림 |
| WAF 소프트 블록 | 200 OK + `checking your browser\|verify you are human` | 실패 — Phase 2/3 에스컬레이션 |
| Akamai behavioral | 200 OK + `behavioral-content\|sec-if-cpt` + `_abck` 쿠키 | 실패 — JS 실행 필수 → Phase 3 직행 (TLS 타겟 변경 무의미) |
| RSS Content-Type 오류 | RSS 기대 + `text/html` 응답 | 실패 — "RSS 미지원" |
| 에러 JSON | 200 OK + JSON `"error"` 키 존재 | 실패 — 에러 내용 로깅 |

## 전부 실패 시

1. 시도한 Phase와 각 실패 신호를 기록
2. 사이드카 결과가 있으면 provenance 태깅하여 채택
3. 사이드카도 없으면 사용자에게 실패 보고 + 시도 결과 공유

# 범용 웹 추출 — Jina Reader

> `r.jina.ai/URL` 한 줄로 거의 모든 공개 URL을 마크다운으로 변환.
> Puppeteer 기반 실제 브라우저 렌더링 — JS SPA까지 처리.
>
> **2026-08-09 probe 기준 무료 무키 경로는 종료됨.** 익명 호출은 401이며,
> 리다이렉트를 따라가면 Cloudflare Turnstile(`Just a moment...`)에 막힌다.
> **이제 `JINA_API_KEY` 환경 변수가 필요**하다 — `Authorization: Bearer <key>` 헤더.
> 엔진에서는 `engine/surrogates.yaml`의 `jina_reader` 엔트리가 키가 있을 때만 활성화된다
> (kind=reader, provenance=live — 서버 측 재렌더링).
> 예전 "무료 500 RPM" 안내는 모두 폐기되었으므로 따르지 않는다.

## 기본 사용

```bash
curl -s -H "Authorization: Bearer ${JINA_API_KEY}" "https://r.jina.ai/{URL}"
```

## 고급 기능

### JSON 구조화 출력

```bash
curl -H "Accept: application/json" "https://r.jina.ai/{URL}"
```

반환: `data.{title, description, url, content, metadata, external, usage}`

**핵심**: `external.alternate`에서 사이트의 **RSS URL을 자동 발견** 가능.

### CSS 선택자 타겟팅

```bash
curl -H "X-Target-Selector: .article-body" "https://r.jina.ai/{URL}"
```

네비게이션/풋터 제거, 본문만 추출. 커뮤니티 게시판에서 특히 효과적.

### SPA 스트리밍 모드

```bash
curl -H "Accept: text/event-stream" "https://r.jina.ai/{URL}"
```

JS 로딩 완료까지 대기. 동적 콘텐츠가 완전히 렌더링된 최종 버전 반환.

### 스크린샷

```bash
curl -H "X-Respond-With: screenshot" "https://r.jina.ai/{URL}"
```

GCS 서명 URL 반환 (4시간 유효). 비주얼 검증 용도.

### PDF 처리

```bash
curl -s "https://r.jina.ai/https://example.com/file.pdf"
```

PDF → 마크다운 자동 변환. 페이지 수 메타데이터 포함.

### 쿠키 전달 (인증 사이트)

```bash
curl -H "X-Set-Cookie: session=abc123" "https://r.jina.ai/{URL}"
```

### 링크 보존

```bash
curl -H "X-With-Links: true" "https://r.jina.ai/{URL}"
```

### 캐시 제어

```bash
# 캐시 우회 (실시간 필요 시)
curl -H "X-No-Cache: true" "https://r.jina.ai/{URL}"

# 캐시 TTL 지정 (초)
curl -H "X-Cache-Tolerance: 600" "https://r.jina.ai/{URL}"
```

### 순수 텍스트 / 원본 HTML

```bash
# body.innerText만
curl -H "X-Respond-With: text" "https://r.jina.ai/{URL}"

# 원본 HTML
curl -H "X-Respond-With: html" "https://r.jina.ai/{URL}"
```

## 검증된 성공 사이트

| 사이트 | 결과 | 비고 |
|--------|------|------|
| Threads | 성공 | 프로필 + 포스트 |
| 클리앙 | 성공 | 게시글 목록 + 본문 |
| 루리웹 | 성공 | 게시글 목록 + 본문 |
| 뽐뿌 | 성공 | 게시글 + RSS도 가능 |
| 네이버 뉴스 | 성공 | 기사 목록 + 본문 완전 |
| 네이버 증권 | 성공 | 실시간 주가 |
| 긱뉴스 | 성공 | 토픽 목록 + 본문 |
| 44bits | 성공 | 기사 목록 |
| 커리어리 | 성공 | JS 렌더링으로 추출 |
| 브런치 | 성공 | 기사 전문 |
| 한경 | 성공 | 뉴스 기사 |
| 다음 뉴스 | 성공 | 뉴스 기사 |
| Medium | 성공 | 기사 전문 (paywall 제외) |
| Substack | 성공 | 뉴스레터 전문 |
| dev.to | 성공 | 기사 전문 |
| PDF (모든 URL) | 성공 | 자동 변환 |

## 실패하는 사이트

| 사이트 | 이유 |
|--------|------|
| X/Twitter | 402 — Syndication/oEmbed 사용 (twitter.md 참조) |
| Reddit | 차단 — JSON API 사용 (json-api.md 참조) |
| 디시인사이드 | 빈 본문 반환 |
| 에펨코리아 | HTTP 430 |
| 요즘IT | CloudFront 403 |
| 네이버 쇼핑 | CAPTCHA |
| 쿠팡 | WAF 차단 |


## RSS 자동 발견

Jina JSON 모드의 `external.alternate`에서 사이트의 RSS URL이 자동 노출됨:

```bash
curl -H "Accept: application/json" "https://r.jina.ai/{URL}" | \
python3 -c "import sys,json; print(json.load(sys.stdin)['data'].get('external',{}))"
```

# JSON API 직접 호출

> URL 변형이나 공개 엔드포인트로 구조화된 JSON을 직접 가져오는 패턴.
> 인증 불필요. Jina Reader보다 빠르고 정확한 구조화 데이터 획득.

## Reddit

**Mobile User-Agent 필수** (없으면 403/429).

```bash
UA="Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15"

# 서브레딧 핫 포스트
curl -sL -H "User-Agent: $UA" "https://www.reddit.com/r/{subreddit}/hot.json?limit=10"

# 검색
curl -sL -H "User-Agent: $UA" "https://www.reddit.com/r/{subreddit}/search.json?q={query}&restrict_sr=1"

# 포스트 + 댓글
curl -sL -H "User-Agent: $UA" "https://www.reddit.com/r/{subreddit}/comments/{post_id}/{slug}/.json"

# 정렬: hot.json / new.json / top.json?t=week
```

데이터: `title`, `author`, `score`, `selftext`(전문), `num_comments`, `created_utc`
댓글: 응답 `[1]` 배열에 재귀적 트리

## Hacker News (Firebase API)

Rate limit 사실상 없음.

```bash
# 탑 스토리 ID 목록
curl -sL "https://hacker-news.firebaseio.com/v0/topstories.json?limitToFirst=10&orderBy=%22%24key%22"

# 개별 아이템
curl -sL "https://hacker-news.firebaseio.com/v0/item/{id}.json"

# 변형: beststories / newstories / askstories / showstories
```

데이터: `title`, `url`, `score`, `by`(작성자), `descendants`(댓글수), `kids`(댓글 ID)

배치 조회:
```bash
python3 -c "
import urllib.request, json
ids = json.load(urllib.request.urlopen('https://hacker-news.firebaseio.com/v0/topstories.json?limitToFirst=5&orderBy=\"\$key\"'))
for id in ids:
    item = json.load(urllib.request.urlopen(f'https://hacker-news.firebaseio.com/v0/item/{id}.json'))
    print(f'[{item.get(\"score\",0)}] {item.get(\"title\")}')
    print(f'  {item.get(\"url\",\"N/A\")[:60]}')
"
```

## Lobste.rs

Rate limit 없음. HN보다 작지만 고품질 큐레이션.

```bash
# 핫 스토리
curl -sL "https://lobste.rs/hottest.json"

# 태그별 (ai, programming, web, security 등)
curl -sL "https://lobste.rs/t/ai.json"

# 최신
curl -sL "https://lobste.rs/newest.json"

# 개별 스토리 + 댓글
curl -sL "https://lobste.rs/s/{short_id}.json"
```

데이터: `title`, `url`, `score`, `comment_count`, `tags`, `submitter_user`

## dev.to

```bash
# 태그별 최신
curl -sL "https://dev.to/api/articles?tag=ai&per_page=5"

# 이번 주 탑
curl -sL "https://dev.to/api/articles?top=7&per_page=5"

# 특정 유저
curl -sL "https://dev.to/api/articles?username={user}&per_page=5"
```

데이터: `title`, `user.name`, `public_reactions_count`, `reading_time_minutes`, `tags`

## npm Registry

```bash
# 패키지 최신 버전
curl -sL "https://registry.npmjs.org/{package}/latest"

# 패키지 검색
curl -sL "https://registry.npmjs.org/-/v1/search?text={query}&size=5"

# 다운로드 통계
curl -sL "https://api.npmjs.org/downloads/range/last-month/{package}"
```

## PyPI

```bash
# 패키지 정보
curl -sL "https://pypi.org/pypi/{package}/json"

# 다운로드 통계
curl -sL "https://pypistats.org/api/packages/{package}/recent"
```

## Wikipedia

```bash
# 페이지 요약
curl -sL "https://en.wikipedia.org/api/rest_v1/page/summary/{title}"
# 한국어: https://ko.wikipedia.org/api/rest_v1/page/summary/{title}

# 검색
curl -sL "https://en.wikipedia.org/w/api.php?action=opensearch&search={query}&limit=5&format=json"
```

## V2EX

```bash
curl -sL "https://www.v2ex.com/api/topics/hot.json" -H "User-Agent: insane-search/1.0"
```

## RSS 피드

→ [rss.md](rss.md)로 이동. 한국 언론 RSS, Google News RSS, feedparser 사용법 등 상세 가이드 참조.

# 미디어 추출 — yt-dlp

> yt-dlp는 YouTube 전용 도구가 아니라 **1,858개 사이트**를 지원하는 범용 미디어 추출 도구.
> 영상, 오디오, 팟캐스트, 라이브 스트리밍 — 미디어 URL이면 yt-dlp를 먼저 시도한다.

## 설치 확인

```bash
which yt-dlp || python3 -m yt_dlp --version
```

- `yt-dlp` 명령어가 PATH에 있으면 그대로 사용
- 없으면 `python3 -m yt_dlp`로 대체 (아래 모든 명령어에서 치환)
- 미설치 시: `pip install yt-dlp`

## 핵심 명령어 (모든 지원 사이트 공통)

### 메타데이터 추출 (가장 범용)

```bash
yt-dlp --dump-json "URL"
```

title, uploader, duration, view_count, description, tags 등 구조화 JSON 반환.
전용 extractor가 있는 사이트에서 ~95% 성공.

### 자막 추출

```bash
yt-dlp --write-sub --write-auto-sub --sub-lang "en,ko" --skip-download -o "/tmp/%(id)s" "URL"
cat /tmp/VIDEO_ID.*.vtt
```

YouTube는 100개 언어 자동자막 지원. 다른 사이트는 자체 자막 제공 시에만 동작.

### 검색

```bash
# YouTube
yt-dlp --dump-json "ytsearch5:{검색어}"

# SoundCloud
yt-dlp --dump-json "scsearch5:{검색어}"

# Dailymotion
yt-dlp --dump-json "dailymotionsearch5:{검색어}"

# Yahoo
yt-dlp --dump-json "yahoosearch5:{검색어}"
```

### 채널/플레이리스트 목록 (다운로드 없이)

```bash
yt-dlp --flat-playlist --dump-json "채널_URL"
```

title, id, url, duration 반환. 채널 전체 영상 목록을 초고속 수집.

### 댓글 추출 (YouTube)

```bash
yt-dlp --write-comments --skip-download --write-info-json \
  --extractor-args "youtube:max_comments=20" \
  -o "/tmp/%(id)s" "URL"
```

## 지원 플랫폼 카테고리

### 영상

| 사이트 | 메타데이터 | 자막 | 검색 | 비고 |
|--------|----------|------|------|------|
| YouTube | O | O (자동생성 포함) | `ytsearch` | 최고 지원 |
| Vimeo | O | O (사이트 제공 시) | X | 학술/다큐 콘텐츠 풍부 |
| Twitch | O (VOD/클립) | X | X | 기술 스트리밍 |
| TikTok | O | X | X | 공개 계정만 |
| Dailymotion | O | O | `dailymotionsearch` | |
| Rumble | O | X | X | |
| PeerTube | O | X | X | 탈중앙화 |

### 오디오/팟캐스트

| 사이트 | 메타데이터 | 검색 | 비고 |
|--------|----------|------|------|
| SoundCloud | O | `scsearch` | 검색까지 가능 — 최고 |
| Apple Podcasts | O | X | RSS 기반 |
| TuneIn | O | X | |
| acast | O | X | 채널 단위 지원 |
| Spreaker | O | X | |
| Audius | O | X | 블록체인 기반 |

### 한국 플랫폼

| 사이트 | Extractor | 비고 |
|--------|-----------|------|
| Naver TV | `Naver`, `Naver:live` | |
| Kakao | `Kakao` | |
| SBS | `SBS`, `sbs.co.kr` | |
| JTBC | `JTBC`, `JTBC:program` | |
| Chzzk | `chzzk:video`, `chzzk:live` | 네이버 스트리밍 |
| Soop (구 AfreecaTV) | `soop`, `soop:live` | |
| Daum | `daum.net`, `daum.net:clip` | |
| Weverse | `Weverse`, `WeverseLive` | K-팝 팬덤 |

### 뉴스 VOD

| 사이트 | 비고 |
|--------|------|
| BBC | 공개 VOD |
| ABC (호주) | iview |
| CBS News | |
| NBC News | 차단 많음 |

> 뉴스 사이트는 직접 URL보다 **YouTube 공식 채널 경유**가 더 안정적.
> 예: `ytsearch:BBC News {키워드}`

## 주의사항

- 자동 생성 자막은 행간 중복 → 후처리 필요
- generic extractor는 성공률 ~30% — 전용 extractor 있는 사이트 우선
- 페이월/로그인 사이트는 대부분 실패
- `--dump-json`이 가장 안전한 범용 명령 (다운로드 없음, 메타데이터만)

# 메타데이터 추출 — OGP / JSON-LD / Schema.org

> HTML을 받았을 때 구조화된 데이터를 추출하는 보조 기법.
> 본문 전체를 못 가져와도 제목, 요약, 가격, 프로필 등 핵심 정보를 확보할 수 있다.

## 의존성

없음 (curl + python3 기본 모듈).

## OGP (Open Graph Protocol) 메타태그

대부분의 사이트가 소셜 공유용으로 삽입. 제목 + 설명 + 이미지 확보 가능.

```bash
curl -sL -H "User-Agent: Mozilla/5.0 ..." "{URL}" | \
  python3 -c "
import sys, re
html = sys.stdin.read()
for m in re.findall(r'<meta property=\"og:(\w+)\" content=\"([^\"]*?)\"', html):
    print(f'og:{m[0]} = {m[1]}')
for m in re.findall(r'<meta name=\"description\" content=\"([^\"]*?)\"', html):
    print(f'description = {m}')
"
```

## JSON-LD (Schema.org 구조화 데이터)

**가장 가치 높은 추출 대상.** 상품, 기사, 프로필 등 구조화된 정보가 JSON으로 들어있다.

```bash
curl -sL "{URL}" | \
  python3 -c "
import sys, re, json
html = sys.stdin.read()
blocks = re.findall(r'<script type=\"application/ld\+json\">(.*?)</script>', html, re.DOTALL)
for b in blocks:
    try:
        data = json.loads(b)
        print(json.dumps(data, ensure_ascii=False, indent=2))
    except:
        pass
"
```

### 실제 사례

**쿠팡 검색 결과** — `CollectionPage` + `ItemList`:
```json
{
  "@type": "CollectionPage",
  "mainEntity": {
    "@type": "ItemList",
    "itemListElement": [
      {
        "@type": "ListItem",
        "item": {
          "@type": "Product",
          "name": "...",
          "offers": { "price": 29900 }
        }
      }
    ]
  }
}
```

**LinkedIn 프로필** — `Person`:
```json
{
  "@type": "Person",
  "name": "...",
  "jobTitle": "...",
  "alumniOf": [
    { "@type": "Organization", "name": "..." }
  ]
}
```

**뉴스 기사** — `NewsArticle`:
```json
{
  "@type": "NewsArticle",
  "headline": "...",
  "datePublished": "2026-04-16",
  "author": { "name": "..." },
  "articleBody": "..."
}
```

## Next.js RSC 페이로드 (요즘IT 등)

Next.js App Router 사이트는 `self.__next_f.push()` 스크립트에 콘텐츠가 포함됨.

```bash
curl -sL "{URL}" | \
  python3 -c "
import sys, re
html = sys.stdin.read()
chunks = re.findall(r'self\.__next_f\.push\(\[1,\"(.*?)\"\]\)', html)
text = ''.join(chunks)
# 한국어 텍스트 추출 (유니코드 이스케이프 디코딩)
decoded = text.encode().decode('unicode_escape', errors='ignore')
print(decoded[:3000])
"
```

## 활용 시점

메타데이터 추출은 **독립 방법이 아니라 보조 기법**이다.
어떤 Phase에서든 HTML을 받으면 같이 실행:

- Phase 1에서 curl로 HTML 받음 → JSON-LD도 추출
- Phase 2에서 curl_cffi로 HTML 받음 → JSON-LD도 추출
- Phase 3에서 Playwright로 DOM 받음 → `browser_evaluate`로 JSON-LD 추출

본문은 못 가져와도 JSON-LD에서 **상품 가격, 기사 요약, 프로필 정보**는 확보될 수 있다.

# 네이버 계열 접근 전략

> 네이버 서비스별로 접근 방법이 다르다. 블로그는 모바일 URL, 뉴스/증권은 Jina Reader.

## 네이버 블로그

WebFetch 차단. 모바일 URL 변환 + iPhone UA로 접근.

```bash
# blog.naver.com/{ID}/{NO} → m.blog.naver.com 변환
curl -sL \
  -H "User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1" \
  -H "Accept-Language: ko-KR,ko;q=0.9" \
  -H "Referer: https://m.naver.com/" \
  "https://m.blog.naver.com/PostView.naver?blogId={ID}&logNo={NO}"
```

RSS도 가능 (최신 50개, 본문 약 300자):
```bash
curl -sL "https://rss.blog.naver.com/{BLOG_ID}.xml"
```

## 네이버 뉴스

Jina Reader로 완전 접근 가능.

```bash
# 기사 목록
curl -s "https://r.jina.ai/https://news.naver.com/"

# 개별 기사
curl -s "https://r.jina.ai/https://n.news.naver.com/article/{press_id}/{article_id}"
```

## 네이버 증권

Jina Reader로 실시간 주가, 주요 뉴스 접근.

```bash
curl -s "https://r.jina.ai/https://finance.naver.com/item/main.naver?code={종목코드}"
```

## 네이버 금융 시세 (비공식, 무인증)

인증 불필요. 주가 시계열 데이터 JSON 반환.

```bash
# 일봉 시세 (삼성전자=005930)
curl -sL "https://api.finance.naver.com/siseJson.naver?symbol=005930&requestType=1&startTime=20240101&endTime=20241231&timeframe=day"

# 분봉
curl -sL "https://api.finance.naver.com/siseJson.naver?symbol=005930&requestType=0&timeframe=minute&count=200"
```

응답: `[[날짜, 시가, 고가, 저가, 종가, 거래량, 외국인거래율], ...]`

## 네이버 검색 (신원위장으로 직접 접근)

curl_cffi + 세션 쿠키 워밍으로 네이버 검색 결과를 직접 크롤링할 수 있다. API 키 불필요.

```python
from curl_cffi import requests
from urllib.parse import quote

s = requests.Session(impersonate="chrome124")
s.headers.update({
    "Accept-Language": "ko-KR,ko;q=0.9",
    "Referer": "https://www.google.com/",
})
s.get("https://www.naver.com/", timeout=10)  # 쿠키 워밍
s.headers["Referer"] = "https://www.naver.com/"

# 통합 검색 (블로그+뉴스+웹 혼합)
r = s.get(f"https://search.naver.com/search.naver?query={quote('검색어')}")

# 블로그 탭
r = s.get(f"https://search.naver.com/search.naver?where=post&query={quote('검색어')}")

# 뉴스 탭
r = s.get(f"https://search.naver.com/search.naver?where=news&query={quote('검색어')}")
```

### 추출 가능한 데이터

| 탭 | URL 패턴 | 추출 |
|---|---|---|
| 통합 | `search.naver?query=` | 블로그 URL, 외부 링크, 뉴스 |
| 블로그 | `where=post&query=` | blog.naver.com URL, 제목, 스니펫 |
| 뉴스 | `where=news&query=` | n.news.naver.com URL, 제목 |

### 한국어 키워드 검색의 핵심 경로

웹 검색 도구는 한국어 신규 콘텐츠 인덱싱이 지연되지만, 네이버 검색은 한국어에 최적화되어 있다.
**한국 사이트 키워드 검색 → 네이버 검색 직접 접근이 가장 정확하고 빠르다.**

## 네이버 카페

로그인 + iframe 이중 장벽. 본문 직접 접근 불가.
fallback 체인에서 Phase 1~3을 시도하되, login/paywall 감지 시 "인증 필요"로 종료.

## 네이버 TV

yt-dlp로 접근 (media.md 참조).

```bash
yt-dlp --dump-json "https://tv.naver.com/v/{video_id}"
```

# Playwright — MCP vs Local Chrome

> JS 렌더링 / JS 챌린지 사이트를 위한 두 가지 접근. **WAF 프로파일의
> `capabilities_needed` 태그가 선택을 결정**한다. 사용자가 직접 고를 필요 없다.

## 두 Approach 요약

| Approach | 실행기 | TLS 스택 | 적합 WAF | 한계 |
|----------|--------|----------|----------|------|
| **1. MCP** | `mcp__playwright__*` 도구 | Playwright 번들 Chromium (BoringSSL) | Cloudflare 기본, CAPTCHA 없는 SPA, JS 챌린지 약한 사이트 | Akamai Bot Manager 등 TLS-감지형 WAF에 **즉시 탐지됨** (`channel` 옵션 없음) |
| **2. Local Node + `channel:'chrome'`** | `engine/templates/playwright_real_chrome.js` | 시스템 설치 실제 Chrome | Akamai Bot Manager, PerimeterX, DataDome 강화 설정 | Node + Chrome 시스템 설치 필요 |

`engine/executor.py`가 프로파일 태그를 보고 자동 라우팅하므로, 이 선택을 스킬 외부에서 의식할 필요는 없다.

## Approach 1 — Playwright MCP

### 의존성

이 항목은 이미 연결된 MCP를 사용하는 엔진 어댑터의 설명이다.
에이전트가 직접 여는 브라우저 세션은 js eval의 omowright(`browser` 스킬에 스테이징)로
띄운다: 직접 소유 브라우저는 `connectPipe`, 스텔스는 `connectCloakProfile`, 사용자 로그인이
필요하면 `connectBrowserSkill`. MCP를 새로 설치하지 않는다.

### 기본 워크플로

```
1. browser_navigate → URL
2. browser_wait_for → 메인 콘텐츠 셀렉터 (SPA는 필수)
3. browser_snapshot    (접근성 트리 — 토큰 효율)
   또는
   browser_evaluate    (특정 셀렉터 데이터 추출)
   또는
   browser_run_code    (스크롤/페이지네이션)
```

### 도구별 용도

| 도구 | 용도 |
|------|------|
| `browser_snapshot` | 접근성 트리 반환 — 텍스트+인터랙티브 요소 구조화. 가장 빠르고 토큰 효율적 |
| `browser_evaluate` | `() => document.querySelector(...).innerText` 등 JS 평가 |
| `browser_run_code` | `async ({ page }) => {...}` 풀 자동화 — 무한 스크롤, 다단계 인터랙션 |
| `browser_network_requests` | XHR/fetch 호출 목록 — **WAF 뒤 진짜 API 엔드포인트 발견용** (→ curl_cffi로 직접 호출) |
| `browser_console_messages` | JS 에러/로그 |

### 주의

- MCP는 Chromium 번들 기반. TLS 지문이 BoringSSL이라 Akamai/DataDome은 **293 바이트 Access Denied** 또는 즉시 403 반환.
- 이 경우 자동으로 Approach 2로 이관되도록 `engine/executor.py`가 처리. 수동 선택 불필요.

## Approach 2 — Local Node + Real Chrome

### 의존성 (최초 1회)

엔진 템플릿의 스크립트 의존성은 `engine/templates/package.json`에 고정되어 있다.
엔진 디렉터리에서 한 번만 설치한다. Chrome은 이미 시스템에 설치되어 있어야 하며,
브라우저 다운로드 명령은 없다.

```bash
cd "$SKILL_DIR/engine"
test -f package.json || cp templates/package.json package.json
bun install
```

### 호출 (engine 내부)

```python
from insane_search.engine.executor import run_playwright_fallback

attempt, html = run_playwright_fallback(
    "https://example.com/path",
    profile_id="akamai_bot_manager",
    success_selectors=["article"],
    device_class="desktop",   # "desktop" | "mobile" | "auto"
)
```

내부에서 `engine/templates/playwright_real_chrome.js` 또는 `playwright_mobile_chrome.js`를 Node로 실행하고 HTML을 받아온다. 템플릿은 **URL과 셀렉터 파라미터만** 받으며 사이트별 분기가 없다. 템플릿은 엔진이 실행하는 프로그램이다. 에이전트가 이 템플릿을 본떠 브라우저 스크립트를 새로 쓰지 않는다.

### 데스크톱 템플릿 (`playwright_real_chrome.js`)

- 번들 Chromium이 아니라 **시스템에 설치된 실제 Chrome**을 띄운다. TLS 지문이 실제 Chrome이 되는 것이 핵심이다.
- stealth 플러그인을 적용하고, 작업 전용 영속 프로필 디렉터리를 쓴다.
- Akamai는 headless를 탐지하므로 **headful**로 실행하고, 뷰포트는 1366×900이다.

### 모바일 템플릿 (`playwright_mobile_chrome.js`)

- TLS는 데스크톱과 같은 실제 Chrome이고, iPhone 13 Pro 디바이스 기술자(UA/viewport/isMobile/hasTouch)만 주입한다. headful로 실행한다.

**주의**: 실제 Chrome + 모바일 디바이스 기술자 조합은 TLS 핑거프린트를 Chrome으로 유지하면서 HTTP 레이어(UA/viewport)만 모바일로 바꾼다. WAF가 실제 Chrome으로 인식해서 관대한 경우가 많다.

### 엔진 밖에서 같은 효과가 필요할 때

엔진 폴백이 아니라 에이전트가 직접 페이지를 조작해야 한다면 js eval에서 omowright(`browser` 스킬에 스테이징)를 쓴다. 지문이 고정된 스텔스 브라우저는 `connectCloakProfile({ profileDir })`, 모바일 뷰는 `emulate(page, "iphone-14")`, 사용자 로그인이 필요하면 `connectBrowserSkill()`이다.

## 선택 규칙 (자동)

`engine/waf_profiles.yaml`의 `capabilities_needed` 태그가 결정한다:

| 태그 조합 | 선택 실행기 | 대표 케이스 |
|----------|-------------|-------------|
| `needs_real_tls_stack` + `needs_js_exec` | Approach 2 (real_chrome) | Akamai Bot Manager |
| `needs_js_exec` only | Approach 1 (MCP) | Cloudflare Turnstile |
| `needs_real_tls_stack` only | Approach 2 (real_chrome) | 일부 DataDome 설정 |
| 둘 다 없음 | curl 체인에서 해결. Playwright 안 씀 | F5 BIG-IP (TLS만 우회 필요) |

`device_class="mobile"`이 지정되면 real_chrome → mobile 변종으로 swap.

## 공통 검증

두 Approach 모두 최종 HTML을 `engine/validators.py:validate()`로 재검증한다. 즉 Playwright가 HTML을 받아와도 **챌린지 페이지 또는 빈 SPA면 여전히 CHALLENGE 판정**. 자동으로 다음 조합이나 failure 보고로 이어진다.

## 디버깅 팁

- 템플릿의 `profileDir`는 작업 전용 경로만 사용한다. 사용자의 실제 프로필을 실행·복제·초기화·삭제하지 않는다.
- 작업 종료 시 브라우저를 닫고 작업 전용 프로필만 정리한다.
- 실패 시 `result.trace`의 `error` 필드에 Node stderr 200자가 포함됨

## 사이트 예시 (독자 이해용, 코드 분기 근거 아님)

> 이 섹션은 **설명 목적**이며 `engine/**` 코드에는 반영되지 않는다.

- **Cloudflare 기본 챌린지**: Approach 1 (MCP) 충분
- **Akamai Bot Manager**: Approach 2 필수. MCP로는 TLS-UA 불일치 탐지됨
- **SSR 블로그 플랫폼**: curl_cffi safari만으로 HTML 수신. Playwright 불필요
- **검색 결과 JS 렌더링 SPA**: Approach 1로 `browser_wait_for` 후 `browser_snapshot`

실제 라우팅은 프로파일 태그가 결정한다. 위 예시는 참고일 뿐 코드 분기 근거로 쓰지 않는다.

# 공개 API 직접 호출

> 인증 없이 구조화된 데이터를 반환하는 공개 API.
> 웹 크롤링이 아니라 공식 API — 안정적이고 정확.

## Bluesky (AT Protocol)

프로필과 피드는 완전 공개. 검색은 403 차단.

```bash
# 프로필
curl -sL "https://public.api.bsky.app/xrpc/app.bsky.actor.getProfile?actor={handle}" | \
python3 -c "import sys,json; d=json.load(sys.stdin); print(f'{d[\"displayName\"]} — Followers: {d[\"followersCount\"]}, Posts: {d[\"postsCount\"]}')"

# 피드 (최근 게시물)
curl -sL "https://public.api.bsky.app/xrpc/app.bsky.feed.getAuthorFeed?actor={handle}&limit=10"
```

Rate limit: ~3,000 req/hour

## Mastodon

인스턴스별 상이. mastodon.social은 공개 타임라인 차단, hachyderm.io/fosstodon.org 등은 허용.

```bash
# 계정 조회
curl -sL "https://{instance}/api/v1/accounts/lookup?acct={username}"

# 계정 타임라인 (ID 획득 후)
curl -sL "https://{instance}/api/v1/accounts/{id}/statuses?limit=10"

# 해시태그 타임라인 (인스턴스에 따라 인증 필요)
curl -sL "https://hachyderm.io/api/v1/timelines/tag/{tag}?limit=10"
```

## Stack Exchange (v2.3)

```bash
# 질문 검색
curl -sL "https://api.stackexchange.com/2.3/search?order=desc&sort=votes&intitle={query}&site=stackoverflow"

# 태그 기반
curl -sL "https://api.stackexchange.com/2.3/questions?tagged={tag1};{tag2}&site=stackoverflow&pagesize=5"

# 답변 포함 (본문 필요 시)
curl -sL "https://api.stackexchange.com/2.3/questions/{id}/answers?order=desc&sort=votes&site=stackoverflow&filter=withbody"
```

Rate limit: 비인증 300 req/day (IP당)

## arXiv (학술 논문)

```bash
# 논문 검색 (ti=제목, au=저자, abs=초록, cat=카테고리)
curl -sL "http://export.arxiv.org/api/query?search_query=ti:{query}&max_results=5&sortBy=submittedDate&sortOrder=descending"
```

**주의**: 3 req/second 제한. 요청 간 1초 sleep 필수.
카테고리: cs.AI, cs.CL, cs.LG, cs.CV 등

## CrossRef (DOI / 피어리뷰 논문)

```bash
# 논문 검색
curl -sL "https://api.crossref.org/works?query={query}&filter=from-pub-date:2025-01&rows=5&sort=relevance"

# DOI로 조회
curl -sL "https://api.crossref.org/works/{DOI}"
```

Rate limit: 50 req/second. User-Agent에 이메일 추가 시 Polite Pool 진입.

## OpenLibrary (도서)

```bash
# ISBN 조회
curl -sL "https://openlibrary.org/api/books?bibkeys=ISBN:{isbn}&jscmd=data&format=json"

# 도서 검색
curl -sL "https://openlibrary.org/search.json?q={query}&limit=5"
```

## Wayback Machine (아카이브)

```bash
# 스냅샷 확인
curl -sL "https://archive.org/wayback/available?url={URL}"

# CDX API (스냅샷 목록)
curl -sL "https://web.archive.org/cdx/search/cdx?url={URL}&output=json&fl=timestamp,statuscode&limit=5"
```

## GitHub REST API (gh CLI 없을 때)

비인증 60 req/hour. gh CLI 우선 사용 권장.

```bash
# 저장소 검색
curl -sL "https://api.github.com/search/repositories?q={query}&sort=stars&per_page=5"

# 릴리즈
curl -sL "https://api.github.com/repos/{owner}/{repo}/releases?per_page=5"

# 코드 검색
curl -sL "https://api.github.com/search/code?q={query}+language:python&per_page=5"
```

## Rate Limit 요약

| API | 비인증 | 비고 |
|-----|-------|------|
| Bluesky | ~3K/hr | 검색 403 |
| Mastodon | 인스턴스별 | |
| Stack Exchange | 300/day | 인증 시 10K |
| arXiv | 3/sec | sleep 필수 |
| CrossRef | 50/sec | |
| OpenLibrary | 무제한 | |
| Wayback | 무제한 | |
| GitHub REST | 60/hr | gh CLI 우선 |

# Insane Search

> Part of the **ultimate-browsing** skill (Tier 1). Routing and tier-escalation live in [../../SKILL.md](../../SKILL.md).
> The engine package it drives is at [../../engine/](../../engine/), invoked from the skill directory as `python3 -m engine "<URL>"`.
> Deep-dives in this folder: TLS impersonation, Playwright routing, fallback, metadata, Jina, cache/archive, RSS, JSON/public APIs, Twitter, Naver, media.


> URL 접근이 차단될 때, **사이트 무관한** 우회 전략을 자동 선택한다.

## 하네스 규칙 (Claude에게 강제되는 지침)

이 규칙은 Claude가 즉흥 판단으로 엇나가지 못하게 하기 위한 **고삐**다. 위반 시 이전 test.md 세션처럼 "chrome 200에서 break → safari 미시도 → Playwright 미설치라 포기" 식의 오판이 재현된다.

**R1 — 일반 웹 URL 차단/403/402 감지 시**:
1. WebFetch, 즉흥 curl, 수동 헤더 조합 **시도 금지**
2. 즉시 다음을 실행:
   ```bash
   python3 -m engine "<URL>" [--selector "<CSS>"] [--device auto|desktop|mobile] [--trace] \
                              [--no-retry] [--no-extract]
   ```
3. 종료코드 0(ok) 또는 1(fail) 받은 뒤 판단. trace를 먼저 읽고 재시도 결정.
4. 실패 시에만 `--trace --json`으로 재호출해서 원인 진단 후 `--device` 또는 `user_hint` 조정.

**R2 — 첫 200에서 탈출 금지**: HTTP 200은 **검사 시작 조건**이지 성공이 아니다. `validate()`의 4-계층 검증을 통과해야 성공 선언. CLI는 이미 강제한다.

**R3 — 편향 금지**: `engine/**`, `waf_profiles.yaml`에 특정 사이트 도메인·셀렉터·브랜드명 하드코딩 금지. `python3 engine/bias_check.py`가 CI 게이트. 자세한 규칙은 **No-Site-Name Rule** 섹션.

**R4 — 힌트는 런타임에만**: 사이트 고유 정보(성공 셀렉터, 우선 Referer)는 CLI 인자 또는 `user_hint`로만 전달, 저장소에 고정 금지.

**R5 — Phase 0 공식 API 우선**: X/Reddit/YouTube/HN/arXiv 등 **공식 공개 엔드포인트**가 있는 플랫폼은 Phase 0 테이블을 먼저 확인하고 해당 API를 쓴다. 이건 편향이 아니라 합의된 접근 경로.

**R6 — 실패 선언은 전수 시도 후에만**: 격자(URL 변환 × TLS impersonate × Referer × Playwright fallback)를 **모두** 돌린 뒤에만 "뚫을 수 없음" 결론. CLI의 `max_attempts` 기본 12가 이를 보장.
단, R7 조건(WAF 조기 감지)이 성립하면 engine 격자는 계속 돌되, Claude가 **병렬로** MCP 정찰 루트를 시도할 수 있다. 빠른 쪽이 이긴다.

**R7 — WAF 조기 감지 시 API-first 병행 분기** (분기 결정은 자동이지만 사용자가 결과에서 확인 가능 — 어떤 우회 경로로 성공/실패했는지 결과 metadata에 명시):
발동 조건 (AND):
1. engine 실행 초기에 첫 2~3회 attempt가 모두 `verdict=challenge`
2. `profile_used`가 `akamai_bot_manager`, `cloudflare_turnstile`, `datadome_probable`, `perimeterx_human`, `f5_big_ip`, `aws_waf` 중 하나로 확정
3. **사용자 요청이 리스트/수집/반복 의도** (여러 페이지, N개 이상, "전부", "크롤링", 페이지네이션 등). 단건 본문 조회는 해당 없음.

세 조건 모두 참일 때 Claude는 **병렬 경로**를 시작한다:

**"병렬"의 실행 의미** (Claude 도구 호출이 순차이므로 명확화):
- engine은 `run_in_background=true`로 Bash 툴에서 띄워둔다 — 격자는 그대로 돌되 블로킹하지 않음
- Claude는 그 사이 foreground에서 MCP Playwright 정찰 루트를 진행
- engine이 먼저 성공해도 좋고, MCP 정찰로 얻은 API가 먼저 성공해도 좋음. 빠른 쪽 결과 채택

**MCP 정찰 루트**:
1. `mcp__playwright__browser_navigate` → 대상 페이지 로드 (브라우저 렌더링)
2. `mcp__playwright__browser_network_requests` → XHR/fetch 호출 목록 수집, `/api/`·`/graphql`·`\.json` 필터로 내부 엔드포인트 식별
3. 식별된 JSON API URL을 `python3 -m engine <API_URL>`로 재호출 (백그라운드 engine과는 별개 호출). 대부분 API 레이어는 페이지 HTML보다 WAF 보호가 얕아 curl_cffi로 바로 수집됨
4. 응답 스키마 파악 후 pagination / query parameter 조합해 반복 수집

**왜**: SPA + WAF 사이트(쇼핑몰·커머스 다수)는 마케팅 페이지(HTML)만 WAF로 중투자하고 내부 API는 gateway 레벨 기본 방어만 쓰는 경우가 많다. HTML 격자 전수 낭비(50회 × 0.5s + Playwright fallback 40s ≈ 65초)보다 **MCP 정찰 1회(5~10초) + API 재호출(0.5초)**가 훨씬 경제적이고 성공률 높음.

**R7을 쓰지 말아야 할 때**: 단일 페이지 본문 읽기만 필요한 단건 조회(문서 하나, 블로그 포스트 하나)는 engine만으로 충분하다 — 발동 조건 #3이 이를 배제한다.

**R7 편향 방지**: 내부 API URL·파라미터는 `engine/**`에 하드코딩 금지. 탐지된 URL은 런타임 호출에만 쓰고 저장소에 고정하지 않는다.

---

이 스킬의 핵심 불변식:

- **단일 진입점**: 일반 웹 페이지는 항상 `python3 -m engine <URL>` 또는 `from engine import fetch; fetch(...)`.
- **편향 금지**: `engine/**`, `waf_profiles.yaml`에 특정 사이트 하드코딩 금지.
- **힌트는 런타임에만**: 사이트 고유 정보는 CLI/`user_hint` 경유.

## 의도 분류 (Phase 0 진입 전)

| 사용자 입력 | 경로 |
|------------|------|
| URL 제공 (`https://...`) | → Phase 0 검사 후 없으면 Phase 1 (generic fetch chain) |
| 핸들 제공 (`@username`) | → Phase 0 syndication/API |
| 키워드만 ("X에서 AI 검색") | → 웹 검색 도구(`site:{domain} {keyword}`) 먼저 → URL 확보 후 재진입 |

> **한국어 신규 콘텐츠 한계**: 네이버/다음/한국 커뮤니티의 키워드 검색은 웹 검색 도구 경유가 유일하며, 신규 콘텐츠 인덱싱이 지연될 수 있다.

## Phase 0 — 플랫폼 공식 API 인덱스

> 플랫폼이 **공식 공개한** 전용 API/CLI만 여기에 둔다. 이건 편향이 아니라 합의된 엔드포인트 사용이다.

### 소셜/커뮤니티 전용 API

| 플랫폼 | 방법 | 상세 |
|--------|------|------|
| X/Twitter | syndication (타임라인) + oEmbed (개별 트윗) + 키워드 검색: 웹 검색 도구 → oEmbed | [twitter.md](twitter.md) |
| Reddit | URL + `.json` + Mobile UA | [json-api.md](json-api.md) |
| Bluesky | AT Protocol (`public.api.bsky.app/xrpc/...`) | [public-api.md](public-api.md) |
| Mastodon | 인스턴스별 공개 API | [public-api.md](public-api.md) |
| Hacker News | Firebase API + Algolia Search | [json-api.md](json-api.md) |
| Stack Overflow | SE API v2.3 | [public-api.md](public-api.md) |
| Lobste.rs / V2EX / dev.to | 공개 JSON API | [json-api.md](json-api.md) |

### 미디어 (CLI 도구 필수)

| 플랫폼 | 방법 | 상세 |
|--------|------|------|
| YouTube/Vimeo/Twitch/TikTok/SoundCloud 등 1,858개 | `yt-dlp --dump-json` | [media.md](media.md) |

### 학술/레지스트리

| 플랫폼 | 방법 | 상세 |
|--------|------|------|
| arXiv | Atom API | [public-api.md](public-api.md) |
| CrossRef | REST API | [public-api.md](public-api.md) |
| Wikipedia | REST API | [json-api.md](json-api.md) |
| OpenLibrary | JSON API | [public-api.md](public-api.md) |
| GitHub | gh CLI / REST API | [public-api.md](public-api.md) |
| npm / PyPI | Registry API | [json-api.md](json-api.md) |
| Wayback Machine | CDX API | [public-api.md](public-api.md) |

### 한국 전용 공식 API

| 플랫폼 | 방법 | 상세 |
|--------|------|------|
| 네이버 검색 | `search.naver.com` (통합/블로그/뉴스탭) | [naver.md](naver.md) |
| 네이버 금융 시세 | `api.finance.naver.com/siseJson.naver` (비공식 JSON) | [naver.md](naver.md) |

**그 외 모든 사이트는 Phase 1(generic fetch chain)이 자동 처리한다.**

## Phase 1 — Generic Fetch Chain

### 단일 진입점

```python
from insane_search.engine import fetch

result = fetch(
    "https://example.com/path",
    success_selectors=["article", "[class*='product-card']"],  # 포지티브 프루프 (선택)
    device_class="auto",      # "auto" | "desktop" | "mobile"
    user_hint=None,           # {"referer_strategy": "self_root", "impersonate_first": "safari"}
    timeout=25,
)

if result.ok:
    print(result.verdict)     # strong_ok | weak_ok
    html = result.content     # raw body — 단, content-rescue가 발동한 경우 구조 텍스트
    # v0.10.0 content-rescue: PDF 응답은 pypdf 추출 텍스트, visible text가 얇은
    # SPA 셸은 JSON-LD articleBody / 렌더된 innerText로 대체될 수 있다.
    # result.extraction_source로 판별: "raw"(원문 그대로) | pdf | json_ld | *+inner_text.
    # 일반 HTML 성공은 항상 raw. 끄기: enable_extraction=False / --no-extract.
    # 429/502/503/504는 probe에서 백오프 재시도(Retry-After 반영, 총 10초 캡);
    # 끄기: enable_retry=False / --no-retry.
else:
    # Phase 3 수동 개입 (Playwright MCP) 필요 — result.trace로 원인 진단
    pass
```

### 내부 단계 (디버깅용 노출)

`fetch()`는 단일 API이지만 내부는 phase로 나뉘어 있다. `result.trace`에서 각 시도를 확인할 수 있다.

```
probe      — curl_cffi + safari + self-referer로 첫 시도
validate   — 4-계층 검증 (marker / size / cookie / success_selectors)
detect     — WAF 제품 감지 ([(profile_id, confidence)] 랭킹)
plan       — 프로파일의 tls_candidates × url_transforms × referer 격자 구성
execute    — 격자 전수 시도 (첫 200에서 탈출하지 않음)
fallback   — capability 태그 기반 Playwright 라우팅 (MCP or local+chrome)
report     — FetchResult(ok, verdict, profile_used, trace, summary)
```

### 검증 원칙

- HTTP 200은 **검사 시작 조건**이지 성공이 아니다.
- 성공 판정은 **4-계층 AND**:
  1. 챌린지 마커 없음 (`sec-if-cpt-container`, `Access Denied`, `Just a moment...`, `DataDome`)
  2. 비정상 크기 아님 (< 3KB 또는 WAF fingerprint 크기)
  3. 쿠키 센서 상태 정상 (`_abck=~-1~` 아님)
  4. `success_selectors` 중 하나 이상 매칭 (caller 제공 시 → `strong_ok`, 미제공 시 → `weak_ok`)

### 격자 축 (profile이 우선순위 추천, 격자는 전수 시도)

| 축 | 값 | 비고 |
|----|-----|------|
| `url_transforms` | `original`, `mobile_subdomain` (`www.→m.`), `am_prefix`, `drop_www` | 사이트명 없음, 규칙만 |
| `tls_impersonate` | `safari`, `safari_ios`, `chrome99`, `chrome119`, `chrome131`, `chrome_android`, `firefox`... | 프로파일별 avoid 리스트 존재 |
| `referer_strategy` | `self_root`, `google_search`, `none` | |

**device_class**:
- `"auto"` (기본) — 프로파일 전략 따름
- `"desktop"` — TLS 데스크톱만 + `mobile_subdomain` 비활성
- `"mobile"` — TLS 모바일만 + `mobile_subdomain` 활성

### Playwright 폴백 (capability-matched)

`engine/executor.py`가 프로파일의 `capabilities_needed`를 읽고 실행기를 자동 선택:

| 태그 | 실행기 | 언제 |
|------|--------|------|
| `needs_real_tls_stack` + `needs_js_exec` | `playwright_real_chrome.js` (로컬 Node) | Akamai Bot Manager 등 — Chromium 번들 TLS는 탐지됨 |
| `needs_js_exec` only | Playwright MCP (`mcp__playwright__*`) | Cloudflare 기본 방어 등 |
| `needs_mobile_context` (+ real_tls) | `playwright_mobile_chrome.js` | 모바일 디바이스 에뮬레이션 필요 |

자세한 선택 기준: [playwright.md](playwright.md).

### Playwright MCP 호출 규칙

`fetch_chain`의 `needs_js_exec only` 케이스는 **Claude 세션에서 MCP 도구를 직접 호출**해야 한다. subprocess 경로 없음. 즉:
1. `result.summary`에 "Playwright MCP must be invoked from the Claude session"이 포함되면
2. `mcp__playwright__browser_navigate` → `browser_wait_for` → `browser_snapshot` 흐름으로 Claude가 직접 처리

## Phase 2 — 수동 개입 (옵션)

Phase 1이 `ok=False`를 반환하면 사용자 힌트를 받아 재시도:

```python
result = fetch(
    url,
    success_selectors=[...],
    user_hint={"impersonate_first": "safari_ios", "referer_strategy": "none"},
)
```

힌트는 **현재 호출 1회에만** 적용되며 저장되지 않는다.

## 의존성 자동 설치

최초 호출 시 필요 패키지를 자동 설치한다:
```bash
python3 -c "import curl_cffi, bs4, yaml" 2>/dev/null || pip install curl_cffi beautifulsoup4 pyyaml -q
```

브라우저를 직접 제어할 때는 js eval에서 omowright를 쓴다(`browser` 스킬에 스테이징됨):
직접 띄우는 브라우저는 `connectPipe` / 스텔스는 `connectCloakProfile`(CloakBrowser),
사용자가 로그인해 둔 브라우저는 `connectBrowserSkill`. 이 엔진의 Playwright 템플릿은
Tier 1 추출 폴백 전용이며 에이전트가 직접 브라우저를 다루는 경로가 아니다.
인증이 필요한 페이지는 사용자 프로필을 복제하지 말고 attached 엔진으로 간다.

## 빠른 참조 — Phase 0 명령어

```bash
# 범용 웹 (Jina Reader — 일반 HTML만, WAF 사이트엔 무효)
curl -s "https://r.jina.ai/{URL}"

# yt-dlp — 1,858 사이트 미디어 메타데이터
yt-dlp --dump-json "URL"

# Reddit
curl -sL -H "User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15" \
  "https://www.reddit.com/r/{sub}/hot.json?limit=10"

# X/Twitter 타임라인
curl -sL "https://syndication.twitter.com/srv/timeline-profile/screen-name/{handle}"

# Hacker News
curl -sL "https://hacker-news.firebaseio.com/v0/topstories.json?limitToFirst=10&orderBy=%22%24key%22"

# YouTube 자막
yt-dlp --write-sub --write-auto-sub --sub-lang "en,ko" --skip-download -o "/tmp/%(id)s" "URL"
```

## No-Site-Name Rule

`engine/**`, `waf_profiles.yaml`, `engine/templates/**` 파일에는 **특정 사이트의 도메인/URL/셀렉터/브랜드명을 하드코딩하지 않는다**.

### 금지

- `"coupang.com": {...}` 같은 사이트별 레지스트리 엔트리
- `if "coupang" in url: ...` 같은 도메인 분기
- WAF 프로파일 `notes`에 특정 사이트 이름이나 경험적 byte 크기 박제

### 허용

- `SKILL.md` / `references/*.md`의 **설명 텍스트**에 사이트 이름 예시 (독자 이해용)
- `Phase 0` 공식 API 인덱스 (플랫폼이 공식 공개한 엔드포인트)
- `observations/*.jsonl` 로그 (append-only 관측 데이터 — 코드 경로에 영향 없음)
- 호출자가 제공하는 `success_selectors`, `user_hint` (현재 호출에만 유효)

### 경계 사례 판단 기준

> "이 엔트리가 다른 사이트에서도 같은 WAF를 쓰면 일반적으로 유효한가?" → YES면 `waf_profiles.yaml`, NO면 runtime hint.

### 새 사이트가 안 뚫릴 때

1. 먼저 `result.trace`에서 어느 phase가 실패했는지 확인
2. 사용자의 `user_hint`로 1회 재시도
3. 반복 성공 패턴이 관측되면 `observations/`에 로그 (아직 자동 기록 없음 — 수동)
4. 3회+ 반복 확인되고 **동일 WAF를 쓰는 다른 사이트에도 유효**하면 `waf_profiles.yaml` 해당 프로파일의 `tls_impersonate_candidates` / `url_transform_order`를 튜닝 (사이트명 절대 넣지 않음)
5. 여전히 안 되면 새 WAF 프로파일 후보 검토 (예: DataDome 세부화, Kasada 등)

## 관련 문서 (references/) — 언제 무엇을 읽을지

이 섹션은 **참조 파일 선택 가이드**다. 문제가 생겼을 때 어떤 `references/*.md`를 열어야 할지 결정하는 기준으로 쓴다. Claude는 필요할 때만 해당 파일을 `Read`하고, 선제적으로 전부 읽지 않는다.

### A. Engine 확장·진단 (하네스 내부)

| 파일 | 언제 읽는가 | 무엇을 다루는가 |
|------|-------------|-----------------|
| [`tls-impersonate.md`](tls-impersonate.md) | curl_cffi 격자가 전부 `challenge`/`blocked`로 끝날 때, 새 impersonate 타겟을 `waf_profiles.yaml`에 추가할 때 | curl_cffi로 Safari/Chrome/Firefox TLS(JA3/JA4) 지문 복제하는 방법, WAF(Akamai/Cloudflare/F5 등)별 최적 타겟 조합, 임퍼소네이션 타겟 버전 목록, `tls_impersonate_avoid`의 실증 근거 |
| [`playwright.md`](playwright.md) | engine이 Playwright fallback으로 넘어가는데 MCP/Local Chrome 중 어디로 갈지 확인 필요할 때 | Approach 1 (`mcp__playwright__*` — Cloudflare급 챌린지), Approach 2 (Local Node + `channel:'chrome'` + stealth — Akamai Bot Manager급), 템플릿 파라미터 규격 |
| [`fallback.md`](fallback.md) | `verdict`가 애매하거나 Phase 전환 타이밍 결정 필요할 때 | engine의 Phase 0→1→2→3 에스컬레이션 원칙, 응답 성공/실패 판정 기준 세부, 각 Phase 종료 조건 |
| [`metadata.md`](metadata.md) | 본문 전체를 못 가져왔지만 제목·요약·가격·저자 같은 핵심만이라도 필요할 때 | OGP 메타 태그, JSON-LD (Schema.org), Twitter Card 파싱, 구조화 데이터 추출 패턴 |

### B. 경량 대안 (engine 말고 다른 도구가 나은 상황)

| 파일 | 언제 읽는가 | 무엇을 다루는가 |
|------|-------------|-----------------|
| [`jina.md`](jina.md) | WAF 없는 일반 웹(블로그·뉴스·Wiki)의 깨끗한 마크다운 추출 필요할 때 | `r.jina.ai/URL` 한 줄로 Puppeteer 기반 JS SPA 렌더링, 마크다운 변환, 무료 500 RPM, API 키 불필요 |
| [`cache-archive.md`](cache-archive.md) | 원본 사이트가 차단됐지만 과거 스냅샷으로라도 접근 필요할 때 | Wayback Machine CDX API, archive.today, AMP Cache (Google Cache는 2024-07 종료됨) |
| [`rss.md`](rss.md) | 뉴스·블로그·커뮤니티의 시계열 업데이트를 구조화해 받고 싶을 때 | RSS/Atom 자동 발견, 피드 파싱, 인증 불필요 — 가장 깔끔한 시계열 데이터 소스 |

### C. 플랫폼별 공식/공개 API (Phase 0 인덱스와 연결)

| 파일 | 언제 읽는가 | 무엇을 다루는가 |
|------|-------------|-----------------|
| [`json-api.md`](json-api.md) | Reddit/Wikipedia/HN/npm/PyPI 등 **URL 변형만으로** JSON을 주는 사이트 | Reddit `/json` suffix + Mobile UA, HN Firebase, Algolia Search, Wikipedia REST, npm/PyPI Registry API |
| [`public-api.md`](public-api.md) | Bluesky/Mastodon/arXiv/Stack Overflow/CrossRef/GitHub/OpenLibrary/Wayback 공식 API 사용 시 | 인증 없이 쓰는 공식 공개 REST/AT/Atom API 엔드포인트, 요청 형식, 공통 파라미터 |
| [`twitter.md`](twitter.md) | X/Twitter 접근 — 프로필 타임라인, 특정 트윗, 키워드 검색 | `syndication.twitter.com` 타임라인, oEmbed 개별 트윗, 검색은 웹 검색 도구로 URL 확보 후 oEmbed |
| [`naver.md`](naver.md) | 네이버 블로그·뉴스·증권·검색 접근 | 서비스별 우회(블로그는 `m.blog.naver.com` 변환, 증권은 비공식 JSON, 검색은 `search.naver.com`), 한글 검색 쿼리 패턴 |
| [`media.md`](media.md) | YouTube/Vimeo/Twitch/TikTok/SoundCloud 등 미디어 메타·자막·오디오 필요 시 | `yt-dlp --dump-json` 기반 1,858개 사이트 커버, 자막 다운로드(`--write-sub`), 포맷 선택, 라이브/팟캐스트 |

### D. Engine 코드 직접 읽을 때

| 파일 | 언제 읽는가 |
|------|-------------|
| `engine/fetch_chain.py` | 체인 단계 로직·`Attempt`/`FetchResult` schema 확인 |
| `engine/validators.py` | 4-계층 검증 세부 (Verdict 분류, 챌린지 마커 목록) |
| `engine/waf_detector.py` | WAF 랭킹 감지 알고리즘, `_LAST_LOAD_ERROR` 처리 |
| `engine/waf_profiles.yaml` | 프로파일별 detectors·tls_candidates·capabilities_needed |
| `engine/url_transforms.py` | URL 변환 규칙 추가할 때 |
| `engine/executor.py` | Playwright MCP vs local capability 매칭 로직 |
| `engine/templates/*.js` | Playwright 템플릿 튜닝 (warmup, reload, devices) |
| `engine/bias_check.py` | 편향 린터 규칙 — brand denylist, URL_PATTERN, excluded dirs |

# RSS/Atom 피드

> 인증 불필요. URL만 알면 바로 구독. 뉴스/블로그/커뮤니티에서 가장 깔끔한 데이터.

## 의존성

```bash
python3 -c "import feedparser" 2>/dev/null || pip install feedparser -q
```

## RSS 자동 발견

Jina Reader JSON 모드로 사이트의 RSS URL을 자동 탐지:

```bash
curl -sH "Accept: application/json" "https://r.jina.ai/{URL}" | \
  python3 -c "import sys,json; print(json.load(sys.stdin)['data'].get('external',{}).get('alternate',[]))"
```

## URL 변형으로 피드 탐색

사이트에 RSS가 명시되지 않아도 시도해볼 패턴:

```bash
curl -sL "{origin}/rss"
curl -sL "{origin}/feed"
curl -sL "{origin}/atom.xml"
curl -sL "{origin}/rss.xml"
curl -sL "{origin}/index.xml"
```

## Google News RSS (무인증)

```bash
# 키워드 검색
curl -sL "https://news.google.com/rss/search?q={검색어}&hl=ko&gl=KR&ceid=KR:ko"

# 토픽별 (TECHNOLOGY, BUSINESS, SCIENCE, SPORTS, HEALTH, WORLD)
curl -sL "https://news.google.com/rss/headlines/section/topic/TECHNOLOGY?hl=ko&gl=KR&ceid=KR:ko"

# 시간 필터: when:1h, when:7d, when:12m, after:YYYY-MM-DD
curl -sL "https://news.google.com/rss/search?q={검색어}+when:7d&hl=ko&gl=KR&ceid=KR:ko"
```

## 한국 언론사 RSS

전부 무인증. 바로 curl로 접근 가능.

```bash
# SBS 뉴스
curl -sL "https://news.sbs.co.kr/news/rss.do"

# 조선일보
curl -sL "http://www.chosun.com/site/data/rss/rss.xml"

# 중앙일보
curl -sL "http://rss.joinsmsn.com/joins_news_list.xml"

# 동아일보
curl -sL "http://rss.donga.com/total.xml"

# 경향신문
curl -sL "http://www.khan.co.kr/rss/rssdata/total_news.xml"

# 매일경제
curl -sL "http://file.mk.co.kr/news/rss/rss_30000001.xml"

# MBC 뉴스
curl -sL "http://imnews.imbc.com/rss/news/news_00.xml"

# 한국경제
curl -sL "https://www.hankyung.com/feed/all-news"

# 연합뉴스
curl -sL "https://www.yonhapnewsagency.com/RSS/headline.xml"
```

## 블로그/플랫폼 RSS

```bash
# 네이버 블로그
curl -sL "https://rss.blog.naver.com/{BLOG_ID}.xml"

# 티스토리
curl -sL "https://{blogname}.tistory.com/rss"

# 벨로그
curl -sL "https://v2.velog.io/rss/@{username}"

# Substack
curl -sL "https://{publication}.substack.com/feed"

# GitHub 릴리즈 (Atom)
curl -sL "https://github.com/{owner}/{repo}/releases.atom"

# YouTube 채널
curl -sL "https://www.youtube.com/feeds/videos.xml?channel_id={id}"

# HN (hnrss.org — 비공식이지만 안정적)
curl -sL "https://hnrss.org/frontpage"
```

## feedparser 파싱

```python
import feedparser

feed = feedparser.parse("FEED_URL")
for e in feed.entries[:10]:
    print(f"{e.title} — {e.link}")
    if hasattr(e, 'summary'):
        print(f"  {e.summary[:200]}")
```

## SearXNG (무인증 메타검색)

공개 인스턴스에서 JSON 검색 가능. 인스턴스별로 JSON 지원 여부 다름.

```bash
# 공개 인스턴스 목록: https://searx.space
curl -sL "https://search.mdosch.de/search?q={검색어}&format=json" \
  -H "User-Agent: insane-search/1.0"
```

# TLS 임퍼소네이션 — curl_cffi

> TLS 핑거프린트(JA3/JA4) 기반 WAF를 우회하는 핵심 방법.
> 일반 curl/requests는 OpenSSL 핑거프린트라 즉시 차단되지만,
> curl_cffi는 실제 브라우저(Chrome/Safari/Firefox)의 TLS 핑거프린트를 복제한다.

## 의존성

```bash
python3 -c "import curl_cffi" 2>/dev/null || pip install curl_cffi -q
```

설치 후 사용 가능. **미설치를 이유로 이 스텝을 건너뛰지 않는다.**

## 다중 타겟 순차 시도

하나의 impersonate 타겟이 실패하면 다른 타겟으로 재시도한다.
**시도 순서: safari → chrome → firefox**

```python
from curl_cffi import requests

TARGETS = ["safari", "chrome", "firefox"]
HEADERS = {
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7",
    "Accept-Encoding": "gzip, deflate, br",
    "Referer": "https://www.google.com/",
}

def cffi_fetch(url, locale="ko-KR"):
    """다중 타겟 순차 시도 + 신원위장. 성공하면 (response, target) 반환."""
    from urllib.parse import urlparse
    origin = f"{urlparse(url).scheme}://{urlparse(url).netloc}"
    for target in TARGETS:
        try:
            session = requests.Session(impersonate=target)
            session.headers.update(HEADERS)
            session.headers["Accept-Language"] = f"{locale},{locale.split('-')[0]};q=0.9"
            session.headers["Referer"] = "https://www.google.com/"
            # 신원위장: 홈페이지 쿠키 워밍 → Referer 체인
            try:
                session.get(origin, timeout=10)
            except Exception:
                pass  # 홈 실패해도 본 요청은 시도
            session.headers["Referer"] = origin
            resp = session.get(url, timeout=20)
            # JS 필수 사이트 감지 → 나머지 타겟 시도 무의미
            if "behavioral-content" in resp.text or "sec-if-cpt" in resp.text:
                return None, None  # → Phase 3 Playwright
            if resp.status_code == 200 and len(resp.text) > 500:
                return resp, target
        except Exception:
            continue
    return None, None
```

## 임퍼소네이션 타겟 목록 (v0.15.0)

generic alias는 항상 최신 버전으로 해석된다. **2026년에 chrome99 같은 옛 버전은 WAF가 의심하므로 generic alias 사용 권장.**

| Alias | 해석 (2026.04) | 용도 |
|-------|---------------|------|
| `safari` | safari260 | **한국 사이트 최적** (쿠팡, 에펨코리아) |
| `chrome` | chrome146 | 범용 (Cloudflare, Akamai) |
| `firefox` | firefox135 | chrome/safari 실패 시 대안 |
| `chrome_android` | chrome131_android | 모바일 API 엔드포인트 |
| `safari_ios` | safari260_ios | iOS 모바일 |

<details>
<summary>핀 버전 전체 (클릭)</summary>

```
chrome99, chrome100, chrome101, chrome104, chrome107, chrome110,
chrome116, chrome119, chrome120, chrome123, chrome124, chrome131,
chrome133a, chrome136, chrome142, chrome145, chrome146,
chrome131_android, edge99, edge101,
safari15_3, safari15_5, safari17_0, safari17_2_ios,
safari18_0, safari18_0_ios, safari260, safari260_ios,
firefox133, firefox135
```

</details>

## WAF별 최적 전략

| WAF | 최적 타겟 | 추가 조건 | 성공률 |
|-----|-----------|-----------|--------|
| F5 BIG-IP (쿠팡) | `safari` | `Referer: https://www.coupang.com/` | ~70% |
| Cloudflare (TLS만) | `chrome` | Sec-Fetch-* 헤더 추가 | ~80% |
| Akamai | `chrome` | 레지덴셜 프록시 병행 | 80-90% |
| AWS WAF | `chrome` | — | ~80% |
| CloudFront (요즘IT) | 불필요 | 일반 curl + Chrome UA로 충분 | 100% |

## 세션과 쿠키

```python
from curl_cffi import requests

# 세션 유지 (쿠키 자동 관리)
session = requests.Session(impersonate="safari")

# 첫 요청으로 세션 쿠키 획득
session.get("https://www.coupang.com/")

# 이후 요청에 쿠키 자동 전달
resp = session.get("https://www.coupang.com/np/search?q=키보드")
```

## 콤보: nodriver/FlareSolverr → curl_cffi

JS 챌린지 사이트는 브라우저로 쿠키를 획득한 뒤 curl_cffi로 고속 처리:

```python
# 1. nodriver로 cf_clearance 쿠키 획득
import nodriver as uc
browser = await uc.start(headless=True)
page = await browser.get("https://cf-protected-site.com")
await page.cf_verify()
cookies = await browser.cookies.get_all()

# 2. curl_cffi Session에 쿠키 전달
from curl_cffi import requests
session = requests.Session(impersonate="chrome")
for c in cookies:
    session.cookies.set(c["name"], c["value"])
resp = session.get("https://cf-protected-site.com/api/data")
```

## 비동기 (async)

```python
import asyncio
from curl_cffi.requests import AsyncSession

async def fetch_many(urls):
    async with AsyncSession(impersonate="chrome") as session:
        tasks = [session.get(url) for url in urls]
        return await asyncio.gather(*tasks)
```

## HTTP/3 (v0.15.0+)

```python
from curl_cffi import requests
from curl_cffi.const import CurlHttpVersion

resp = requests.get(
    "https://www.cloudflare.com/",
    impersonate="chrome",
    http_version=CurlHttpVersion.V3,
)
```

WAF 벤더들이 아직 HTTP/3 핑거프린트를 적극 활용하지 않아 우회 효과가 높다.

## 대안 라이브러리

curl_cffi 실패 시 대안:

| 라이브러리 | 설치 | 특징 |
|-----------|------|------|
| primp | `pip install primp` | Rust 기반, Firefox 148까지, 고성능 |
| wreq/rnet | `pip install wreq` | Rust 기반, 100+ 디바이스 프로필 |
| tls-client2 | `pip install tls-client2` | Go 기반 포크, 동기만 |

```python
# primp 예시
import primp
client = primp.Client(impersonate="chrome_146")
resp = client.get("https://example.com")
```

## curl_cffi가 못 뚫는 것

| 방어 수단 | curl_cffi | 대응 |
|-----------|-----------|------|
| TLS/JA3 핑거프린트 | 우회 가능 | 핵심 기능 |
| HTTP/2 SETTINGS 핑거프린트 | 우회 가능 | impersonate에 포함 |
| HTTP/3 QUIC 핑거프린트 | 우회 가능 (v0.15+) | 신규 |
| JS 챌린지 (Turnstile 등) | **불가** | → nodriver 또는 Playwright |
| CAPTCHA | **불가** | → 2captcha/CapSolver |
| IP 평판 (데이터센터) | **불가** | → 프록시/VPN |
| 행동 분석 (마우스/타이밍) | **불가** | → 실제 브라우저 |

JS 챌린지가 걸린 사이트는 → [playwright.md](playwright.md) 로 넘긴다.

# X/Twitter 접근 전략

> WebFetch는 402로 차단됨. 아래 방법으로 우회한다. 모두 API 키/인증 불필요.

## 검색 (트윗 발견)

```python
<사용 가능한 web search tool>(query="site:x.com {검색어}")  # Claude Code: WebSearch / OpenCode 계열: websearch_web_search_exa 등 — 하네스마다 실제 tool 이름이 다르므로 현재 세션의 tool 목록에서 확인할 것
```

웹 검색 도구는 X 포스트를 검색 결과로 반환한다. 제목, snippet, URL을 획득할 수 있지만 트윗 전문이나 engagement 수치는 없다.

## 타임라인 조회 — Syndication API

특정 핸들의 최근 ~100개 트윗 + engagement 수치(likes, RTs) 제공.

### 엔드포인트

```
https://syndication.twitter.com/srv/timeline-profile/screen-name/{handle}
```

### 원샷 스크립트

```bash
curl -sL "https://syndication.twitter.com/srv/timeline-profile/screen-name/{handle}" | \
python3 -c "
import sys, json, re, html
content = sys.stdin.read()
match = re.search(r'__NEXT_DATA__.*?>(.*?)</script>', content)
if match:
    data = json.loads(match.group(1))
    for e in data['props']['pageProps']['timeline']['entries']:
        if e['type'] == 'tweet':
            t = e['content']['tweet']
            print(f\"@{t['user']['screen_name']} ({t.get('created_at','?')})\")
            print(f\"  {html.unescape(t.get('full_text',''))[:300]}\")
            print(f\"  Likes: {t.get('favorite_count',0)} | RTs: {t.get('retweet_count',0)}\")
            print('---')
"
```

### 가져올 수 있는 데이터

| 필드 | 경로 | 예시 |
|------|------|------|
| 트윗 전문 | `tweet.full_text` | "Give your agent the..." |
| 작성자 핸들 | `tweet.user.screen_name` | "openclaw" |
| 작성자 이름 | `tweet.user.name` | "OpenClaw" |
| 좋아요 수 | `tweet.favorite_count` | 1929 |
| RT 수 | `tweet.retweet_count` | 169 |
| 작성 시각 | `tweet.created_at` | "Mon Apr 06 04:04:08 +0000 2026" |
| 트윗 ID | `tweet.id_str` | "2041003999856406714" |
| 미디어 URL | `tweet.entities.media[].media_url_https` | 이미지/동영상 URL |

### 제한

- 최근 ~100개 반환 (페이지네이션 불가)
- 비공개 계정 접근 불가
- 검색 기능 없음 (타임라인만)
- **저팔로워/신규 계정**: `hasResults: false` 반환 가능. 이 경우 oEmbed 개별 트윗 접근은 정상 동작하므로 "조합 패턴"으로 폴백.
- 비공식 엔드포인트 — X가 변경/차단 가능

## 개별 트윗 조회 — oEmbed API

특정 트윗 URL을 알 때 전문 가져오기.

### 엔드포인트

```
https://publish.twitter.com/oembed?url=https://x.com/{user}/status/{tweet_id}
```

### 사용법

```bash
curl -sL "https://publish.twitter.com/oembed?url=https://x.com/{user}/status/{tweet_id}"
```

### 응답 (JSON)

| 필드 | 설명 |
|------|------|
| `author_name` | 작성자 표시 이름 |
| `author_url` | 작성자 프로필 URL |
| `html` | 트윗 전문이 포함된 HTML blockquote |
| `url` | 트윗 원본 URL |

## 조합 패턴 (검색 → 상세)

```
1단계: 웹 검색 도구(query="site:x.com {키워드}") → 트윗 URL 획득
2단계: curl oEmbed API → 트윗 전문 획득
```

## 실패하는 방법 (사용하지 말 것)

| 방법 | 결과 | 원인 |
|------|------|------|
| WebFetch | 402 Payment Required | Claude Code의 WebFetch 제한 |
| Nitter | 빈 응답 | Nitter 인스턴스 대부분 종료됨 |
| Wayback Machine | OG 메타태그만 | SPA 렌더링 안 됨 |
| Mobile UA curl | OG 메타태그만 | SPA 렌더링 안 됨 |
| RSS | 엔드포인트 없음 | X는 RSS 지원 중단 |

#!/usr/bin/env python3
from __future__ import annotations

import shutil
import sqlite3
import sys
import tempfile
import unittest
from collections.abc import Callable
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from extract_cookies import extract_cookies  # noqa: E402


def _make_chromium_db(path: Path, rows: list[tuple[str, bytes, str]]) -> None:
    conn = sqlite3.connect(str(path))
    conn.execute(
        "CREATE TABLE cookies (name TEXT, encrypted_value BLOB, host_key TEXT, path TEXT, "
        "expires_utc INTEGER, is_secure INTEGER, is_httponly INTEGER, samesite INTEGER)"
    )
    conn.executemany(
        "INSERT INTO cookies VALUES (?,?,?,?,?,?,?,?)",
        [
            (name, value, host, "/", 13_300_000_000_000_000, 1, 1, 1)
            for name, value, host in rows
        ],
    )
    conn.commit()
    conn.close()


def _make_firefox_db(path: Path, rows: list[tuple[str, str, str]]) -> None:
    conn = sqlite3.connect(str(path))
    conn.execute(
        "CREATE TABLE moz_cookies (name TEXT, value TEXT, host TEXT, path TEXT, "
        "expiry INTEGER, isSecure INTEGER, isHttpOnly INTEGER, sameSite INTEGER)"
    )
    conn.executemany(
        "INSERT INTO moz_cookies VALUES (?,?,?,?,?,?,?,?)",
        [
            (name, value, host, "/", 9999999999, 1, 1, 1)
            for name, value, host in rows
        ],
    )
    conn.commit()
    conn.close()


class DomainFilter(unittest.TestCase):
    def _base_with_db(self, rel: str, make: Callable[[Path], None]) -> Path:
        base = Path(tempfile.mkdtemp())
        self.addCleanup(lambda: shutil.rmtree(str(base), ignore_errors=True))
        db = base / rel
        db.parent.mkdir(parents=True)
        make(db)
        return base

    def test_chromium_does_not_overmatch_suffix_text(self) -> None:
        base = self._base_with_db(
            "Google/Chrome/User Data/Default/Cookies",
            lambda p: _make_chromium_db(
                p,
                [
                    ("exact", b"exact-value", "example.com"),
                    ("sub", b"sub-value", ".login.example.com"),
                    ("near", b"near-value", ".example.com"),
                ],
            ),
        )

        near = extract_cookies(
            "chrome", ["ample.com"], platform="win32",
            keyring_reader=lambda _s: b"k" * 32, base_override=base,
        )
        exact = extract_cookies(
            "chrome", ["example.com"], platform="win32",
            keyring_reader=lambda _s: b"k" * 32, base_override=base,
        )

        self.assertEqual(near, [])
        self.assertEqual({cookie["name"] for cookie in exact}, {"exact", "near", "sub"})

    def test_firefox_does_not_overmatch_suffix_text(self) -> None:
        base = self._base_with_db(
            "Firefox/Profiles/abc.default/cookies.sqlite",
            lambda p: _make_firefox_db(
                p,
                [
                    ("exact", "exact-value", "example.com"),
                    ("sub", "sub-value", ".login.example.com"),
                    ("near", "near-value", ".example.com"),
                ],
            ),
        )

        near = extract_cookies(
            "firefox", ["ample.com"], platform="darwin", base_override=base,
        )
        exact = extract_cookies(
            "firefox", ["example.com"], platform="darwin", base_override=base,
        )

        self.assertEqual(near, [])
        self.assertEqual({cookie["name"] for cookie in exact}, {"exact", "near", "sub"})


if __name__ == "__main__":
    unittest.main(verbosity=2)

#!/usr/bin/env python3
"""Synthetic-fixture tests for cross-platform cookie extraction (no live browser)."""
from __future__ import annotations

import shutil
import sqlite3
import sys
import tempfile
import unittest
from pathlib import Path
from typing import Callable
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from cookie_crypto import decrypt_chromium_value, derive_key  # noqa: E402
from cookie_paths import UnsupportedPlatform, resolve_cookie_db  # noqa: E402
from extract_cookies import extract_cookies, inject_cookies, write_cookie_file  # noqa: E402


def _make_chromium_db(path: Path, name: str, encrypted_value: bytes, host: str) -> None:
    conn = sqlite3.connect(str(path))
    conn.execute(
        "CREATE TABLE cookies (name TEXT, encrypted_value BLOB, host_key TEXT, path TEXT, "
        "expires_utc INTEGER, is_secure INTEGER, is_httponly INTEGER, samesite INTEGER)"
    )
    conn.execute(
        "INSERT INTO cookies VALUES (?,?,?,?,?,?,?,?)",
        (name, encrypted_value, host, "/", 13_300_000_000_000_000, 1, 1, 1),
    )
    conn.commit()
    conn.close()


def _make_firefox_db(path: Path, name: str, value: str, host: str) -> None:
    conn = sqlite3.connect(str(path))
    conn.execute(
        "CREATE TABLE moz_cookies (name TEXT, value TEXT, host TEXT, path TEXT, "
        "expiry INTEGER, isSecure INTEGER, isHttpOnly INTEGER, sameSite INTEGER)"
    )
    conn.execute("INSERT INTO moz_cookies VALUES (?,?,?,?,?,?,?,?)", (name, value, host, "/", 9999999999, 1, 1, 1))
    conn.commit()
    conn.close()


def _encrypt_cbc_v10(key: bytes, plaintext: str) -> bytes:
    from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

    data = plaintext.encode()
    pad = 16 - (len(data) % 16)
    padded = data + bytes([pad]) * pad
    encryptor = Cipher(algorithms.AES128(key), modes.CBC(b" " * 16)).encryptor()
    return b"v10" + encryptor.update(padded) + encryptor.finalize()


def _encrypt_gcm_v10(key: bytes, plaintext: str) -> bytes:
    from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

    nonce = b"n" * 12
    encryptor = Cipher(algorithms.AES(key), modes.GCM(nonce)).encryptor()
    ciphertext = encryptor.update(plaintext.encode()) + encryptor.finalize()
    return b"v10" + nonce + ciphertext + encryptor.tag


class PathResolution(unittest.TestCase):
    def test_macos_chromium_path(self) -> None:
        base = Path(self.tmp())
        target = base / "Google/Chrome/Default/Cookies"
        target.parent.mkdir(parents=True)
        target.write_bytes(b"")
        self.assertEqual(resolve_cookie_db("chrome", "darwin", base_override=base), target)

    def test_linux_chromium_network_path(self) -> None:
        base = Path(self.tmp())
        target = base / "google-chrome/Default/Network/Cookies"
        target.parent.mkdir(parents=True)
        target.write_bytes(b"")
        self.assertEqual(resolve_cookie_db("chrome", "linux", base_override=base), target)

    def test_windows_chromium_path(self) -> None:
        base = Path(self.tmp())
        target = base / "Google/Chrome/User Data/Default/Cookies"
        target.parent.mkdir(parents=True)
        target.write_bytes(b"")
        self.assertEqual(resolve_cookie_db("chrome", "win32", base_override=base), target)

    def test_unsupported_platform_raises(self) -> None:
        with self.assertRaises(UnsupportedPlatform):
            resolve_cookie_db("chrome", "sunos")

    def test_unsupported_browser_raises(self) -> None:
        with self.assertRaises(UnsupportedPlatform):
            resolve_cookie_db("nonexistent", "darwin")

    def tmp(self) -> str:
        import tempfile

        d = tempfile.mkdtemp()
        self.addCleanup(lambda: __import__("shutil").rmtree(d, ignore_errors=True))
        return d


class KeyDerivation(unittest.TestCase):
    def test_macos_iters_differ_from_linux(self) -> None:
        secret = b"some-keychain-secret"
        self.assertNotEqual(derive_key("darwin", secret), derive_key("linux", secret))
        self.assertEqual(len(derive_key("darwin", secret)), 16)

    def test_windows_passthrough_32_bytes(self) -> None:
        key = b"k" * 32
        self.assertEqual(derive_key("win32", key), key)

    def test_windows_rejects_wrong_length(self) -> None:
        with self.assertRaises(ValueError):
            derive_key("win32", b"short")

    def test_unsupported_platform_raises(self) -> None:
        with self.assertRaises(UnsupportedPlatform):
            derive_key("sunos", b"x")


class Decryption(unittest.TestCase):
    def test_macos_cbc_roundtrip(self) -> None:
        key = derive_key("darwin", b"secret")
        blob = _encrypt_cbc_v10(key, "session-token-123")
        self.assertEqual(decrypt_chromium_value("darwin", key, blob), "session-token-123")

    def test_windows_gcm_roundtrip(self) -> None:
        key = b"k" * 32
        blob = _encrypt_gcm_v10(key, "win-token-xyz")
        self.assertEqual(decrypt_chromium_value("win32", key, blob), "win-token-xyz")


class EndToEnd(unittest.TestCase):
    def _base_with_db(self, rel: str, make: Callable[[Path], None]) -> Path:
        base = Path(tempfile.mkdtemp())
        self.addCleanup(lambda: shutil.rmtree(str(base), ignore_errors=True))
        db = base / rel
        db.parent.mkdir(parents=True)
        make(db)
        return base

    def test_chromium_extract_with_injected_keyring(self) -> None:
        key = derive_key("darwin", b"secret")
        blob = _encrypt_cbc_v10(key, "logged-in")
        base = self._base_with_db(
            "Google/Chrome/Default/Cookies",
            lambda p: _make_chromium_db(p, "SID", blob, ".youtube.com"),
        )
        cookies = extract_cookies(
            "chrome", ["youtube.com"], platform="darwin",
            keyring_reader=lambda _s: b"secret", base_override=base,
        )
        self.assertEqual(len(cookies), 1)
        self.assertEqual(cookies[0]["value"], "logged-in")
        self.assertEqual(cookies[0]["name"], "SID")

    def test_firefox_extract_unencrypted(self) -> None:
        base = self._base_with_db(
            "Firefox/Profiles/abc.default/cookies.sqlite",
            lambda p: _make_firefox_db(p, "auth", "plain-value", ".example.com"),
        )
        cookies = extract_cookies("firefox", ["example.com"], platform="darwin", base_override=base)
        self.assertEqual(len(cookies), 1)
        self.assertEqual(cookies[0]["value"], "plain-value")


class SecretHandling(unittest.TestCase):
    def test_cookie_output_file_is_owner_only(self) -> None:
        base = Path(tempfile.mkdtemp())
        self.addCleanup(lambda: shutil.rmtree(str(base), ignore_errors=True))
        output = base / "cookies.json"

        write_cookie_file(output, [{"name": "SID", "value": "secret"}])

        self.assertEqual(output.stat().st_mode & 0o777, 0o600)
        self.assertIn("secret", output.read_text())

    def test_cookie_output_replaces_existing_file_only_after_private_temp_write(self) -> None:
        base = Path(tempfile.mkdtemp())
        self.addCleanup(lambda: shutil.rmtree(str(base), ignore_errors=True))
        output = base / "cookies.json"
        output.write_text("old\n")
        output.chmod(0o644)

        def _dump(_cookies, f, indent: int) -> None:
            self.assertEqual(output.read_text(), "old\n")
            self.assertEqual(output.stat().st_mode & 0o777, 0o644)
            temp_files = list(base.glob(".cookies.json.*.tmp"))
            self.assertEqual(len(temp_files), 1)
            self.assertEqual(temp_files[0].stat().st_mode & 0o777, 0o600)
            f.write('[{"name": "SID", "value": "secret"}]')

        with patch("extract_cookies.json.dump", side_effect=_dump):
            write_cookie_file(output, [{"name": "SID", "value": "secret"}])

        self.assertEqual(output.stat().st_mode & 0o777, 0o600)
        self.assertIn("secret", output.read_text())

    def test_cookie_output_refuses_symlinks(self) -> None:
        base = Path(tempfile.mkdtemp())
        self.addCleanup(lambda: shutil.rmtree(str(base), ignore_errors=True))
        target = base / "target.json"
        target.write_text("{}")
        link = base / "cookies.json"
        link.symlink_to(target)

        with self.assertRaises(ValueError):
            write_cookie_file(link, [{"name": "SID", "value": "secret"}])

    def test_cookie_output_refuses_dangling_symlinks(self) -> None:
        base = Path(tempfile.mkdtemp())
        self.addCleanup(lambda: shutil.rmtree(str(base), ignore_errors=True))
        link = base / "cookies.json"
        link.symlink_to(base / "missing.json")

        with self.assertRaises(ValueError):
            write_cookie_file(link, [{"name": "SID", "value": "secret"}])

    def test_inject_cookies_sends_values_over_stdin_not_argv(self) -> None:
        cookie = {
            "name": "SID",
            "value": "secret-token",
            "domain": ".youtube.com",
            "path": "/",
            "expires": 9999999999,
            "secure": True,
            "httpOnly": True,
            "sameSite": "Lax",
        }

        with patch("extract_cookies.subprocess.run") as run:
            run.return_value.returncode = 0
            run.return_value.stdout = "1"
            run.return_value.stderr = ""

            inject_cookies([cookie], 9242)

        command = run.call_args.args[0]
        self.assertNotIn("secret-token", command)
        self.assertIn("secret-token", run.call_args.kwargs["input"])


if __name__ == "__main__":
    unittest.main(verbosity=2)