#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ BIZ-145 / S2 竞品价格采集脚本 合规红线（法务 BIZ-147 复核意见 C1–C7，全部硬编码，不可被命令行覆盖）： C1 频率封顶：单店每周 ≤1 次、单次 ≤10 页面、请求间隔 ≥5 秒 C2 遵守 robots.txt：被禁抓即跳过 C3 不绕反爬：固定常规 UA、不用 IP 池、遇验证码/登录墙即放弃该页面 C4 字段白名单：商品名称 / 页面展示标价 / 页面展示促销信息 / 页面可见销量数 C5 数据留存：原始快照 ≤90 天（--prune 清理） C7 不使用代理/账号池：直连、无登录态 输出：data/YYYY-MM-DD/prices.json + prices.csv + fetch.log 失败告警：整体或单店失败时写日志；环境变量 MULTICA_PARENT_ISSUE 存在时 通过 multica CLI 在父单留一条失败通知评论。 """ import csv import json import logging import os import re import subprocess import sys import time import urllib.error import urllib.parse import urllib.request import urllib.robotparser from datetime import datetime, timedelta, timezone from pathlib import Path # ---------------- 合规常量（C1–C7，硬编码） ---------------- MAX_PAGES_PER_SHOP = 10 # C1 MIN_INTERVAL_SEC = 5.0 # C1 RETENTION_DAYS = 90 # C5 USER_AGENT = ( # C3/C6：常规 UA，不带伪装、附采集说明联系方式 "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 " "(+internal price monitoring; contact: ops@bizwingsinc.com)" ) TIMEOUT_SEC = 20 ALLOWED_FIELDS = ("name", "list_price", "promo_text", "sales_display") # C4 白名单 # 验证码 / 登录墙特征（C3：命中即放弃，不尝试绕过） BLOCK_PATTERNS = [ "login.taobao.com", "passport", "验证码", "verify", "captcha", "slide.tsx", "punish", "sec.taobao", ] BASE_DIR = Path(__file__).resolve().parent DATA_DIR = BASE_DIR / "data" TARGETS_FILE = BASE_DIR / "targets.yaml" CST = timezone(timedelta(hours=8)) log = logging.getLogger("competitor-price") def setup_logging(log_file: Path): log.setLevel(logging.INFO) fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s") sh = logging.StreamHandler(sys.stdout) sh.setFormatter(fmt) fh = logging.FileHandler(log_file, encoding="utf-8") fh.setFormatter(fmt) log.addHandler(sh) log.addHandler(fh) def load_targets() -> list[dict]: """极简 YAML 解析（targets.yaml 结构固定，避免引入第三方依赖）。""" text = TARGETS_FILE.read_text(encoding="utf-8") competitors, cur, in_sku, in_monitor = [], None, False, False for raw in text.splitlines(): line = raw.split("#", 1)[0].rstrip() if not line.strip(): continue stripped = line.strip() if stripped.startswith("- id:"): cur = {"id": stripped.split(":", 1)[1].strip(), "sku_urls": [], "monitor_skus": []} competitors.append(cur) in_sku = in_monitor = False elif cur is None: continue elif re.match(r"^- \w+:", stripped) is None and ":" in stripped and not stripped.startswith("-"): key, _, val = stripped.partition(":") val = val.strip().strip('"') if key == "sku_urls": in_sku, in_monitor = True, False if val.startswith("[") and val != "[]": cur["sku_urls"] = [u.strip().strip('"') for u in val.strip("[]").split(",") if u.strip()] elif key == "monitor_skus": in_monitor, in_sku = True, False else: cur[key] = val in_sku = in_monitor = False elif stripped.startswith("-") and in_sku: cur["sku_urls"].append(stripped.lstrip("- ").strip('"')) elif stripped.startswith("-") and in_monitor: cur["monitor_skus"].append(stripped.lstrip("- ").strip('"')) return competitors def robots_allows(url: str) -> bool: """C2：遵守 robots.txt；出错时保守处理为允许但记日志。""" parsed = urllib.parse.urlparse(url) rp_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt" rp = urllib.robotparser.RobotFileParser() try: rp.set_url(rp_url) rp.read() return rp.can_fetch(USER_AGENT, url) except Exception as e: # robots 不可达时不放大风险，记录并放行（人工每周低频） log.warning("robots.txt 读取失败 %s: %s（保守放行，人工复核）", rp_url, e) return True def fetch_page(url: str) -> tuple[int | None, str]: """返回 (http_status, html)。不跟随登录跳转以外的异常。""" req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept-Language": "zh-CN,zh;q=0.9"}) try: with urllib.request.urlopen(req, timeout=TIMEOUT_SEC) as resp: html = resp.read(2_000_000).decode("utf-8", errors="replace") return resp.status, html except urllib.error.HTTPError as e: return e.code, "" except Exception as e: log.warning("请求异常 %s: %s", url, e) return None, "" def looks_blocked(html: str, final_status: int | None) -> bool: """C3：命中反爬/登录墙特征 → 放弃该页面。""" if final_status in (401, 403): return True low = html.lower() return any(p.lower() in low for p in BLOCK_PATTERNS) PRICE_RE = re.compile(r'"price"\s*:\s*"?([\d.]+)"?|¥\s*([\d]+(?:\.\d+)?)|￥\s*([\d]+(?:\.\d+)?)') PROMO_RE = re.compile(r"(满\d+减\d+|第二件半价|买一送一|买赠|券后)") SALES_RE = re.compile(r"(?:月销|已售|销量)\s*[:：]?\s*([\d,]+\+?)") TITLE_RE = re.compile(r"(.*?)", re.S) def parse_fields(url: str, html: str) -> dict: """C4：只抽白名单字段，抽不到就留空并在 status 标注。""" out = {f: "" for f in ALLOWED_FIELDS} m = TITLE_RE.search(html) if m: out["name"] = re.sub(r"-(天猫|淘宝网).*$", "", m.group(1)).strip()[:120] m = PRICE_RE.search(html) if m: out["list_price"] = next(g for g in m.groups() if g) promos = PROMO_RE.findall(html) if promos: out["promo_text"] = "；".join(sorted(set(promos))[:5]) m = SALES_RE.search(html) if m: out["sales_display"] = m.group(1) return {k: v for k, v in out.items() if k in ALLOWED_FIELDS} def notify_parent_issue(summary: str): """失败告警：在父单留评论。无 MULTICA_PARENT_ISSUE 时只写日志。""" issue = os.environ.get("MULTICA_PARENT_ISSUE") or "448cdf6f-ac97-4c7f-953f-0d92d03c6e85" body = f"⚠️ 竞品价格采集失败告警（{datetime.now(CST):%Y-%m-%d %H:%M} GMT+8）\n\n{summary}\n\n详见监控目录 data/ 下当日 fetch.log。" try: subprocess.run(["multica", "issue", "comment", "add", issue, "--content", body], check=True, timeout=60, capture_output=True) log.info("已在父单 %s 留言告警", issue) except Exception as e: log.error("父单告警发送失败: %s", e) def prune_old_snapshots(): """C5：删除 90 天前的采集目录。""" cutoff = datetime.now(CST).date() - timedelta(days=RETENTION_DAYS) for d in DATA_DIR.iterdir() if DATA_DIR.exists() else []: if d.is_dir(): try: day = datetime.strptime(d.name, "%Y-%m-%d").date() except ValueError: continue if day < cutoff: import shutil shutil.rmtree(d) log.info("C5 留存清理：删除 %s", d) def collect_one(comp: dict, today_dir: Path, results: list, failures: list): urls = [u for u in comp.get("sku_urls", []) if u] urls = urls or ([comp["shop_url"]] if comp.get("shop_url") else []) if not urls: log.warning("%s：无可用 URL（白牌未回填），跳过", comp["name"]) results.append({"competitor_id": comp["id"], "competitor": comp["name"], "url": "", "status": "skipped_no_url", **{f: "" for f in ALLOWED_FIELDS}}) return urls = urls[:MAX_PAGES_PER_SHOP] # C1 for i, url in enumerate(urls): if i > 0: time.sleep(MIN_INTERVAL_SEC) # C1 if not robots_allows(url): # C2 log.warning("robots.txt 禁止抓取，跳过：%s", url) failures.append(f"{comp['name']} robots 禁止 {url}") results.append({"competitor_id": comp["id"], "competitor": comp["name"], "url": url, "status": "skipped_robots", **{f: "" for f in ALLOWED_FIELDS}}) continue status, html = fetch_page(url) if looks_blocked(html, status): # C3 log.warning("疑似登录墙/验证码（HTTP %s），放弃：%s", status, url) failures.append(f"{comp['name']} 被反爬拦截 {url} (HTTP {status})") results.append({"competitor_id": comp["id"], "competitor": comp["name"], "url": url, "status": f"blocked_anti_bot_http_{status}", **{f: "" for f in ALLOWED_FIELDS}}) continue if not html: failures.append(f"{comp['name']} 抓取失败 {url} (HTTP {status})") results.append({"competitor_id": comp["id"], "competitor": comp["name"], "url": url, "status": f"fetch_failed_http_{status}", **{f: "" for f in ALLOWED_FIELDS}}) continue fields = parse_fields(url, html) row = {"competitor_id": comp["id"], "competitor": comp["name"], "url": url, "status": "ok" if fields.get("list_price") else "ok_no_price_field", **fields} results.append(row) log.info("✅ %s %s → %s", comp["name"], url, row["status"]) def main() -> int: today = datetime.now(CST).strftime("%Y-%m-%d") today_dir = DATA_DIR / today today_dir.mkdir(parents=True, exist_ok=True) setup_logging(today_dir / "fetch.log") log.info("=== BIZ-145 竞品价格采集启动 %s ===", today) prune_old_snapshots() competitors = load_targets() results: list[dict] = [] failures: list[str] = [] for comp in competitors: collect_one(comp, today_dir, results, failures) time.sleep(MIN_INTERVAL_SEC) # C1：店铺间同样限速 run_meta = { "run_date": today, "tz": "Asia/Shanghai", "compliance": {"C1_interval_sec": MIN_INTERVAL_SEC, "C1_max_pages_per_shop": MAX_PAGES_PER_SHOP, "C2_robots_respected": True, "C3_user_agent": USER_AGENT, "C4_field_whitelist": list(ALLOWED_FIELDS), "C5_retention_days": RETENTION_DAYS, "C7_no_proxy_no_login": True}, "results": results, "failures": failures, } (today_dir / "prices.json").write_text( json.dumps(run_meta, ensure_ascii=False, indent=2), encoding="utf-8") with open(today_dir / "prices.csv", "w", newline="", encoding="utf-8-sig") as f: w = csv.DictWriter(f, fieldnames=["competitor_id", "competitor", "url", "status", *ALLOWED_FIELDS]) w.writeheader() w.writerows(results) log.info("=== 完成：%d 条记录，%d 条失败 ===", len(results), len(failures)) if failures: notify_parent_issue("\n".join(f"- {x}" for x in failures)) return 1 return 0 if __name__ == "__main__": sys.exit(main())