Update pipelines/anonymize.py

pipelines/anonymize.py

@@ -1,63 +1,107 @@
-from typing import Tuple, Dict
 import re
-from reportlab.lib.pagesizes import A4
+from typing import Dict, Tuple, List
 from reportlab.pdfgen import canvas
-from io import BytesIO
+from reportlab.lib.pagesizes import A4
+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_LINE_RE = re.compile(r"^(?:氏名|Name)\s*[:：]?\s*(.+)$", re.MULTILINE)
 
-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]+)"),
-}
+
+def _unique(seq: List[str]) -> List[str]:
+    s = set()
+    out = []
+    for x in seq:
+        if x not in s:
+            s.add(x)
+            out.append(x)
+    return out
 
 
 def anonymize_text(text: str) -> Tuple[str, Dict[str, str]]:
-    mapping: Dict[str, str] = {}
-    redacted = text
-
-    # 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)
-
-    # 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)
-
-    # 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)
-
-    return redacted, mapping
+    """
+    氏名・メール・電話を難読化。置換マップも返す。
+    - 氏名： '氏名: 山田太郎' / 'Name: Taro Yamada' などの行を検出
+    - メール/電話：正規表現で検出
+    """
+    replace_map: Dict[str, str] = {}
+
+    # emails
+    emails = _unique(EMAIL_RE.findall(text))
+    for i, e in enumerate(emails, start=1):
+        token = f"<EMAIL_{i}>"
+        replace_map[e] = token
+
+    # phones
+    phones = _unique([p.strip() for p in PHONE_RE.findall(text)])
+    for i, p in enumerate(phones, start=1):
+        token = f"<PHONE_{i}>"
+        replace_map[p] = token
+
+    # names (line-based)
+    names = []
+    for m in NAME_LINE_RE.finditer(text):
+        nm = m.group(1).strip()
+        if nm:
+            names.append(nm)
+    names = _unique(names)
+    for i, n in enumerate(names, start=1):
+        # 長すぎる場合を適当にトリム
+        n_short = n[:80]
+        replace_map[n_short] = f"<NAME_{i}>"
+
+    # 置換（長い文字列から先に）
+    def _repl_all(t: str, mapping: Dict[str, str]) -> str:
+        for k in sorted(mapping.keys(), key=len, reverse=True):
+            t = t.replace(k, mapping[k])
+        return t
+
+    anonymized = _repl_all(text, replace_map)
+    return anonymized, replace_map
 
 
 def render_anonymized_pdf(text: str) -> bytes:
-    buf = BytesIO()
+    """
+    単純にA4 1カラムでテキストを書き出す簡易PDF。
+    （元PDFのレイアウト再現は行わないが、“匿名化済み本文”を配布できる）
+    """
+    buf = io.BytesIO()
     c = canvas.Canvas(buf, pagesize=A4)
     width, height = A4
 
-    # 簡易レイアウト：1ページあたり約60行
-    x_margin = 50
-    y = height - 50
-    line_height = 12
+    left = 15 * mm
+    top = height - 15 * mm
+    line_height = 6 * mm
+    x = left
+    y = top
+
+    c.setFont("Helvetica", 10)
 
     for line in text.splitlines():
-        if y < 50:
+        # 改ページ
+        if y < 20 * mm:
             c.showPage()
-            y = height - 50
-        c.drawString(x_margin, y, line[:120])
-        y -= line_height
+            c.setFont("Helvetica", 10)
+            y = top
+        # 長い行を適当に折り返し
+        max_chars = 110  # おおよその幅
+        if len(line) <= max_chars:
+            c.drawString(x, y, line)
+            y -= line_height
+        else:
+            start = 0
+            while start < len(line):
+                seg = line[start:start + max_chars]
+                c.drawString(x, y, seg)
+                y -= line_height
+                if y < 20 * mm:
+                    c.showPage()
+                    c.setFont("Helvetica", 10)
+                    y = top
+                start += max_chars
 
     c.showPage()
     c.save()