from __future__ import annotations import csv import json import os import re import time from datetime import date from pathlib import Path from urllib.parse import urlparse import pandas as pd import requests from bs4 import BeautifulSoup ROOT = Path(__file__).resolve().parent ACCESSED_AT = date(2026, 7, 13).isoformat() HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; SHWRP-research/1.0; +https://bento24235111.com/)"} MIN_REQUEST_INTERVAL_SECONDS = 1.0 PATTERNS = { "marketplace_signal": r"shopee|蝦皮|momo|pchome|pinkoi|rakuten|樂天|東森購物|etmall|yahoo.{0,5}購物", # A bare @handle also matches e-mail domains, so require LINE-specific context. "line_signal": r"line\.me|lin\.ee|line@|line\s*[:：]?\s*@[-_a-z0-9]{4,}|line官方|加入好友", "social_signal_crawl": r"facebook\.com|instagram\.com|youtube\.com|youtu\.be|tiktok\.com|threads\.net", "payment_signal": r"信用卡|line\s*pay|街口|綠界|ecpay|藍新|newebpay|apple\s*pay|google\s*pay|全支付|信用卡付款", "logistics_signal": r"黑貓|宅急便|宅配|超商取貨|7-11取貨|全家取貨|店到店|郵局|常溫配送|低溫配送", "membership_signal": r"會員登入|會員註冊|註冊會員|加入會員|member|login|register", "direct_checkout_signal": r"購物車|立即購買|前往結帳|加入購物車|shopping.?cart|add.?to.?cart|checkout", "subscription_signal": r"定期購|定期配送|訂閱制|每月配送|週期配送|subscription", "owned_content_signal": r"品牌故事|關於我們|最新消息|部落格|食譜|米知識|專欄|news|blog|story|recipe", "custom_service_signal": r"客製|企業採購|團體訂購|大量訂購|婚禮|彌月|禮盒訂製|報價", "review_signal": r"顧客評價|商品評價|購買評價|review|rating|評論", } def fetch_public_text(url: str) -> tuple[int | None, str, str]: if not isinstance(url, str) or not url.startswith("http"): return None, "", "invalid" try: response = requests.get(url, headers=HEADERS, timeout=18, allow_redirects=True) if response.status_code in {401, 403}: raise PermissionError(f"Public access denied ({response.status_code}); collection stopped") if response.status_code == 429: raise RuntimeError("Rate limited (429); collection stopped without retry or identity rotation") content_type = response.headers.get("content-type", "").lower() if response.status_code != 200: return response.status_code, "", content_type[:80] if len(response.content) > 4_000_000: return response.status_code, "", "oversize" if "html" not in content_type and "text" not in content_type: return response.status_code, "", content_type[:80] response.encoding = response.apparent_encoding or response.encoding soup = BeautifulSoup(response.text, "html.parser") for tag in soup(["script", "style", "noscript", "svg"]): tag.decompose() text = " ".join(soup.get_text(" ", strip=True).split()) links = " ".join(anchor.get("href", "") for anchor in soup.find_all("a")) return response.status_code, (text + " " + links).lower()[:500_000], content_type[:80] except Exception as exc: return None, "", type(exc).__name__ if os.environ.get("PUBLIC_WEB_COLLECTION_AUTHORIZED") != "YES": raise RuntimeError( "Public web collection is disabled by default. Recheck each source's current terms and obtain " "any required permission before setting PUBLIC_WEB_COLLECTION_AUTHORIZED=YES." ) cases = pd.read_csv(ROOT / "cases-seed-60.csv") audit_rows: list[dict] = [] coded_rows: list[dict] = [] for _, row in cases.iterrows(): combined = "" success_count = 0 for role, column in [("primary", "primary_url"), ("secondary", "secondary_url")]: url = str(row.get(column, "")) status, text, content_type = fetch_public_text(url) combined += " " + text success_count += int(status == 200 and bool(text)) audit_rows.append({ "id": row["id"], "operator": row["operator"], "role": role, "url": url, "status": status or "", "text_available": int(bool(text)), "content_type_or_error": content_type, "accessed_at": ACCESSED_AT, }) time.sleep(MIN_REQUEST_INTERVAL_SECONDS) signals = {name: int(bool(re.search(pattern, combined, flags=re.I))) for name, pattern in PATTERNS.items()} # Preserve stronger hand-coded evidence from the prior fixed 60-case dataset. signals["marketplace_signal"] = max(signals["marketplace_signal"], int(row["third_party_channel"])) signals["social_signal"] = max(signals.pop("social_signal_crawl"), int(row["social_channel"])) signals["direct_checkout_signal"] = max(signals["direct_checkout_signal"], int(row["own_ecommerce"])) signals["custom_service_signal"] = max(signals["custom_service_signal"], int(row["b2b_service"])) signals["owned_content_signal"] = max(signals["owned_content_signal"], int(row["heritage_story"])) platform_components = [signals["marketplace_signal"], signals["social_signal"], signals["line_signal"], signals["payment_signal"], signals["logistics_signal"]] owned_components = [int(row["direct_web"]), int(row["own_ecommerce"]), signals["membership_signal"], signals["direct_checkout_signal"], signals["subscription_signal"], signals["owned_content_signal"]] differentiation_components = [int(row["flexible_bulk"]), int(row["quality_assurance"]), int(row["origin_variety"]), int(row["b2b_service"]), int(row["heritage_story"]), int(row["experience_education"]), signals["custom_service_signal"]] platform_count = sum(platform_components) owned_count = sum(owned_components) differentiation_count = sum(differentiation_components) public_channel_concentration = sum([ int(signals["marketplace_signal"] and not row["own_ecommerce"]), int(signals["social_signal"] and not row["direct_web"]), int(platform_count >= 2 and not signals["membership_signal"]), int(platform_count >= 2 and not signals["direct_checkout_signal"]), int(not row["direct_web"]), ]) coded = row.to_dict() | signals | { "pages_with_text": success_count, "platform_leverage_count": platform_count, "platform_leverage_index": round(platform_count / 5 * 100, 1), "owned_asset_count": owned_count, "owned_asset_index": round(owned_count / 6 * 100, 1), "differentiation_count": differentiation_count, "differentiation_index": round(differentiation_count / 7 * 100, 1), "public_channel_concentration_count": public_channel_concentration, "public_channel_concentration_index": round(public_channel_concentration / 5 * 100, 1), "coding_date": ACCESSED_AT, } coded_rows.append(coded) coded_df = pd.DataFrame(coded_rows) lev_median = coded_df["platform_leverage_index"].median() own_median = coded_df["owned_asset_index"].median() def strategy(row): high_l = row["platform_leverage_index"] >= lev_median high_o = row["owned_asset_index"] >= own_median if high_l and high_o: return "orchestrated_hybrid" if high_l and not high_o: return "platform_first" if not high_l and high_o: return "direct_first" return "low_public_digital_footprint" coded_df["strategy_type"] = coded_df.apply(strategy, axis=1) coded_df.to_csv(ROOT / "cases-coded-60.csv", index=False, encoding="utf-8-sig") pd.DataFrame(audit_rows).to_csv(ROOT / "crawl-audit-120.csv", index=False, encoding="utf-8-sig") metadata = { "design": "secondary coding of a fixed stratified purposive 60-case dataset", "case_count": int(len(coded_df)), "source_urls": int(len(audit_rows)), "coding_date": ACCESSED_AT, "platform_index_components": ["marketplace", "social", "LINE", "payment", "logistics"], "owned_asset_components": ["direct website", "own e-commerce", "membership", "direct checkout", "subscription", "owned content"], "concentration_proxy_note": "Public-channel concentration proxy only; it is not a measure of revenue dependence, contractual dependence, switching cost, or legal compliance.", "platform_median": float(lev_median), "owned_asset_median": float(own_median), "successful_text_pages": int(pd.DataFrame(audit_rows)["text_available"].sum()), } (ROOT / "collection-metadata.json").write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(metadata, ensure_ascii=False, indent=2))