diff --git a/README.md b/README.md index 6668c1ca..e7c25b22 100644 --- a/README.md +++ b/README.md @@ -366,6 +366,8 @@ dot-skill/ - **colleague** Feishu auto-collection: requires adding the App bot to relevant group chats - **relationship**: longer time spans are better; material covering both conflict and repair is ideal + - Before any analysis runs, `tools/privacy_screen.py` scans the collected material for ID numbers, contact details, third-party speakers, and sensitive topics, then stops and asks you what to keep / redact / exclude. It never edits or deletes your files, and it prints what it does **not** cover — a `CLEAR` result means "no configured pattern matched", not "nothing sensitive is in here" + - Writing to disk, installing into a host, and publishing to the gallery are three separate confirmations. Nothing is installed or published unless you say so - **celebrity**: avoid feeding only second-hand interpretations - This is still a demo version — please file issues if you find bugs! diff --git a/SKILL.md b/SKILL.md index ebb8df1b..bb28fcf8 100644 --- a/SKILL.md +++ b/SKILL.md @@ -65,6 +65,7 @@ allowed-tools: Read, Write, Edit, Bash | 飞书文档（MCP App Token） | `Bash` → `python3 tools/feishu_mcp_client.py` | | 钉钉全自动采集 | `Bash` → `python3 tools/dingtalk_auto_collector.py` | | 解析邮件 .eml/.mbox | `Bash` → `python3 tools/email_parser.py` | +| 隐私筛查（relationship 必跑） | `Bash` → `python3 tools/privacy_screen.py` | | 写入/更新 Skill 文件 | `Write` / `Edit` 工具 | | 版本管理 | `Bash` → `python3 tools/version_manager.py` | | 列出已有 Skill | `Bash` → `python3 tools/skill_writer.py --action list` | @@ -376,6 +377,52 @@ python3 tools/feishu_mcp_client.py \ 如果用户说"没有文件"或"跳过"，仅凭 Step 1 的手动信息生成 Skill。 +### Step 2.5：隐私筛查关卡（仅 relationship） + +`relationship` 的原材料是私人对话，里面经常混进第三方的消息、证件号、联系方式和敏感话题。 +**在进入 Step 3 分析之前，必须先跑隐私筛查，并等用户逐类确认。** + +1. 对已落盘的材料跑筛查： + ```bash + python3 tools/privacy_screen.py \ + --knowledge-dir ./knowledge/{slug} \ + --target-name "{对方称呼}" \ + --participant "{用户自己的聊天昵称}" + ``` + 报告写入 `knowledge/{slug}/privacy_report.json`，同时在终端打印分类汇总表。 + +2. 把工具打印的汇总表**原样**展示给用户，包含： + - 每个类别的命中数量与严重级别 + - 每类 1-2 条**已脱敏**的示例（工具已自动遮蔽，不要自己把原值贴回来） + - 工具输出的"未覆盖范围"清单 —— 不要省略，用户需要知道筛查漏了什么 + +3. 逐类询问处置方式，等待用户明确回答： + ``` + 每个类别你想怎么处理？ + + [保留] 这部分对刻画这个人是必要的，照常进入分析 + [脱敏] 保留语义但抹掉具体值（分析时忽略原值） + [排除] 这个文件/这段内容完全不进入分析 + + 逐类告诉我，或者说"全部排除"。 + ``` + +4. **状态为 `NEEDS_REVIEW` 且用户尚未逐类给出处置意见时，不允许进入 Step 3。** + 没有回答不等于同意；沉默、"随便"、"你看着办"都要再确认一次。 + +5. 用户选择"脱敏"或"排除"的内容，在后续分析中要真正不使用： + - 排除的文件不要读进上下文 + - 脱敏的字段不要写进 `persona.md` / `work.md`，也不要作为"记忆细节"复述 + +6. `third_party.speaker` 命中意味着材料里有第三方在说话。默认**排除**第三方发言， + 除非用户明确说这些内容对刻画对方是必要的。 + +7. `privacy_report.json` 只留在本地，属于敏感文件：不要复制进生成的 Skill 目录， + 分享或发布时也不要带上。 + +> 筛查是提示，不是许可。工具报 `CLEAR` 只代表"配置的规则没有命中"， +> 不代表材料里没有敏感信息；该问用户的仍然要问。 + ### Step 3：分析原材料 先根据 character family 解析本次的执行矩阵： @@ -636,6 +683,42 @@ Persona 摘要： 告知用户时，文件位置必须按当前 family 返回，不要默认写成 colleague。 +### Step 5.5：落盘 / 安装 / 分享关卡（仅 relationship） + +`relationship` Skill 承载的是私人关系材料，因此它的每一次"扩散"都要单独确认。 +以下三个动作各自需要一次**独立的**明确同意，不能用同一句"确认"一次性覆盖： + +1. **写入磁盘**（Step 4 的预览确认） + - 用户的回答是最终依据，不是走过场：说"再看看""等等"就停在预览，不要落盘 + - 只有明确的肯定答复才继续 + +2. **安装到宿主**（`--install-claude-skill` / `--install-openclaw-skill` / `--install-codex-skill` + / 放进 `~/.dsh/skills/`） + ``` + 要把这个 Skill 安装到宿主吗？ + + 安装后，这个宿主里的任何会话都可以直接调用它， + 也就是说这段关系材料会脱离当前对话的范围。 + + [安装到 {宿主}] / [先不安装，只留在本地目录] + ``` + - 默认**不安装**。用户没提安装，就不要主动加 `--install-*` 参数 + +3. **分享 / 发布到 gallery** + ``` + ⚠️ 这是一个 relationship Skill，里面包含私人对话提炼出的内容。 + + 发布到 gallery 意味着任何人都能看到并安装它。 + 对方本人是否知情并同意公开？ + + [确认发布] / [不发布] + ``` + - 默认**不发布**，且必须是单独的一次确认——Step 4 的"确认生成"不构成发布同意 + - 用户没有明确说要公开分享时，不要提议发布，也不要生成 gallery 提交内容 + - 发布前确认 `knowledge/` 与 `privacy_report.json` 没有被打包进去 + +> 三个关卡的共同原则：**沉默不是同意**。用户没回答就停下来问，不要替他做决定。 + --- ## 进化模式：追加文件 @@ -643,6 +726,8 @@ Persona 摘要： 用户提供新文件或文本时： 1. 按 Step 2 的方式读取新内容 + - 如果当前是 `relationship`，新材料同样要过 Step 2.5 的隐私筛查关卡， + 并等用户逐类确认后再继续；不要因为"只是追加"就跳过 2. 根据当前 family 解析 base dir 3. 用 `Read` 读取现有 `{resolved_base_dir}/{slug}/work.md` 和 `persona.md` 4. 使用当前 family 对应的 merger prompt 分析增量内容 @@ -675,6 +760,21 @@ Persona 摘要： 1. 参考 `prompts/correction_handler.md` 识别纠正内容 2. 判断属于 Work（技术/流程）还是 Persona（性格/沟通） +2.5. **如果当前是 `relationship`，且这条纠正会改动 `persona.md`：先展示、后应用。** + - 把准备写入的内容原样摊开给用户看，再等一次明确同意： + ``` + 这条纠正我准备这样写进 persona.md： + + 场景：{scene} + 原来的行为：{wrong} + 改成：{correct} + + 确认写入？(确认 / 改一下 / 算了) + ``` + - 用户确认前**不要**调用 `skill_writer.py` + - 一次提出多条纠正时，逐条展示逐条确认，不要打包成一次"全部确认" + - 纠正会改变这个 Skill 对一个真实的人的刻画，写错了就是替对方说了他没说过的话—— + 所以这里宁可多问一次 3. 如果属于 Work： - 生成 `/tmp/dot_skill_{slug}_work_patch.md` - patch 必须是可替换的 `##` section，不要直接手改最终文件 @@ -792,6 +892,7 @@ This Skill runs in any compatible host that can read local files and execute Bas | Feishu docs (MCP App Token) | `Bash` → `python3 tools/feishu_mcp_client.py` | | DingTalk auto-collect | `Bash` → `python3 tools/dingtalk_auto_collector.py` | | Parse email .eml/.mbox | `Bash` → `python3 tools/email_parser.py` | +| Privacy screening (required for relationship) | `Bash` → `python3 tools/privacy_screen.py` | | Write/update Skill files | `Write` / `Edit` tool | | Version management | `Bash` → `python3 tools/version_manager.py` | | List existing Skills | `Bash` → `python3 tools/skill_writer.py --action list` | @@ -1103,6 +1204,60 @@ User-pasted content is used directly as text material. No tools needed. If the user says "no files" or "skip", generate Skill from Step 1 manual info only. +### Step 2.5: Privacy Screening Gate (relationship only) + +`relationship` material is private conversation. It routinely carries third-party +messages, ID numbers, contact details, and sensitive topics. +**Run the privacy screen and get per-category confirmation before Step 3.** + +1. Screen the material that has been written to disk: + ```bash + python3 tools/privacy_screen.py \ + --knowledge-dir ./knowledge/{slug} \ + --target-name "{name of the person being distilled}" \ + --participant "{the user's own chat handle}" + ``` + The report is written to `knowledge/{slug}/privacy_report.json`, and a category + summary table is printed to the terminal. + +2. Show the printed summary to the user **as-is**, including: + - the hit count and severity for each category + - one or two **already-masked** examples per category (the tool masks them; never + paste the raw values back in) + - the "not covered by this screen" list the tool prints — do not omit it, the user + needs to know what the screen misses + +3. Ask how to handle each flagged category and wait for an explicit answer: + ``` + How do you want to handle each category? + + [keep] necessary for capturing this person, analyze as normal + [redact] keep the meaning, drop the specific values + [exclude] this file / passage does not enter analysis at all + + Tell me per category, or say "exclude everything". + ``` + +4. **While the status is `NEEDS_REVIEW` and the user has not given a per-category + decision, do not enter Step 3.** A non-answer is not consent: silence, "whatever", + and "you decide" all mean ask again. + +5. Honour "redact" and "exclude" for real in the analysis that follows: + - do not read excluded files into context + - do not write redacted values into `persona.md` / `work.md`, and do not recite them + back as "remembered details" + +6. A `third_party.speaker` hit means someone else is talking in the material. **Exclude + third-party messages by default**, unless the user explicitly says they are necessary + for capturing the target person. + +7. `privacy_report.json` stays local and is itself sensitive: never copy it into the + generated skill directory, and never include it when sharing or publishing. + +> The screen is a prompt for judgement, not a permission slip. A `CLEAR` status only +> means no configured pattern matched — it does not mean the material is free of +> sensitive information. Ask the user anyway. + ### Step 3: Analyze Source Material First resolve the execution matrix for the selected character family: @@ -1364,6 +1519,49 @@ After user confirmation, do not hand-build a `skills/colleague/{slug}`-style tre When reporting success, return the correct family-specific location instead of assuming colleague storage. +### Step 5.5: Write / Install / Share Gates (relationship only) + +A `relationship` skill carries private relationship material, so every step that spreads +it further needs its own confirmation. These three actions each require a **separate** +explicit yes — one blanket "confirm" does not cover all of them: + +1. **Writing to disk** (the Step 4 preview confirmation) + - The user's answer is authoritative, not a formality: "let me look again" or "hold on" + means stop at the preview and do not write + - Only an explicit yes continues + +2. **Installing into a host** (`--install-claude-skill` / `--install-openclaw-skill` / + `--install-codex-skill`, or dropping it into `~/.dsh/skills/`) + ``` + Install this skill into a host? + + Once installed, any session in that host can invoke it directly — this relationship + material leaves the scope of the current conversation. + + [install into {host}] / [don't install, keep it in the local directory] + ``` + - Default to **not installing**. If the user did not ask for installation, do not add + an `--install-*` flag on your own + +3. **Sharing / publishing to the gallery** + ``` + ⚠️ This is a relationship skill. It contains material distilled from private + conversations. + + Publishing to the gallery means anyone can see and install it. + Does the other person know about this and agree to it being public? + + [confirm publish] / [do not publish] + ``` + - Default to **not publishing**, and require a distinct confirmation — the Step 4 + "confirm generation" answer is not consent to publish + - Unless the user has explicitly asked to share publicly, do not propose publishing and + do not prepare gallery submission content + - Before publishing, verify that `knowledge/` and `privacy_report.json` are not bundled + +> The shared rule across all three gates: **silence is not consent.** If the user has not +> answered, stop and ask rather than deciding for them. + --- ## Evolution Mode: Append Files @@ -1371,6 +1569,9 @@ When reporting success, return the correct family-specific location instead of a When user provides new files or text: 1. Read new content using Step 2 methods + - If the current family is `relationship`, the new material goes through the Step 2.5 + privacy screening gate as well, with per-category confirmation before continuing. + Do not skip it just because this is "only an append" 2. Resolve the base dir for the current family 3. `Read` existing `{resolved_base_dir}/{slug}/work.md` and `persona.md` 4. Use the family-specific merger prompt for incremental analysis @@ -1403,6 +1604,23 @@ When user expresses "that's wrong" / "he should be": 1. Refer to `prompts/correction_handler.md` to identify correction content 2. Determine if it belongs to Work (technical/workflow) or Persona (personality/communication) +2.5. **If the current family is `relationship` and the correction touches `persona.md`, + show it before applying it.** + - Lay out exactly what you are about to write, then wait for an explicit yes: + ``` + Here is how I'd write this correction into persona.md: + + Scene: {scene} + Was: {wrong} + Becomes: {correct} + + Write it? (confirm / adjust / cancel) + ``` + - Do **not** call `skill_writer.py` before the user confirms + - When several corrections arrive at once, show and confirm them one at a time rather + than bundling them into a single "confirm all" + - A correction changes how this skill portrays a real person; getting it wrong puts + words in their mouth. Ask the extra question here 3. If it belongs to Work: - Generate `/tmp/dot_skill_{slug}_work_patch.md` - The patch must be one or more replaceable `##` sections diff --git a/docs/lang/README_ZH.md b/docs/lang/README_ZH.md index 67e13c47..c76199da 100644 --- a/docs/lang/README_ZH.md +++ b/docs/lang/README_ZH.md @@ -357,6 +357,8 @@ dot-skill/ - **colleague** 飞书自动采集：需将 App bot 加入相关群聊 - **relationship**：时间跨度越长越好，能覆盖冲突与和解更佳 + - 分析开始前，`tools/privacy_screen.py` 会先扫描采集到的材料，找出证件号、联系方式、第三方发言和敏感话题，然后停下来问你每一类要保留 / 脱敏 / 排除。它不会修改或删除你的文件，并且会明确列出**没有覆盖**的范围 —— `CLEAR` 只代表"配置的规则没有命中"，不代表材料里没有敏感信息 + - 落盘、安装到宿主、发布到 gallery 是三次独立确认。你不点头，就不会安装，也不会发布 - **celebrity**：避免只喂二手解读 - 目前还是 demo 版本，如果有 bug 请多多提 issue！ diff --git a/prompts/persona_analyzer.md b/prompts/persona_analyzer.md index 3441399a..5d7803f3 100644 --- a/prompts/persona_analyzer.md +++ b/prompts/persona_analyzer.md @@ -127,7 +127,7 @@ emoji：[无/偶尔/频繁，类型] ## 输出要求 -- 语言：中文 +- 语言：与用户当前使用的语言保持一致（用户用韩语交流就输出韩语，用中文就输出中文，用英语就输出英语）。不要默认中文。 - 原材料不足的维度：标注 `（原材料不足）` - 有原文依据的结论：引用原话（加引号） - 手动标签与文件分析冲突时：输出两个版本并注明，供 persona_builder 处理 diff --git a/prompts/relationship/intake.md b/prompts/relationship/intake.md index 76a568cc..286da490 100644 --- a/prompts/relationship/intake.md +++ b/prompts/relationship/intake.md @@ -84,12 +84,27 @@ What materials can you provide for this person? If you upload files, save them under this skill’s `knowledge/` folders before analyzing them. For WeChat chat history import, you can try WeFlow first. + +Two more things I need before I read any of it: + +- What name or handle do they appear under in these chats? +- What name or handle do you appear under? + +Anyone else who shows up under a different name is treated as a third party, and +I'll ask you before using anything they said. ``` Capture: - available source types - whether files were uploaded - which folders should receive the material +- the target's speaker label (passed to `privacy_screen.py --target-name`) +- the user's own speaker label (passed to `privacy_screen.py --participant`) + +If files were uploaded, the next step is the Step 2.5 privacy screening gate in +`SKILL.md` — run it before any analysis and wait for the user's per-category +decision. Do not start analysis on a `NEEDS_REVIEW` screen that has not been +resolved. --- @@ -104,6 +119,8 @@ Summary: Basic profile: {profile_summary} Closest memory: {memory_summary} Materials: {source_summary} + Their chat handle: {target_speaker_label} + Your chat handle: {user_speaker_label} Confirm? (confirm / edit [field]) ``` diff --git a/prompts/work_analyzer.md b/prompts/work_analyzer.md index dd963f01..0416a44b 100644 --- a/prompts/work_analyzer.md +++ b/prompts/work_analyzer.md @@ -68,7 +68,21 @@ ## 职位专项提取 -根据 {name} 的职位，重点提取对应维度： +根据 {name} 的职位，重点提取对应维度。 + +### 职位不在下列清单时 + +下面的清单不可能覆盖所有职位。如果 {name} 的职位不在清单里（财务、法务、人事、销售、医疗、教育、制造现场、研究等），**不要跳过本节，也不要套用不匹配的 IT 维度**。改为自行推导： + +1. 从原材料中找出他**反复做的 3-5 类工作**（例如"审批某类申请""每月产出某份报告""处理某类异常"） +2. 对每一类提取三件事： + - **判断标准**：依据什么做决定，门槛值具体是多少 + - **动作顺序**：先做什么、后做什么 + - **例外**：什么情况下他会偏离常规做法 +3. **升级阈值**：什么程度的问题他自己处理，到什么程度会上报或转交他人 +4. **红线**：他明确拒绝的事，以及不肯让步的坚持 + +这四项对任何职位都成立，比硬套模板有价值得多。 --- @@ -173,9 +187,69 @@ --- +### 📦 供应链 / SCM（需求计划 · 采购 · 库存 · 物流 · 生产计划 · S&OP） + +**需求计划与预测**： +- 预测基线的产出方式（统计模型 / 销售输入 / 两者混合） +- 手动覆盖预测的时机与理由类型（促销、新品、季节性、一次性事件） +- 预测颗粒度（SKU / 品类 / 渠道）与滚动周期 +- 他看的准确度指标（MAPE / bias / WMAPE）以及他认为可接受的范围 + +**库存判断**： +- 安全库存的设定逻辑（目标服务水平、需求变异、补货前置期） +- 缺货与呆滞之间的取舍倾向：哪一边他更不能接受 +- 呆滞 / 临期库存的处理路径（调拨、促销、报废）分别在什么条件下触发 +- 他常用的库存指标（周转天数、DOS、库龄结构） + +**采购与供应商**： +- 供应商选择与分级标准 +- 议价时的让步顺序：价格 / 账期 / MOQ / 前置期中，哪些可以让，哪些绝不让 +- 供应商延迟时的升级阈值：延迟几天、影响多大才上报 +- 单一供应商与多源采购之间的判断依据 + +**物流与履约**： +- 运输模式选择的权衡（成本 vs 时效） +- 加急发货（空运、专车等）的批准门槛 +- 他关注的履约指标（OTIF、准时交付率、订单满足率） + +**异常与升级**： +- 断供、品质事故、需求暴涨时的第一步动作 +- 跨部门拉齐的顺序（销售 / 生产 / 财务，先找谁） +- 什么情况下他会推翻系统给出的建议 + +**S&OP 与会议**： +- S&OP 会议上他先看哪几个数字 +- 计划与实际出现差异时的解释方式 +- 决策与假设的留存方式 + +--- + +### 🧭 咨询顾问 / 专业服务 + +**诊断与问题定义**： +- 接到课题后的诊断顺序（先看数据 / 先访谈 / 先走流程） +- 界定问题边界与项目范围的方式 +- 反复使用的分析框架 + +**交付物**： +- 报告结构习惯（结论先行 / 逻辑树 / 单页摘要） +- 图表与数据的使用偏好 +- 中间汇报的频率与形式 + +**客户沟通**： +- 向客户高层汇报与向实务层沟通的差异 +- 遇到客户反对意见时的处理方式 +- 什么会答应、什么会明确拒绝 + +**方法论与经验**： +- 他反复套用的模板或方法论 +- 他明确说过的"这类项目一定要先做 X"这类经验结论 + +--- + ## 输出要求 -- 语言：中文 +- 语言：与用户当前使用的语言保持一致（用户用韩语交流就输出韩语，用中文就输出中文，用英语就输出英语）。不要默认中文。 - 没有信息的维度：标注 `（原材料不足，建议追加相关文档）` - 有原文依据的结论：加引号标注原话 - 输出结果直接用于生成 work.md，要求具体可执行，不要写"可能""倾向于"这类模糊表述 diff --git a/prompts/work_builder.md b/prompts/work_builder.md index f4074ff5..a6581d48 100644 --- a/prompts/work_builder.md +++ b/prompts/work_builder.md @@ -26,43 +26,62 @@ --- -## 技术规范 +## 专业规范与判断标准 +> 本节的小节标题**随职位变化**。只写与该职位真实相关的小节，删掉不适用的， +> 不要为了填满模板而保留空壳标题。下面按职位给出常见组合： + +**工程/技术类**（后端、前端、算法…）： ### 技术栈 {主要技术栈列表} - -### 代码风格 -{代码风格描述} - -### 命名规范 -{命名规范描述} - +### 代码风格 / 命名规范 +{描述} ### 接口设计 {接口设计规范描述} - -{如果有前端内容则加：} -### 前端规范 -{前端规范描述} - ### Code Review 重点 你在 CR 时特别关注： {CR 重点列表} +**供应链 / SCM 类**： +### 计划与预测口径 +{预测基线方法、覆盖预测的条件、颗粒度与滚动周期、准确度指标与可接受范围} +### 库存判断标准 +{安全库存设定逻辑、缺货与呆滞的取舍倾向、呆滞处理触发条件、常看指标} +### 采购与供应商标准 +{分级标准、议价让步顺序、延迟升级阈值、单一/多源判断} +### 异常升级阈值 +{什么程度自己处理、什么程度上报、先拉齐谁} + +**咨询 / 专业服务类**： +### 诊断顺序 +{接到课题后的第一步、问题边界界定方式、常用框架} +### 交付物标准 +{报告结构、图表偏好、中间汇报节奏} +### 客户沟通原则 +{对高层与实务层的差异、面对反对意见的处理、答应与拒绝的边界} + +**其他职位**： +按 work_analyzer 推导出的"反复做的工作类别"建小节，每个小节写清 +{判断标准 · 动作顺序 · 例外}。 + --- ## 工作流程 -### 接到需求时 -{需求处理步骤} +> 同样按职位调整小节名。工程类用下面的默认组合，其他职位替换为该职位真实的 +> 高频场景（例如 SCM 的"月度 S&OP 时""供应商延迟时"，咨询的"项目启动时""交付汇报前"）。 + +### 接到任务时 +{任务处理步骤} -### 写技术方案时 -{方案文档结构描述} +### 产出主要文档时 +{文档结构描述} -### 处理线上问题时 -{线上问题处理流程} +### 处理异常 / 紧急情况时 +{异常处理流程} -### 做 Code Review 时 -{CR 流程描述} +### 审阅他人产出时 +{审阅流程与重点} --- @@ -83,10 +102,10 @@ 当用户要求你完成以下任务时，严格按照上述规范执行： - 写代码（CRUD / 接口 / 前端组件）→ 遵循技术规范和代码风格 -- 写文档（技术方案 / 接口文档）→ 遵循输出风格 -- 做 Code Review → 遵循 CR 重点 -- 处理需求 → 遵循工作流程 -- 回答技术问题 → 优先使用经验知识库中的结论 +- 产出该职位的主要交付物 → 遵循输出风格 +- 审阅他人的产出 → 遵循上面的审阅重点 +- 处理任务与异常 → 遵循工作流程 +- 回答专业问题 → 优先使用经验知识库中的结论，并说明判断依据与门槛值 如果被问到职责范围外的问题，以该同事的方式回应（参见 Persona 部分）。 ``` @@ -96,6 +115,11 @@ ## 生成注意事项 1. 如果原材料信息不足某个维度，该维度用"（暂无足够信息，建议追加相关文档）"占位 -2. 知识结论要具体，避免泛泛而谈（错误示例："注重代码质量"；正确示例："函数单一职责，超过 50 行必须拆分"） -3. 技术栈和规范要直接可执行，不要写成"可能使用"或"倾向于" -4. 整个文件用 Markdown 格式，标题层级清晰 +2. 知识结论要具体，避免泛泛而谈。 + - 工程类 —— 错误："注重代码质量"；正确："函数单一职责，超过 50 行必须拆分" + - SCM 类 —— 错误："重视库存管理"；正确："供应商延迟 3 天以内自己协调，超过 3 天且影响出货就当天上报" + - 咨询类 —— 错误："以客户为中心"；正确："诊断阶段先看 3 个月实绩数据，访谈放在数据看完之后" +3. 规范要直接可执行，不要写成"可能使用"或"倾向于"。凡是有门槛值、顺序、条件的，一律写出具体数字与条件 +4. 只保留与该职位相关的小节，删除不适用的标题，不要留空壳 +5. 整个文件用 Markdown 格式，标题层级清晰 +6. 语言与用户当前使用的语言保持一致，不要默认中文 diff --git a/tests/test_privacy_screen.py b/tests/test_privacy_screen.py new file mode 100644 index 00000000..a48d9b86 --- /dev/null +++ b/tests/test_privacy_screen.py @@ -0,0 +1,306 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +from pathlib import Path +import sys + + +TOOLS_DIR = Path(__file__).resolve().parents[1] / "tools" +if str(TOOLS_DIR) not in sys.path: + sys.path.insert(0, str(TOOLS_DIR)) + +from privacy_screen import ( # noqa: E402 + STATUS_CLEAR, + STATUS_NEEDS_REVIEW, + chinese_id_valid, + format_summary, + korean_rrn_valid, + luhn_valid, + mask_value, + screen_directory, + screen_text, + write_report, +) + + +# Structurally valid fixtures (generated to satisfy each checksum) so the +# detectors are exercised on realistic input rather than arbitrary digits. +VALID_KR_RRN = "900101-1000006" +VALID_CN_ID = "110101199003070003" +VALID_CARD = "4111 1111 1111 1111" + + +def categories(findings) -> set[str]: + return {finding["category"] for finding in findings} + + +class ValidatorTest(unittest.TestCase): + def test_luhn_accepts_valid_and_rejects_invalid(self) -> None: + self.assertTrue(luhn_valid("4111111111111111")) + self.assertFalse(luhn_valid("4111111111111112")) + self.assertFalse(luhn_valid("123")) + + def test_korean_rrn_checksum(self) -> None: + self.assertTrue(korean_rrn_valid("9001011000006")) + self.assertFalse(korean_rrn_valid("9001011000007")) + # Impossible birth month is rejected before the checksum runs. + self.assertFalse(korean_rrn_valid("9099011000006")) + + def test_chinese_id_checksum(self) -> None: + self.assertTrue(chinese_id_valid(VALID_CN_ID)) + self.assertFalse(chinese_id_valid("110101199003070004")) + + +class MaskingTest(unittest.TestCase): + def test_mask_value_hides_middle(self) -> None: + masked = mask_value("4111111111111111") + self.assertTrue(masked.startswith("41")) + self.assertTrue(masked.endswith("11")) + self.assertIn("*", masked) + self.assertNotEqual(masked, "4111111111111111") + + def test_short_values_are_fully_masked(self) -> None: + self.assertEqual(mask_value("abc"), "***") + + +class DetectorTest(unittest.TestCase): + def test_detects_identity_grade_identifiers(self) -> None: + text = "\n".join( + [ + f"주민번호는 {VALID_KR_RRN} 이야", + f"身份证 {VALID_CN_ID}", + f"card {VALID_CARD}", + ] + ) + found = categories(screen_text(text)) + self.assertIn("pii.national_id.kr", found) + self.assertIn("pii.national_id.cn", found) + self.assertIn("pii.credit_card", found) + + def test_detects_contact_details(self) -> None: + text = "call 010-1234-5678 or mail me at someone@example.com" + found = categories(screen_text(text)) + self.assertIn("pii.phone", found) + self.assertIn("pii.email", found) + + def test_bank_account_requires_context_keyword(self) -> None: + with_keyword = screen_text("계좌 110-234-567890 으로 보내줘") + self.assertIn("pii.bank_account", categories(with_keyword)) + + # The same digit shape with no banking context is not an account hit. + without_keyword = screen_text("주문번호 110-234-567890") + self.assertNotIn("pii.bank_account", categories(without_keyword)) + + def test_detects_sensitive_topics(self) -> None: + text = "\n".join( + [ + "작년에 우울증 진단을 받았어", + "he mentioned his salary and some debt", + ] + ) + found = categories(screen_text(text)) + self.assertIn("sensitive.health", found) + self.assertIn("sensitive.financial", found) + + def test_sensitive_category_reported_once_per_line(self) -> None: + findings = screen_text("depression anxiety disorder diagnosis therapy") + health = [f for f in findings if f["category"] == "sensitive.health"] + self.assertEqual(len(health), 1) + + +class FalsePositiveGuardTest(unittest.TestCase): + def test_plain_digit_runs_are_not_flagged(self) -> None: + # An order id and a bare six-digit code should not become an RRN or card. + found = categories(screen_text("order 1234567890123 code 900101")) + self.assertNotIn("pii.national_id.kr", found) + self.assertNotIn("pii.credit_card", found) + + def test_one_span_is_reported_by_one_detector_only(self) -> None: + # A Korean RRN is also a Luhn-valid digit run; the more specific + # detector claims the span and the card detector must stay out. + found = categories(screen_text(f"주민번호 {VALID_KR_RRN}")) + self.assertIn("pii.national_id.kr", found) + self.assertNotIn("pii.credit_card", found) + + def test_ordinary_prose_with_colon_is_not_a_speaker(self) -> None: + text = "Note: this sentence merely contains a colon and should stay quiet." + self.assertNotIn("third_party.speaker", categories(screen_text(text))) + + +class ThirdPartyTest(unittest.TestCase): + def test_flags_unexpected_speaker_labels(self) -> None: + transcript = "\n".join( + [ + "Alex: are you coming tonight", + "me: yes", + "Jordan: I'll bring the cake", + "Alex: nice", + "Jordan: see you at eight", + ] + ) + findings = screen_text( + transcript, + target_name="Alex", + known_participants=("me",), + ) + third_party = [f for f in findings if f["category"] == "third_party.speaker"] + self.assertEqual(len(third_party), 1) + # The speaker name is reported in the clear so the user can identify who + # it is; identifier values elsewhere stay masked. + self.assertEqual(third_party[0]["speaker"], "Jordan") + + def test_timestamped_lines_capture_the_name_not_the_clock(self) -> None: + # A loose date prefix backtracks into the clock's colon and captures a + # digit; this guards the explicit date/time prefix. + transcript = "\n".join( + [ + "2024-01-02 10:00 Alex: hi", + "2024-01-02 10:02 Jordan: I'll be there", + "2024-01-02 10:03 Jordan: see you at eight", + ] + ) + findings = screen_text(transcript, target_name="Alex") + third_party = [f for f in findings if f["category"] == "third_party.speaker"] + self.assertEqual(len(third_party), 1) + self.assertEqual(third_party[0]["speaker"], "Jordan") + self.assertEqual(third_party[0]["message_count"], 2) + + def test_kakaotalk_desktop_export_format(self) -> None: + transcript = "\n".join( + [ + "2024년 1월 2일 오후 3:15, 민준 : 오늘 시간 돼?", + "2024년 1월 2일 오후 3:16, 나 : 응 괜찮아", + "2024년 1월 2일 오후 3:17, 서연 : 나도 갈래", + "2024년 1월 2일 오후 3:18, 서연 : 8시에 봐", + ] + ) + findings = screen_text(transcript, target_name="민준", known_participants=("나",)) + third_party = [f for f in findings if f["category"] == "third_party.speaker"] + self.assertEqual(len(third_party), 1) + self.assertEqual(third_party[0]["speaker"], "서연") + + def test_kakaotalk_bracketed_export_format(self) -> None: + transcript = "\n".join( + [ + "[민준] [오후 3:20] 오늘 시간 돼?", + "[서연] [오후 3:21] 나도 갈래", + "[서연] [오후 3:22] 8시에 봐", + ] + ) + findings = screen_text(transcript, target_name="민준") + third_party = [f for f in findings if f["category"] == "third_party.speaker"] + self.assertEqual(len(third_party), 1) + self.assertEqual(third_party[0]["speaker"], "서연") + + def test_comma_separated_timestamp_does_not_leak_into_the_name(self) -> None: + transcript = "2024-01-02 15:15, Jordan : hi\n2024-01-02 15:16, Jordan : again" + findings = screen_text(transcript, target_name="Alex") + third_party = [f for f in findings if f["category"] == "third_party.speaker"] + self.assertEqual(len(third_party), 1) + self.assertEqual(third_party[0]["speaker"], "Jordan") + + def test_single_line_label_is_not_treated_as_a_speaker(self) -> None: + # One "Name:" line is indistinguishable from prose, so it stays quiet. + transcript = "Alex: hi\nAlex: still there\nCasey: one off line" + findings = screen_text(transcript, target_name="Alex") + self.assertNotIn("third_party.speaker", categories(findings)) + + def test_target_and_known_participants_are_not_flagged(self) -> None: + transcript = "Alex: hi\nme: hello\nAlex: how are you" + findings = screen_text( + transcript, + target_name="Alex", + known_participants=("me",), + ) + self.assertNotIn("third_party.speaker", categories(findings)) + + +class ScreenDirectoryTest(unittest.TestCase): + def test_clean_material_reports_clear(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + knowledge = Path(tmp_dir) / "knowledge" / "alex" + knowledge.mkdir(parents=True) + (knowledge / "messages.txt").write_text( + "we talked about the weather and a movie\n", + encoding="utf-8", + ) + report = screen_directory(knowledge) + self.assertEqual(report["status"], STATUS_CLEAR) + self.assertEqual(report["total_findings"], 0) + self.assertIn("messages.txt", report["files_scanned"]) + + def test_sensitive_material_reports_needs_review(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + knowledge = Path(tmp_dir) / "knowledge" / "alex" + knowledge.mkdir(parents=True) + (knowledge / "messages.txt").write_text( + f"주민번호 {VALID_KR_RRN}\ncall 010-1234-5678\n", + encoding="utf-8", + ) + report = screen_directory(knowledge) + self.assertEqual(report["status"], STATUS_NEEDS_REVIEW) + self.assertGreaterEqual(report["counts_by_category"]["pii.national_id.kr"], 1) + self.assertTrue(report["limitations"]) + + def test_report_does_not_contain_raw_sensitive_values(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + knowledge = Path(tmp_dir) / "knowledge" / "alex" + knowledge.mkdir(parents=True) + (knowledge / "messages.txt").write_text( + f"내 주민번호는 {VALID_KR_RRN} 이고 카드는 {VALID_CARD} 야\n" + "메일은 secret.person@example.com\n", + encoding="utf-8", + ) + report = screen_directory(knowledge) + serialized = json.dumps(report, ensure_ascii=False) + + self.assertNotIn(VALID_KR_RRN, serialized) + self.assertNotIn("4111 1111 1111 1111", serialized) + self.assertNotIn("secret.person@example.com", serialized) + + def test_write_report_round_trips(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + knowledge = Path(tmp_dir) / "knowledge" / "alex" + knowledge.mkdir(parents=True) + (knowledge / "messages.txt").write_text("call 010-1234-5678\n", encoding="utf-8") + + report = screen_directory(knowledge) + report_path = write_report(report, knowledge) + + self.assertTrue(report_path.exists()) + reloaded = json.loads(report_path.read_text(encoding="utf-8")) + self.assertEqual(reloaded["status"], STATUS_NEEDS_REVIEW) + + def test_existing_report_is_not_rescanned(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + knowledge = Path(tmp_dir) / "knowledge" / "alex" + knowledge.mkdir(parents=True) + (knowledge / "messages.txt").write_text("call 010-1234-5678\n", encoding="utf-8") + write_report(screen_directory(knowledge), knowledge) + + second = screen_directory(knowledge) + self.assertNotIn("privacy_report.json", second["files_scanned"]) + + +class SummaryTest(unittest.TestCase): + def test_summary_lists_categories_and_limitations(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + knowledge = Path(tmp_dir) / "knowledge" / "alex" + knowledge.mkdir(parents=True) + (knowledge / "messages.txt").write_text( + f"주민번호 {VALID_KR_RRN}\n우울증 진단 이야기\n", + encoding="utf-8", + ) + report = screen_directory(knowledge) + + summary = format_summary(report) + self.assertIn(STATUS_NEEDS_REVIEW, summary) + self.assertIn("pii.national_id.kr", summary) + self.assertIn("Not covered by this screen:", summary) + self.assertNotIn(VALID_KR_RRN, summary) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/privacy_screen.py b/tools/privacy_screen.py new file mode 100644 index 00000000..b7b06bed --- /dev/null +++ b/tools/privacy_screen.py @@ -0,0 +1,655 @@ +#!/usr/bin/env python3 +"""Screen collected relationship material for sensitive personal data. + +This tool runs over the raw material gathered under ``knowledge/{slug}/`` before +that material is handed to the persona/work analyzers. It never edits, redacts, +or deletes the source files: it only produces a structured report so the user +can decide, per category, whether to keep / redact / exclude the material. + +The screen is intentionally conservative about what it claims to find. Detectors +that can be structurally validated (Korean RRN, Chinese resident ID, credit-card +numbers) are checksum-verified so they rarely fire on ordinary digit runs. +Detectors that cannot be validated (addresses, sensitive-topic keywords) are +reported as weaker signals and are documented with their known false-positive +modes below. + +Usage: + python3 tools/privacy_screen.py --knowledge-dir ./knowledge/{slug} \ + --target-name "Alex" --user-name "me" + python3 tools/privacy_screen.py --knowledge-dir ./knowledge/{slug} --json +""" + +from __future__ import annotations + +import argparse +import json +import re +from pathlib import Path + + +# --------------------------------------------------------------------------- +# Status values +# --------------------------------------------------------------------------- + +STATUS_CLEAR = "CLEAR" +STATUS_NEEDS_REVIEW = "NEEDS_REVIEW" + +# Files under knowledge/ that are worth screening. Everything else (images, +# binaries, the report itself) is skipped rather than guessed at. +DEFAULT_SUFFIXES = (".txt", ".md", ".json", ".csv", ".eml", ".log") + +REPORT_FILENAME = "privacy_report.json" + +# What this screen does NOT catch. Kept next to the code so it stays honest and +# so the same wording can be surfaced in the report and in the README. +KNOWN_LIMITATIONS = [ + "Free-text mentions of other people's names are not detected; only speaker " + "labels in chat-style exports are.", + "A speaker who leads only one line is not reported, because a single " + "'Name:' line cannot be told apart from prose such as 'Note:'.", + "Sensitive-topic detection is keyword-based, so it misses paraphrase, slang, " + "and any language whose vocabulary is not listed.", + "Addresses are matched on shape (road/city markers plus a number) and will " + "both over-fire on ordinary sentences and miss unusual formats.", + "Images, audio, and other binary attachments are not inspected at all.", + "A CLEAR status means no configured pattern matched. It is not a guarantee " + "that the material contains no sensitive information.", + "Detected identifier values are always masked in this report, but third-party " + "speaker names are shown in full so you can tell who they are.", +] + + +# --------------------------------------------------------------------------- +# Structural validators +# --------------------------------------------------------------------------- + + +def luhn_valid(digits: str) -> bool: + """Return True when a digit string passes the Luhn checksum. + + Used to keep the credit-card detector from firing on arbitrary 13-19 digit + runs such as order numbers or concatenated timestamps. + """ + if not digits.isdigit() or not 13 <= len(digits) <= 19: + return False + total = 0 + parity = len(digits) % 2 + for index, char in enumerate(digits): + value = int(char) + if index % 2 == parity: + value *= 2 + if value > 9: + value -= 9 + total += value + return total % 10 == 0 + + +def korean_rrn_valid(digits: str) -> bool: + """Return True when 13 digits form a valid Korean resident registration number. + + Validates the birth-date portion and the trailing check digit. Without this, + the pattern ``\\d{6}-\\d{7}`` would match plenty of ordinary date-like text. + """ + if not digits.isdigit() or len(digits) != 13: + return False + + month = int(digits[2:4]) + day = int(digits[4:6]) + if not 1 <= month <= 12 or not 1 <= day <= 31: + return False + # The 7th digit encodes century + gender; 0 and 9 are not assigned. + if digits[6] not in "12345678": + return False + + weights = (2, 3, 4, 5, 6, 7, 8, 9, 2, 3, 4, 5) + total = sum(int(char) * weight for char, weight in zip(digits[:12], weights)) + return (11 - (total % 11)) % 10 == int(digits[12]) + + +def chinese_id_valid(value: str) -> bool: + """Return True when an 18-character string is a valid PRC resident ID. + + The final character may be ``X``. The ISO 7064 MOD 11-2 check keeps this from + matching long numeric identifiers generally. + """ + value = value.upper() + if len(value) != 18 or not value[:17].isdigit(): + return False + if value[17] not in "0123456789X": + return False + + month = int(value[10:12]) + day = int(value[12:14]) + if not 1 <= month <= 12 or not 1 <= day <= 31: + return False + + weights = (7, 9, 10, 5, 8, 4, 2, 1, 6, 3, 7, 9, 10, 5, 8, 4, 2) + checks = "10X98765432" + total = sum(int(char) * weight for char, weight in zip(value[:17], weights)) + return checks[total % 11] == value[17] + + +# --------------------------------------------------------------------------- +# Detector definitions +# --------------------------------------------------------------------------- +# +# Each detector is (category, severity, compiled pattern, validator). The +# validator receives the matched text and returns True to keep the finding. It +# exists so structural checks can suppress the bulk of false positives. +# +# severity is advisory only: it orders the summary table so the user sees +# identity-grade identifiers before softer topical signals. + +DETECTORS: list[tuple[str, str, re.Pattern[str], object]] = [ + ( + "pii.national_id.kr", + "high", + # Korean RRN, hyphenated or bare. Checksum-validated below. + re.compile(r"\b(\d{6})[-\s]?(\d{7})\b"), + lambda text: korean_rrn_valid(re.sub(r"\D", "", text)), + ), + ( + "pii.national_id.cn", + "high", + re.compile(r"\b\d{17}[\dXx]\b"), + lambda text: chinese_id_valid(text), + ), + ( + "pii.credit_card", + "high", + # Groups of 4 separated by space or hyphen, or a bare 13-19 digit run. + re.compile(r"\b(?:\d[ -]?){12,18}\d\b"), + lambda text: luhn_valid(re.sub(r"\D", "", text)), + ), + ( + "pii.bank_account", + "high", + # Account numbers are only recognised when a bank keyword sits next to + # them; the digit shape alone is far too generic to match on its own. + re.compile( + r"(?:계좌|예금주|은행|송금|账户|银行卡|转账|account\s*(?:no\.?|number)|IBAN)" + r"[^\n]{0,20}?\b\d[\d-]{7,19}\d\b", + re.IGNORECASE, + ), + None, + ), + ( + "pii.email", + "medium", + re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b"), + None, + ), + ( + "pii.phone", + "medium", + # Requires a country prefix or separators. A bare 10-digit run is not + # matched, which is what keeps order IDs and timestamps out. + re.compile( + r"(?:\+\d{1,3}[-.\s]?)?(?:0\d{1,2}|\(\d{2,3}\))[-.\s]\d{3,4}[-.\s]\d{4}\b" + r"|\b1[3-9]\d{9}\b" + ), + None, + ), + ( + "pii.address", + "low", + # Shape-based: a Korean/Chinese administrative or road marker followed by + # a number. Documented as noisy; treated as a low-severity hint. + re.compile( + r"[가-힣A-Za-z0-9]+(?:시|군|구|동|읍|면)\s*[가-힣A-Za-z0-9]*\s*" + r"[가-힣A-Za-z0-9]+(?:로|길)\s*\d+" + r"|[一-鿿]{2,}(?:省|市|区|县)[一-鿿]{0,10}(?:路|街|号)\s*\d*" + r"|\b\d{1,5}\s+[A-Z][a-z]+\s+(?:Street|St|Avenue|Ave|Road|Rd|Boulevard|Blvd)\b" + ), + None, + ), +] + +# Keyword lexicons for topical categories. These cannot be validated, so they are +# reported as signals to review rather than as confirmed sensitive content. +SENSITIVE_LEXICONS: dict[str, tuple[str, tuple[str, ...]]] = { + "sensitive.health": ( + "medium", + ( + "우울증", "공황", "정신과", "진단", "수술", "입원", "처방", "항우울제", + "임신", "유산", "장애", "암 ", "치료", + "抑郁", "焦虑症", "精神科", "住院", "手术", "确诊", "怀孕", "流产", + "depression", "anxiety disorder", "diagnosis", "diagnosed", + "hospitalized", "surgery", "prescription", "therapy", "miscarriage", + "pregnant", "disability", + ), + ), + "sensitive.sexual": ( + "high", + ( + "성관계", "섹스", "야한", "누드", + "做爱", "性关系", "裸照", + "sex", "sexual", "nude", "nsfw", "intimate photo", + ), + ), + "sensitive.minor": ( + "high", + ( + "미성년", "초등학생", "중학생", "고등학생", "만 14세", "만14세", + "未成年", "小学生", "初中生", + "minor", "underage", "under 18", "middle schooler", "elementary school", + ), + ), + "sensitive.financial": ( + "medium", + ( + "연봉", "빚", "대출", "월세", "보증금", "카드값", "파산", + "年薪", "贷款", "债务", "破产", + "salary", "debt", "loan", "mortgage", "bankrupt", "net worth", + ), + ), + "sensitive.affiliation": ( + "medium", + ( + "교회", "성당", "절에", "불교", "기독교", "천주교", "이슬람", + "정당", "지지하는 후보", "보수", "진보", + "教会", "佛教", "基督教", "党员", + "church", "mosque", "synagogue", "buddhist", "christian", "muslim", + "voted for", "political party", "conservative party", "labour party", + ), + ), +} + +# Chat-export speaker labels, e.g. "Alex: hi", "[2024-01-01 10:00] Alex: hi", +# "2024-01-01 10:00 Alex : hi". Only these structured labels are used for +# third-party detection; free-text name mentions are explicitly out of scope. +# The leading date/time is consumed explicitly rather than with a wildcard: a +# loose prefix backtracks into the clock's own colon and captures a digit as the +# speaker name. +SPEAKER_PATTERNS = ( + # ISO-ish timestamp, optionally followed by a comma: "2024-01-02 10:00 Alex: hi", + # "2024-01-02 15:15, 민준 : hi". + re.compile( + r"^\s*\[?\d{4}[-/.]\d{1,2}[-/.]\d{1,2}" + r"(?:[ T]\d{1,2}:\d{2}(?::\d{2})?)?\s*\]?\s*[,\-–]?\s*" + r"([^\s:：][^:：]{0,30}?)\s*[:：]\s" + ), + # KakaoTalk desktop export: "2024년 1월 2일 오후 3:15, 민준 : hi". + re.compile( + r"^\s*\d{4}년\s*\d{1,2}월\s*\d{1,2}일\s*(?:오전|오후)?\s*\d{1,2}:\d{2}\s*[,\-–]?\s*" + r"([^\s:：][^:：]{0,30}?)\s*[:：]\s" + ), + # Bracketed label followed by a bracketed time, as exported from KakaoTalk on + # iOS: "[민준] [오후 3:20] hi". + re.compile(r"^\s*\[([^\]\n]{1,30})\]\s*\[[^\]\n]{0,20}\]\s*\S"), + re.compile(r"^\s*([^\s:：][^:：]{0,30}?)\s*[:：]\s"), +) + +# Document and email-header lead-ins that share the "Word:" shape with speaker +# labels but are never participants. +SPEAKER_STOPWORDS = frozenset( + { + "note", "notes", "summary", "warning", "caution", "todo", "tip", "update", + "example", "output", "input", "result", "status", "source", "sources", + "http", "https", "re", "fw", "fwd", + "from", "to", "cc", "bcc", "subject", "date", "sent", "reply-to", + "참고", "요약", "주의", "발신", "수신", "제목", "날짜", + "备注", "摘要", "注意", "发件人", "收件人", "主题", "日期", + } +) + +# A label must lead at least this many lines before it counts as a speaker. Real +# participants recur; prose lead-ins like "Note:" usually appear once. This is +# the main defence against flagging ordinary sentences as third parties. +MIN_SPEAKER_LINES = 2 + + +# --------------------------------------------------------------------------- +# Masking +# --------------------------------------------------------------------------- + + +def mask_value(value: str, keep: int = 2) -> str: + """Mask the middle of a matched value so the report is safe(r) to keep. + + Keeps at most ``keep`` leading and trailing characters. Short values are + masked entirely rather than partially revealed. + """ + stripped = value.strip() + if len(stripped) <= keep * 2: + return "*" * len(stripped) + return f"{stripped[:keep]}{'*' * (len(stripped) - keep * 2)}{stripped[-keep:]}" + + +def build_preview(line: str, match_text: str, window: int = 20) -> str: + """Return a short context preview with the matched span masked out.""" + masked = mask_value(match_text) + index = line.find(match_text) + if index == -1: + return masked + start = max(0, index - window) + end = min(len(line), index + len(match_text) + window) + prefix = "…" if start > 0 else "" + suffix = "…" if end < len(line) else "" + body = line[start:index] + masked + line[index + len(match_text):end] + return f"{prefix}{body.strip()}{suffix}" + + +# --------------------------------------------------------------------------- +# Screening +# --------------------------------------------------------------------------- + + +def _normalize_name(name: str) -> str: + return re.sub(r"\s+", " ", name).strip().casefold() + + +def _speaker_label(line: str) -> tuple[str, int] | None: + """Return the (label, column) of a chat-style speaker prefix, if any. + + Only shape is judged here; whether the label is an actual participant is + decided later by the recurrence rule in :func:`collect_speaker_lines`. + """ + for pattern in SPEAKER_PATTERNS: + match = pattern.match(line) + if not match: + continue + # Strip separator punctuation a timestamp prefix may have left attached. + label = match.group(1).strip().strip(",-–·") + label = label.strip() + if not label or len(label) > 30: + return None + if len(label.split()) > 4: + return None + # Trailing sentence punctuation means this was prose, not a label. + if label[-1] in ".!?,;": + return None + if _normalize_name(label) in SPEAKER_STOPWORDS: + return None + return label, match.start(1) + 1 + return None + + +def collect_speaker_lines(text: str) -> dict[str, list[tuple[str, int, int]]]: + """Map each candidate speaker label to the lines where it leads. + + Keyed by the normalized label so casing and spacing variants collapse + together; each value keeps the original label plus line/column positions. + """ + speakers: dict[str, list[tuple[str, int, int]]] = {} + for line_number, line in enumerate(text.splitlines(), start=1): + found = _speaker_label(line) + if found is None: + continue + label, column = found + speakers.setdefault(_normalize_name(label), []).append((label, line_number, column)) + return speakers + + +def screen_text( + text: str, + source: str = "", + target_name: str | None = None, + known_participants: tuple[str, ...] = (), + include_previews: bool = True, +) -> list[dict]: + """Screen one blob of text and return a list of findings. + + ``target_name`` and ``known_participants`` are the people the user has + already accounted for; any other speaker label found in chat-style lines is + reported under ``third_party.speaker``. + """ + allowed_speakers = { + _normalize_name(name) + for name in (*known_participants, target_name or "") + if name + } + findings: list[dict] = [] + + for line_number, line in enumerate(text.splitlines(), start=1): + # DETECTORS is ordered most-specific first, so the first detector to + # claim a span wins and looser patterns cannot double-report it (a + # national ID, for instance, is also a Luhn-valid digit run). + claimed: list[tuple[int, int]] = [] + for category, severity, pattern, validator in DETECTORS: + for match in pattern.finditer(line): + matched = match.group(0) + if validator is not None and not validator(matched): + continue + start, end = match.span() + if any(start < claimed_end and claimed_start < end for claimed_start, claimed_end in claimed): + continue + claimed.append((start, end)) + findings.append( + { + "category": category, + "severity": severity, + "source": source, + "line": line_number, + "column": start + 1, + "masked_preview": build_preview(line, matched) + if include_previews + else mask_value(matched), + } + ) + + lowered = line.casefold() + for category, (severity, keywords) in SENSITIVE_LEXICONS.items(): + for keyword in keywords: + index = lowered.find(keyword.casefold()) + if index == -1: + continue + findings.append( + { + "category": category, + "severity": severity, + "source": source, + "line": line_number, + "column": index + 1, + "matched_term": keyword.strip(), + "masked_preview": build_preview(line, line[index : index + len(keyword)]) + if include_previews + else mask_value(keyword), + } + ) + break # one finding per category per line is enough to flag it + + # Third parties are judged over the whole text rather than line by line: a + # label only counts as a participant once it has led several lines. + for normalized, occurrences in collect_speaker_lines(text).items(): + if normalized in allowed_speakers or len(occurrences) < MIN_SPEAKER_LINES: + continue + label, first_line, first_column = occurrences[0] + findings.append( + { + "category": "third_party.speaker", + "severity": "high", + "source": source, + "line": first_line, + "column": first_column, + "message_count": len(occurrences), + # Shown in the clear on purpose: the user has to recognise who + # this is to decide whether their messages may be used. Unlike an + # ID number, a chat display name is not an identifier value the + # report would otherwise be leaking — it sits in the source file + # right next to this report, which never leaves the machine. + "speaker": label, + } + ) + + return findings + + +def iter_source_files(knowledge_dir: Path, suffixes: tuple[str, ...] = DEFAULT_SUFFIXES): + """Yield screenable files under a knowledge directory, skipping the report.""" + for path in sorted(knowledge_dir.rglob("*")): + if not path.is_file() or path.name == REPORT_FILENAME: + continue + if path.suffix.lower() in suffixes: + yield path + + +def screen_directory( + knowledge_dir: Path, + target_name: str | None = None, + known_participants: tuple[str, ...] = (), + include_previews: bool = True, +) -> dict: + """Screen every readable file under ``knowledge_dir`` and build a report.""" + knowledge_dir = Path(knowledge_dir) + findings: list[dict] = [] + scanned: list[str] = [] + skipped: list[str] = [] + + for path in iter_source_files(knowledge_dir): + try: + text = path.read_text(encoding="utf-8") + except (UnicodeDecodeError, OSError): + skipped.append(str(path.relative_to(knowledge_dir))) + continue + relative = str(path.relative_to(knowledge_dir)) + scanned.append(relative) + findings.extend( + screen_text( + text, + source=relative, + target_name=target_name, + known_participants=known_participants, + include_previews=include_previews, + ) + ) + + counts: dict[str, int] = {} + for finding in findings: + counts[finding["category"]] = counts.get(finding["category"], 0) + 1 + + return { + "status": STATUS_NEEDS_REVIEW if findings else STATUS_CLEAR, + "knowledge_dir": str(knowledge_dir), + "target_name": target_name, + "files_scanned": scanned, + "files_skipped": skipped, + "counts_by_category": dict(sorted(counts.items())), + "total_findings": len(findings), + "findings": findings, + "limitations": list(KNOWN_LIMITATIONS), + "note": ( + "This report itself references sensitive locations in your source " + "material. Treat it as sensitive: it stays local and is never " + "published with the generated skill." + ), + } + + +def write_report(report: dict, knowledge_dir: Path) -> Path: + """Write the report next to the screened material and return its path.""" + knowledge_dir = Path(knowledge_dir) + knowledge_dir.mkdir(parents=True, exist_ok=True) + report_path = knowledge_dir / REPORT_FILENAME + report_path.write_text( + json.dumps(report, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + return report_path + + +SEVERITY_ORDER = {"high": 0, "medium": 1, "low": 2} + + +def format_summary(report: dict, max_examples: int = 2) -> str: + """Render the human-facing summary table shown at the confirmation gate.""" + lines = [ + f"Privacy screen: {report['status']}", + f"Files scanned: {len(report['files_scanned'])}" + + (f" (skipped: {len(report['files_skipped'])})" if report["files_skipped"] else ""), + "", + ] + if not report["findings"]: + lines.append("No configured pattern matched.") + else: + by_category: dict[str, list[dict]] = {} + for finding in report["findings"]: + by_category.setdefault(finding["category"], []).append(finding) + + ordered = sorted( + by_category.items(), + key=lambda item: ( + SEVERITY_ORDER.get(item[1][0]["severity"], 3), + -len(item[1]), + item[0], + ), + ) + lines.append(f"{'category':<26}{'sev':<8}{'count':>6}") + lines.append("-" * 40) + for category, items in ordered: + lines.append(f"{category:<26}{items[0]['severity']:<8}{len(items):>6}") + lines.append("") + lines.append("Examples (masked):") + for category, items in ordered: + for finding in items[:max_examples]: + if "speaker" in finding: + detail = f"{finding['speaker']} ({finding['message_count']} messages)" + else: + detail = finding["masked_preview"] + lines.append( + f" [{category}] {finding['source']}:{finding['line']} {detail}" + ) + + lines.extend(["", "Not covered by this screen:"]) + lines.extend(f" - {item}" for item in report["limitations"]) + return "\n".join(lines) + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Screen collected relationship material for sensitive personal data" + ) + parser.add_argument( + "--knowledge-dir", + required=True, + help="Directory holding the collected material, e.g. ./knowledge/{slug}", + ) + parser.add_argument( + "--target-name", + default=None, + help="Name of the person being distilled; their speaker label is not flagged", + ) + parser.add_argument( + "--participant", + action="append", + default=[], + dest="participants", + help="Additional expected speaker label (repeatable), e.g. the user themselves", + ) + parser.add_argument( + "--no-previews", + action="store_true", + help="Omit context previews and report masked matches only", + ) + parser.add_argument( + "--fail-on-review", + action="store_true", + help="Exit with code 1 when the status is NEEDS_REVIEW (for gating scripts)", + ) + parser.add_argument("--json", action="store_true", help="Print the report as JSON") + args = parser.parse_args() + + knowledge_dir = Path(args.knowledge_dir).expanduser() + if not knowledge_dir.exists(): + parser.error(f"knowledge dir not found: {knowledge_dir}") + + report = screen_directory( + knowledge_dir, + target_name=args.target_name, + known_participants=tuple(args.participants), + include_previews=not args.no_previews, + ) + report_path = write_report(report, knowledge_dir) + + if args.json: + print(json.dumps(report, ensure_ascii=False, indent=2)) + else: + print(format_summary(report)) + print(f"\nReport written to {report_path}") + + if args.fail_on_review and report["status"] == STATUS_NEEDS_REVIEW: + raise SystemExit(1) + + +if __name__ == "__main__": + main()