#!/usr/bin/env python3 """ TFS 2013 工作项批量导入工具 目标环境（来自截图）: 服务器 : http://pdtsf01.dev.goldenstand.cn:8080/tfs/DefaultCollection 团队项目 : Tools_Dev 团队 : DataTechnologies 迭代路径 : 2026年Sprint\\2026年Sprint10 (URL: .../_backlogs/Iteration/2026%E5%B9%B4Sprint/2026%E5%B9%B4Sprint10) 数据源: data/work_items.csv（Title/State/AssignedTo/FinancialProject/Completed/LogDate/CreatedDate/ClosedDate） 字段映射: config.json -> field_refs；留空的自定义字段按显示名自动解析。 用法: python tfs_import.py --dry-run # 只校验数据与字段，不写入 python tfs_import.py # 正式导入 python tfs_import.py --link-existing # 已导入的项补挂父链接(不新建) python tfs_import.py --input other.csv # 指定数据文件 python tfs_import.py --skip-existing-titles # 跳过已存在同标题工作项 python tfs_import.py --list-fields # 列出服务器全部字段(显示名/引用名) 认证(config.json auth.mode): ntlm : 域\\用户 + 密码（TFS 2013 典型配置） basic : 服务器开了基本身份验证时使用 默认凭据: username 留空则尝试当前 Windows 凭据(仅 basic/ntlm 均需显式账号，TFS 2013 一般用 ntlm) """ from __future__ import annotations import argparse import csv import json import sys import time from datetime import datetime from pathlib import Path try: import requests except ImportError: print("缺少依赖 requests，请先: pip install -r requirements.txt", file=sys.stderr) sys.exit(2) BASE_DIR = Path(__file__).resolve().parent API_VER = "1.0" DATE_FMT_IN = ["%Y-%m-%d %H:%M:%S", "%Y/%m/%d %H:%M:%S", "%Y-%m-%d", "%Y/%m/%d", "%Y-%m-%d %H:%M"] # ---------------------------------------------------------------- 配置与数据 def load_config(path: Path) -> dict: with path.open("r", encoding="utf-8") as f: return json.load(f) def load_rows(path: Path) -> list[dict]: # 依次尝试 utf-8-sig / gbk，兼容新旧 CSV 编码 rows: list[dict] | None = None last_err: Exception | None = None for enc in ("utf-8-sig", "gbk"): try: with path.open("r", encoding=enc, newline="") as f: reader = csv.DictReader(f) rows = [r for r in reader if (r.get("Title") or "").strip()] break except UnicodeDecodeError as e: last_err = e if rows is None: raise SystemExit(f"无法读取 CSV {path}（utf-8/gbk 均失败）: {last_err}") return rows def parse_dt(value: str) -> str | None: """把 CSV 里的时间统一成 TFS 接受的 ISO 格式；空值返回 None。""" value = (value or "").strip() if not value: return None for fmt in DATE_FMT_IN: try: return datetime.strptime(value, fmt).strftime("%Y-%m-%dT%H:%M:%S") except ValueError: continue raise ValueError(f"无法解析时间: {value!r}") # ---------------------------------------------------------------- HTTP 会话 def build_session(cfg: dict) -> requests.Session: auth = cfg.get("auth", {}) mode = (auth.get("mode") or "ntlm").lower() sess = requests.Session() sess.verify = False sess.headers.update({"Accept": "application/json"}) user = auth.get("username") or "" pwd = auth.get("password") or "" domain = auth.get("domain") or "" if mode == "ntlm": from requests_ntlm import HttpNtlmAuth if not user or not pwd: raise SystemExit("ntlm 模式需要 config.json 里配置 auth.username / auth.password") login = f"{domain}\\{user}" if domain else user sess.auth = HttpNtlmAuth(login, pwd) elif mode == "basic": if not user or not pwd: raise SystemExit("basic 模式需要 config.json 里配置 auth.username / auth.password") sess.auth = (f"{domain}\\{user}" if domain else user, pwd) else: raise SystemExit(f"未知 auth.mode: {mode}") return sess def tfs_send(sess: requests.Session, method: str, url: str, *, retries: int = 3, **kwargs) -> requests.Response: """带退避重试的请求：502/503/504 网关错误或网络异常时最多重试 retries 次。""" kwargs.setdefault("timeout", 60) resp: requests.Response | None = None for attempt in range(1, retries + 1): try: resp = sess.request(method, url, **kwargs) except requests.RequestException as e: if attempt == retries: raise wait = 3 * attempt print(f"[重试] {method.upper()} {url} 网络异常: {e}；{wait}s 后第 {attempt + 1} 次尝试") time.sleep(wait) continue if resp.status_code in (502, 503, 504) and attempt < retries: wait = 3 * attempt print(f"[重试] {method.upper()} {url} -> {resp.status_code}；{wait}s 后第 {attempt + 1} 次尝试") time.sleep(wait) continue return resp raise RuntimeError(f"请求重试 {retries} 次后仍失败: {method.upper()} {url}") def tfs_get(sess: requests.Session, url: str, params: dict | None = None) -> dict: resp = tfs_send(sess, "get", url, params=params) if resp.status_code >= 400: raise RuntimeError(f"GET {url} -> {resp.status_code}: {resp.text[:500]}") return resp.json() def tfs_wiql(sess: requests.Session, url: str, wiql: str) -> dict: resp = tfs_send(sess, "post", url, params={"api-version": API_VER}, json={"query": wiql}) if resp.status_code >= 400: raise RuntimeError(f"WIQL 查询失败 {resp.status_code}: {resp.text[:500]}") return resp.json() # ---------------------------------------------------------------- 字段解析 def resolve_fields(sess: requests.Session, base: str, cfg: dict) -> dict: """返回 {语义名: 引用名}。优先用 config.field_refs 显式值，其次按显示名查服务器字段表。""" refs = dict(cfg.get("field_refs") or {}) by_display = cfg.get("resolve_fields_by_display_name") or {} need_lookup = [k for k, v in by_display.items() if not refs.get(k)] if not need_lookup: return refs data = tfs_get(sess, f"{base}/_apis/wit/fields", params={"api-version": API_VER}) catalog = data.get("value", []) lower_map = {} for f in catalog: lower_map.setdefault(f.get("name", "").strip().lower(), f.get("referenceName")) for key in need_lookup: display = by_display.get(key) or key ref = lower_map.get(display.strip().lower()) if not ref: # 宽松匹配: 显示名以配置词开头(如 Completed -> Completed Work) for f in catalog: name = f.get("name", "").strip().lower() if name.startswith(display.strip().lower()): ref = f.get("referenceName") print(f"[字段] {key}: {display!r} 模糊匹配到 {f.get('name')!r} -> {ref}") break if not ref: avail = ", ".join(sorted({f.get("name", "") for f in catalog if f.get("name")}))[:800] raise SystemExit( f"字段 {key} 的显示名 {display!r} 在服务器上找不到，请在 config.json 的 " f"field_refs 里显式配置引用名。\n服务器字段(截断): {avail}" ) refs[key] = ref print(f"[字段] {key}: {display!r} -> {ref}") return refs # ---------------------------------------------------------------- 重复检测 def existing_titles(sess: requests.Session, base: str, project: str, iteration: str, type_name: str) -> set[str]: wiql = ( "SELECT [System.Title] FROM WorkItems " f"WHERE [System.TeamProject] = '{project}' " f"AND [System.WorkItemType] = '{type_name}' " f"AND [System.IterationPath] = '{iteration}'" ) data = tfs_wiql(sess, f"{base}/{project}/_apis/wit/wiql", wiql) ids = [w["id"] for w in data.get("workItems", [])] titles: set[str] = set() for i in range(0, len(ids), 50): chunk = ids[i:i + 50] data = tfs_get( sess, f"{base}/_apis/wit/workitems", params={"ids": ",".join(map(str, chunk)), "fields": "System.Title", "api-version": API_VER}, ) for item in data.get("value", []): titles.add(item.get("fields", {}).get("System.Title", "")) return titles # ---------------------------------------------------------------- 构造与提交 def build_patch(row: dict, refs: dict, cfg: dict, iteration_full: str, area: str) -> list[dict]: ops: list[dict] = [] def add(ref: str | None, value): if not ref or value is None or value == "": return ops.append({"op": "add", "path": f"/fields/{ref}", "value": value}) add(refs["title"], row["Title"].strip()) add(refs["state"], (row.get("State") or cfg.get("default_state") or "To Do").strip()) add(refs["assigned_to"], (row.get("AssignedTo") or "").strip()) add(refs["iteration_path"], iteration_full) if area: add(refs["area_path"], area) add(refs.get("financial_project"), (row.get("FinancialProject") or "").strip()) completed = (row.get("Completed") or "").strip() if completed: add(refs.get("completed"), float(completed) if "." in completed else int(completed)) log_date = parse_dt(row.get("LogDate")) if log_date: add(refs.get("log_date"), log_date) # CreatedDate / ClosedDate 是系统字段，常规接口不可写；bypassRules=true 时部分模板可写。 created = parse_dt(row.get("CreatedDate")) if created: add(refs.get("created_date"), created) closed = parse_dt(row.get("ClosedDate")) if closed: add(refs.get("closed_date"), closed) return ops def create_work_item(sess: requests.Session, base: str, project: str, type_name: str, ops: list[dict], bypass_rules: bool) -> dict: url = f"{base}/{project}/_apis/wit/workitems/${requests.utils.quote(type_name)}" params = {"api-version": API_VER} if bypass_rules: params["bypassRules"] = "true" headers = {"Content-Type": "application/json-patch+json"} resp = tfs_send(sess, "patch", url, params=params, headers=headers, data=json.dumps(ops, ensure_ascii=False)) if resp.status_code >= 400: raise RuntimeError(f"PATCH {url} -> {resp.status_code}: {resp.text[:800]}") return resp.json() # ---------------------------------------------------------------- 父项挂接 def query_work_item_ids(sess: requests.Session, base: str, project: str, type_name: str, iteration: str) -> list[int]: wiql = ( "SELECT [System.Id] FROM WorkItems " f"WHERE [System.TeamProject] = '{project}' " f"AND [System.WorkItemType] = '{type_name}' " f"AND [System.IterationPath] = '{iteration}'" ) data = tfs_wiql(sess, f"{base}/{project}/_apis/wit/wiql", wiql) return [w["id"] for w in data.get("workItems", [])] def find_parent_id(sess: requests.Session, base: str, project: str, cfg: dict) -> int | None: """解析父工作项：优先 config.parent_id，其次按 parent_title 全项目精确匹配标题。""" pid = str(cfg.get("parent_id") or "").strip() if pid: return int(pid) title = (cfg.get("parent_title") or "").strip() if not title: return None safe_title = title.replace("'", "''") wiql = ( "SELECT [System.Id] FROM WorkItems " f"WHERE [System.TeamProject] = '{project}' " f"AND [System.Title] = '{safe_title}' " "AND [System.WorkItemType] <> 'Task'" ) data = tfs_wiql(sess, f"{base}/{project}/_apis/wit/wiql", wiql) items = data.get("workItems", []) if not items: raise SystemExit( f"按标题找不到父工作项: {title!r}\n" f"请把 config.json 的 parent_title 改成积压工作里已有的条目标题，" f"或直接填 parent_id。" ) if len(items) > 1: print(f"[父项] 警告: 标题 {title!r} 匹配到 {len(items)} 条，取最新创建的一条") # 取 CreatedDate 最大（最新）的一条 return max(int(w["id"]) for w in items) def task_has_parent(item: dict) -> bool: for rel in item.get("relations") or []: if rel.get("rel") == "System.LinkTypes.Hierarchy-Reverse": return True return False def link_parent(sess: requests.Session, base: str, task_id: int, parent_id: int) -> None: ops = [{ "op": "add", "path": "/relations/-", "value": { "rel": "System.LinkTypes.Hierarchy-Reverse", "url": f"{base}/_apis/wit/workitems/{parent_id}", }, }] url = f"{base}/_apis/wit/workitems/{task_id}" resp = tfs_send( sess, "patch", url, params={"api-version": API_VER}, headers={"Content-Type": "application/json-patch+json"}, data=json.dumps(ops), ) if resp.status_code >= 400: raise RuntimeError(f"挂接 #{task_id} -> #{parent_id} 失败 {resp.status_code}: {resp.text[:500]}") def link_existing_tasks(sess: requests.Session, base: str, project: str, cfg: dict, iteration_full: str, parent_id: int) -> int: """把已导入的同类型/同迭代 Task 补挂到父项下；已有父的跳过。""" type_name = cfg["work_item_type"] ids = query_work_item_ids(sess, base, project, type_name, iteration_full) print(f"[补挂] 找到 {type_name} {len(ids)} 条，父项 #{parent_id}") linked = skipped = failed = 0 for tid in ids: try: data = tfs_get(sess, f"{base}/_apis/wit/workitems/{tid}", params={"api-version": API_VER}) if task_has_parent(data): skipped += 1 continue link_parent(sess, base, tid, parent_id) linked += 1 print(f"[补挂] OK #{tid} -> #{parent_id}") except Exception as e: failed += 1 print(f"[补挂] FAIL #{tid} -> {e}") print(f"[补挂] 汇总: 成功 {linked} / 已有父跳过 {skipped} / 失败 {failed}") return 0 if not failed else 1 # ---------------------------------------------------------------- 主流程 def main() -> int: ap = argparse.ArgumentParser(description="TFS 2013 工作项批量导入") ap.add_argument("--config", default=str(BASE_DIR / "config.json")) ap.add_argument("--input", default=str(BASE_DIR / "data" / "work_items.csv")) ap.add_argument("--dry-run", action="store_true", help="只校验不写入") ap.add_argument("--skip-existing-titles", action="store_true", help="跳过服务器已存在的同标题项") ap.add_argument("--list-fields", action="store_true", help="列出服务器字段表后退出") ap.add_argument("--link-existing", action="store_true", help="把已导入的同类型/同迭代工作项补挂到父项下(不新建)") ap.add_argument("--limit", type=int, default=0, help="只处理前 N 条(调试用)") args = ap.parse_args() cfg = load_config(Path(args.config)) base = cfg["tfs_url"].rstrip("/") project = cfg["project"] type_name = cfg["work_item_type"] iteration_cfg = cfg.get("iteration_path") or "" iteration_full = f"{project}\\{iteration_cfg}" if iteration_cfg and not iteration_cfg.startswith(project) else iteration_cfg area = cfg.get("area_path") or "" bypass_rules = bool(cfg.get("bypass_rules", True)) # --link-existing 不读数据文件，直接进补挂流程 if args.link_existing: sess = build_session(cfg) parent_id = find_parent_id(sess, base, project, cfg) if not parent_id: raise SystemExit("--link-existing 需要配置 parent_title 或 parent_id") print(f"[父项] #{parent_id}") print(f"[迭代] {iteration_full}") return link_existing_tasks(sess, base, project, cfg, iteration_full, parent_id) rows = load_rows(Path(args.input)) if args.limit: rows = rows[: args.limit] print(f"[数据] 共 {len(rows)} 行待导入: {args.input}") sess = build_session(cfg) if args.list_fields: data = tfs_get(sess, f"{base}/_apis/wit/fields", params={"api-version": API_VER}) for f in sorted(data.get("value", []), key=lambda x: x.get("name", "")): print(f"{f.get('referenceName', ''):<55} {f.get('name', '')}") return 0 refs = resolve_fields(sess, base, cfg) print(f"[迭代] {iteration_full}") parent_id = find_parent_id(sess, base, project, cfg) if parent_id: print(f"[父项] #{parent_id}") else: print("[父项] 未配置 parent_title/parent_id，导入的项不挂父链接") known: set[str] = set() if args.skip_existing_titles or args.dry_run: try: known = existing_titles(sess, base, project, iteration_full, type_name) print(f"[去重] 服务器同类型/同迭代已有 {len(known)} 条") except Exception as e: print(f"[去重] 查询失败(继续): {e}") ok = fail = skipped = 0 report_path = BASE_DIR / "data" / f"import_report_{time.strftime('%Y%m%d_%H%M%S')}.csv" with report_path.open("w", encoding="utf-8-sig", newline="") as rf: writer = csv.writer(rf) writer.writerow(["行号", "Title", "结果", "WorkItemId", "错误"]) for idx, row in enumerate(rows, 1): title = row["Title"].strip() try: if args.skip_existing_titles and title in known: skipped += 1 writer.writerow([idx, title, "SKIP", "", "标题已存在"]) print(f"[{idx}/{len(rows)}] SKIP {title}") continue ops = build_patch(row, refs, cfg, iteration_full, area) if args.dry_run: ok += 1 writer.writerow([idx, title, "DRY", "", json.dumps(ops, ensure_ascii=False)[:200]]) print(f"[{idx}/{len(rows)}] DRY {title} ({len(ops)} ops)") continue result = create_work_item(sess, base, project, type_name, ops, bypass_rules) wid = result.get("id") if parent_id and wid: try: link_parent(sess, base, int(wid), parent_id) print(f"[挂接] #{wid} -> 父项 #{parent_id}") except Exception as e: print(f"[挂接] FAIL #{wid} -> {e}") ok += 1 writer.writerow([idx, title, "OK", wid, ""]) print(f"[{idx}/{len(rows)}] OK #{wid} {title}") except Exception as e: fail += 1 writer.writerow([idx, title, "FAIL", "", str(e)[:500]]) print(f"[{idx}/{len(rows)}] FAIL {title} -> {e}") print(f"\n[汇总] 成功 {ok} / 跳过 {skipped} / 失败 {fail}；报告: {report_path}") return 1 if fail else 0 if __name__ == "__main__": sys.exit(main())