Update pipelines/anonymize.py

pipelines/anonymize.py

@@ -1,45 +1,66 @@
 import re
-from typing import Dict, Tuple
-from reportlab.platypus import SimpleDocTemplate, Preformatted
-from reportlab.lib.styles import getSampleStyleSheet
+from typing import Tuple, Dict, List
+from reportlab.pdfgen import canvas
 from reportlab.lib.pagesizes import A4
+from reportlab.lib.units import mm
 import io
 
-_EMAIL = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
-_PHONE = re.compile(r"(?:\+?\d{1,3}[ -]?)?(?:\(\d{2,4}\)[ -]?)?\d{2,4}[ -]?\d{2,4}[ -]?\d{3,4}")
+# 簡易パターン（必要に応じて精緻化してください）
+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 = re.compile(r"(?:氏名[:：]?\s*|Name[:：]?\s*)([^\s　]+)")
+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 _mask(match, prefix):
-        val = match.group(0)
-        if val not in mapping:
-            mapping[val] = f"[{prefix}_{len(mapping)+1}]"
-        return mapping[val]
+    # 電話
+    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)
 
-    # Email / Phone
-    text = _EMAIL.sub(lambda m: _mask(m, "EMAIL"), text)
-    text = _PHONE.sub(lambda m: _mask(m, "PHONE"), text)
+    # 氏名行
+    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)
 
-    # 氏名（ヒューリスティック）
-    def _name_mask(m):
-        full = m.group(0)
-        name = m.group(1)
-        if name not in mapping:
-            mapping[name] = f"[NAME_{len(mapping)+1}]"
-        return full.replace(name, mapping[name])
+    return text, replacements
 
-    text = _NAME.sub(_name_mask, text)
-
-    return text, mapping
 
 def render_anonymized_pdf(text: str) -> bytes:
     buf = io.BytesIO()
-    doc = SimpleDocTemplate(buf, pagesize=A4)
-    styles = getSampleStyleSheet()
-    # 改行維持
-    flow = [Preformatted(text, styles["Code"])]
-    doc.build(flow)
+    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:
+        # ページ送り
+        if y < 20 * mm:
+            c.showPage()
+            c.setFont("Helvetica", 10)
+            y = y0
+        c.drawString(x0, y, line[:2000])  # 超長行は切り詰め
+        y -= line_h
+
+    c.showPage()
+    c.save()
     return buf.getvalue()