#!/usr/bin/env python3 """Build an auditable WeDrive source universe for digital-business research.""" from __future__ import annotations import argparse import csv import hashlib import io import subprocess from collections import Counter, defaultdict from pathlib import Path DEFAULT_INDEX = Path("work_company_knowledge/indexes/document-index.tsv") RELEVANT_AREAS = { "smart-canteen", "nutrition", "food-safety", "inventory", "security-compliance", } DOMAIN_KEYWORDS = ( "智慧食堂", "智慧餐厅", "食堂", "餐厅", "餐饮", "团餐", "营养餐", "膳食", "食安", "食品安全", "餐台", "称重", "结算", "餐补", "补贴", "订餐", "消费机", "进销存", "后厨", "菜品", "留样", "晨检", "验收秤", "餐盘", "就餐", "供餐", ) SEGMENT_RULES = ( ("体育系统/训练基地", ("体科所", "运动员", "训练基地", "体育训练", "体校", "科训")), ("学校/教委/校园餐", ("学校", "校园", "教委", "教育局", "大学", "学院", "中学", "小学")), ("医院/康养员工餐", ("医院", "医疗", "康养", "康复辅具")), ("政府/机关/央国企", ("政府", "机关", "国信", "网信办", "央企", "国企", "油田", "炼油厂", "烟草", "银行")), ("团餐运营商/渠道", ("团餐", "餐饮公司", "渠道", "运营商")), ("企业园区/集团食堂", ("集团", "园区", "员工餐厅", "职工食堂", "职工餐厅", "大厦")), ) TOPIC_RULES = ( ("商业金额/报价", ("预算", "报价", "费用", "金额", "合同", "中标", "经费", "成本", "收入")), ("采购/招投标", ("采购", "招标", "投标", "竞争性", "需求", "磋商")), ("验收/交付/运维", ("验收", "交付", "实施", "运维", "上线", "测试", "售后")), ("财务/账户/结算", ("财务", "对账", "结算", "补贴", "充值", "退款", "销户", "支付")), ("客户调研/会议", ("会议纪要", "沟通会", "需求会", "调研", "问题汇总", "工作规划")), ("竞品/替代方案", ("竞品", "对比", "友商", "满客宝", "乐牛", "优信", "智盘", "雄伟")), ("食品安全", ("食安", "食品安全", "晨检", "留样", "明厨亮灶", "溯源")), ("营养健康", ("营养", "健康", "膳食", "食谱", "菜谱")), ("产品/功能/设备", ("产品", "方案", "手册", "设备", "功能", "PRD", "系统")), ("安全/信创/接口", ("等保", "密评", "信创", "国产化", "接口", "数据安全", "隐私")), ) def read_tsv(path: Path) -> list[dict[str, str]]: with path.open("r", newline="", encoding="utf-8-sig") as handle: return list(csv.DictReader(handle, delimiter="\t")) def read_git_tsv(ref: str, path: Path) -> list[dict[str, str]]: try: raw = subprocess.check_output(["git", "show", f"{ref}:{path.as_posix()}"]) except subprocess.CalledProcessError: return [] return list(csv.DictReader(io.StringIO(raw.decode("utf-8-sig")), delimiter="\t")) def matches(text: str, keywords: tuple[str, ...]) -> bool: return any(keyword.lower() in text.lower() for keyword in keywords) def classify_many(text: str, rules: tuple[tuple[str, tuple[str, ...]], ...], fallback: str) -> list[str]: labels = [label for label, keywords in rules if matches(text, keywords)] return labels or [fallback] def candidate_id(abs_path: str) -> str: return "MRW-" + hashlib.sha1(abs_path.encode("utf-8")).hexdigest()[:10].upper() def relevance_score( row: dict[str, str], text: str, topics: list[str], segments: list[str], direct_match: bool ) -> int: score = 0 if direct_match and row["knowledge_area"] in RELEVANT_AREAS: score += 3 if direct_match and matches(text, DOMAIN_KEYWORDS): score += 3 if not direct_match: score += 1 if any(topic in topics for topic in ("商业金额/报价", "采购/招投标")): score += 3 if "验收/交付/运维" in topics: score += 2 if "客户调研/会议" in topics: score += 2 if segments != ["通用产品/未识别场景"]: score += 1 if row.get("year_bucket") in {"2025", "2026"}: score += 1 return min(score, 10) def write_csv(path: Path, fieldnames: list[str], rows: list[dict[str, object]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", newline="", encoding="utf-8-sig") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def build_universe(rows: list[dict[str, str]]) -> tuple[list[dict[str, object]], Counter[str]]: universe: list[dict[str, object]] = [] exclusions: Counter[str] = Counter() direct_matches: dict[str, bool] = {} direct_project_counts: Counter[tuple[str, str]] = Counter() for row in rows: text = " ".join((row.get("project_or_topic", ""), row.get("rel_path", ""), row.get("name", ""))) direct = row.get("knowledge_area") in RELEVANT_AREAS or matches(text, DOMAIN_KEYWORDS) direct_matches[row["abs_path"]] = direct if direct: direct_project_counts[(row["source"], row["project_or_topic"])] += 1 relevant_delivery_projects = { key for key, count in direct_project_counts.items() if key[0] == "delivery" and (count >= 2 or matches(key[1], DOMAIN_KEYWORDS)) } for row in rows: text = " ".join((row.get("project_or_topic", ""), row.get("rel_path", ""), row.get("name", ""))) direct = direct_matches[row["abs_path"]] project_context = (row["source"], row["project_or_topic"]) in relevant_delivery_projects relevant = direct or project_context if not relevant: exclusions["非餐饮/营养/食安研究域"] += 1 continue segments = classify_many(text, SEGMENT_RULES, "通用产品/未识别场景") topics = classify_many(text, TOPIC_RULES, "未识别证据主题") score = relevance_score(row, text, topics, segments, direct) universe.append( { "candidate_id": candidate_id(row["abs_path"]), "source": row["source"], "project_or_topic": row["project_or_topic"], "year_bucket": row["year_bucket"], "stage": row["stage"], "role_lane": row["role_lane"], "knowledge_area": row["knowledge_area"], "inclusion_basis": "direct_keyword_or_area" if direct else "relevant_project_context", "market_segments": " | ".join(segments), "evidence_topics": " | ".join(topics), "relevance_score": score, "deep_read_priority": "P0" if score >= 8 else "P1" if score >= 5 else "P2", "mtime": row["mtime"], "name": row["name"], "rel_path": row["rel_path"], "abs_path": row["abs_path"], } ) universe.sort(key=lambda item: (-int(item["relevance_score"]), str(item["source"]), str(item["rel_path"]))) return universe, exclusions def build_project_coverage(universe: list[dict[str, object]]) -> list[dict[str, object]]: grouped: dict[tuple[str, str, str], list[dict[str, object]]] = defaultdict(list) for row in universe: grouped[(str(row["source"]), str(row["project_or_topic"]), str(row["year_bucket"]))].append(row) output: list[dict[str, object]] = [] for (source, project, year), items in grouped.items(): segments = Counter() topics = Counter() for item in items: segments.update(str(item["market_segments"]).split(" | ")) topics.update(str(item["evidence_topics"]).split(" | ")) output.append( { "source": source, "project_or_topic": project, "year_bucket": year, "candidate_documents": len(items), "p0_deep_read_documents": sum(item["deep_read_priority"] == "P0" for item in items), "top_segments": " | ".join(label for label, _ in segments.most_common(3)), "top_evidence_topics": " | ".join(label for label, _ in topics.most_common(5)), "newest_mtime": max(str(item["mtime"]) for item in items), "sample_path": items[0]["abs_path"], } ) output.sort(key=lambda item: (-int(item["p0_deep_read_documents"]), -int(item["candidate_documents"]), str(item["project_or_topic"]))) return output def build_refresh_audit(current: list[dict[str, str]], previous: list[dict[str, str]]) -> list[dict[str, object]]: current_by_path = {row["abs_path"]: row for row in current} previous_by_path = {row["abs_path"]: row for row in previous} current_paths = set(current_by_path) previous_paths = set(previous_by_path) changed = sum(current_by_path[path] != previous_by_path[path] for path in current_paths & previous_paths) return [ {"metric": "previous_documents", "value": len(previous), "definition": "baseline Git ref document rows"}, {"metric": "current_documents", "value": len(current), "definition": "current WeDrive document rows"}, {"metric": "added_documents", "value": len(current_paths - previous_paths), "definition": "new absolute paths"}, {"metric": "removed_documents", "value": len(previous_paths - current_paths), "definition": "paths absent from current scan"}, {"metric": "changed_documents", "value": changed, "definition": "same path with metadata/classification change"}, ] def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--index", type=Path, default=DEFAULT_INDEX) parser.add_argument("--baseline-ref", default="HEAD") parser.add_argument("--output-dir", type=Path, required=True) args = parser.parse_args() current = read_tsv(args.index) previous = read_git_tsv(args.baseline_ref, args.index) universe, exclusions = build_universe(current) coverage = build_project_coverage(universe) refresh = build_refresh_audit(current, previous) write_csv( args.output_dir / "wedrive-research-universe.csv", [ "candidate_id", "source", "project_or_topic", "year_bucket", "stage", "role_lane", "knowledge_area", "inclusion_basis", "market_segments", "evidence_topics", "relevance_score", "deep_read_priority", "mtime", "name", "rel_path", "abs_path", ], universe, ) write_csv( args.output_dir / "wedrive-project-coverage.csv", [ "source", "project_or_topic", "year_bucket", "candidate_documents", "p0_deep_read_documents", "top_segments", "top_evidence_topics", "newest_mtime", "sample_path", ], coverage, ) write_csv(args.output_dir / "wedrive-source-refresh-audit.csv", ["metric", "value", "definition"], refresh) filter_rows = [ {"classification": "all_documents", "document_count": len(current)}, {"classification": "research_candidates", "document_count": len(universe)}, {"classification": "p0_deep_read", "document_count": sum(row["deep_read_priority"] == "P0" for row in universe)}, {"classification": "p1_deep_read", "document_count": sum(row["deep_read_priority"] == "P1" for row in universe)}, {"classification": "p2_reference", "document_count": sum(row["deep_read_priority"] == "P2" for row in universe)}, ] filter_rows.extend( {"classification": f"excluded:{reason}", "document_count": count} for reason, count in exclusions.items() ) write_csv(args.output_dir / "wedrive-filter-audit.csv", ["classification", "document_count"], filter_rows) print( f"documents={len(current)} candidates={len(universe)} " f"p0={sum(row['deep_read_priority'] == 'P0' for row in universe)} projects={len(coverage)}" ) return 0 if __name__ == "__main__": raise SystemExit(main())