#!/usr/bin/env python3 from __future__ import annotations import argparse import csv import difflib import json import re import subprocess from dataclasses import dataclass from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo ORG_ID = "60069db88deaa14d9e02b875" PROFILE = "ai-nutrition-audit" SPACE_TYPE = "Project" CATEGORIES = ("Req", "Task") TZ = ZoneInfo("Asia/Shanghai") @dataclass class WedriveProject: source: str top_dir: str project_or_topic: str year_bucket: str file_count: str document_count: str size_mb: str sample_path: str folder_path: str normalized: str def run_aliyun(api: str, args: list[str]) -> dict: cmd = ["aliyun", "devops", api, "--profile", PROFILE, "--organizationId", ORG_ID, *args] proc = subprocess.run(cmd, text=True, capture_output=True, check=False) if proc.returncode != 0: raise RuntimeError(f"{api} failed: {proc.stderr.strip() or proc.stdout.strip()}") return json.loads(proc.stdout) def ms_to_local(ms: object) -> str: try: val = int(ms) except (TypeError, ValueError): return "" return datetime.fromtimestamp(val / 1000, TZ).strftime("%Y-%m-%d %H:%M:%S %z") def normalize_name(value: str) -> str: value = value or "" value = re.sub(r"【\d{4}】", "", value) value = re.sub(r"\d{4}年?", "", value) value = re.sub(r"[（）()\[\]【】\-_/·:：,，.。+&\s]", "", value) for word in ("自研", "项目", "系统", "标准版", "专业版", "智慧", "管理", "平台", "软件", "相关"): value = value.replace(word, "") return value.lower() def folder_from_sample(sample_path: str) -> str: p = Path(sample_path) if p.name in {".DS_Store", ".WeDrive"}: return str(p.parent) if p.suffix: return str(p.parent) return str(p) def load_wedrive_projects(path: Path) -> list[WedriveProject]: rows: list[WedriveProject] = [] with path.open(newline="", encoding="utf-8") as f: reader = csv.DictReader(f, delimiter="\t") for row in reader: project = row.get("project_or_topic", "") if not project or project == ".": continue sample = row.get("sample_path", "") rows.append( WedriveProject( source=row.get("source", ""), top_dir=row.get("top_dir", ""), project_or_topic=project, year_bucket=row.get("year_bucket", ""), file_count=row.get("file_count", ""), document_count=row.get("document_count", ""), size_mb=row.get("size_mb", ""), sample_path=sample, folder_path=folder_from_sample(sample), normalized=normalize_name(project), ) ) return rows def score_match(name: str, candidate: WedriveProject) -> float: n = normalize_name(name) c = candidate.normalized if not n or not c: return 0 if n == c: score = 100 elif n in c or c in n: score = 86 + min(len(n), len(c)) / max(len(n), len(c)) else: ratio = difflib.SequenceMatcher(None, n, c).ratio() * 80 common = set(n) & set(c) overlap = len(common) / max(len(set(n)), 1) * 45 score = max(ratio, overlap) # Very short normalized names such as "食安" or "进销存" are topic-level hints, # not enough evidence to bind a Yunxiao project to a specific WeDrive folder. if min(len(n), len(c)) < 5: score = min(score, 60) return score def list_projects() -> list[dict]: data = run_aliyun( "ListProjects", ["--category", "Project", "--maxResults", "100"], ) return data.get("projects", []) def list_workitems(project_id: str, category: str) -> list[dict]: items: list[dict] = [] next_token = "" while True: args = [ "--spaceIdentifier", project_id, "--spaceType", SPACE_TYPE, "--category", category, "--maxResults", "100", ] if next_token: args.extend(["--nextToken", next_token]) data = run_aliyun("ListWorkitems", args) items.extend(data.get("workitems", [])) next_token = data.get("nextToken") or "" if not next_token: break return items def write_csv(path: Path, fieldnames: list[str], rows: list[dict]) -> None: with path.open("w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--out", default="work/2026-06-01-yunxiao-wedrive-traceability") parser.add_argument("--wedrive-index", default="work_company_knowledge/indexes/project-index.tsv") parser.add_argument("--limit-projects", type=int, default=0) args = parser.parse_args() out = Path(args.out) raw = out / "raw" raw.mkdir(parents=True, exist_ok=True) wedrive = load_wedrive_projects(Path(args.wedrive_index)) projects = list_projects() if args.limit_projects: projects = projects[: args.limit_projects] (raw / "yunxiao-projects.json").write_text( json.dumps(projects, ensure_ascii=False, indent=2), encoding="utf-8" ) project_rows: list[dict] = [] workitem_rows: list[dict] = [] mapping_rows: list[dict] = [] candidate_rows: list[dict] = [] workitem_raw: dict[str, dict[str, list[dict]]] = {} for project in projects: project_id = project.get("identifier", "") custom_code = project.get("customCode", "") project_name = project.get("name", "") project_rows.append( { "yunxiao_project_id": project_id, "custom_code": custom_code, "project_name": project_name, "scope": project.get("scope", ""), "gmt_create": ms_to_local(project.get("gmtCreate")), "description": project.get("description", ""), } ) ranked = sorted( ((score_match(project_name, w), w) for w in wedrive), key=lambda item: item[0], reverse=True, )[:5] best_score, best = ranked[0] if ranked else (0, None) for rank, (score, w) in enumerate(ranked, 1): if score < 25: continue candidate_rows.append( { "yunxiao_project_id": project_id, "custom_code": custom_code, "project_name": project_name, "candidate_rank": rank, "match_score": f"{score:.1f}", "wedrive_project_or_topic": w.project_or_topic, "wedrive_source": w.source, "wedrive_top_dir": w.top_dir, "wedrive_folder_path": w.folder_path, "wedrive_file_count": w.file_count, "wedrive_document_count": w.document_count, "wedrive_size_mb": w.size_mb, "match_status": "auto_candidate" if score < 86 else "strong_candidate", } ) mapping_rows.append( { "yunxiao_project_id": project_id, "custom_code": custom_code, "project_name": project_name, "match_score": f"{best_score:.1f}" if best else "0", "match_status": "strong" if best and best_score >= 86 else ("weak_review_needed" if best and best_score >= 45 else "unmatched"), "wedrive_project_or_topic": best.project_or_topic if best and best_score >= 45 else "", "wedrive_source": best.source if best and best_score >= 45 else "", "wedrive_folder_path": best.folder_path if best and best_score >= 45 else "", "wedrive_file_count": best.file_count if best and best_score >= 45 else "", "wedrive_document_count": best.document_count if best and best_score >= 45 else "", "wedrive_size_mb": best.size_mb if best and best_score >= 45 else "", } ) workitem_raw[project_id] = {} for category in CATEGORIES: try: items = list_workitems(project_id, category) except RuntimeError as exc: workitem_raw[project_id][category] = [{"error": str(exc)}] continue workitem_raw[project_id][category] = items for item in items: workitem_rows.append( { "yunxiao_project_id": project_id, "custom_code": custom_code, "project_name": project_name, "category": category, "serial_number": item.get("serialNumber", ""), "workitem_id": item.get("identifier", ""), "subject": item.get("subject", ""), "status": item.get("status", ""), "status_id": item.get("statusIdentifier", ""), "workitem_type_id": item.get("workitemTypeIdentifier", ""), "assigned_to": item.get("assignedTo", ""), "parent_id": item.get("parentIdentifier", ""), "logical_status": item.get("logicalStatus", ""), "gmt_create": ms_to_local(item.get("gmtCreate")), "gmt_modified": ms_to_local(item.get("gmtModified")), "wedrive_match_status": mapping_rows[-1]["match_status"], "wedrive_folder_path": mapping_rows[-1]["wedrive_folder_path"], } ) (raw / "yunxiao-workitems-by-project.json").write_text( json.dumps(workitem_raw, ensure_ascii=False, indent=2), encoding="utf-8" ) write_csv( out / "yunxiao-projects.csv", ["yunxiao_project_id", "custom_code", "project_name", "scope", "gmt_create", "description"], project_rows, ) write_csv( out / "yunxiao-workitems.csv", [ "yunxiao_project_id", "custom_code", "project_name", "category", "serial_number", "workitem_id", "subject", "status", "status_id", "workitem_type_id", "assigned_to", "parent_id", "logical_status", "gmt_create", "gmt_modified", "wedrive_match_status", "wedrive_folder_path", ], workitem_rows, ) write_csv( out / "yunxiao-wedrive-project-map.csv", [ "yunxiao_project_id", "custom_code", "project_name", "match_score", "match_status", "wedrive_project_or_topic", "wedrive_source", "wedrive_folder_path", "wedrive_file_count", "wedrive_document_count", "wedrive_size_mb", ], mapping_rows, ) write_csv( out / "yunxiao-wedrive-match-candidates.csv", [ "yunxiao_project_id", "custom_code", "project_name", "candidate_rank", "match_score", "wedrive_project_or_topic", "wedrive_source", "wedrive_top_dir", "wedrive_folder_path", "wedrive_file_count", "wedrive_document_count", "wedrive_size_mb", "match_status", ], candidate_rows, ) summary = { "generated_at": datetime.now(TZ).strftime("%Y-%m-%d %H:%M:%S %z"), "yunxiao_project_count": len(project_rows), "yunxiao_workitem_count": len(workitem_rows), "requirement_count": sum(1 for r in workitem_rows if r["category"] == "Req"), "task_count": sum(1 for r in workitem_rows if r["category"] == "Task"), "strong_project_matches": sum(1 for r in mapping_rows if r["match_status"] == "strong"), "weak_project_matches": sum(1 for r in mapping_rows if r["match_status"] == "weak_review_needed"), "unmatched_projects": sum(1 for r in mapping_rows if r["match_status"] == "unmatched"), } (out / "run-summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(summary, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()