feat(worklog): 移入工作紀錄並補 Copilot 說明
This commit is contained in:
Jeffery
Executable
+241
@@ -0,0 +1,241 @@
|
||||
#!/usr/bin/env python3
|
||||
# ==============================================================================
|
||||
# 用途：worklog 的 transcript 處理工具。負責 (1) 從 Claude Code transcript
|
||||
# JSONL 抽出「本輪」對話片段（最後一筆使用者訊息之後的全部內容），
|
||||
# (2) 估算本輪花費時間，(3) 對文字做機密遮蔽（token／密碼／PII），
|
||||
# 作為寫入 wiki 前的第二道防線。
|
||||
# 更新時間：2026/07/27 15:09:00
|
||||
# 相依：Python 3 標準庫。全程僅走 stdin／stdout，不寫任何檔案。
|
||||
# ==============================================================================
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
|
||||
# 單則工具結果／參數的擷取上限，避免整份 transcript 塞進摘要輸入
|
||||
TOOL_RESULT_LIMIT = 200
|
||||
TOOL_INPUT_LIMIT = 160
|
||||
TOTAL_LIMIT = 24000
|
||||
|
||||
# ------------------------------------------------------------------------------
|
||||
# 機密遮蔽規則：命中一律換成 ***
|
||||
# ------------------------------------------------------------------------------
|
||||
REDACT_PATTERNS = [
|
||||
(r"[A-Za-z0-9_\-]*:[A-Za-z0-9_\-]{16,}@", "***@"), # URL 內嵌憑證 user:token@
|
||||
(r"\b[0-9a-f]{40}\b", "***"), # Gitea 40 字元 token
|
||||
(r"\bgh[pousr]_[A-Za-z0-9_]{16,}\b", "***"), # GitHub token
|
||||
(r"\bsk-[A-Za-z0-9\-_]{16,}\b", "***"), # API key
|
||||
(r"(?i)\b(token|password|passwd|pwd|secret|api[_-]?key)\b\s*[:=]\s*\S+", r"\1=***"),
|
||||
(r"(?i)Authorization:\s*(token|bearer)\s+\S+", r"Authorization: \1 ***"),
|
||||
(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}", "***"), # Email
|
||||
(r"\b09\d{2}[-\s]?\d{3}[-\s]?\d{3}\b", "***"), # 台灣手機
|
||||
(r"\b[A-Z][12]\d{8}\b", "***"), # 身分證字號
|
||||
]
|
||||
|
||||
|
||||
def redact(text):
|
||||
"""對文字套用全部機密遮蔽規則，回傳遮蔽後的結果。"""
|
||||
for pattern, replacement in REDACT_PATTERNS:
|
||||
text = re.sub(pattern, replacement, text)
|
||||
return text
|
||||
|
||||
|
||||
def _is_real_user_message(entry):
|
||||
"""判斷 transcript 條目是否為真正的使用者輸入（排除 tool_result 回填的 user 條目）。"""
|
||||
if entry.get("type") != "user":
|
||||
return False
|
||||
content = entry.get("message", {}).get("content")
|
||||
if isinstance(content, str):
|
||||
return bool(content.strip())
|
||||
if isinstance(content, list):
|
||||
return any(b.get("type") == "text" for b in content if isinstance(b, dict))
|
||||
return False
|
||||
|
||||
|
||||
def _blocks(entry):
|
||||
"""取出條目的 content blocks，統一為 list 形式。"""
|
||||
content = entry.get("message", {}).get("content")
|
||||
if isinstance(content, str):
|
||||
return [{"type": "text", "text": content}]
|
||||
return content if isinstance(content, list) else []
|
||||
|
||||
|
||||
def _render(entry):
|
||||
"""將單一 transcript 條目轉為摘要輸入用的純文字行（工具結果僅取前段）。"""
|
||||
role = entry.get("type")
|
||||
lines = []
|
||||
for block in _blocks(entry):
|
||||
if not isinstance(block, dict):
|
||||
continue
|
||||
kind = block.get("type")
|
||||
if kind == "text":
|
||||
text = (block.get("text") or "").strip()
|
||||
if text:
|
||||
lines.append(f"[{role}] {text}")
|
||||
elif kind == "tool_use":
|
||||
name = block.get("name", "?")
|
||||
raw = json.dumps(block.get("input", {}), ensure_ascii=False)
|
||||
lines.append(f"[tool:{name}] {raw[:TOOL_INPUT_LIMIT]}")
|
||||
elif kind == "tool_result":
|
||||
raw = block.get("content")
|
||||
if isinstance(raw, list):
|
||||
raw = " ".join(
|
||||
b.get("text", "") for b in raw if isinstance(b, dict) and b.get("type") == "text"
|
||||
)
|
||||
raw = str(raw or "").strip().replace("\n", " ")
|
||||
if raw:
|
||||
lines.append(f"[result] {raw[:TOOL_RESULT_LIMIT]}")
|
||||
return lines
|
||||
|
||||
|
||||
def _read_entries(path):
|
||||
"""讀取 transcript JSONL，忽略無法解析的列。"""
|
||||
try:
|
||||
with open(path, encoding="utf-8") as fh:
|
||||
entries = []
|
||||
for line in fh:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
entries.append(json.loads(line))
|
||||
except ValueError:
|
||||
continue
|
||||
except OSError:
|
||||
return []
|
||||
return entries
|
||||
|
||||
|
||||
def _turn_start_index(entries):
|
||||
"""找出本輪起點：最後一筆真正使用者訊息的位置。"""
|
||||
start = 0
|
||||
for index in range(len(entries) - 1, -1, -1):
|
||||
if _is_real_user_message(entries[index]):
|
||||
start = index
|
||||
break
|
||||
return start
|
||||
|
||||
|
||||
def _parse_timestamp(value):
|
||||
"""解析常見 transcript timestamp 格式，失敗回 None。"""
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
return None
|
||||
raw = value.strip()
|
||||
if raw.endswith("Z"):
|
||||
raw = raw[:-1] + "+00:00"
|
||||
try:
|
||||
dt = datetime.fromisoformat(raw)
|
||||
except ValueError:
|
||||
return None
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=timezone.utc)
|
||||
return dt
|
||||
|
||||
|
||||
def _entry_timestamp(entry):
|
||||
"""取出 transcript 條目的時間欄位。"""
|
||||
for key in ("timestamp", "created_at", "time"):
|
||||
dt = _parse_timestamp(entry.get(key))
|
||||
if dt:
|
||||
return dt
|
||||
message = entry.get("message")
|
||||
if isinstance(message, dict):
|
||||
for key in ("timestamp", "created_at", "time"):
|
||||
dt = _parse_timestamp(message.get(key))
|
||||
if dt:
|
||||
return dt
|
||||
return None
|
||||
|
||||
|
||||
def format_duration(seconds):
|
||||
"""把秒數格式化為精簡中文耗時。"""
|
||||
if seconds < 0:
|
||||
return "未判定"
|
||||
minutes = int(round(seconds / 60))
|
||||
if minutes <= 0:
|
||||
return "1 分鐘內"
|
||||
hours, mins = divmod(minutes, 60)
|
||||
if hours and mins:
|
||||
return f"{hours} 小時 {mins} 分鐘"
|
||||
if hours:
|
||||
return f"{hours} 小時"
|
||||
return f"{mins} 分鐘"
|
||||
|
||||
|
||||
def turn_duration(path):
|
||||
"""
|
||||
估算本輪花費時間：取本輪起點到最後一筆可解析 timestamp 的差距。
|
||||
|
||||
transcript 無時間欄位或本輪少於兩個時間點時回「未判定」，避免臆測。
|
||||
"""
|
||||
entries = _read_entries(path)
|
||||
if not entries:
|
||||
return "未判定"
|
||||
start = _turn_start_index(entries)
|
||||
stamps = [dt for dt in (_entry_timestamp(e) for e in entries[start:]) if dt]
|
||||
if len(stamps) < 2:
|
||||
return "未判定"
|
||||
return format_duration((max(stamps) - min(stamps)).total_seconds())
|
||||
|
||||
|
||||
def extract_turn(path):
|
||||
"""
|
||||
從 transcript JSONL 抽出本輪內容：最後一筆真正使用者訊息（含該筆）之後的全部條目。
|
||||
|
||||
不需任何狀態檔即可界定「本輪」，符合工作內容不落地的要求。
|
||||
回傳純文字字串；讀取失敗或無內容時回空字串。
|
||||
"""
|
||||
entries = _read_entries(path)
|
||||
if not entries:
|
||||
return ""
|
||||
start = _turn_start_index(entries)
|
||||
|
||||
|
||||
lines = []
|
||||
for entry in entries[start:]:
|
||||
lines.extend(_render(entry))
|
||||
|
||||
text = "\n".join(lines).strip()
|
||||
if len(text) > TOTAL_LIMIT:
|
||||
head = text[: TOTAL_LIMIT // 2]
|
||||
tail = text[-TOTAL_LIMIT // 2 :]
|
||||
text = f"{head}\n…（中段省略）…\n{tail}"
|
||||
return text
|
||||
|
||||
|
||||
USAGE = """用法：transcript.py <子命令> [參數]
|
||||
|
||||
extract <transcript 路徑> 抽出本輪內容並遮蔽機密後輸出到 stdout
|
||||
duration <transcript 路徑> 估算本輪花費時間，無法判定時輸出「未判定」
|
||||
redact 自 stdin 讀取文字，遮蔽機密後輸出到 stdout
|
||||
"""
|
||||
|
||||
|
||||
def main(argv):
|
||||
"""CLI 進入點：解析子命令並執行抽取或遮蔽。"""
|
||||
if not argv or argv[0] in ("-h", "--help"):
|
||||
print(USAGE)
|
||||
return 0
|
||||
if argv[0] == "extract":
|
||||
if len(argv) < 2:
|
||||
return 2
|
||||
text = extract_turn(argv[1])
|
||||
if not text:
|
||||
return 1
|
||||
sys.stdout.write(redact(text))
|
||||
return 0
|
||||
if argv[0] == "duration":
|
||||
if len(argv) < 2:
|
||||
return 2
|
||||
sys.stdout.write(turn_duration(argv[1]))
|
||||
return 0
|
||||
if argv[0] == "redact":
|
||||
sys.stdout.write(redact(sys.stdin.read()))
|
||||
return 0
|
||||
print(USAGE)
|
||||
return 2
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main(sys.argv[1:]))
|
||||
Reference in New Issue
Block a user