init commit

src/application/services/kyc_personal_data.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+from datetime import date,datetime
+from typing import Any
+from src.application.domain.dto import KycPersonalData
+from src.application.domain.exceptions import ApplicationException
+
+
+FIELD_ALIASES = {
+    'first_name': {'first_name','name','given_name','имя'},
+    'last_name': {'last_name','surname','family_name','фамилия'},
+    'middle_name': {'middle_name','patronymic','отчество'},
+    'birth_date': {'birth_date','birthdate','date_birth','birthday','дата рождения'},
+    'inn': {'inn','tax_id','инн'},
+}
+
+
+def extract_personal_data(data: Any) -> KycPersonalData:
+    values: dict[str,str] = {}
+    for key,value in _walk(data):
+        normalized = _normalize_key(key)
+        for field,aliases in FIELD_ALIASES.items():
+            if field not in values and normalized in aliases and value not in (None,''):
+                values[field] = str(value).strip()
+
+    missing = [field for field in ('first_name','last_name','birth_date') if not values.get(field)]
+    if missing:
+        raise ApplicationException(status_code=422,message='KYC personal data is incomplete')
+
+    return KycPersonalData(
+        first_name=values['first_name'],
+        last_name=values['last_name'],
+        middle_name=values.get('middle_name'),
+        birth_date=str(_parse_date(values['birth_date'])),
+        inn=values.get('inn'),
+    )
+
+
+def ensure_adult(birth_date: date) -> None:
+    today = date.today()
+    try:
+        adult_from = date(today.year - 18,today.month,today.day)
+    except ValueError:
+        adult_from = date(today.year - 18,2,28)
+    if birth_date > adult_from:
+        raise ApplicationException(status_code=403,message='KYC is unavailable for users under 18')
+
+
+def parse_birth_date(value: str) -> date:
+    return _parse_date(value)
+
+
+def _walk(data: Any) -> list[tuple[str,Any]]:
+    items: list[tuple[str,Any]] = []
+    if isinstance(data,dict):
+        for key,value in data.items():
+            if isinstance(value,dict | list):
+                items.extend(_walk(value))
+            else:
+                items.append((str(key),value))
+    elif isinstance(data,list):
+        for item in data:
+            items.extend(_walk(item))
+    return items
+
+
+def _normalize_key(key: str) -> str:
+    return key.strip().lower().replace('-','_').replace(' ','_')
+
+
+def _parse_date(value: str) -> date:
+    clean = value.strip()
+    formats = ('%Y-%m-%d','%d.%m.%Y','%d-%m-%Y','%d/%m/%Y','%Y.%m.%d')
+    for date_format in formats:
+        try:
+            return datetime.strptime(clean,date_format).date()
+        except ValueError:
+            continue
+    raise ApplicationException(status_code=422,message='KYC birth date has invalid format')