c91e7f8c20
移除结构化 audience 字段，强化 voc_业务_2 源评论归因匹配与 Persona 引用展示，更新 README 与流水线默认清理 SQLite。 Co-authored-by: Cursor <cursoragent@cursor.com>
191 lines
7 KiB
Python
191 lines
7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
步骤 1：运行 main_voc分析.py 的 step 1–6（跳过 step 7 报告生成），产出 SQLite 数据库。
|
||
|
||
用法::
|
||
|
||
# 全流程（config.yaml 留空则 LLM 自动识别产品名和行业）
|
||
../310py/bin/python run_pipeline.py --input-dir "../Bikini trimmer voc"
|
||
|
||
# 手动指定产品/行业（跳过 LLM 识别）
|
||
../310py/bin/python run_pipeline.py --input-dir "../Bikini trimmer voc" --product "Bikini Trimmer" --industry "个人护理"
|
||
|
||
# 断点续跑（从 step 4 向量化开始；前提是 step 1–3 的 sqlite 产物已存在）
|
||
../310py/bin/python run_pipeline.py --from-step 4
|
||
|
||
# 仅重跑词频（step 6）
|
||
../310py/bin/python run_pipeline.py --from-step 6
|
||
|
||
# 仅重跑聚类（step 5）
|
||
../310py/bin/python run_pipeline.py --from-step 5
|
||
|
||
内部调用等价于（以全流程为例）::
|
||
|
||
../310py/bin/python ../main_voc分析.py --only-step 1 --input-dir "..." --product "..." --industry "..."
|
||
../310py/bin/python ../main_voc分析.py --only-step 2 --product "..." --industry "..."
|
||
...（依此类推到 step 6）
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import logging
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import yaml
|
||
|
||
logger = logging.getLogger("voc.run_pipeline")
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
PROJECT_ROOT = SCRIPT_DIR.parent
|
||
CONFIG_FILE = SCRIPT_DIR / "config.yaml"
|
||
|
||
|
||
def load_config() -> dict:
|
||
with CONFIG_FILE.open(encoding="utf-8") as f:
|
||
return yaml.safe_load(f) or {}
|
||
|
||
|
||
def run_step(python_bin: str, main_script: str, step: int, **kwargs) -> bool:
|
||
"""运行单个 --only-step。"""
|
||
cmd = [
|
||
python_bin, main_script,
|
||
"--only-step", str(step),
|
||
"--product", kwargs.get("product", "亚马逊商品"),
|
||
"--industry", kwargs.get("industry", "亚马逊电商"),
|
||
]
|
||
if step == 1 and "input_dir" in kwargs:
|
||
cmd.extend(["--input-dir", kwargs["input_dir"]])
|
||
|
||
logger.info("执行 step %s: %s", step, " ".join(cmd))
|
||
# 不 capture 输出，子进程日志实时可见，避免长时间步骤看起来像卡死
|
||
result = subprocess.run(cmd, cwd=str(PROJECT_ROOT))
|
||
|
||
if result.returncode != 0:
|
||
logger.error("Step %s 失败 (exit %s)", step, result.returncode)
|
||
return False
|
||
logger.info("Step %s 完成", step)
|
||
return True
|
||
|
||
|
||
def run_build_report(
|
||
python_bin: str,
|
||
build_script: str,
|
||
*,
|
||
product: str,
|
||
industry: str,
|
||
) -> bool:
|
||
"""运行 build_report.py，子进程日志实时输出到当前终端。"""
|
||
cmd = [
|
||
python_bin, "-u", build_script,
|
||
"--product", product,
|
||
"--industry", industry,
|
||
]
|
||
logger.info("执行报告: %s", " ".join(cmd))
|
||
env = os.environ.copy()
|
||
env["PYTHONUNBUFFERED"] = "1"
|
||
# 显式继承 stdout/stderr，避免长时间 LLM 步骤看起来像卡死
|
||
result = subprocess.run(
|
||
cmd,
|
||
cwd=str(SCRIPT_DIR),
|
||
env=env,
|
||
stdout=sys.stdout,
|
||
stderr=sys.stderr,
|
||
)
|
||
if result.returncode != 0:
|
||
logger.error("build_report.py 失败 (exit %s)", result.returncode)
|
||
return False
|
||
return True
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="运行 VOC 分析流水线 step 1-6")
|
||
parser.add_argument("--input-dir", help="原始 CSV 目录")
|
||
parser.add_argument("--product", help="产品名")
|
||
parser.add_argument("--industry", help="行业名")
|
||
parser.add_argument("--from-step", type=int, default=1, choices=range(1, 7))
|
||
args = parser.parse_args()
|
||
|
||
cfg = load_config()
|
||
|
||
product = args.product or cfg.get("product_name", "").strip()
|
||
industry = args.industry or cfg.get("industry", "").strip()
|
||
|
||
# 如果产品名或行业为空，用 LLM 从原始评论中自动识别
|
||
need_detect = (not product or product == "亚马逊商品" or not industry or industry == "亚马逊电商")
|
||
if need_detect:
|
||
input_dir_raw = args.input_dir or cfg.get("input_dir", "")
|
||
input_dir_path_raw = (SCRIPT_DIR / input_dir_raw).resolve() if input_dir_raw else None
|
||
if input_dir_path_raw and input_dir_path_raw.is_dir():
|
||
from data_loader import DataLoader
|
||
samples, dir_name = DataLoader.load_raw_review_samples(input_dir_path_raw, max_samples=50)
|
||
if samples:
|
||
import sys as _sys
|
||
_sys.path.insert(0, str(SCRIPT_DIR))
|
||
from llm_analyzer import detect_product_and_industry
|
||
detected_product, detected_industry = detect_product_and_industry(samples, dir_name)
|
||
if not product or product == "亚马逊商品":
|
||
product = detected_product
|
||
logger.info("LLM 自动识别产品名: %s", product)
|
||
if not industry or industry == "亚马逊电商":
|
||
industry = detected_industry
|
||
logger.info("LLM 自动识别行业: %s", industry)
|
||
else:
|
||
if not product:
|
||
product = "亚马逊商品"
|
||
if not industry:
|
||
industry = "亚马逊电商"
|
||
else:
|
||
if not product:
|
||
product = "亚马逊商品"
|
||
if not industry:
|
||
industry = "亚马逊电商"
|
||
input_dir = args.input_dir or cfg.get("input_dir", "")
|
||
main_script = cfg.get("main_script", "../main_voc分析.py")
|
||
python_bin = cfg.get("python_bin", "../310py/bin/python")
|
||
|
||
# 解析相对路径
|
||
main_script_path = (SCRIPT_DIR / main_script).resolve()
|
||
python_bin_path = (SCRIPT_DIR / python_bin) # 不 resolve：保留符号链接以确保 uv venv 正确激活
|
||
input_dir_path = str((SCRIPT_DIR / input_dir).resolve()) if input_dir else ""
|
||
|
||
if not python_bin_path.exists():
|
||
logger.error("Python 解释器不存在: %s", python_bin_path)
|
||
sys.exit(1)
|
||
if not main_script_path.is_file():
|
||
logger.error("main_voc分析.py 不存在: %s", main_script_path)
|
||
sys.exit(1)
|
||
|
||
logger.info("产品: %s | 行业: %s | 输入目录: %s", product, industry, input_dir_path)
|
||
logger.info("从 step %s 开始", args.from_step)
|
||
|
||
steps = list(range(args.from_step, 7)) # step 1-6 only
|
||
for step in steps:
|
||
kwargs = {"product": product, "industry": industry}
|
||
if step == 1:
|
||
kwargs["input_dir"] = input_dir_path
|
||
if not run_step(str(python_bin_path), str(main_script_path), step, **kwargs):
|
||
logger.error("流水线中止于 step %s", step)
|
||
sys.exit(1)
|
||
|
||
logger.info("流水线 step 1-6 全部完成。数据库已就绪。")
|
||
|
||
# 自动运行 build_report.py（日志实时打印到终端）
|
||
logger.info("自动运行 build_report.py 生成报告...")
|
||
build_script = str(SCRIPT_DIR / "build_report.py")
|
||
if not run_build_report(
|
||
str(python_bin_path),
|
||
build_script,
|
||
product=product,
|
||
industry=industry,
|
||
):
|
||
sys.exit(1)
|
||
logger.info("报告生成完成。")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|