# handlers/user/kyc.py
from __future__ import annotations
import asyncio
import logging
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
from html import escape
from pathlib import Path
from typing import Any, Final, Optional
from zoneinfo import ZoneInfo
from aiogram import F, Router
from aiogram.enums import ParseMode
from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from aiogram.types import (
CallbackQuery,
InlineKeyboardButton,
InlineKeyboardMarkup,
KeyboardButton,
Message,
ReplyKeyboardMarkup,
ReplyKeyboardRemove,
)
from core.config import settings
logger = logging.getLogger(__name__)
router = Router(
name="user_kyc_complete"
)
# ============================================================
# Constants
# ============================================================
TEHRAN: Final = ZoneInfo("Asia/Tehran")
STATUS_PENDING: Final[str] = "pending"
STATUS_APPROVED: Final[str] = "approved"
STATUS_REJECTED: Final[str] = "rejected"
DOC_NATIONAL_CARD: Final[str] = "national_card"
DOC_PASSPORT: Final[str] = "passport"
DOCUMENT_LABELS: Final[dict[str, str]] = {
DOC_NATIONAL_CARD: "کارت ملی",
DOC_PASSPORT: "پاسپورت",
}
# Custom Emoji IDs currently used elsewhere in this project.
EMOJI: Final[dict[str, str]] = {
"shield": "5987803826512466184",
"success": "5951861774947979620",
"danger": "6030430277413638534",
"user": "5431577498364158238",
"id": "5316809461044623413",
"back": "5447183459602669338",
"home": "5305265301917549162",
"file": "5305265301917549162",
"phone": "5929038627974680867",
"warning": "5987803826512466184",
}
# ============================================================
# Exceptions
# ============================================================
class KYCError(Exception):
"""Base error for KYC flow."""
class KYCValidationError(KYCError):
"""Invalid user input or incomplete KYC data."""
class KYCAlreadyPending(KYCError):
"""User already has a pending KYC case."""
class KYCAlreadyHandled(KYCError):
"""Case was already approved or rejected."""
class KYCNotFound(KYCError):
"""Requested KYC case does not exist."""
# ============================================================
# Models
# ============================================================
@dataclass(slots=True, frozen=True)
class KYCRequest:
id: int
user_id: int
username: Optional[str]
full_name: Optional[str]
phone_number: str
phone_verified: bool
phone_verified_at: Optional[str]
document_type: str
document_front_file_id: str
document_back_file_id: Optional[str]
bank_card_file_id: str
status: str
rejection_reason: Optional[str]
submitted_at: str
reviewed_at: Optional[str]
reviewed_by: Optional[int]
created_at: str
updated_at: str
attempt_no: int = 1
@property
def document_label(self) -> str:
return DOCUMENT_LABELS.get(
self.document_type,
self.document_type,
)
# ============================================================
# FSM
# ============================================================
class KYCStates(StatesGroup):
waiting_for_phone = State()
waiting_for_document_type = State()
waiting_for_front = State()
waiting_for_back = State()
waiting_for_bank_card = State()
waiting_for_confirmation = State()
class OwnerKYCStates(StatesGroup):
waiting_for_reject_reason = State()
# ============================================================
# Configuration helpers
# ============================================================
def database_path() -> str:
return str(
getattr(settings, "DB_PATH", None)
or getattr(settings, "DATABASE_PATH", None)
or "matrix_bot.db"
)
def admin_ids() -> list[int]:
"""Read owner/admin IDs without hardcoding them here."""
getter = getattr(
settings,
"get_admin_list",
None,
)
if callable(getter):
try:
return sorted({
int(value)
for value in getter()
})
except Exception:
logger.exception(
"Could not read KYC admin list."
)
raw = getattr(
settings,
"ADMIN_IDS",
None,
)
result: set[int] = set()
if isinstance(raw, str):
values = raw.split(",")
elif isinstance(raw, (list, tuple, set)):
values = raw
else:
values = []
for value in values:
try:
result.add(
int(str(value).strip())
)
except (TypeError, ValueError):
continue
return sorted(result)
def is_admin(user_id: int) -> bool:
return int(user_id) in set(
admin_ids()
)
# ============================================================
# Time / display helpers
# ============================================================
def utc_now_iso() -> str:
return datetime.now(
timezone.utc
).isoformat(
timespec="seconds"
)
def tehran_now_text() -> str:
return datetime.now(
TEHRAN
).strftime(
"%Y/%m/%d - %H:%M:%S"
)
def to_tehran_text(
value: Optional[str],
) -> str:
raw = str(
value
or ""
).strip()
if not raw:
return "نامشخص"
try:
normalized = raw.replace(
"Z",
"+00:00",
)
dt = datetime.fromisoformat(
normalized
)
if dt.tzinfo is None:
dt = dt.replace(
tzinfo=timezone.utc
)
return dt.astimezone(
TEHRAN
).strftime(
"%Y/%m/%d - %H:%M:%S"
)
except Exception:
return raw
def normalize_phone(
phone_number: str,
) -> str:
raw = str(
phone_number
or ""
).strip()
for ch in (
" ",
"-",
"(",
")",
):
raw = raw.replace(
ch,
"",
)
if raw.startswith("00"):
raw = "+" + raw[2:]
if not raw.startswith("+"):
raw = "+" + raw
digits = raw[1:]
if (
not digits.isdigit()
or len(digits) < 7
or len(digits) > 15
):
raise KYCValidationError(
"شماره تلفن معتبر نیست."
)
return raw
def mask_phone(
phone_number: str,
) -> str:
phone = str(
phone_number
or ""
)
if len(phone) <= 7:
return phone
hidden = max(
3,
len(phone) - 7,
)
return (
phone[:4]
+ ("*" * hidden)
+ phone[-3:]
)
def status_label(
status: str,
) -> str:
return {
STATUS_PENDING:
"⏳ در انتظار بررسی مالک",
STATUS_APPROVED:
"✅ تأیید شده",
STATUS_REJECTED:
"❌ رد شده",
}.get(
str(status),
str(status),
)
def document_label(
document_type: str,
) -> str:
return DOCUMENT_LABELS.get(
document_type,
document_type,
)
# ============================================================
# Safe Telegram UI helpers
# ============================================================
def _supported_button_fields() -> set[str]:
fields = getattr(
InlineKeyboardButton,
"model_fields",
None,
)
if not fields:
fields = getattr(
InlineKeyboardButton,
"__fields__",
{},
)
return set(
getattr(
fields,
"keys",
lambda: [],
)()
)
def btn(
text: str,
callback_data: str,
*,
style: Optional[str] = None,
icon: Optional[str] = None,
) -> InlineKeyboardButton:
"""
Premium-button compatible with both newer and older aiogram builds.
"""
kwargs: dict[str, Any] = {
"text": text,
"callback_data": callback_data,
}
supported = _supported_button_fields()
if (
style
and "style" in supported
):
kwargs["style"] = style
if (
icon
and "icon_custom_emoji_id" in supported
):
kwargs[
"icon_custom_emoji_id"
] = EMOJI.get(
icon,
icon,
)
return InlineKeyboardButton(
**kwargs
)
async def safe_callback_answer(
callback: CallbackQuery,
text: Optional[str] = None,
show_alert: bool = False,
) -> None:
try:
await callback.answer(
text=text,
show_alert=show_alert,
)
except TelegramBadRequest:
pass
async def safe_edit(
callback: CallbackQuery,
text: str,
reply_markup: Optional[
InlineKeyboardMarkup
] = None,
) -> None:
if not callback.message:
return
try:
if (
getattr(
callback.message,
"photo",
None,
)
or getattr(
callback.message,
"document",
None,
)
):
await callback.message.edit_caption(
caption=text,
reply_markup=reply_markup,
parse_mode=ParseMode.HTML,
)
else:
await callback.message.edit_text(
text=text,
reply_markup=reply_markup,
parse_mode=ParseMode.HTML,
)
return
except TelegramBadRequest as exc:
if (
"message is not modified"
in str(exc).lower()
):
return
try:
await callback.message.answer(
text,
reply_markup=reply_markup,
parse_mode=ParseMode.HTML,
)
except Exception:
logger.exception(
"Could not render KYC callback message."
)
# ============================================================
# Keyboards
# ============================================================
def phone_share_keyboard() -> ReplyKeyboardMarkup:
"""
Official Telegram contact button.
The handler still verifies contact.user_id == message.from_user.id.
"""
return ReplyKeyboardMarkup(
keyboard=[
[
KeyboardButton(
text="📱 ارسال شماره همین اکانت",
request_contact=True,
)
]
],
resize_keyboard=True,
one_time_keyboard=True,
selective=True,
input_field_placeholder=(
"برای ادامه شماره همین اکانت را ارسال کنید"
),
)
def remove_reply_keyboard() -> ReplyKeyboardRemove:
return ReplyKeyboardRemove()
def cancel_inline_keyboard() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
inline_keyboard=[
[
btn(
"انصراف از احراز هویت",
"kyc_cancel",
style="danger",
icon="danger",
)
]
]
)
def document_type_keyboard() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
inline_keyboard=[
[
btn(
"کارت ملی",
"kyc_doc_national_card",
style="primary",
icon="id",
),
btn(
"پاسپورت",
"kyc_doc_passport",
style="primary",
icon="file",
),
],
[
btn(
"انصراف",
"kyc_cancel",
style="danger",
icon="danger",
)
],
]
)
def confirmation_keyboard() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
inline_keyboard=[
[
btn(
"ثبت و ارسال برای مالک",
"kyc_submit_case",
style="success",
icon="success",
)
],
[
btn(
"ارسال دوباره مدارک",
"kyc_restart_documents",
style="primary",
icon="file",
)
],
[
btn(
"انصراف",
"kyc_cancel",
style="danger",
icon="danger",
)
],
]
)
def user_status_keyboard(
status: str,
) -> InlineKeyboardMarkup:
rows: list[list[
InlineKeyboardButton
]] = []
if status == STATUS_REJECTED:
rows.append(
[
btn(
"شروع مجدد احراز هویت",
"user_kyc_start",
style="success",
icon="shield",
)
]
)
rows.append(
[
btn(
"حساب کاربری",
"user_profile",
style="primary",
icon="user",
)
]
)
rows.append(
[
btn(
"منوی اصلی",
"main_menu",
style="primary",
icon="home",
)
]
)
return InlineKeyboardMarkup(
inline_keyboard=rows
)
def owner_case_keyboard(
request_id: int,
) -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
inline_keyboard=[
[
btn(
"تأیید احراز هویت",
f"kyc_case_approve_{request_id}",
style="success",
icon="success",
),
btn(
"رد احراز هویت",
f"kyc_case_reject_{request_id}",
style="danger",
icon="danger",
),
],
[
btn(
"مشاهده همه مدارک",
f"kyc_case_view_{request_id}",
style="primary",
icon="file",
)
],
[
btn(
"صف احراز هویت",
"admin_kyc_queue",
style="primary",
icon="back",
)
],
]
)
def owner_menu_keyboard() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
inline_keyboard=[
[
btn(
"درخواستهای منتظر",
"admin_kyc_queue",
style="primary",
icon="file",
)
],
[
btn(
"بازگشت به مدیریت",
"admin_main",
style="primary",
icon="back",
)
],
]
)
def owner_queue_keyboard(
requests: list[KYCRequest],
) -> InlineKeyboardMarkup:
rows: list[list[
InlineKeyboardButton
]] = []
for item in requests:
display = (
f"#{item.id} • "
f"{item.full_name or item.user_id}"
)
rows.append(
[
btn(
display[:55],
f"kyc_case_view_{item.id}",
style="primary",
icon="user",
)
]
)
rows.append(
[
btn(
"بروزرسانی",
"admin_kyc_queue",
style="primary",
icon="shield",
),
btn(
"بازگشت",
"admin_kyc_menu",
style="danger",
icon="back",
),
]
)
return InlineKeyboardMarkup(
inline_keyboard=rows
)
def owner_reject_cancel_keyboard(
request_id: int,
) -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
inline_keyboard=[
[
btn(
"لغو رد درخواست",
f"kyc_case_view_{request_id}",
style="danger",
icon="back",
)
]
]
)
# Compatibility names used by older code.
def back_keyboard() -> InlineKeyboardMarkup:
return cancel_inline_keyboard()
def success_keyboard() -> InlineKeyboardMarkup:
return user_status_keyboard(
STATUS_PENDING
)
def kyc_retry_keyboard() -> InlineKeyboardMarkup:
return user_status_keyboard(
STATUS_REJECTED
)
def admin_kyc_keyboard(
request_id: int,
) -> InlineKeyboardMarkup:
return owner_case_keyboard(
request_id
)
# ============================================================
# Database service
# ============================================================
class KYCService:
"""
Self-contained SQLite KYC persistence.
Only Telegram file IDs are stored for document images; this module does
not download identity document bytes to local disk.
"""
def __init__(
self,
db_path: Optional[str] = None,
) -> None:
self.db_path = str(
db_path
or database_path()
)
self._schema_lock = asyncio.Lock()
self._schema_ready = False
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(
self.db_path,
timeout=20,
check_same_thread=False,
)
conn.row_factory = sqlite3.Row
conn.execute(
"PRAGMA busy_timeout = 20000"
)
conn.execute(
"PRAGMA journal_mode = WAL"
)
return conn
@staticmethod
def _columns(
conn: sqlite3.Connection,
table: str,
) -> set[str]:
return {
str(row["name"])
for row in conn.execute(
f"PRAGMA table_info({table})"
).fetchall()
}
def _ensure_schema_sync(self) -> None:
conn = self._connect()
try:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS kyc_requests (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
username TEXT,
full_name TEXT,
phone_number TEXT NOT NULL DEFAULT '',
phone_verified INTEGER NOT NULL DEFAULT 0,
phone_verified_at TEXT,
document_type TEXT NOT NULL DEFAULT '',
document_front_file_id TEXT NOT NULL DEFAULT '',
document_back_file_id TEXT,
bank_card_file_id TEXT NOT NULL DEFAULT '',
selfie_file_id TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'pending',
rejection_reason TEXT,
submitted_at TEXT NOT NULL DEFAULT '',
reviewed_at TEXT,
reviewed_by INTEGER,
attempt_no INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL DEFAULT '',
updated_at TEXT NOT NULL DEFAULT ''
)
"""
)
# Lightweight migration for older/minimal kyc_requests tables.
required: dict[str, str] = {
"username": "TEXT",
"full_name": "TEXT",
"phone_number": "TEXT NOT NULL DEFAULT ''",
"phone_verified": "INTEGER NOT NULL DEFAULT 0",
"phone_verified_at": "TEXT",
"document_type": "TEXT NOT NULL DEFAULT ''",
"document_front_file_id": "TEXT NOT NULL DEFAULT ''",
"document_back_file_id": "TEXT",
"bank_card_file_id": "TEXT NOT NULL DEFAULT ''",
# legacy-only: old versions used selfie; kept only
# so existing databases remain compatible.
"selfie_file_id": "TEXT NOT NULL DEFAULT ''",
"status": "TEXT NOT NULL DEFAULT 'pending'",
"rejection_reason": "TEXT",
"submitted_at": "TEXT NOT NULL DEFAULT ''",
"reviewed_at": "TEXT",
"reviewed_by": "INTEGER",
"attempt_no": "INTEGER NOT NULL DEFAULT 1",
"created_at": "TEXT NOT NULL DEFAULT ''",
"updated_at": "TEXT NOT NULL DEFAULT ''",
}
columns = self._columns(
conn,
"kyc_requests",
)
for column, sql_type in required.items():
if column in columns:
continue
conn.execute(
f"ALTER TABLE kyc_requests "
f"ADD COLUMN {column} {sql_type}"
)
conn.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_kyc_requests_user_id
ON kyc_requests(user_id)
"""
)
conn.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_kyc_requests_status
ON kyc_requests(status)
"""
)
conn.execute(
"""
CREATE INDEX IF NOT EXISTS
idx_kyc_requests_user_status
ON kyc_requests(user_id, status)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS kyc_verified_phones (
user_id INTEGER PRIMARY KEY,
phone_number TEXT NOT NULL,
username TEXT,
full_name TEXT,
verified_via_contact INTEGER NOT NULL DEFAULT 1,
verified_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS kyc_audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
request_id INTEGER,
user_id INTEGER NOT NULL,
action TEXT NOT NULL,
actor_user_id INTEGER,
details TEXT,
created_at TEXT NOT NULL
)
"""
)
conn.commit()
finally:
conn.close()
async def ensure_schema(self) -> None:
if self._schema_ready:
return
async with self._schema_lock:
if self._schema_ready:
return
await asyncio.to_thread(
self._ensure_schema_sync
)
self._schema_ready = True
@staticmethod
def _row_to_request(
row: sqlite3.Row,
) -> KYCRequest:
return KYCRequest(
id=int(row["id"]),
user_id=int(row["user_id"]),
username=row["username"],
full_name=row["full_name"],
phone_number=str(
row["phone_number"]
or ""
),
phone_verified=bool(
int(
row["phone_verified"]
or 0
)
),
phone_verified_at=row[
"phone_verified_at"
],
document_type=str(
row["document_type"]
or ""
),
document_front_file_id=str(
row[
"document_front_file_id"
]
or ""
),
document_back_file_id=row[
"document_back_file_id"
],
bank_card_file_id=str(
row["bank_card_file_id"]
or ""
),
status=str(
row["status"]
or STATUS_PENDING
),
rejection_reason=row[
"rejection_reason"
],
submitted_at=str(
row["submitted_at"]
or ""
),
reviewed_at=row[
"reviewed_at"
],
reviewed_by=(
int(row["reviewed_by"])
if row["reviewed_by"]
is not None
else None
),
created_at=str(
row["created_at"]
or ""
),
updated_at=str(
row["updated_at"]
or ""
),
attempt_no=int(
row["attempt_no"]
or 1
),
)
def _audit_sync(
self,
conn: sqlite3.Connection,
*,
request_id: Optional[int],
user_id: int,
action: str,
actor_user_id: Optional[int],
details: Optional[str] = None,
) -> None:
conn.execute(
"""
INSERT INTO kyc_audit_log (
request_id,
user_id,
action,
actor_user_id,
details,
created_at
)
VALUES (?, ?, ?, ?, ?, ?)
""",
(
request_id,
int(user_id),
str(action),
actor_user_id,
details,
utc_now_iso(),
),
)
def _save_phone_sync(
self,
*,
user_id: int,
phone_number: str,
username: Optional[str],
full_name: Optional[str],
) -> None:
now = utc_now_iso()
conn = self._connect()
try:
conn.execute(
"BEGIN IMMEDIATE"
)
conn.execute(
"""
INSERT INTO kyc_verified_phones (
user_id,
phone_number,
username,
full_name,
verified_via_contact,
verified_at,
updated_at
)
VALUES (?, ?, ?, ?, 1, ?, ?)
ON CONFLICT(user_id) DO UPDATE SET
phone_number = excluded.phone_number,
username = excluded.username,
full_name = excluded.full_name,
verified_via_contact = 1,
verified_at = excluded.verified_at,
updated_at = excluded.updated_at
""",
(
int(user_id),
phone_number,
username,
full_name,
now,
now,
),
)
self._audit_sync(
conn,
request_id=None,
user_id=user_id,
action="phone_verified_via_telegram_contact",
actor_user_id=user_id,
details=None,
)
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
async def save_verified_phone(
self,
*,
user_id: int,
phone_number: str,
username: Optional[str],
full_name: Optional[str],
) -> None:
await self.ensure_schema()
await asyncio.to_thread(
self._save_phone_sync,
user_id=int(user_id),
phone_number=phone_number,
username=username,
full_name=full_name,
)
def _get_verified_phone_sync(
self,
user_id: int,
) -> Optional[str]:
conn = self._connect()
try:
row = conn.execute(
"""
SELECT phone_number
FROM kyc_verified_phones
WHERE user_id = ?
AND verified_via_contact = 1
LIMIT 1
""",
(
int(user_id),
),
).fetchone()
return (
str(row["phone_number"])
if row
else None
)
finally:
conn.close()
async def get_verified_phone(
self,
user_id: int,
) -> Optional[str]:
await self.ensure_schema()
return await asyncio.to_thread(
self._get_verified_phone_sync,
int(user_id),
)
def _latest_for_user_sync(
self,
user_id: int,
) -> Optional[KYCRequest]:
conn = self._connect()
try:
row = conn.execute(
"""
SELECT *
FROM kyc_requests
WHERE user_id = ?
ORDER BY id DESC
LIMIT 1
""",
(
int(user_id),
),
).fetchone()
return (
self._row_to_request(row)
if row
else None
)
finally:
conn.close()
async def latest_for_user(
self,
user_id: int,
) -> Optional[KYCRequest]:
await self.ensure_schema()
return await asyncio.to_thread(
self._latest_for_user_sync,
int(user_id),
)
async def get_status(
self,
user_id: int,
) -> Optional[str]:
item = await self.latest_for_user(
user_id
)
return (
item.status
if item
else None
)
def _create_request_sync(
self,
*,
user_id: int,
username: Optional[str],
full_name: Optional[str],
phone_number: str,
document_type: str,
document_front_file_id: str,
document_back_file_id: Optional[str],
bank_card_file_id: str,
) -> KYCRequest:
now = utc_now_iso()
conn = self._connect()
try:
conn.execute(
"BEGIN IMMEDIATE"
)
pending = conn.execute(
"""
SELECT id
FROM kyc_requests
WHERE user_id = ?
AND status = ?
ORDER BY id DESC
LIMIT 1
""",
(
int(user_id),
STATUS_PENDING,
),
).fetchone()
if pending:
raise KYCAlreadyPending(
"یک درخواست در انتظار بررسی دارید."
)
row = conn.execute(
"""
SELECT COALESCE(MAX(attempt_no), 0) AS max_attempt
FROM kyc_requests
WHERE user_id = ?
""",
(
int(user_id),
),
).fetchone()
attempt_no = int(
row["max_attempt"]
or 0
) + 1
cursor = conn.execute(
"""
INSERT INTO kyc_requests (
user_id,
username,
full_name,
phone_number,
phone_verified,
phone_verified_at,
document_type,
document_front_file_id,
document_back_file_id,
bank_card_file_id,
selfie_file_id,
status,
rejection_reason,
submitted_at,
reviewed_at,
reviewed_by,
attempt_no,
created_at,
updated_at
)
VALUES (
?, ?, ?, ?, 1, ?, ?, ?, ?, ?, '', ?, NULL,
?, NULL, NULL, ?, ?, ?
)
""",
(
int(user_id),
username,
full_name,
phone_number,
now,
document_type,
document_front_file_id,
document_back_file_id,
bank_card_file_id,
STATUS_PENDING,
now,
attempt_no,
now,
now,
),
)
request_id = int(
cursor.lastrowid
)
self._audit_sync(
conn,
request_id=request_id,
user_id=user_id,
action="kyc_submitted",
actor_user_id=user_id,
details=(
f"document_type={document_type};"
f"attempt={attempt_no}"
),
)
row = conn.execute(
"""
SELECT *
FROM kyc_requests
WHERE id = ?
""",
(
request_id,
),
).fetchone()
conn.commit()
if not row:
raise KYCError(
"پرونده پس از ثبت پیدا نشد."
)
return self._row_to_request(
row
)
except Exception:
conn.rollback()
raise
finally:
conn.close()
async def create_request(
self,
*,
user_id: int,
username: Optional[str],
full_name: Optional[str],
phone_number: str,
document_type: str,
document_front_file_id: str,
document_back_file_id: Optional[str],
bank_card_file_id: str,
) -> KYCRequest:
await self.ensure_schema()
if document_type not in DOCUMENT_LABELS:
raise KYCValidationError(
"نوع مدرک نامعتبر است."
)
if not document_front_file_id:
raise KYCValidationError(
"تصویر اصلی مدرک ثبت نشده است."
)
if (
document_type
== DOC_NATIONAL_CARD
and not document_back_file_id
):
raise KYCValidationError(
"تصویر پشت کارت ملی ثبت نشده است."
)
if not bank_card_file_id:
raise KYCValidationError(
"تصویر کارت بانکی ثبت نشده است."
)
normalized_phone = normalize_phone(
phone_number
)
return await asyncio.to_thread(
self._create_request_sync,
user_id=int(user_id),
username=username,
full_name=full_name,
phone_number=normalized_phone,
document_type=document_type,
document_front_file_id=document_front_file_id,
document_back_file_id=document_back_file_id,
bank_card_file_id=bank_card_file_id,
)
def _get_request_sync(
self,
request_id: int,
) -> KYCRequest:
conn = self._connect()
try:
row = conn.execute(
"""
SELECT *
FROM kyc_requests
WHERE id = ?
LIMIT 1
""",
(
int(request_id),
),
).fetchone()
if not row:
raise KYCNotFound(
"پرونده پیدا نشد."
)
return self._row_to_request(
row
)
finally:
conn.close()
async def get_request(
self,
request_id: int,
) -> KYCRequest:
await self.ensure_schema()
return await asyncio.to_thread(
self._get_request_sync,
int(request_id),
)
def _list_pending_sync(
self,
limit: int,
) -> list[KYCRequest]:
conn = self._connect()
try:
rows = conn.execute(
"""
SELECT *
FROM kyc_requests
WHERE status = ?
ORDER BY id ASC
LIMIT ?
""",
(
STATUS_PENDING,
max(
1,
min(
int(limit),
50,
),
),
),
).fetchall()
return [
self._row_to_request(row)
for row in rows
]
finally:
conn.close()
async def list_pending(
self,
limit: int = 20,
) -> list[KYCRequest]:
await self.ensure_schema()
return await asyncio.to_thread(
self._list_pending_sync,
int(limit),
)
def _stats_sync(self) -> dict[str, int]:
conn = self._connect()
try:
row = conn.execute(
"""
SELECT
COUNT(*) AS total,
SUM(CASE WHEN status = 'pending' THEN 1 ELSE 0 END) AS pending,
SUM(CASE WHEN status = 'approved' THEN 1 ELSE 0 END) AS approved,
SUM(CASE WHEN status = 'rejected' THEN 1 ELSE 0 END) AS rejected
FROM kyc_requests
"""
).fetchone()
return {
"total": int(
row["total"]
or 0
),
"pending": int(
row["pending"]
or 0
),
"approved": int(
row["approved"]
or 0
),
"rejected": int(
row["rejected"]
or 0
),
}
finally:
conn.close()
async def stats(self) -> dict[str, int]:
await self.ensure_schema()
return await asyncio.to_thread(
self._stats_sync
)
def _mirror_user_kyc_status(
self,
conn: sqlite3.Connection,
*,
user_id: int,
status: str,
) -> None:
"""
Best-effort compatibility with projects that already have users.kyc_status.
It never changes the users schema automatically.
"""
try:
cols = self._columns(
conn,
"users",
)
if "kyc_status" not in cols:
return
key = (
"user_id"
if "user_id" in cols
else "id"
if "id" in cols
else None
)
if not key:
return
conn.execute(
f"""
UPDATE users
SET kyc_status = ?
WHERE {key} = ?
""",
(
status,
int(user_id),
),
)
except Exception:
logger.debug(
"Could not mirror KYC status to users table.",
exc_info=True,
)
def _review_sync(
self,
*,
request_id: int,
admin_id: int,
status: str,
reason: Optional[str],
) -> KYCRequest:
if status not in {
STATUS_APPROVED,
STATUS_REJECTED,
}:
raise KYCValidationError(
"وضعیت بررسی نامعتبر است."
)
now = utc_now_iso()
conn = self._connect()
try:
conn.execute(
"BEGIN IMMEDIATE"
)
row = conn.execute(
"""
SELECT *
FROM kyc_requests
WHERE id = ?
LIMIT 1
""",
(
int(request_id),
),
).fetchone()
if not row:
raise KYCNotFound(
"پرونده پیدا نشد."
)
current = self._row_to_request(
row
)
if current.status != STATUS_PENDING:
raise KYCAlreadyHandled(
"این پرونده قبلاً بررسی شده است."
)
cleaned_reason = (
str(reason).strip()
if reason
else None
)
if (
status == STATUS_REJECTED
and (
not cleaned_reason
or len(cleaned_reason) < 3
)
):
raise KYCValidationError(
"دلیل رد باید حداقل ۳ کاراکتر باشد."
)
conn.execute(
"""
UPDATE kyc_requests
SET
status = ?,
rejection_reason = ?,
reviewed_at = ?,
reviewed_by = ?,
updated_at = ?
WHERE id = ?
AND status = ?
""",
(
status,
cleaned_reason,
now,
int(admin_id),
now,
int(request_id),
STATUS_PENDING,
),
)
self._mirror_user_kyc_status(
conn,
user_id=current.user_id,
status=status,
)
self._audit_sync(
conn,
request_id=current.id,
user_id=current.user_id,
action=(
"kyc_approved"
if status == STATUS_APPROVED
else "kyc_rejected"
),
actor_user_id=admin_id,
details=cleaned_reason,
)
row = conn.execute(
"""
SELECT *
FROM kyc_requests
WHERE id = ?
""",
(
int(request_id),
),
).fetchone()
conn.commit()
if not row:
raise KYCNotFound(
"پرونده بعد از بررسی پیدا نشد."
)
return self._row_to_request(
row
)
except Exception:
conn.rollback()
raise
finally:
conn.close()
async def approve(
self,
*,
request_id: int,
admin_id: int,
) -> KYCRequest:
await self.ensure_schema()
return await asyncio.to_thread(
self._review_sync,
request_id=int(request_id),
admin_id=int(admin_id),
status=STATUS_APPROVED,
reason=None,
)
async def reject(
self,
*,
request_id: int,
admin_id: int,
reason: str,
) -> KYCRequest:
await self.ensure_schema()
return await asyncio.to_thread(
self._review_sync,
request_id=int(request_id),
admin_id=int(admin_id),
status=STATUS_REJECTED,
reason=reason,
)
_kyc_service: Optional[
KYCService
] = None
def service() -> KYCService:
global _kyc_service
path = database_path()
if (
_kyc_service is None
or _kyc_service.db_path != path
):
_kyc_service = KYCService(
path
)
return _kyc_service
# Compatibility helpers for older project code.
async def get_kyc_status(
user_id: int,
) -> Optional[str]:
return await service().get_status(
user_id
)
async def get_verified_phone(
user_id: int,
) -> Optional[str]:
return await service().get_verified_phone(
user_id
)
async def save_verified_phone(
*,
user_id: int,
phone_number: str,
username: Optional[str],
full_name: Optional[str],
) -> None:
await service().save_verified_phone(
user_id=user_id,
phone_number=phone_number,
username=username,
full_name=full_name,
)
# ============================================================
# User texts
# ============================================================
def kyc_start_text() -> str:
return (
"🛡️ احراز هویت حساب کاربری\n"
"━━━━━━━━━━━━━━━━━━━━\n\n"
"احراز هویت در چند مرحله کوتاه انجام میشود:\n\n"
"1️⃣ تأیید شماره همین اکانت\n"
"شماره فقط از طریق دکمه رسمی Telegram دریافت میشود.\n\n"
"2️⃣ انتخاب و ارسال مدرک\n"
"کارت ملی یا پاسپورت.\n\n"
"3️⃣ کارت بانکی به نام صاحب مدرک\n"
"برای تطبیق نام صاحب حساب بانکی با اطلاعات هویتی.\n\n"
"4️⃣ بررسی توسط مالک\n"
"نتیجه تأیید یا رد از طریق همین ربات اعلام میشود.\n\n"
"🔐 امنیت مرحله شماره\n"
"• شماره تایپی پذیرفته نمیشود.\n"
"• Contact متعلق به شخص دیگر پذیرفته نمیشود.\n"
"• فقط وقتی شماره قبول میشود که Telegram تأیید کند "
"Contact متعلق به همین User ID است.\n\n"
"برای شروع، شماره همین اکانت را با دکمه پایین ارسال کنید."
)
def phone_verified_text(
phone_number: str,
) -> str:
return (
"✅ شماره اکانت تأیید شد\n"
"━━━━━━━━━━━━━━━━━━━━\n\n"
f"📱 شماره: {escape(mask_phone(phone_number))}\n"
"🔐 روش تأیید: Telegram Contact / Same User ID\n\n"
"حالا نوع مدرک شناسایی را انتخاب کنید."
)
def front_request_text(
document_type: str,
) -> str:
label = escape(
document_label(
document_type
)
)
if document_type == DOC_PASSPORT:
detail = (
"صفحه اصلی پاسپورت که عکس و مشخصات شما در آن قرار دارد."
)
else:
detail = (
"تصویر روی کارت ملی؛ همه گوشههای کارت داخل تصویر باشد."
)
return (
"🪪 ارسال تصویر اصلی مدرک\n"
"━━━━━━━━━━━━━━━━━━━━\n\n"
f"نوع مدرک: {label}\n\n"
f"📸 {detail}\n\n"
"شرایط:\n"
"• واضح و خوانا باشد.\n"
"• نور کافی داشته باشد.\n"
"• اطلاعات مدرک بریده نشده باشد.\n"
"• فقط به صورت Photo ارسال شود."
)
def back_request_text() -> str:
return (
"🪪 تصویر پشت کارت ملی\n"
"━━━━━━━━━━━━━━━━━━━━\n\n"
"حالا تصویر واضح پشت کارت ملی را ارسال کنید.\n\n"
"• همه گوشههای کارت مشخص باشد.\n"
"• تصویر تار یا بریده نباشد.\n"
"• فقط Photo ارسال شود."
)
def bank_card_request_text() -> str:
return (
"💳 مرحله کارت بانکی\n"
"━━━━━━━━━━━━━━━━━━━━\n\n"
"یک عکس واضح از کارت بانکی به نام خودتان ارسال کنید "
"تا مالک بتواند نام صاحب کارت را با مدرک هویتی تطبیق دهد.\n\n"
"🔐 قبل از ارسال حتماً اطلاعات حساس را بپوشانید:\n"
"• CVV2 کاملاً پوشانده شود.\n"
"• تاریخ انقضا کاملاً پوشانده شود.\n"
"• ارقام میانی شماره کارت پوشانده شود.\n"
"• فقط ۶ رقم اول و ۴ رقم آخر شماره کارت مشخص باشد.\n"
"• نام صاحب کارت باید خوانا باشد.\n\n"
"⛔ هرگز ارسال نکنید:\n"
"رمز اول، رمز پویا/رمز دوم، CVV2، کد پیامکی یا اطلاعات ورود بانکی.\n\n"
"📸 تصویر را فقط به صورت Photo ارسال کنید."
)
def build_confirmation_text(
*,
user_id: int,
phone_number: str,
document_type: str,
has_back: bool,
) -> str:
return (
"📋 بررسی نهایی اطلاعات KYC\n"
"━━━━━━━━━━━━━━━━━━━━\n\n"
f"🆔 User ID: {user_id}\n"
f"📱 شماره تأییدشده: {escape(mask_phone(phone_number))}\n"
f"🪪 نوع مدرک: {escape(document_label(document_type))}\n"
"📸 تصویر اصلی مدرک: دریافت شد\n"
f"🔄 تصویر پشت مدرک: {'دریافت شد' if has_back else 'نیاز ندارد'}\n"
"💳 کارت بانکی: دریافت شد\n"
"🔐 اطلاعات حساس کارت: باید پوشانده شده باشد\n\n"
"اگر همه موارد درست است، «ثبت و ارسال برای مالک» را بزنید."
)
def build_user_status_text(
item: Optional[KYCRequest],
) -> str:
if not item:
return (
"🛡️ وضعیت احراز هویت\n"
"━━━━━━━━━━━━━━━━━━━━\n\n"
"هنوز پرونده احراز هویتی ثبت نکردهاید."
)
lines = [
"🛡️ وضعیت احراز هویت",
"━━━━━━━━━━━━━━━━━━━━",
"",
f"🧾 شماره پرونده: #{item.id}",
f"🔁 نوبت ارسال: {item.attempt_no}",
f"📌 وضعیت: {status_label(item.status)}",
f"📱 شماره: {escape(mask_phone(item.phone_number))}",
f"🪪 مدرک: {escape(item.document_label)}",
f"🗓 زمان ثبت: {escape(to_tehran_text(item.submitted_at))}",
]
if item.reviewed_at:
lines.append(
f"🕒 زمان بررسی: {escape(to_tehran_text(item.reviewed_at))}"
)
if (
item.status == STATUS_REJECTED
and item.rejection_reason
):
lines.extend(
[
"",
"📝 دلیل رد:",
escape(
item.rejection_reason
),
]
)
if item.status == STATUS_PENDING:
lines.extend(
[
"",
"⏳ پرونده برای مالک ارسال شده و در انتظار بررسی است.",
]
)
if item.status == STATUS_APPROVED:
lines.extend(
[
"",
"✅ احراز هویت این حساب با موفقیت تأیید شده است.",
]
)
return "\n".join(
lines
)
# ============================================================
# Owner texts
# ============================================================
def build_owner_case_text(
item: KYCRequest,
) -> str:
username = (
f"@{item.username}"
if item.username
else "ثبت نشده"
)
lines = [
"🛡️ پرونده احراز هویت",
"━━━━━━━━━━━━━━━━━━━━",
"",
f"🧾 پرونده: #{item.id}",
f"🔁 نوبت ارسال: {item.attempt_no}",
f"👤 نام: {escape(item.full_name or 'کاربر')}",
f"🔗 یوزرنیم: {escape(username)}",
f"🆔 User ID: {item.user_id}",
"",
"📱 شماره تأییدشده",
f"{escape(item.phone_number)}",
"🔐 Telegram Contact: تأیید شده برای همین User ID",
"",
f"🪪 نوع مدرک: {escape(item.document_label)}",
"📸 تصویر اصلی: دریافت شده",
f"🔄 تصویر پشت: {'دریافت شده' if item.document_back_file_id else 'نیاز ندارد'}",
"💳 کارت بانکی: دریافت شده",
"🔐 CVV2 / انقضا / ارقام میانی: باید پوشانده شده باشد",
"",
f"🗓 ثبت درخواست: {escape(to_tehran_text(item.submitted_at))}",
f"📌 وضعیت: {status_label(item.status)}",
]
if item.reviewed_at:
lines.append(
f"🕒 زمان بررسی: {escape(to_tehran_text(item.reviewed_at))}"
)
if item.reviewed_by:
lines.append(
f"🛡 بررسیکننده: {item.reviewed_by}"
)
if item.rejection_reason:
lines.extend(
[
"",
"📝 دلیل رد:",
escape(
item.rejection_reason
),
]
)
return "\n".join(
lines
)
# ============================================================
# Admin notification
# ============================================================
async def notify_owner_new_case(
bot,
item: KYCRequest,
) -> int:
targets = admin_ids()
if not targets:
logger.error(
"No owner/admin configured for KYC."
)
return 0
sent = 0
caption = build_owner_case_text(
item
)
for admin_id in targets:
try:
await bot.send_photo(
chat_id=admin_id,
photo=item.document_front_file_id,
caption=caption,
reply_markup=owner_case_keyboard(
item.id
),
parse_mode=ParseMode.HTML,
)
if item.document_back_file_id:
await bot.send_photo(
chat_id=admin_id,
photo=item.document_back_file_id,
caption=(
f"🔄 پشت مدرک پرونده #{item.id}"
),
parse_mode=ParseMode.HTML,
)
await bot.send_photo(
chat_id=admin_id,
photo=item.bank_card_file_id,
caption=(
f"💳 کارت بانکی پرونده #{item.id}\n"
"🔐 برای بررسی فقط نام صاحب کارت، ۶ رقم اول و ۴ رقم آخر لازم است."
),
parse_mode=ParseMode.HTML,
)
sent += 1
except TelegramForbiddenError:
logger.warning(
"KYC owner notification blocked | admin=%s request=%s",
admin_id,
item.id,
)
except Exception:
logger.exception(
"KYC owner notification failed | admin=%s request=%s",
admin_id,
item.id,
)
return sent
async def send_case_media_to_owner(
callback: CallbackQuery,
item: KYCRequest,
) -> None:
chat_id = callback.from_user.id
await callback.bot.send_photo(
chat_id=chat_id,
photo=item.document_front_file_id,
caption=(
f"📸 تصویر اصلی مدرک پرونده #{item.id}"
),
parse_mode=ParseMode.HTML,
)
if item.document_back_file_id:
await callback.bot.send_photo(
chat_id=chat_id,
photo=item.document_back_file_id,
caption=(
f"🔄 تصویر پشت مدرک پرونده #{item.id}"
),
parse_mode=ParseMode.HTML,
)
await callback.bot.send_photo(
chat_id=chat_id,
photo=item.bank_card_file_id,
caption=(
f"💳 کارت بانکی پرونده #{item.id}\n"
"🔐 نام صاحب کارت را با مدرک تطبیق دهید؛ "
"CVV2، تاریخ انقضا و ارقام میانی نباید قابل مشاهده باشند."
),
reply_markup=(
owner_case_keyboard(item.id)
if item.status == STATUS_PENDING
else owner_menu_keyboard()
),
parse_mode=ParseMode.HTML,
)
# ============================================================
# User: start / status
# ============================================================
@router.message(
Command("kyc")
)
async def kyc_command(
message: Message,
state: FSMContext,
) -> None:
if not message.from_user:
return
await state.clear()
latest = await service().latest_for_user(
message.from_user.id
)
if latest and latest.status in {
STATUS_PENDING,
STATUS_APPROVED,
}:
await message.answer(
build_user_status_text(
latest
),
reply_markup=user_status_keyboard(
latest.status
),
parse_mode=ParseMode.HTML,
)
return
await state.set_state(
KYCStates.waiting_for_phone
)
await message.answer(
kyc_start_text(),
reply_markup=phone_share_keyboard(),
parse_mode=ParseMode.HTML,
)
@router.callback_query(
F.data == "user_kyc_start"
)
async def user_kyc_start_handler(
callback: CallbackQuery,
state: FSMContext,
) -> None:
user_id = callback.from_user.id
try:
await service().ensure_schema()
await state.clear()
latest = await service().latest_for_user(
user_id
)
if latest and latest.status in {
STATUS_PENDING,
STATUS_APPROVED,
}:
await safe_edit(
callback,
build_user_status_text(
latest
),
user_status_keyboard(
latest.status
),
)
await safe_callback_answer(
callback
)
return
# Rejected cases may start a fresh attempt.
await state.set_state(
KYCStates.waiting_for_phone
)
await safe_edit(
callback,
kyc_start_text(),
cancel_inline_keyboard(),
)
if callback.message:
await callback.message.answer(
"📱 برای مرحله اول، شماره همین اکانت را با دکمه زیر ارسال کنید.",
reply_markup=phone_share_keyboard(),
parse_mode=ParseMode.HTML,
)
await safe_callback_answer(
callback
)
except Exception:
logger.exception(
"KYC start failed | user=%s",
user_id,
)
await state.clear()
await safe_callback_answer(
callback,
"❌ شروع احراز هویت با خطا روبهرو شد.",
True,
)
@router.callback_query(
F.data == "user_kyc_status"
)
async def user_kyc_status_handler(
callback: CallbackQuery,
state: FSMContext,
) -> None:
await state.clear()
latest = await service().latest_for_user(
callback.from_user.id
)
status = (
latest.status
if latest
else STATUS_REJECTED
)
await safe_edit(
callback,
build_user_status_text(
latest
),
user_status_keyboard(
status
),
)
await safe_callback_answer(
callback
)
# ============================================================
# User: phone verification
# ============================================================
@router.message(
KYCStates.waiting_for_phone,
F.contact,
)
async def process_kyc_phone(
message: Message,
state: FSMContext,
) -> None:
user = message.from_user
contact = message.contact
if not user or not contact:
return
# Critical ownership check.
if (
contact.user_id is None
or int(contact.user_id)
!= int(user.id)
):
await message.answer(
"⛔ این Contact متعلق به اکانت شما نیست.\n\n"
"فقط شمارهای پذیرفته میشود که Telegram آن را برای همین User ID ارسال کند.\n"
"از دکمه «ارسال شماره همین اکانت» استفاده کنید.",
reply_markup=phone_share_keyboard(),
parse_mode=ParseMode.HTML,
)
return
try:
phone = normalize_phone(
contact.phone_number
)
await service().save_verified_phone(
user_id=user.id,
phone_number=phone,
username=user.username,
full_name=user.full_name,
)
await state.update_data(
verified_phone=phone,
phone_verified=True,
)
await state.set_state(
KYCStates.waiting_for_document_type
)
await message.answer(
"✅ شماره همین اکانت با موفقیت تأیید شد.",
reply_markup=remove_reply_keyboard(),
)
await message.answer(
phone_verified_text(
phone
),
reply_markup=document_type_keyboard(),
parse_mode=ParseMode.HTML,
)
logger.info(
"KYC phone verified | user=%s",
user.id,
)
except Exception:
logger.exception(
"KYC phone verify/save failed | user=%s",
user.id,
)
await message.answer(
"❌ تأیید شماره با خطا روبهرو شد. دوباره از دکمه زیر استفاده کنید.",
reply_markup=phone_share_keyboard(),
)
@router.message(
KYCStates.waiting_for_phone,
)
async def invalid_kyc_phone_input(
message: Message,
) -> None:
await message.answer(
"📱 شماره را دستی تایپ نکنید.\n\n"
"برای احراز مالکیت شماره فقط Contact رسمی Telegram پذیرفته میشود. "
"روی دکمه پایین بزنید.",
reply_markup=phone_share_keyboard(),
parse_mode=ParseMode.HTML,
)
# ============================================================
# User: document type
# ============================================================
async def _select_document_type(
callback: CallbackQuery,
state: FSMContext,
document_type: str,
) -> None:
if document_type not in DOCUMENT_LABELS:
await safe_callback_answer(
callback,
"نوع مدرک نامعتبر است.",
True,
)
return
data = await state.get_data()
phone = data.get(
"verified_phone"
)
if not phone:
phone = await service().get_verified_phone(
callback.from_user.id
)
if not phone:
await state.clear()
await state.set_state(
KYCStates.waiting_for_phone
)
if callback.message:
await callback.message.answer(
"⚠️ ابتدا شماره همین اکانت را تأیید کنید.",
reply_markup=phone_share_keyboard(),
)
await safe_callback_answer(
callback,
"ابتدا شماره را تأیید کنید.",
True,
)
return
await state.update_data(
verified_phone=phone,
document_type=document_type,
document_front_file_id=None,
document_back_file_id=None,
bank_card_file_id=None,
)
await state.set_state(
KYCStates.waiting_for_front
)
await safe_edit(
callback,
front_request_text(
document_type
),
cancel_inline_keyboard(),
)
await safe_callback_answer(
callback
)
@router.callback_query(
F.data == "kyc_doc_national_card"
)
async def choose_national_card(
callback: CallbackQuery,
state: FSMContext,
) -> None:
await _select_document_type(
callback,
state,
DOC_NATIONAL_CARD,
)
@router.callback_query(
F.data == "kyc_doc_passport"
)
async def choose_passport(
callback: CallbackQuery,
state: FSMContext,
) -> None:
await _select_document_type(
callback,
state,
DOC_PASSPORT,
)
@router.message(
KYCStates.waiting_for_document_type,
)
async def invalid_document_type_message(
message: Message,
) -> None:
await message.answer(
"🪪 نوع مدرک را فقط از دکمههای زیر انتخاب کنید.",
reply_markup=document_type_keyboard(),
)
# ============================================================
# User: document front
# ============================================================
@router.message(
KYCStates.waiting_for_front,
F.photo,
)
async def receive_document_front(
message: Message,
state: FSMContext,
) -> None:
if not message.photo:
return
file_id = message.photo[-1].file_id
data = await state.get_data()
doc_type = str(
data.get(
"document_type",
"",
)
)
if doc_type not in DOCUMENT_LABELS:
await state.set_state(
KYCStates.waiting_for_document_type
)
await message.answer(
"⚠️ نوع مدرک مشخص نیست؛ دوباره انتخاب کنید.",
reply_markup=document_type_keyboard(),
)
return
await state.update_data(
document_front_file_id=file_id
)
if doc_type == DOC_NATIONAL_CARD:
await state.set_state(
KYCStates.waiting_for_back
)
await message.answer(
back_request_text(),
reply_markup=cancel_inline_keyboard(),
parse_mode=ParseMode.HTML,
)
return
await state.update_data(
document_back_file_id=None
)
await state.set_state(
KYCStates.waiting_for_bank_card
)
await message.answer(
bank_card_request_text(),
reply_markup=cancel_inline_keyboard(),
parse_mode=ParseMode.HTML,
)
@router.message(
KYCStates.waiting_for_front,
)
async def invalid_front_document(
message: Message,
) -> None:
await message.answer(
"📸 تصویر اصلی مدرک را فقط به صورت Photo ارسال کنید.\n"
"Document، PDF، ویدئو یا متن پذیرفته نمیشود.",
reply_markup=cancel_inline_keyboard(),
parse_mode=ParseMode.HTML,
)
# ============================================================
# User: document back
# ============================================================
@router.message(
KYCStates.waiting_for_back,
F.photo,
)
async def receive_document_back(
message: Message,
state: FSMContext,
) -> None:
if not message.photo:
return
await state.update_data(
document_back_file_id=(
message.photo[-1].file_id
)
)
await state.set_state(
KYCStates.waiting_for_bank_card
)
await message.answer(
bank_card_request_text(),
reply_markup=cancel_inline_keyboard(),
parse_mode=ParseMode.HTML,
)
@router.message(
KYCStates.waiting_for_back,
)
async def invalid_back_document(
message: Message,
) -> None:
await message.answer(
"📸 تصویر پشت کارت را فقط به صورت Photo ارسال کنید.",
reply_markup=cancel_inline_keyboard(),
parse_mode=ParseMode.HTML,
)
# ============================================================
# User: bank card
# ============================================================
@router.message(
KYCStates.waiting_for_bank_card,
F.photo,
)
async def receive_kyc_bank_card(
message: Message,
state: FSMContext,
) -> None:
"""
دریافت تصویر کارت بانکی.
تشخیص خودکار اطلاعات روی تصویر انجام نمیشود؛ کاربر موظف است
CVV2، تاریخ انقضا و ارقام میانی را قبل از ارسال بپوشاند.
"""
if not message.from_user or not message.photo:
return
await state.update_data(
bank_card_file_id=(
message.photo[-1].file_id
)
)
data = await state.get_data()
phone = str(
data.get(
"verified_phone",
"",
)
)
doc_type = str(
data.get(
"document_type",
"",
)
)
front = data.get(
"document_front_file_id"
)
bank_card = data.get(
"bank_card_file_id"
)
if (
not phone
or doc_type not in DOCUMENT_LABELS
or not front
or not bank_card
):
await state.clear()
await message.answer(
"❌ اطلاعات مراحل KYC ناقص شده است. لطفاً دوباره شروع کنید.",
reply_markup=user_status_keyboard(
STATUS_REJECTED
),
)
return
if (
doc_type == DOC_NATIONAL_CARD
and not data.get(
"document_back_file_id"
)
):
await state.set_state(
KYCStates.waiting_for_back
)
await message.answer(
"⚠️ تصویر پشت کارت ملی هنوز دریافت نشده است.",
reply_markup=cancel_inline_keyboard(),
)
return
await state.set_state(
KYCStates.waiting_for_confirmation
)
await message.answer(
build_confirmation_text(
user_id=message.from_user.id,
phone_number=phone,
document_type=doc_type,
has_back=bool(
data.get(
"document_back_file_id"
)
),
)
+ (
"\n\n🔐 یادآوری امنیتی:\n"
"قبل از ثبت نهایی مطمئن شوید CVV2، تاریخ انقضا و "
"ارقام میانی کارت در تصویر پوشانده شدهاند."
),
reply_markup=confirmation_keyboard(),
parse_mode=ParseMode.HTML,
)
@router.message(
KYCStates.waiting_for_bank_card,
)
async def invalid_bank_card(
message: Message,
) -> None:
await message.answer(
"💳 عکس کارت بانکی را فقط به صورت Photo ارسال کنید.\n\n"
"قبل از ارسال: CVV2، تاریخ انقضا و ارقام میانی کارت را بپوشانید؛ "
"فقط نام صاحب کارت، ۶ رقم اول و ۴ رقم آخر مشخص باشد.",
reply_markup=cancel_inline_keyboard(),
parse_mode=ParseMode.HTML,
)
# ============================================================
# User: restart documents / cancel
# ============================================================
@router.callback_query(
F.data == "kyc_restart_documents"
)
async def restart_kyc_documents(
callback: CallbackQuery,
state: FSMContext,
) -> None:
data = await state.get_data()
phone = data.get(
"verified_phone"
)
if not phone:
phone = await service().get_verified_phone(
callback.from_user.id
)
await state.clear()
if not phone:
await state.set_state(
KYCStates.waiting_for_phone
)
if callback.message:
await callback.message.answer(
"📱 دوباره شماره همین اکانت را ارسال کنید.",
reply_markup=phone_share_keyboard(),
)
else:
await state.update_data(
verified_phone=phone,
phone_verified=True,
)
await state.set_state(
KYCStates.waiting_for_document_type
)
await safe_edit(
callback,
phone_verified_text(
str(phone)
),
document_type_keyboard(),
)
await safe_callback_answer(
callback
)
@router.callback_query(
F.data == "kyc_cancel"
)
async def cancel_kyc(
callback: CallbackQuery,
state: FSMContext,
) -> None:
await state.clear()
await safe_edit(
callback,
"❌ فرآیند احراز هویت لغو شد.\n\n"
"هر زمان خواستید میتوانید از حساب کاربری دوباره شروع کنید.",
user_status_keyboard(
STATUS_REJECTED
),
)
if callback.message:
try:
await callback.message.answer(
"کیبورد احراز هویت بسته شد.",
reply_markup=remove_reply_keyboard(),
)
except Exception:
pass
await safe_callback_answer(
callback,
"احراز هویت لغو شد."
)
# ============================================================
# User: submit case
# ============================================================
@router.callback_query(
F.data == "kyc_submit_case"
)
async def submit_kyc_case(
callback: CallbackQuery,
state: FSMContext,
) -> None:
if not callback.from_user:
return
data = await state.get_data()
phone = str(
data.get(
"verified_phone",
"",
)
).strip()
doc_type = str(
data.get(
"document_type",
"",
)
).strip()
front = str(
data.get(
"document_front_file_id",
"",
)
).strip()
back_value = data.get(
"document_back_file_id"
)
back = (
str(back_value).strip()
if back_value
else None
)
bank_card = str(
data.get(
"bank_card_file_id",
"",
)
).strip()
try:
item = await service().create_request(
user_id=callback.from_user.id,
username=callback.from_user.username,
full_name=callback.from_user.full_name,
phone_number=phone,
document_type=doc_type,
document_front_file_id=front,
document_back_file_id=back,
bank_card_file_id=bank_card,
)
except KYCAlreadyPending:
await state.clear()
latest = await service().latest_for_user(
callback.from_user.id
)
await safe_edit(
callback,
build_user_status_text(
latest
),
user_status_keyboard(
STATUS_PENDING
),
)
await safe_callback_answer(
callback,
"یک درخواست در انتظار بررسی دارید.",
True,
)
return
except KYCValidationError as exc:
await safe_callback_answer(
callback,
str(exc),
True,
)
return
except Exception:
logger.exception(
"Could not create KYC request | user=%s",
callback.from_user.id,
)
await safe_callback_answer(
callback,
"❌ ثبت پرونده با خطا روبهرو شد.",
True,
)
return
await state.clear()
sent_to_admins = await notify_owner_new_case(
callback.bot,
item,
)
notification_text = (
"📨 پرونده برای مالک ارسال شد."
if sent_to_admins > 0
else (
"⚠️ پرونده ثبت شد، اما ارسال فوری برای مالک ناموفق بود؛ "
"پرونده در صف بررسی باقی مانده است."
)
)
text = (
"✅ درخواست احراز هویت ثبت شد\n"
"━━━━━━━━━━━━━━━━━━━━\n\n"
f"🧾 شماره پرونده: #{item.id}\n"
f"🔁 نوبت ارسال: {item.attempt_no}\n"
f"🆔 User ID: {item.user_id}\n"
f"📱 شماره: {escape(mask_phone(item.phone_number))}\n"
f"🪪 نوع مدرک: {escape(item.document_label)}\n"
f"📸 تصویر اصلی: دریافت شد\n"
f"🔄 تصویر پشت: {'دریافت شد' if item.document_back_file_id else 'نیاز ندارد'}\n"
f"💳 کارت بانکی: دریافت شد\n"
"🔐 اطلاعات حساس کارت: پوشانده شده طبق دستورالعمل\n"
f"🗓 تاریخ ثبت: {escape(to_tehran_text(item.submitted_at))}\n"
f"📌 وضعیت: {status_label(item.status)}\n\n"
f"{notification_text}\n"
"بعد از بررسی، نتیجه از طریق همین ربات به شما اعلام میشود."
)
await safe_edit(
callback,
text,
user_status_keyboard(
STATUS_PENDING
),
)
await safe_callback_answer(
callback,
"درخواست ثبت شد."
)
# ============================================================
# Owner/admin guards
# ============================================================
async def ensure_owner_callback(
callback: CallbackQuery,
) -> bool:
if is_admin(
callback.from_user.id
):
return True
await safe_callback_answer(
callback,
"⛔ دسترسی ندارید.",
True,
)
return False
async def ensure_owner_message(
message: Message,
) -> bool:
if (
message.from_user
and is_admin(
message.from_user.id
)
):
return True
await message.answer(
"⛔ دسترسی ندارید."
)
return False
def parse_id(
data: Optional[str],
prefix: str,
) -> Optional[int]:
raw = str(
data
or ""
)
if not raw.startswith(
prefix
):
return None
try:
return int(
raw[
len(prefix):
]
)
except ValueError:
return None
# ============================================================
# Owner/admin: menu and queue
# ============================================================
@router.message(
Command("kyc_pending")
)
async def owner_kyc_pending_command(
message: Message,
state: FSMContext,
) -> None:
if not await ensure_owner_message(
message
):
return
await state.clear()
items = await service().list_pending(
20
)
text = (
"🛡️ صف احراز هویت\n"
"━━━━━━━━━━━━━━━━━━━━\n\n"
f"⏳ درخواستهای منتظر: {len(items)}\n\n"
"برای مشاهده پرونده از دکمههای زیر استفاده کنید."
)
await message.answer(
text,
reply_markup=owner_queue_keyboard(
items
),
parse_mode=ParseMode.HTML,
)
@router.callback_query(
F.data == "admin_kyc_menu"
)
async def admin_kyc_menu_handler(
callback: CallbackQuery,
state: FSMContext,
) -> None:
if not await ensure_owner_callback(
callback
):
return
await state.clear()
stats = await service().stats()
text = (
"🛡️ مدیریت احراز هویت\n"
"━━━━━━━━━━━━━━━━━━━━\n\n"
f"⏳ در انتظار: {stats['pending']:,}\n"
f"✅ تأییدشده: {stats['approved']:,}\n"
f"❌ ردشده: {stats['rejected']:,}\n"
f"📊 کل پروندهها: {stats['total']:,}"
)
await safe_edit(
callback,
text,
owner_menu_keyboard(),
)
await safe_callback_answer(
callback
)
@router.callback_query(
F.data == "admin_kyc_queue"
)
async def admin_kyc_queue_handler(
callback: CallbackQuery,
state: FSMContext,
) -> None:
if not await ensure_owner_callback(
callback
):
return
await state.clear()
items = await service().list_pending(
20
)
lines = [
"📋 درخواستهای منتظر KYC",
"━━━━━━━━━━━━━━━━━━━━",
"",
]
if not items:
lines.append(
"✅ در حال حاضر پروندهای در انتظار بررسی نیست."
)
else:
for item in items:
lines.extend(
[
f"🧾 #{item.id} • {escape(item.full_name or 'کاربر')}",
f"🆔 {item.user_id}",
f"🪪 {escape(item.document_label)}",
f"🗓 {escape(to_tehran_text(item.submitted_at))}",
"",
]
)
await safe_edit(
callback,
"\n".join(lines),
owner_queue_keyboard(
items
),
)
await safe_callback_answer(
callback
)
# ============================================================
# Owner/admin: view case
# ============================================================
@router.callback_query(
F.data.startswith(
"kyc_case_view_"
)
)
async def owner_view_kyc_case(
callback: CallbackQuery,
state: FSMContext,
) -> None:
if not await ensure_owner_callback(
callback
):
return
request_id = parse_id(
callback.data,
"kyc_case_view_",
)
if not request_id:
await safe_callback_answer(
callback,
"شناسه پرونده نامعتبر است.",
True,
)
return
await state.clear()
try:
item = await service().get_request(
request_id
)
except KYCNotFound:
await safe_callback_answer(
callback,
"پرونده پیدا نشد.",
True,
)
return
await safe_edit(
callback,
build_owner_case_text(
item
),
(
owner_case_keyboard(item.id)
if item.status == STATUS_PENDING
else owner_menu_keyboard()
),
)
try:
await send_case_media_to_owner(
callback,
item,
)
except Exception:
logger.exception(
"Could not send KYC case media | request=%s",
item.id,
)
await safe_callback_answer(
callback
)
# ============================================================
# Owner/admin: approve
# ============================================================
@router.callback_query(
F.data.startswith(
"kyc_case_approve_"
)
)
async def owner_approve_kyc_case(
callback: CallbackQuery,
state: FSMContext,
) -> None:
if not await ensure_owner_callback(
callback
):
return
request_id = parse_id(
callback.data,
"kyc_case_approve_",
)
if not request_id:
await safe_callback_answer(
callback,
"شناسه پرونده نامعتبر است.",
True,
)
return
await state.clear()
try:
item = await service().approve(
request_id=request_id,
admin_id=callback.from_user.id,
)
except KYCAlreadyHandled:
await safe_callback_answer(
callback,
"این پرونده قبلاً بررسی شده است.",
True,
)
return
except Exception:
logger.exception(
"KYC approve failed | request=%s admin=%s",
request_id,
callback.from_user.id,
)
await safe_callback_answer(
callback,
"❌ ثبت تأیید با خطا روبهرو شد.",
True,
)
return
await safe_edit(
callback,
build_owner_case_text(
item
)
+ "\n\n✅ این پرونده توسط شما تأیید شد.",
owner_menu_keyboard(),
)
try:
await callback.bot.send_message(
chat_id=item.user_id,
text=(
"🎉 احراز هویت شما تأیید شد\n"
"━━━━━━━━━━━━━━━━━━━━\n\n"
f"🧾 پرونده: #{item.id}\n"
f"🗓 زمان تأیید: {escape(to_tehran_text(item.reviewed_at))}\n"
"📌 وضعیت: ✅ تأیید شده\n\n"
"حساب شما با موفقیت احراز هویت شد."
),
reply_markup=user_status_keyboard(
STATUS_APPROVED
),
parse_mode=ParseMode.HTML,
)
except TelegramForbiddenError:
logger.warning(
"Approved KYC user blocked bot | user=%s request=%s",
item.user_id,
item.id,
)
except Exception:
logger.exception(
"Could not notify approved KYC user | user=%s request=%s",
item.user_id,
item.id,
)
await safe_callback_answer(
callback,
"✅ احراز هویت تأیید شد.",
True,
)
# ============================================================
# Owner/admin: reject
# ============================================================
@router.callback_query(
F.data.startswith(
"kyc_case_reject_"
)
)
async def owner_reject_kyc_prompt(
callback: CallbackQuery,
state: FSMContext,
) -> None:
if not await ensure_owner_callback(
callback
):
return
request_id = parse_id(
callback.data,
"kyc_case_reject_",
)
if not request_id:
await safe_callback_answer(
callback,
"شناسه پرونده نامعتبر است.",
True,
)
return
try:
item = await service().get_request(
request_id
)
except KYCNotFound:
await safe_callback_answer(
callback,
"پرونده پیدا نشد.",
True,
)
return
if item.status != STATUS_PENDING:
await safe_callback_answer(
callback,
"این پرونده قبلاً بررسی شده است.",
True,
)
return
await state.set_state(
OwnerKYCStates.waiting_for_reject_reason
)
await state.update_data(
kyc_reject_request_id=request_id
)
await safe_edit(
callback,
"❌ رد احراز هویت\n"
"━━━━━━━━━━━━━━━━━━━━\n\n"
f"🧾 پرونده: #{request_id}\n"
f"👤 کاربر: {item.user_id}\n\n"
"دلیل رد را واضح بنویسید. این متن برای کاربر ارسال میشود.\n\n"
"حداقل ۳ و حداکثر ۱۰۰۰ کاراکتر.",
owner_reject_cancel_keyboard(
request_id
),
)
await safe_callback_answer(
callback
)
@router.message(
OwnerKYCStates.waiting_for_reject_reason,
F.text,
)
async def owner_reject_kyc_reason(
message: Message,
state: FSMContext,
) -> None:
if not await ensure_owner_message(
message
):
await state.clear()
return
data = await state.get_data()
request_id = data.get(
"kyc_reject_request_id"
)
if not request_id:
await state.clear()
await message.answer(
"⚠️ نشست بررسی منقضی شده است."
)
return
reason = str(
message.text
or ""
).strip()
if len(reason) < 3:
await message.answer(
"❌ دلیل رد خیلی کوتاه است؛ حداقل ۳ کاراکتر وارد کنید."
)
return
if len(reason) > 1000:
await message.answer(
"❌ دلیل رد نباید بیشتر از ۱۰۰۰ کاراکتر باشد."
)
return
try:
item = await service().reject(
request_id=int(request_id),
admin_id=message.from_user.id,
reason=reason,
)
except KYCAlreadyHandled:
await state.clear()
await message.answer(
"⚠️ این پرونده قبلاً بررسی شده است."
)
return
except Exception:
logger.exception(
"KYC reject failed | request=%s admin=%s",
request_id,
message.from_user.id,
)
await message.answer(
"❌ ثبت رد پرونده با خطا روبهرو شد."
)
return
await state.clear()
await message.answer(
"❌ احراز هویت رد شد\n\n"
f"🧾 پرونده: #{item.id}\n"
f"👤 کاربر: {item.user_id}\n"
f"📝 دلیل: {escape(reason)}\n"
f"🗓 زمان: {escape(to_tehran_text(item.reviewed_at))}",
reply_markup=owner_menu_keyboard(),
parse_mode=ParseMode.HTML,
)
try:
await message.bot.send_message(
chat_id=item.user_id,
text=(
"❌ درخواست احراز هویت شما رد شد\n"
"━━━━━━━━━━━━━━━━━━━━\n\n"
f"🧾 پرونده: #{item.id}\n"
f"🗓 زمان بررسی: {escape(to_tehran_text(item.reviewed_at))}\n\n"
"📝 دلیل رد:\n"
f"{escape(reason)}\n\n"
"بعد از اصلاح موارد اعلامشده میتوانید دوباره احراز هویت را شروع کنید."
),
reply_markup=user_status_keyboard(
STATUS_REJECTED
),
parse_mode=ParseMode.HTML,
)
except TelegramForbiddenError:
logger.warning(
"Rejected KYC user blocked bot | user=%s request=%s",
item.user_id,
item.id,
)
except Exception:
logger.exception(
"Could not notify rejected KYC user | user=%s request=%s",
item.user_id,
item.id,
)
# ============================================================
# Compatibility wrapper for older caller signatures
# ============================================================
async def save_kyc_request(
user_id: int,
file_id: str,
status: str = STATUS_PENDING,
) -> bool:
"""
Compatibility only.
The complete KYC flow should use service().create_request(), because a
real case requires verified phone + identity document + masked bank card.
This wrapper does
not create an incomplete KYC case and therefore returns False.
"""
logger.warning(
"Legacy save_kyc_request called for user=%s; full KYC flow is required.",
user_id,
)
return False
__all__ = [
"router",
"KYCStates",
"OwnerKYCStates",
"KYCRequest",
"KYCService",
"KYCError",
"KYCValidationError",
"KYCAlreadyPending",
"KYCAlreadyHandled",
"KYCNotFound",
"STATUS_PENDING",
"STATUS_APPROVED",
"STATUS_REJECTED",
"DOC_NATIONAL_CARD",
"DOC_PASSPORT",
"service",
"get_kyc_status",
"get_verified_phone",
"save_verified_phone",
"save_kyc_request",
"phone_share_keyboard",
"document_type_keyboard",
"admin_kyc_keyboard",
]