Browser Execution Observer

Library core untuk ekstraksi data lewat browser Chrome (CDP). Satu mesin yang sama dipakai untuk banyak sosial media: platform baru cukup dideskripsikan (URL, pola API, mapping field), tanpa menulis ulang sesi browser, scroll, capture, maupun parser.

Instalasi

pip install beo-mcdp
# opsional: pip install "beo-mcdp[kafka]"  /  "beo-mcdp[patchright]"

Cara kerja

1. Browser membuka halaman (Endpoint.url).
2. Library mendengarkan response API yang dipanggil halaman (Endpoint.api_match).
3. JSON-nya diubah jadi list dict oleh parser (biasanya FieldMapper), di-dedupe lewat key, lalu halaman di-scroll sampai limit/data habis.

Menambah platform baru

from cdp_scraper import (Endpoint, F, FieldMapper, Platform, Scraper,
                         const, make_scroller, register_platform, end_when_falsy)

parser = FieldMapper(
    items="data.posts",                      # path ke list item
    fields={
        "id":    ("id", "str", ""),          # (path, transform, default)
        "url":   F(template="https://site.com/{author.handle}/{id}"),
        "likes": ("stats.likes", "int", 0),
        "tags":  "tags[].name",              # semua elemen
        "posted": F("created", transform="iso_utc"),
        "platform": const("site"),
    },
)

site = Platform(
    name="site",                              # juga jadi label log: [site] ...
    cdp={"cdp_url": "http://localhost:9223"}, # opsional: Chrome/slot terpisah per platform
    overlay_selectors=('div.popup-close',),   # popup yang ditutup otomatis
    login_url_hints=("/login",),              # deteksi browser belum login
    endpoints=[Endpoint(
        "search", url="https://site.com/search?q={keyword}", api_match="api/search",
        parser=parser, params=("keyword",), label="keyword", limit=20, max_limit=50,
        end=end_when_falsy("has_more"),
        scroll=make_scroller(wheel=2500, click_pattern="load more"),
    )],
)
register_platform(site)

result = await Scraper("site").run("search", {"keyword": "kopi"}, limit=30)     # satu query
results = await Scraper("site").run_many("search", [{"keyword": "a"}, {"keyword": "b"}])

run() melempar exception (PageCrashed, BrowserBusy, ValueError untuk parameter salah). run_many() memakai satu tab untuk banyak query dan menjadikan error per-item status="error". Contoh lengkap TikTok dan X/Twitter ada di examples/.

Mapping parsing (FieldMapper)

Bentuk spec field: "path" · ("path", "transform", default) · F(path | any_of=[...] | template=... | fn=..., default=, transform=, required=) · const(x) · callable(record).

Opsi FieldMapper: item_path (sub-path record di tiap item), where / record_where ({path: nilai | EXISTS | callable} atau callable), unwrap (membuka pembungkus), items boleh list path atau callable.

Transform bawaan: int float str bool url lower upper strip clean_text username hashtag iso_utc date_id join first count unique drop_empty dan each:<transform> untuk list. Argumen format tanggal: iso_utc:%a %b %d %H:%M:%S %z %Y. Tambah sendiri: register_transform("nama", fn).

Jika logika terlalu rumit untuk mapping, gunakan F(fn=...) atau parser Python biasa (parser= menerima callable apa pun yang mengembalikan list dict).

Hook & aksi halaman

prepare (setelah halaman terbuka) dan scroll (tiap putaran) adalah fungsi async def (page). Bawaan: make_scroller(...), make_wait_for(selector), close_overlays(page, selectors). Endpoint.capture_options meneruskan kwargs tambahan ke capture (poll_interval, settle, dst).

Konfigurasi browser

from cdp_scraper import CDPConfig, setup_cdp
setup_cdp(cdp_url="http://localhost:9222", max_sessions=2, lock_timeout=60)   # default global
cfg = CDPConfig.from_env("TWITTER_")   # TWITTER_CDP_URL, TWITTER_MAX_CONCURRENT_SESSIONS, ...

Setiap request memakai tab baru sendiri (ditutup saat selesai) dan dibatasi max_sessions bersamaan per konfigurasi. Konfigurasi yang sama berbagi slot; cdp_url berbeda = slot berbeda. Screenshot debug disimpan di CDPConfig.debug_dir (default folder kerja); matikan dengan debug_screenshots=False.

Logging

Format seragam [job] Status message: detail, job = nama platform:

[2026-11-10 08:19:38] [INFO] [site] Success message: Capture api/search selesai: 20 data

Library hanya memakai logger cdp_scraper (tanpa handler). Aktifkan dengan setup_logging() atau injeksi logger sendiri: setup_cdp(custom_logger=...). get_logger("kafka").error("...") untuk log aplikasi dengan format yang sama.

Kafka / Redpanda

from cdp_scraper import RedpandaProducer, publish_chunks
producer = RedpandaProducer()            # KAFKA_BOOTSTRAP_SERVERS; koneksi dibuat saat dipakai
if producer.available:
    stats = publish_chunks(producer, "topic-x", keywords, chunk_size=2,
                           build_payload=lambda chunk: {"keywords": chunk})
    # {"berhasil": n, "gagal": n}

Kontrak sumber data (keyword/akun) ada di cdp_scraper.utils.database (KeywordSource, AccountSource); implementasinya milik aplikasi.

Pengujian

python -m unittest discover -s tests -t .

Tes memakai browser palsu, jadi tidak butuh Chrome/Kafka.

Perubahan dari 0.1.x (breaking)

• Konfigurasi sekarang CDPConfig (immutable, per platform); setup_cdp(...) tetap ada. Variabel modul config.CDP_URL dst dihapus.
• get_session_slots dan PW_ENGINE dihapus; pakai CDPSessionManager / manager.engine.
• BrowserBusy (subclass TimeoutError) menggantikan TimeoutError polos.
• helpers.close_login_modal(page, selectors) kini wajib diberi selector (tidak ada lagi selector TikTok bawaan).
• utils: tidak lagi mengimpor Flask; extract_challenges (khusus TikTok) dihapus; calculate_engagement_rate menerima interaction_keys / view_key (default sama dengan perilaku lama).
• utils.database kini berupa Protocol; utils.redpanda_client tidak lagi membuat producer saat import (gunakan cdp_scraper.messaging).