Update pipelines/anonymize.py

pipelines/anonymize.py

@@ -1,66 +1,59 @@
 import re
-from typing import Tuple, Dict, List
-from reportlab.pdfgen import canvas
+from typing import Tuple, Dict
 from reportlab.lib.pagesizes import A4
+from reportlab.pdfgen import canvas
 from reportlab.lib.units import mm
-import io
 
-# 簡易パターン（必要に応じて精緻化してください）
 EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
 PHONE_RE = re.compile(r"(?:\+?\d{1,3}[ -]?)?(?:\(\d{2,4}\)[ -]?)?\d{2,4}[ -]?\d{2,4}[ -]?\d{3,4}")
-# 氏名候補（行頭・「氏名:」「Name:」付近など）
-NAME_LINE_RE = re.compile(r"^(?:氏名|Name)\s*[:：]\s*([^\s　].{0,40})$", re.MULTILINE)
+NAME_HINT = re.compile(r"^(氏名|Name)[:：]?\s*(.+)$", re.MULTILINE)
+
 
-def anonymize_text(text: str) -> Tuple[str, Dict[str, List[str]]]:
-    replacements = {"EMAIL": [], "PHONE": [], "NAME": []}
-    # メール
-    def _email_sub(m):
-        val = m.group(0)
-        replacements["EMAIL"].append(val)
-        return "[EMAIL REDACTED]"
-    text = EMAIL_RE.sub(_email_sub, text)
+def anonymize_text(text: str) -> Tuple[str, Dict[str, str]]:
+    mapping: Dict[str, str] = {}
 
-    # 電話
-    def _phone_sub(m):
-        val = m.group(0)
-        # 数字が少なすぎるものはノイズとして除外
-        digits = re.sub(r"\D", "", val)
-        if len(digits) < 8:
-            return val
-        replacements["PHONE"].append(val)
-        return "[PHONE REDACTED]"
-    text = PHONE_RE.sub(_phone_sub, text)
+    def _mask_all(pattern: re.Pattern, label: str, s: str) -> str:
+        idx = 1
+        def _repl(m):
+            nonlocal idx
+            key = m.group(0)
+            if key not in mapping:
+                mapping[key] = f"<{label}{idx}>"
+                idx += 1
+            return mapping[key]
+        return pattern.sub(_repl, s)
 
-    # 氏名行
-    for m in NAME_LINE_RE.finditer(text):
-        name = m.group(1).strip()
-        if name and name not in replacements["NAME"]:
-            replacements["NAME"].append(name)
-    text = NAME_LINE_RE.sub(lambda m: m.group(0).split(":")[0] + ": [NAME REDACTED]", text)
+    out = text
+    out = _mask_all(EMAIL_RE, "EMAIL", out)
+    out = _mask_all(PHONE_RE, "TEL", out)
 
-    return text, replacements
+    # 名前行の簡易マスク
+    for m in NAME_HINT.finditer(text):
+        full = m.group(0)
+        name = m.group(2).strip()
+        if name and name not in mapping:
+            mapping[name] = "<NAME>"
+        out = out.replace(full, f"{m.group(1)}: <NAME>")
+
+    return out, mapping
 
 
 def render_anonymized_pdf(text: str) -> bytes:
-    buf = io.BytesIO()
+    from io import BytesIO
+    buf = BytesIO()
     c = canvas.Canvas(buf, pagesize=A4)
     width, height = A4
-    # マージン
-    x0, y0 = 15 * mm, height - 20 * mm
-    line_h = 6.0 * mm
-    c.setFont("Helvetica", 10)
 
-    lines = text.splitlines()
-    y = y0
-    for line in lines:
-        # ページ送り
+    line_height = 6 * mm
+    x = 15 * mm
+    y = height - 20 * mm
+
+    for line in text.splitlines():
         if y < 20 * mm:
             c.showPage()
-            c.setFont("Helvetica", 10)
-            y = y0
-        c.drawString(x0, y, line[:2000])  # 超長行は切り詰め
-        y -= line_h
+            y = height - 20 * mm
+        c.drawString(x, y, line[:2000])
+        y -= line_height
 
-    c.showPage()
     c.save()
     return buf.getvalue()