Update pipelines/anonymize.py

pipelines/anonymize.py

@@ -1,55 +1,64 @@
-import io
+from typing import Tuple, Dict
 import re
-from reportlab.pdfgen import canvas
 from reportlab.lib.pagesizes import A4
-from reportlab.lib.units import mm
+from reportlab.pdfgen import canvas
+from io import BytesIO
+
+
+PII_PATTERNS = {
+    "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}"),
+    "NAME_HINT": re.compile(r"(?:氏名|Name)\s*[:：]?\s*([^\n]+)"),
+}
 
-RE_EMAIL = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
-RE_PHONE = re.compile(r"(?:\+?\d{1,3}[ -]?)?(?:\(\d{2,4}\)[ -]?)?\d{2,4}[ -]?\d{2,4}[ -]?\d{3,4}")
-RE_NAME = re.compile(r"(?:(?:氏名[:：]?)[\t ]*)(\S+)")
 
-def anonymize_text(text: str):
-    anon_map = {}
+def anonymize_text(text: str) -> Tuple[str, Dict[str, str]]:
+    mapping: Dict[str, str] = {}
+    redacted = text
 
-    def repl_email(m):
-        val = m.group(0)
-        key = "<EMAIL>"
-        anon_map.setdefault(key, []).append(val)
-        return key
+    # Email
+    for m in set(PII_PATTERNS["EMAIL"].findall(text)):
+        token = f"[EMAIL_{abs(hash(m))%10000:04d}]"
+        mapping[m] = token
+        redacted = redacted.replace(m, token)
 
-    def repl_phone(m):
-        val = m.group(0)
-        key = "<PHONE>"
-        anon_map.setdefault(key, []).append(val)
-        return key
+    # Phone
+    for m in set(PII_PATTERNS["PHONE"].findall(text)):
+        m_str = m if isinstance(m, str) else "".join(m)
+        if len(m_str.strip()) < 7:
+            continue
+        token = f"[TEL_{abs(hash(m_str))%10000:04d}]"
+        mapping[m_str] = token
+        redacted = redacted.replace(m_str, token)
 
-    def repl_name(m):
-        val = m.group(1)
-        key = "<NAME>"
-        anon_map.setdefault(key, []).append(val)
-        return m.group(0).replace(val, key)
+    # Name line (hint-based)
+    for nm in set(PII_PATTERNS["NAME_HINT"].findall(text)):
+        nm_clean = nm.strip()
+        if nm_clean:
+            token = f"[NAME_{abs(hash(nm_clean))%10000:04d}]"
+            mapping[nm_clean] = token
+            redacted = redacted.replace(nm_clean, token)
 
-    t = RE_EMAIL.sub(repl_email, text)
-    t = RE_PHONE.sub(repl_phone, t)
-    t = RE_NAME.sub(repl_name, t)
+    return redacted, mapping
 
-    return t, anon_map
 
 def render_anonymized_pdf(text: str) -> bytes:
-    buf = io.BytesIO()
+    buf = BytesIO()
     c = canvas.Canvas(buf, pagesize=A4)
+    width, height = A4
 
-    margin = 15 * mm
-    line_height = 5 * mm
-    x = margin
-    y = A4[1] - margin
+    # 簡易レイアウト：1ページあたり約60行
+    x_margin = 50
+    y = height - 50
+    line_height = 12
 
     for line in text.splitlines():
-        if y < margin:
+        if y < 50:
             c.showPage()
-            y = A4[1] - margin
-        c.drawString(x, y, line[:1200])
+            y = height - 50
+        c.drawString(x_margin, y, line[:120])
         y -= line_height
 
+    c.showPage()
     c.save()
     return buf.getvalue()