Introduce Email and CPF Value Objects

api/src/bracc/models/identifier.py

@@ -0,0 +1,123 @@
+import re
+
+
+def clean_identifier(raw: str) -> str:
+    return re.sub(r"\D", "", raw)
+
+
+class Document:
+    def __init__(self, value: str):
+        self._value = clean_identifier(value)
+
+    def get_value(self) -> str:
+        return self._value
+
+
+class Cpf(Document):
+    _PATTERN = re.compile(r"^\d{11}$")
+
+    def __init__(self, value: str):
+        super().__init__(value)
+
+        if not self.is_valid(self._value):
+            raise ValueError("CPF inválido")
+
+    @classmethod
+    def is_valid(cls, value: str) -> bool:
+        value = clean_identifier(value)
+
+        if not cls._PATTERN.match(value):
+            return False
+
+        if value == value[0] * 11:
+            return False
+
+        if not cls._validate_digit(value, 9):
+            return False
+
+        return cls._validate_digit(value, 10)
+
+    @staticmethod
+    def _validate_digit(cpf: str, position: int) -> bool:
+        total = 0
+        weight = position + 1
+
+        for i in range(position):
+            total += int(cpf[i]) * weight
+            weight -= 1
+
+        remainder = total % 11
+        digit = 0 if remainder < 2 else 11 - remainder
+
+        return digit == int(cpf[position])
+
+    def pretty(self) -> str:
+        return f"{self._value[:3]}.{self._value[3:6]}.{self._value[6:9]}-{self._value[9:]}"
+
+    def mask(self) -> str:
+        return f"***.***.{self._value[6:9]}-{self._value[9:]}"
+
+    def mask_raw(self) -> str:
+        return f"*******{self._value[7:]}"
+
+
+class Cnpj(Document):
+    _PATTERN = re.compile(r"^\d{14}$")
+
+    def __init__(self, value: str):
+        super().__init__(value)
+
+        if not self.is_valid(self._value):
+            raise ValueError("CNPJ inválido")
+
+    @classmethod
+    def is_valid(cls, value: str) -> bool:
+        value = clean_identifier(value)
+
+        if not cls._PATTERN.match(value):
+            return False
+
+        if value == value[0] * 14:
+            return False
+
+        if not cls._validate_digit(value, 12):
+            return False
+
+        return cls._validate_digit(value, 13)
+
+    @staticmethod
+    def _validate_digit(cnpj: str, position: int) -> bool:
+        if position == 12:
+            weights = [5,4,3,2,9,8,7,6,5,4,3,2]
+        else:
+            weights = [6,5,4,3,2,9,8,7,6,5,4,3,2]
+
+        total = sum(int(cnpj[i]) * weights[i] for i in range(position))
+
+        remainder = total % 11
+        digit = 0 if remainder < 2 else 11 - remainder
+
+        return digit == int(cnpj[position])
+
+    def pretty(self) -> str:
+        return (
+    f"{self._value[:2]}.{self._value[2:5]}."
+    f"{self._value[5:8]}/{self._value[8:12]}-"
+    f"{self._value[12:]}"
+)
+    def mask(self) -> str:
+        return f"**.***.***/{self._value[8:12]}-{self._value[12:]}"
+
+
+def get_identifier(value: str) -> Cpf | Cnpj | None:
+    try:
+        return Cpf(value)
+    except ValueError:
+        pass
+
+    try:
+        return Cnpj(value)
+    except ValueError:
+        pass
+
+    return None

api/src/bracc/routers/entity.py

@@ -15,6 +15,7 @@
     TimelineEvent,
     TimelineResponse,
 )
+from bracc.models.identifier import get_identifier
 from bracc.services.intelligence_provider import IntelligenceProvider
 from bracc.services.neo4j_service import execute_query, execute_query_single, sanitize_props
 from bracc.services.public_guard import (
@@ -29,13 +30,8 @@
 
 router = APIRouter(prefix="/api/v1/entity", tags=["entity"])
 
-CPF_PATTERN = re.compile(r"^\d{11}$")
-CNPJ_PATTERN = re.compile(r"^\d{14}$")
 
 
-def _clean_identifier(raw: str) -> str:
-    return re.sub(r"[.\-/]", "", raw)
-
 
 def _is_pep(properties: dict[str, Any]) -> bool:
     role = str(properties.get("role", "")).lower()
@@ -86,43 +82,43 @@ def _node_to_entity(
     )
 
 
-def _format_cpf(digits: str) -> str:
-    """Format an 11-digit string as CPF: 123.456.789-00."""
-    return f"{digits[:3]}.{digits[3:6]}.{digits[6:9]}-{digits[9:]}"
-
-
-def _format_cnpj(digits: str) -> str:
-    """Format a 14-digit string as CNPJ: 12.345.678/0001-00."""
-    return f"{digits[:2]}.{digits[2:5]}.{digits[5:8]}/{digits[8:12]}-{digits[12:]}"
-
 
 @router.get("/{cpf_or_cnpj}", response_model=EntityResponse)
 async def get_entity(
     cpf_or_cnpj: str,
     session: Annotated[AsyncSession, Depends(get_session)],
 ) -> EntityResponse:
     enforce_entity_lookup_policy(cpf_or_cnpj)
-    identifier = _clean_identifier(cpf_or_cnpj)
 
-    if not CPF_PATTERN.match(identifier) and not CNPJ_PATTERN.match(identifier):
-        raise HTTPException(status_code=400, detail="Invalid CPF or CNPJ format")
+    identifier = get_identifier(cpf_or_cnpj)
 
-    if CPF_PATTERN.match(identifier):
-        identifier_formatted = _format_cpf(identifier)
-    else:
-        identifier_formatted = _format_cnpj(identifier)
+    if identifier is None:
+        raise HTTPException(
+            status_code=400,
+            detail="Invalid CPF or CNPJ"
+        )
 
     record = await execute_query_single(
         session,
         "entity_lookup",
-        {"identifier": identifier, "identifier_formatted": identifier_formatted},
+        {
+            "identifier": identifier.get_value(),
+            "identifier_formatted": identifier.pretty(),
+        },
     )
+
     if record is None:
-        raise HTTPException(status_code=404, detail="Entity not found")
+        raise HTTPException(
+            status_code=404,
+            detail="Entity not found"
+        )
+
     enforce_person_access_policy(record["entity_labels"])
 
     return _node_to_entity(
-        record["e"], record["entity_labels"], record["entity_id"]
+        record["e"],
+        record["entity_labels"],
+        record["entity_id"],
     )