91e6c47fc0
统一 voc_llm 密钥解析与默认模型；向量化改为本地 mlx 模型；更新 README、gitignore 与流水线文档。 Co-authored-by: Cursor <cursoragent@cursor.com>
434 lines
15 KiB
Python
434 lines
15 KiB
Python
"""
|
||
VOC 全流程：合并 → 清洗 → 结构化 → 向量化 →（聚类 ∥ 词频）→ 分析报告 HTML。
|
||
|
||
用法::
|
||
|
||
# 全流程（--industry 默认 Pet Supplies；写入 sqlite 前默认清理旧库，keep-db不清理）
|
||
./310py/bin/python main_voc分析.py --input-dir 目录 --product "产品名" --keep-db
|
||
--industry "行业名"
|
||
# 断点续跑（步骤 4 起可省略 --product，自动读 voc_structured.sqlite）
|
||
./310py/bin/python main_voc分析.py --from-step 5
|
||
./310py/bin/python main_voc分析.py --from-step 6 --skip-wordfreq-llm # 仅重跑词频
|
||
./310py/bin/python main_voc分析.py --only-step 7 # 仅生成报告
|
||
|
||
# 保留已有 voc_structured / voc_embeddings / voc_clustering.sqlite，不清理覆盖
|
||
./310py/bin/python main_voc分析.py --from-step 4 --keep-db
|
||
|
||
报告 HTML：output/{product_name}/{product_name}_voc_report.html
|
||
|
||
Chat 默认 deepseek-v4-pro、思考关闭（DEEPSEEK_API_KEY）；报告主 LLM 见 voc_report（Pro + max）；向量化本地 Qwen3-Embedding-4B-mxfp8（推荐 ./310py/bin/python）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import logging
|
||
import re
|
||
import sqlite3
|
||
import sys
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from pathlib import Path
|
||
|
||
from content清洗 import process_reviews, save_cleaned_reviews
|
||
from 合并评论数据 import merge_csv_directory
|
||
from 向量化 import EMBED_DEFAULT_WORKERS, run_embed
|
||
from 结构化_server import STRUCT_DEFAULT_WORKERS, run_analysis
|
||
from 聚类 import run_clustering
|
||
from voc_llm import require_chat_api_key
|
||
from voc_report import DEFAULT_CLUSTER_MIN_REVIEW_RATIO, generate_report
|
||
from 词频 import run as run_wordfreq
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(levelname)s] %(message)s",
|
||
stream=sys.stderr,
|
||
)
|
||
logger = logging.getLogger("voc_analysis")
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent
|
||
OUTPUT_DIR = PROJECT_ROOT / "output"
|
||
|
||
STRUCTURED_DB = PROJECT_ROOT / "voc_structured.sqlite"
|
||
EMBED_DB = PROJECT_ROOT / "voc_embeddings.sqlite"
|
||
CLUSTER_DB = PROJECT_ROOT / "voc_clustering.sqlite"
|
||
TERMS_JSON = OUTPUT_DIR / "voc_terms.json"
|
||
WORD_FREQ_CSV = OUTPUT_DIR / "word_freq.csv"
|
||
|
||
DEFAULT_MERGED = PROJECT_ROOT / "merged_reviews.csv"
|
||
DEFAULT_CLEANED = PROJECT_ROOT / "merged_reviews_cleaned.csv"
|
||
DEFAULT_INDUSTRY = "-"
|
||
|
||
|
||
def _safe_product_dir_name(product_name: str) -> str:
|
||
"""用于 output 子目录名；去除路径非法字符。"""
|
||
name = (product_name or "product").strip()
|
||
name = re.sub(r'[<>:"/\\|?*]', "_", name)
|
||
return name or "product"
|
||
|
||
|
||
def _product_output_dir(product_name: str) -> Path:
|
||
return OUTPUT_DIR / _safe_product_dir_name(product_name)
|
||
|
||
|
||
def _report_html_path(product_name: str) -> Path:
|
||
slug = _safe_product_dir_name(product_name)
|
||
return _product_output_dir(product_name) / f"{slug}_voc_report.html"
|
||
|
||
|
||
def _latest_job_meta(structured_db: Path) -> tuple[int, str, str]:
|
||
conn = sqlite3.connect(structured_db)
|
||
try:
|
||
row = conn.execute(
|
||
"""
|
||
SELECT id, industry, product_name
|
||
FROM analysis_jobs ORDER BY id DESC LIMIT 1
|
||
"""
|
||
).fetchone()
|
||
if not row:
|
||
raise RuntimeError("无结构化任务，请先执行步骤 3")
|
||
return int(row[0]), str(row[1]), str(row[2])
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _should_run(step: int, from_step: int, only_step: int | None) -> bool:
|
||
if only_step is not None:
|
||
return step == only_step
|
||
return step >= from_step
|
||
|
||
|
||
def _clean_intermediates(
|
||
*,
|
||
merged: Path,
|
||
cleaned: Path,
|
||
structured: Path,
|
||
embed: Path,
|
||
cluster: Path,
|
||
terms: Path,
|
||
word_freq: Path,
|
||
) -> None:
|
||
for p in (merged, cleaned, structured, embed, cluster, terms, word_freq):
|
||
if p.is_file():
|
||
p.unlink()
|
||
logger.info("已删除中间文件 %s", p)
|
||
|
||
|
||
def run_voc_analysis(
|
||
*,
|
||
input_dir: Path,
|
||
industry: str,
|
||
product_name: str,
|
||
merged_csv: Path,
|
||
cleaned_csv: Path,
|
||
from_step: int = 1,
|
||
only_step: int | None = None,
|
||
clean_intermediates: bool = False,
|
||
clean_databases: bool = True,
|
||
skip_wordfreq_llm: bool = False,
|
||
min_cluster_review_ratio: float | None = None,
|
||
save_llm_raw: bool = False,
|
||
struct_workers: int | None = None,
|
||
embed_workers: int | None = None,
|
||
) -> dict:
|
||
result: dict = {}
|
||
|
||
if _should_run(1, from_step, only_step):
|
||
logger.info("步骤 1/7：合并 CSV")
|
||
merge_csv_directory(input_dir, merged_csv)
|
||
result["merged_csv"] = str(merged_csv)
|
||
|
||
if _should_run(2, from_step, only_step):
|
||
logger.info("步骤 2/7：清洗评论")
|
||
if not merged_csv.is_file():
|
||
raise FileNotFoundError(f"缺少 {merged_csv}，请先执行合并")
|
||
df = process_reviews(merged_csv)
|
||
save_cleaned_reviews(df, cleaned_csv)
|
||
result["cleaned_csv"] = str(cleaned_csv)
|
||
|
||
need_chat = (
|
||
_should_run(3, from_step, only_step)
|
||
or _should_run(5, from_step, only_step)
|
||
or _should_run(6, from_step, only_step)
|
||
or _should_run(7, from_step, only_step)
|
||
)
|
||
if need_chat:
|
||
require_chat_api_key()
|
||
|
||
if _should_run(3, from_step, only_step):
|
||
logger.info("步骤 3/7：结构化分析")
|
||
if not cleaned_csv.is_file():
|
||
raise FileNotFoundError(f"缺少 {cleaned_csv}，请先执行清洗")
|
||
ar = run_analysis(
|
||
industry=industry,
|
||
product_name=product_name,
|
||
file_path=str(cleaned_csv),
|
||
clean_databases=clean_databases,
|
||
workers=struct_workers,
|
||
)
|
||
result["structured_job_id"] = ar.get("job_id")
|
||
result["structured_db"] = str(STRUCTURED_DB)
|
||
n_ext = len(ar.get("extractions") or {})
|
||
if n_ext == 0:
|
||
raise RuntimeError("步骤 3 结构化无有效结果，已中止（不会继续向量化）")
|
||
|
||
job_id: int | None = None
|
||
ind, prod = industry, product_name
|
||
if STRUCTURED_DB.is_file():
|
||
try:
|
||
job_id, ind, prod = _latest_job_meta(STRUCTURED_DB)
|
||
except RuntimeError:
|
||
pass
|
||
|
||
if _should_run(4, from_step, only_step):
|
||
if clean_databases and not _should_run(3, from_step, only_step):
|
||
logger.info("写入向量化库前清理 voc_embeddings.sqlite / voc_clustering.sqlite")
|
||
for p in (EMBED_DB, CLUSTER_DB):
|
||
if p.is_file():
|
||
p.unlink()
|
||
logger.info("已清理 SQLite: %s", p.name)
|
||
logger.info("步骤 4/7：向量化")
|
||
er = run_embed(
|
||
job_id=job_id,
|
||
csv_path=cleaned_csv,
|
||
structured_db=STRUCTURED_DB,
|
||
embed_db=EMBED_DB,
|
||
reset_db=clean_databases,
|
||
workers=embed_workers,
|
||
)
|
||
result["embed"] = er
|
||
job_id = int(er.get("job_id", job_id or 0))
|
||
|
||
if job_id is None and STRUCTURED_DB.is_file():
|
||
job_id, ind, prod = _latest_job_meta(STRUCTURED_DB)
|
||
|
||
run_cluster = _should_run(5, from_step, only_step)
|
||
run_wf = _should_run(6, from_step, only_step)
|
||
|
||
if run_cluster and run_wf:
|
||
logger.info("步骤 5–6/7：聚类与词频（并行）")
|
||
with ThreadPoolExecutor(max_workers=2) as pool:
|
||
f_cluster = pool.submit(
|
||
run_clustering,
|
||
job_id=job_id,
|
||
structured_db=STRUCTURED_DB,
|
||
embed_db=EMBED_DB,
|
||
cluster_db=CLUSTER_DB,
|
||
reset_db=clean_databases,
|
||
)
|
||
f_wf = pool.submit(run_wordfreq, skip_llm=skip_wordfreq_llm)
|
||
result["clustering"] = f_cluster.result()
|
||
result["wordfreq"] = f_wf.result()
|
||
else:
|
||
if run_cluster:
|
||
logger.info("步骤 5/7：聚类")
|
||
result["clustering"] = run_clustering(
|
||
job_id=job_id,
|
||
structured_db=STRUCTURED_DB,
|
||
embed_db=EMBED_DB,
|
||
cluster_db=CLUSTER_DB,
|
||
reset_db=clean_databases,
|
||
)
|
||
if run_wf:
|
||
logger.info("步骤 6/7：词频")
|
||
result["wordfreq"] = run_wordfreq(skip_llm=skip_wordfreq_llm)
|
||
|
||
if _should_run(7, from_step, only_step):
|
||
logger.info("步骤 7/7：生成分析报告")
|
||
if not CLUSTER_DB.is_file():
|
||
raise FileNotFoundError(f"缺少 {CLUSTER_DB}")
|
||
if not WORD_FREQ_CSV.is_file():
|
||
raise FileNotFoundError(f"缺少 {WORD_FREQ_CSV}")
|
||
if not cleaned_csv.is_file():
|
||
raise FileNotFoundError(f"缺少 {cleaned_csv}")
|
||
if STRUCTURED_DB.is_file():
|
||
_, ind, prod = _latest_job_meta(STRUCTURED_DB)
|
||
report_html = _report_html_path(prod)
|
||
result["report_html"] = str(report_html)
|
||
result["report"] = generate_report(
|
||
product_name=prod,
|
||
industry=ind,
|
||
cleaned_csv=cleaned_csv,
|
||
cluster_db=CLUSTER_DB,
|
||
embed_db=EMBED_DB,
|
||
word_freq_csv=WORD_FREQ_CSV,
|
||
output_html=report_html,
|
||
structured_db=STRUCTURED_DB,
|
||
min_cluster_review_ratio=min_cluster_review_ratio,
|
||
save_llm_raw=save_llm_raw,
|
||
)
|
||
|
||
if clean_intermediates and _should_run(7, from_step, only_step):
|
||
_clean_intermediates(
|
||
merged=merged_csv,
|
||
cleaned=cleaned_csv,
|
||
structured=STRUCTURED_DB,
|
||
embed=EMBED_DB,
|
||
cluster=CLUSTER_DB,
|
||
terms=TERMS_JSON,
|
||
word_freq=WORD_FREQ_CSV,
|
||
)
|
||
|
||
return result
|
||
|
||
|
||
# 兼容旧名
|
||
run_pipeline = run_voc_analysis
|
||
|
||
|
||
def _resolve_industry_product(
|
||
*,
|
||
industry: str | None,
|
||
product_name: str | None,
|
||
from_step: int,
|
||
only_step: int | None,
|
||
) -> tuple[str, str]:
|
||
"""步骤 4 起可从 structured 库自动读取 product；industry 默认 Pet Supplies。"""
|
||
ind = (industry or DEFAULT_INDUSTRY).strip() or DEFAULT_INDUSTRY
|
||
if product_name:
|
||
return ind, product_name
|
||
start_step = only_step if only_step is not None else from_step
|
||
if start_step >= 4:
|
||
if not STRUCTURED_DB.is_file():
|
||
raise SystemExit(
|
||
f"缺少 {STRUCTURED_DB.name}，无法自动读取 --product"
|
||
)
|
||
_, db_ind, prod = _latest_job_meta(STRUCTURED_DB)
|
||
logger.info("未指定 product，已从 structured 库读取: %r（industry=%r）", prod, db_ind)
|
||
return db_ind, prod
|
||
raise SystemExit(
|
||
f"步骤 1–3 需要 --product；--industry 可省略（默认 {DEFAULT_INDUSTRY}）；"
|
||
"从步骤 4 起 product 也可省略（自动读 voc_structured.sqlite）"
|
||
)
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="VOC 分析全流程")
|
||
parser.add_argument(
|
||
"--input-dir",
|
||
type=Path,
|
||
default=None,
|
||
help="原始 VOC CSV 目录（步骤 1 合并输入；续跑后续步骤可不传）",
|
||
)
|
||
parser.add_argument(
|
||
"--industry",
|
||
default=DEFAULT_INDUSTRY,
|
||
help=f"行业（默认 {DEFAULT_INDUSTRY}）",
|
||
)
|
||
parser.add_argument("--product", default=None, help="产品名（步骤 4 起可省略，自动读库）")
|
||
parser.add_argument(
|
||
"--merged-csv",
|
||
type=Path,
|
||
default=DEFAULT_MERGED,
|
||
help=f"合并输出（默认 {DEFAULT_MERGED.name}）",
|
||
)
|
||
parser.add_argument(
|
||
"--cleaned-csv",
|
||
type=Path,
|
||
default=DEFAULT_CLEANED,
|
||
help=f"清洗输出（默认 {DEFAULT_CLEANED.name}）",
|
||
)
|
||
parser.add_argument(
|
||
"--from-step",
|
||
type=int,
|
||
default=1,
|
||
choices=range(1, 8),
|
||
metavar="N",
|
||
help="从第 N 步开始执行（1–7，默认 1）",
|
||
)
|
||
parser.add_argument(
|
||
"--only-step",
|
||
type=int,
|
||
default=None,
|
||
choices=range(1, 8),
|
||
metavar="N",
|
||
help="仅执行第 N 步（需已有前置产物）",
|
||
)
|
||
parser.add_argument(
|
||
"--keep-db",
|
||
action="store_true",
|
||
help="保留已有 voc_structured / voc_embeddings / voc_clustering.sqlite，不覆盖清理（默认写入前会清理）",
|
||
)
|
||
parser.add_argument(
|
||
"--clean-intermediates",
|
||
action="store_true",
|
||
help="报告生成成功后删除中间 sqlite/csv（仅保留各产品子目录下的报告 HTML）",
|
||
)
|
||
parser.add_argument(
|
||
"--skip-wordfreq-llm",
|
||
action="store_true",
|
||
help="词频步骤复用已有 voc_terms.json",
|
||
)
|
||
parser.add_argument(
|
||
"--filter-small-clusters",
|
||
action="store_true",
|
||
help=(
|
||
"报告仅纳入本 stage 内去重评论占比 ≥ "
|
||
f"{DEFAULT_CLUSTER_MIN_REVIEW_RATIO * 100:.0f}%% 的簇；默认不筛选、全部簇参与"
|
||
),
|
||
)
|
||
parser.add_argument(
|
||
"--save-llm-raw",
|
||
action="store_true",
|
||
help="步骤 7 将报告 LLM 完整原文写入报告同目录下的 report_llm_raw.txt（默认不保存）",
|
||
)
|
||
parser.add_argument(
|
||
"--struct-workers",
|
||
type=int,
|
||
default=None,
|
||
metavar="N",
|
||
help=(
|
||
f"步骤 3 结构化批间并行数（默认 {STRUCT_DEFAULT_WORKERS}；"
|
||
"环境变量 VOC_STRUCT_WORKERS）"
|
||
),
|
||
)
|
||
parser.add_argument(
|
||
"--embed-workers",
|
||
type=int,
|
||
default=None,
|
||
metavar="N",
|
||
help=(
|
||
f"步骤 4 向量化批间并行数（默认 {EMBED_DEFAULT_WORKERS}；"
|
||
"环境变量 VOC_EMBED_WORKERS）"
|
||
),
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
need_input = args.only_step in (None, 1) and args.from_step <= 1
|
||
if need_input:
|
||
if not args.input_dir:
|
||
raise SystemExit("步骤 1 需要 --input-dir")
|
||
if not args.input_dir.is_dir():
|
||
raise SystemExit(f"输入目录不存在: {args.input_dir}")
|
||
input_dir = (args.input_dir or DEFAULT_MERGED.parent).resolve()
|
||
industry, product_name = _resolve_industry_product(
|
||
industry=args.industry,
|
||
product_name=args.product,
|
||
from_step=args.from_step,
|
||
only_step=args.only_step,
|
||
)
|
||
|
||
out = run_voc_analysis(
|
||
input_dir=input_dir,
|
||
industry=industry,
|
||
product_name=product_name,
|
||
merged_csv=args.merged_csv.resolve(),
|
||
cleaned_csv=args.cleaned_csv.resolve(),
|
||
from_step=args.from_step,
|
||
only_step=args.only_step,
|
||
clean_intermediates=args.clean_intermediates,
|
||
clean_databases=not args.keep_db,
|
||
skip_wordfreq_llm=args.skip_wordfreq_llm,
|
||
min_cluster_review_ratio=(
|
||
DEFAULT_CLUSTER_MIN_REVIEW_RATIO
|
||
if args.filter_small_clusters
|
||
else None
|
||
),
|
||
save_llm_raw=args.save_llm_raw,
|
||
struct_workers=args.struct_workers,
|
||
embed_workers=args.embed_workers,
|
||
)
|
||
print(json.dumps(out, ensure_ascii=False, indent=2))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|