yhm-shop-data/对标试点/02_采集脚本/collect_douyin.py

# -*- coding: utf-8 -*-
"""
抖音门店公开信息采集器 v0.1
不装任何第三方库，直接调用本机 Chrome 无头模式，读取渲染后的页面。
只采集前台公开可见的信息（评分、评价数、团购商品与已售数），不登录、不破解、不采个人信息。

用法（在 对标试点 目录下执行）:
    python 02_采集脚本/collect_douyin.py --dump
        读取 01_对标店清单.csv，逐个打开门店链接，把渲染后的页面存到 02_采集脚本/dump/
        每页间隔约 4 秒，控制频率

    python 02_采集脚本/collect_douyin.py --parse
        解析 dump/ 里已保存的页面，生成 03_分析/采集快照_<日期>.csv
        注意: 解析规则需要先用 1-2 个真实页面校准，第一次跑请只放 1-2 条链接
"""
import csv
import glob
import hashlib
import os
import re
import subprocess
import sys
import time
from datetime import date

BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DUMP_DIR = os.path.join(BASE, '02_采集脚本', 'dump')
LIST_CSV = os.path.join(BASE, '01_对标店清单.csv')
SNAP_DIR = os.path.join(BASE, '03_分析')

CHROME_CANDIDATES = [
    r'C:\Program Files\Google\Chrome\Application\chrome.exe',
    r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe',
]
CHROME = next((p for p in CHROME_CANDIDATES if os.path.exists(p)), None)


def load_links():
    """从清单里取出所有不重复的门店链接"""
    if not os.path.exists(LIST_CSV):
        sys.exit('找不到 01_对标店清单.csv')
    seen, out = set(), []
    with open(LIST_CSV, encoding='utf-8-sig') as f:
        for r in csv.DictReader(f):
            for key in ('本店POI链接', '竞对POI链接'):
                link = (r.get(key) or '').strip()
                if link and link not in seen:
                    seen.add(link)
                    out.append((r.get('本店编码', ''), r.get('竞对门店名', ''), link))
    return out


def dump_pages():
    if not CHROME:
        sys.exit('本机没有找到 Chrome / Edge，请确认安装路径')
    os.makedirs(DUMP_DIR, exist_ok=True)
    links = load_links()
    print(f'共 {len(links)} 个待采集链接')
    for i, (code, name, link) in enumerate(links, 1):
        fp = os.path.join(DUMP_DIR, f'{code}_{hashlib.md5(link.encode()).hexdigest()[:8]}.html')
        if os.path.exists(fp) and os.path.getsize(fp) > 50000:
            print(f'[{i}/{len(links)}] 已存在，跳过 {name or code}')
            continue
        print(f'[{i}/{len(links)}] {name or code} {link}')
        try:
            subprocess.run([CHROME, '--headless=new', '--disable-gpu',
                            '--window-size=1200,2400', '--virtual-time-budget=8000',
                            '--dump-dom', link], capture_output=True, timeout=90)
            # dump-dom 输出到 stdout，这里重跑一次拿文本
            res = subprocess.run([CHROME, '--headless=new', '--disable-gpu',
                                  '--window-size=1200,2400', '--virtual-time-budget=8000',
                                  '--dump-dom', link], capture_output=True, timeout=90)
            html = (res.stdout or b'').decode('utf-8', 'ignore')
            if len(html) < 20000:
                print('    页面内容过短，可能需要登录或被风控，请人工核对这个链接')
            with open(fp, 'w', encoding='utf-8') as f:
                f.write(html)
        except subprocess.TimeoutExpired:
            print('    超时，跳过')
        time.sleep(4)


def num_cn(s):
    """把 '1.2万' '1000+' 转成数字，解析不了返回 None"""
    s = s.replace('+', '').replace(',', '').strip()
    m = re.match(r'^([\d.]+)(万|w)?$', s)
    if not m:
        return None
    v = float(m.group(1))
    if m.group(2):
        v *= 10000
    return v


def parse_page(html):
    """首次校准前的草稿规则。跑一次真实 dump 后按实际页面结构重写这里。"""
    items = []
    rating = None
    m = re.search(r'(\d\.\d)\s*分', html)
    if m:
        rating = float(m.group(1))
    reviews = None
    m = re.search(r'(\d+)\s*条评价', html)
    if m:
        reviews = int(m.group(1))
    # 团购商品: 匹配 "商品名 ... 已售 xxx" 的松散结构，实际结构以真实 dump 为准
    for blk in re.findall(r'<div[^>]*>([^<>]{2,30})</div>.{0,600}?已售\s*([\d.万亿w+]*)', html, re.S):
        name, sold = blk[0].strip(), num_cn(blk[1] or '')
        items.append((name, sold))
    return rating, reviews, items


def parse_dumps():
    today = date.today().isoformat()
    files = sorted(glob.glob(os.path.join(DUMP_DIR, '*.html')))
    if not files:
        sys.exit('dump/ 目录为空，先执行 --dump')
    out = os.path.join(SNAP_DIR, f'采集快照_{today.replace("-", "")}.csv')
    with open(out, 'w', encoding='utf-8-sig', newline='') as f:
        w = csv.writer(f)
        w.writerow(['采集日期', '门店编码', '门店名', '门店类型', 'POI链接',
                    '商品名', '原价', '团购价', '已售', '评分', '评价数'])
        for fp in files:
            stem = os.path.splitext(os.path.basename(fp))[0]
            code = stem.split('_')[0]
            with open(fp, encoding='utf-8') as fh:
                html = fh.read()
            rating, reviews, items = parse_page(html)
            print(f'{stem}: 评分={rating} 评价={reviews} 商品={len(items)}')
            w.writerow([today, code, stem, '本店', '', '', '', '', '', rating or '', reviews or ''])
            for name, sold in items:
                w.writerow([today, code, stem, '本店', '', name, '', '', sold or '', rating or '', reviews or ''])
    print(f'快照已写入: {out}')
    print('提示: 字段解析规则是草稿，请人工抽查几行，再决定是否重写 parse_page()')


if __name__ == '__main__':
    if '--dump' in sys.argv:
        dump_pages()
    elif '--parse' in sys.argv:
        parse_dumps()
    else:
        print(__doc__)