969 lines
40 KiB
Python
969 lines
40 KiB
Python
# ======= E-TAXES AUTHENTICATION MODULE =======
|
||
|
||
import frappe
|
||
import requests
|
||
import json
|
||
from frappe.utils import nowdate, now, cint, now_datetime, get_datetime
|
||
from frappe.model.document import Document
|
||
import re
|
||
from difflib import SequenceMatcher
|
||
import datetime
|
||
import random
|
||
import time
|
||
|
||
# Константы
|
||
RENEW_URL = "https://new.e-taxes.gov.az/api/po/auth/public/v1/renew"
|
||
MAX_RETRY_COUNT = 3
|
||
DEFAULT_HEADERS = {
|
||
"Accept": "application/json, text/plain, */*",
|
||
"Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",
|
||
"Cache-Control": "no-cache",
|
||
"Pragma": "no-cache",
|
||
"Referer": "https://new.e-taxes.gov.az/eportal/az/login/asan"
|
||
}
|
||
ACTIVITY_TIMEOUT_MINUTES = 10000000000000000 # Настраиваемый параметр времени неактивности
|
||
|
||
@frappe.whitelist()
|
||
def record_etaxes_activity(asan_login_name=None):
|
||
"""Записывает время последней активности с e-taxes напрямую"""
|
||
try:
|
||
# Если имя не указано, получаем дефолтный профиль
|
||
if not asan_login_name:
|
||
default_settings = frappe.get_all(
|
||
"Asan Login",
|
||
filters={"is_default": 1},
|
||
fields=["name"],
|
||
limit=1
|
||
)
|
||
|
||
if not default_settings:
|
||
return
|
||
|
||
asan_login_name = default_settings[0].name
|
||
|
||
# Обновляем поле last_activity_time напрямую через БД
|
||
frappe.db.set_value("Asan Login", asan_login_name, "last_activity_time", now_datetime(), update_modified=False)
|
||
frappe.db.commit()
|
||
|
||
except Exception as e:
|
||
frappe.log_error(f"Error recording activity: {str(e)}", "Activity Recording Error")
|
||
|
||
@frappe.whitelist()
|
||
def update_activity_time(asan_login_name):
|
||
"""Выполняет фактическое обновление поля last_activity_time - упрощенная версия"""
|
||
try:
|
||
# Обновляем поле last_activity_time напрямую через БД
|
||
frappe.db.set_value("Asan Login", asan_login_name, "last_activity_time", now_datetime(), update_modified=False)
|
||
frappe.db.commit()
|
||
except Exception as e:
|
||
frappe.log_error(f"Error updating activity time: {str(e)}", "Activity Update Error")
|
||
|
||
def check_recent_activity(asan_login_name=None):
|
||
"""Проверяет, была ли активность пользователя за последние N минут"""
|
||
try:
|
||
# Если имя не указано, получаем дефолтный профиль
|
||
if not asan_login_name:
|
||
default_settings = frappe.get_all(
|
||
"Asan Login",
|
||
filters={"is_default": 1},
|
||
fields=["name"],
|
||
limit=1
|
||
)
|
||
|
||
if not default_settings:
|
||
return False
|
||
|
||
asan_login_name = default_settings[0].name
|
||
|
||
# Получаем время последней активности напрямую из БД
|
||
last_activity = frappe.db.get_value("Asan Login", asan_login_name, "last_activity_time")
|
||
|
||
# Если последней активности не было, возвращаем False
|
||
if not last_activity:
|
||
return False
|
||
|
||
# Проверяем, была ли активность за последние N минут
|
||
time_diff = now_datetime() - get_datetime(last_activity)
|
||
minutes_diff = time_diff.total_seconds() / 60
|
||
|
||
return minutes_diff <= ACTIVITY_TIMEOUT_MINUTES
|
||
except Exception as e:
|
||
frappe.log_error(f"Error checking activity: {str(e)}", "Activity Check Error")
|
||
return False # В случае ошибки считаем, что активности не было
|
||
|
||
@frappe.whitelist()
|
||
def renew_token(asan_login_name=None, retry_count=0):
|
||
"""Renews the main token with optimized performance and error handling"""
|
||
try:
|
||
# Получаем документ Asan Login
|
||
if not asan_login_name:
|
||
# Простой прямой запрос для получения дефолтного логина
|
||
default_settings = frappe.get_all(
|
||
"Asan Login",
|
||
filters={"is_default": 1},
|
||
fields=["name"],
|
||
limit=1
|
||
)
|
||
|
||
if not default_settings:
|
||
frappe.log_error("No default Asan Login found", "Token Renewal Error")
|
||
return {"success": False, "message": "No default Asan Login found"}
|
||
|
||
asan_login_name = default_settings[0].name
|
||
|
||
# Проверяем наличие активности
|
||
if not check_recent_activity(asan_login_name):
|
||
frappe.logger().info(f"Token renewal skipped due to inactivity for {asan_login_name}")
|
||
return {"success": True, "message": "Token renewal skipped due to inactivity"}
|
||
|
||
# Получаем документ (без блокировки, поскольку запуск один)
|
||
asan_login = frappe.get_doc("Asan Login", asan_login_name)
|
||
|
||
# Проверяем наличие токена
|
||
if not asan_login.main_token:
|
||
frappe.log_error(f"No main token found for {asan_login_name}", "Token Renewal Error")
|
||
return {"success": False, "message": "No main token found"}
|
||
|
||
# Формируем заголовок авторизации
|
||
auth_header = f"Bearer {asan_login.main_token}"
|
||
|
||
# Подготавливаем заголовки запроса
|
||
headers = DEFAULT_HEADERS.copy()
|
||
headers["x-authorization"] = auth_header
|
||
|
||
# Отправляем запрос с таймаутом
|
||
response = requests.post(RENEW_URL, headers=headers, timeout=10)
|
||
|
||
# Check for specific HTTP status codes
|
||
if response.status_code == 500:
|
||
return {
|
||
"error": "server_error",
|
||
"message": "Service temporarily unavailable. Please try again in a few minutes.",
|
||
"status_code": 500
|
||
}
|
||
|
||
# Обрабатываем ответ
|
||
if response.status_code == 200:
|
||
# Получаем новый токен из заголовков ответа
|
||
new_token = response.headers.get('x-authorization')
|
||
|
||
if new_token:
|
||
frappe.logger().info(f"Token renewal successful for {asan_login_name}")
|
||
|
||
# Проверяем, изменился ли токен
|
||
if new_token != asan_login.main_token:
|
||
# Обновляем токен напрямую через БД, чтобы избежать конфликта версий
|
||
frappe.db.set_value("Asan Login", asan_login_name, "main_token", new_token)
|
||
frappe.db.commit()
|
||
frappe.logger().info(f"Token updated for {asan_login_name}")
|
||
else:
|
||
frappe.logger().info(f"Token unchanged for {asan_login_name}")
|
||
|
||
return {"success": True, "message": "Token renewed successfully"}
|
||
else:
|
||
frappe.log_error(
|
||
"Token renewal received 200 but no token in response headers",
|
||
f"Token Renewal for {asan_login_name}"
|
||
)
|
||
return {"success": False, "message": "No token in response"}
|
||
|
||
elif response.status_code == 401:
|
||
# В случае ошибки авторизации меняем статус
|
||
frappe.db.set_value("Asan Login", asan_login_name, "auth_status", "Not Authenticated")
|
||
frappe.db.commit()
|
||
|
||
frappe.log_error(
|
||
f"Authorization error (401) for {asan_login_name}",
|
||
"Token Renewal Error"
|
||
)
|
||
return {
|
||
"error": "unauthorized",
|
||
"message": "Authentication required. Please login again.",
|
||
"status_code": 401
|
||
}
|
||
|
||
else:
|
||
# Механизм повторных попыток при временных ошибках
|
||
if retry_count < MAX_RETRY_COUNT and response.status_code >= 500:
|
||
# Экспоненциальная задержка между попытками
|
||
wait_time = (2 ** retry_count) * (0.5 + random.random())
|
||
time.sleep(wait_time)
|
||
return renew_token(asan_login_name, retry_count + 1)
|
||
|
||
frappe.log_error(
|
||
f"Failed with status {response.status_code}: {response.text[:200]}",
|
||
f"Token Renewal for {asan_login_name}"
|
||
)
|
||
return {"success": False, "message": f"Failed with status {response.status_code}"}
|
||
|
||
except requests.RequestException as e:
|
||
# Сетевые ошибки подключения
|
||
if retry_count < MAX_RETRY_COUNT:
|
||
wait_time = (2 ** retry_count) * (0.5 + random.random())
|
||
time.sleep(wait_time)
|
||
return renew_token(asan_login_name, retry_count + 1)
|
||
|
||
# Check if the error is HTTP 401 or 500
|
||
if hasattr(e, 'response'):
|
||
if e.response.status_code == 401:
|
||
return {
|
||
"error": "unauthorized",
|
||
"message": "Authentication required. Please login again.",
|
||
"status_code": 401
|
||
}
|
||
elif e.response.status_code == 500:
|
||
return {
|
||
"error": "server_error",
|
||
"message": "Service temporarily unavailable. Please try again in a few minutes.",
|
||
"status_code": 500
|
||
}
|
||
|
||
frappe.log_error(
|
||
f"Network error: {str(e)}",
|
||
f"Token Renewal for {asan_login_name}"
|
||
)
|
||
return {"success": False, "message": f"Network error: {str(e)}"}
|
||
|
||
except Exception as e:
|
||
frappe.log_error(
|
||
f"Unexpected error: {str(e)}\n{frappe.get_traceback()}",
|
||
f"Token Renewal for {asan_login_name}"
|
||
)
|
||
return {"success": False, "message": "An unknown error occurred, please try again in a few minutes."}
|
||
|
||
@frappe.whitelist()
|
||
def setup_token_renewal():
|
||
"""Настраивает задание планировщика для обновления токенов"""
|
||
try:
|
||
# Проверка существования задания
|
||
job_exists = frappe.db.exists("Scheduled Job Type", "renew_etaxes_token")
|
||
|
||
if not job_exists:
|
||
# Создаем новое задание
|
||
doc = frappe.new_doc("Scheduled Job Type")
|
||
doc.update({
|
||
"name": "renew_etaxes_token",
|
||
"method": "invoice_az.auth.renew_token", # Обновляем путь
|
||
"frequency": "Cron",
|
||
"cron_format": "*/4 * * * *", # Каждые 4 минуты
|
||
"create_log": 1,
|
||
"execute_alone": 1,
|
||
"enabled": 1
|
||
})
|
||
doc.insert(ignore_permissions=True)
|
||
frappe.db.commit()
|
||
|
||
# Добавляем в планировщик немедленно
|
||
try:
|
||
from frappe.utils.background_jobs import get_scheduler
|
||
scheduler = get_scheduler()
|
||
if scheduler:
|
||
scheduler.add_scheduled_job(doc)
|
||
except Exception as e:
|
||
frappe.log_error(f"Could not add job to scheduler: {str(e)}", "Scheduler Error")
|
||
|
||
return {"success": True, "message": "Scheduler job created"}
|
||
else:
|
||
# Обновляем существующее задание если нужно
|
||
job = frappe.get_doc("Scheduled Job Type", "renew_etaxes_token")
|
||
|
||
if job.method != "invoice_az.auth.renew_token" or job.cron_format != "*/4 * * * *":
|
||
job.method = "invoice_az.auth.renew_token"
|
||
job.cron_format = "*/4 * * * *"
|
||
job.save(ignore_permissions=True)
|
||
frappe.db.commit()
|
||
|
||
# Обновляем в планировщике
|
||
try:
|
||
from frappe.utils.background_jobs import get_scheduler
|
||
scheduler = get_scheduler()
|
||
if scheduler:
|
||
scheduler.remove_scheduled_job(job.name)
|
||
scheduler.add_scheduled_job(job)
|
||
except Exception as e:
|
||
frappe.log_error(f"Could not update scheduler: {str(e)}", "Scheduler Error")
|
||
|
||
return {"success": True, "message": "Scheduler job updated"}
|
||
|
||
return {"success": True, "message": "Scheduler job already exists"}
|
||
|
||
except Exception as e:
|
||
frappe.log_error(f"Error setting up token renewal: {str(e)}\n{frappe.get_traceback()}", "Token Renewal Setup Error")
|
||
return {"success": False, "message": str(e)}
|
||
|
||
@frappe.whitelist()
|
||
def get_auth_token(phone, user_id):
|
||
"""Getting authorization token"""
|
||
# Записываем активность
|
||
record_etaxes_activity()
|
||
|
||
url = "https://new.e-taxes.gov.az/api/po/auth/public/v1/asanImza/start"
|
||
|
||
payload = {
|
||
"phone": phone,
|
||
"userId": user_id
|
||
}
|
||
|
||
headers = {
|
||
"Accept": "application/json, text/plain, */*",
|
||
"Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",
|
||
"Content-Type": "application/json",
|
||
"Cache-Control": "no-cache",
|
||
"Pragma": "no-cache",
|
||
"Referer": "https://new.e-taxes.gov.az/eportal/az/login/asan"
|
||
}
|
||
|
||
try:
|
||
response = requests.post(url, data=json.dumps(payload), headers=headers)
|
||
|
||
# Check for specific HTTP status codes
|
||
if response.status_code == 500:
|
||
return {
|
||
"error": "server_error",
|
||
"message": "Service temporarily unavailable. Please try again in a few minutes.",
|
||
"status_code": 500
|
||
}
|
||
|
||
if response.status_code == 401:
|
||
return {
|
||
"error": "unauthorized",
|
||
"message": "Authentication required. Please login again.",
|
||
"status_code": 401
|
||
}
|
||
|
||
response.raise_for_status()
|
||
|
||
# Getting token from the response header
|
||
bearer_token = response.headers.get('x-authorization', '')
|
||
|
||
# Getting response data
|
||
resp_data = response.json()
|
||
|
||
return {
|
||
'bearer_token': bearer_token,
|
||
'response_data': resp_data
|
||
}
|
||
except requests.exceptions.HTTPError as e:
|
||
# Check if the error is HTTP 401 or 500
|
||
if e.response.status_code == 401:
|
||
return {
|
||
"error": "unauthorized",
|
||
"message": "Authentication required. Please login again.",
|
||
"status_code": 401
|
||
}
|
||
elif e.response.status_code == 500:
|
||
return {
|
||
"error": "server_error",
|
||
"message": "Service temporarily unavailable. Please try again in a few minutes.",
|
||
"status_code": 500
|
||
}
|
||
frappe.log_error(f"HTTP error in get_auth_token: {str(e)}", "API Error")
|
||
return {"error": "http_error", "message": "An unknown error occurred, please try again in a few minutes."}
|
||
except Exception as e:
|
||
frappe.log_error(f"Error getting auth token: {str(e)}", "API Error")
|
||
return {"error": "unknown_error", "message": "An unknown error occurred, please try again in a few minutes."}
|
||
|
||
@frappe.whitelist()
|
||
def check_auth_status(bearer_token):
|
||
"""Checking authorization status"""
|
||
# Записываем активность
|
||
record_etaxes_activity()
|
||
|
||
url = "https://new.e-taxes.gov.az/api/po/auth/public/v1/asanImza/status"
|
||
|
||
headers = {
|
||
"Accept": "application/json, text/plain, */*",
|
||
"Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",
|
||
"Cache-Control": "no-cache",
|
||
"Pragma": "no-cache",
|
||
"Referer": "https://new.e-taxes.gov.az/eportal/az/login/asan"
|
||
}
|
||
|
||
# Adding token to headers if it exists
|
||
if bearer_token:
|
||
headers["x-authorization"] = f"Bearer {bearer_token}"
|
||
|
||
try:
|
||
response = requests.get(url, headers=headers)
|
||
|
||
# Check for specific HTTP status codes
|
||
if response.status_code == 500:
|
||
return {
|
||
"error": "server_error",
|
||
"message": "Service temporarily unavailable. Please try again in a few minutes.",
|
||
"status_code": 500
|
||
}
|
||
|
||
if response.status_code == 401:
|
||
return {
|
||
"error": "unauthorized",
|
||
"message": "Authentication required. Please login again.",
|
||
"status_code": 401
|
||
}
|
||
|
||
response.raise_for_status()
|
||
|
||
# Getting status data
|
||
status_data = response.json()
|
||
|
||
return status_data
|
||
except requests.exceptions.HTTPError as e:
|
||
# Check if the error is HTTP 401 or 500
|
||
if e.response.status_code == 401:
|
||
return {
|
||
"error": "unauthorized",
|
||
"message": "Authentication required. Please login again.",
|
||
"status_code": 401
|
||
}
|
||
elif e.response.status_code == 500:
|
||
return {
|
||
"error": "server_error",
|
||
"message": "Service temporarily unavailable. Please try again in a few minutes.",
|
||
"status_code": 500
|
||
}
|
||
frappe.log_error(f"HTTP error in check_auth_status: {str(e)}", "API Error")
|
||
return {"error": "http_error", "message": "An unknown error occurred, please try again in a few minutes."}
|
||
except Exception as e:
|
||
frappe.log_error(f"Error checking auth status: {str(e)}", "API Error")
|
||
return {"error": "unknown_error", "message": "An unknown error occurred, please try again in a few minutes."}
|
||
|
||
@frappe.whitelist()
|
||
def get_certificates(bearer_token):
|
||
"""Getting available certificates"""
|
||
# Записываем активность
|
||
record_etaxes_activity()
|
||
|
||
url = "https://new.e-taxes.gov.az/api/po/auth/public/v1/asanImza/certificates"
|
||
|
||
headers = {
|
||
"Accept": "application/json, text/plain, */*",
|
||
"Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",
|
||
"Cache-Control": "no-cache",
|
||
"Pragma": "no-cache",
|
||
"Referer": "https://new.e-taxes.gov.az/eportal/az/verification/asan",
|
||
"x-authorization": f"Bearer {bearer_token}"
|
||
}
|
||
|
||
try:
|
||
response = requests.get(url, headers=headers)
|
||
|
||
if response.status_code == 500:
|
||
return {
|
||
"error": "server_error",
|
||
"message": "Service temporarily unavailable. Please try again in a few minutes.",
|
||
"status_code": 500
|
||
}
|
||
|
||
# Check for 401 error
|
||
if response.status_code == 401:
|
||
return {
|
||
"error": "unauthorized",
|
||
"message": "Authentication required. Please login again."
|
||
}
|
||
|
||
response.raise_for_status()
|
||
|
||
return response.json()
|
||
except requests.exceptions.HTTPError as e:
|
||
|
||
# Check if the error is 401 Unauthorized
|
||
if e.response.status_code == 401:
|
||
return {
|
||
"error": "unauthorized",
|
||
"message": "Authentication required. Please login again."
|
||
}
|
||
elif e.response.status_code == 500:
|
||
return {
|
||
"error": "server_error",
|
||
"message": "Service temporarily unavailable. Please try again in a few minutes.",
|
||
"status_code": 500
|
||
}
|
||
|
||
frappe.log_error(f"HTTP error in get_certificates: {str(e)}", "API Error")
|
||
return {"error": "http_error", "message": "An unknown error occurred, please try again in a few minutes."}
|
||
except Exception as e:
|
||
frappe.log_error(f"Error getting certificates: {str(e)}", "API Error")
|
||
return {"error": "unknown_error", "message": "An unknown error occurred, please try again in a few minutes."}
|
||
|
||
@frappe.whitelist()
|
||
def choose_taxpayer(bearer_token, owner_type, tin):
|
||
"""Selecting taxpayer after certificate selection"""
|
||
# Записываем активность
|
||
record_etaxes_activity()
|
||
|
||
url = "https://new.e-taxes.gov.az/api/po/auth/public/v1/asanImza/chooseTaxpayer"
|
||
|
||
payload = {
|
||
"ownerType": owner_type, # "legal" or "individual"
|
||
"legalTin": tin if owner_type == "legal" else None,
|
||
"individualFin": tin if owner_type == "individual" else None
|
||
}
|
||
|
||
# Remove None values from payload
|
||
payload = {k: v for k, v in payload.items() if v is not None}
|
||
|
||
headers = {
|
||
"Accept": "application/json, text/plain, */*",
|
||
"Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",
|
||
"Content-Type": "application/json",
|
||
"Cache-Control": "no-cache",
|
||
"Pragma": "no-cache",
|
||
"Referer": "https://new.e-taxes.gov.az/eportal/az/login/asan",
|
||
"x-authorization": f"Bearer {bearer_token}"
|
||
}
|
||
|
||
try:
|
||
response = requests.post(url, data=json.dumps(payload), headers=headers)
|
||
|
||
if response.status_code == 500:
|
||
return {
|
||
"error": "server_error",
|
||
"message": "Service temporarily unavailable. Please try again in a few minutes.",
|
||
"status_code": 500
|
||
}
|
||
|
||
# Check for 401 error
|
||
if response.status_code == 401:
|
||
return {
|
||
"error": "unauthorized",
|
||
"message": "Authentication required. Please login again."
|
||
}
|
||
|
||
response.raise_for_status()
|
||
|
||
# Get main token from response header
|
||
main_token = response.headers.get('x-authorization', '')
|
||
|
||
# Get response data
|
||
resp_data = response.json()
|
||
|
||
return {
|
||
'main_token': main_token,
|
||
'response_data': resp_data
|
||
}
|
||
except requests.exceptions.HTTPError as e:
|
||
|
||
# Check if the error is 401 Unauthorized
|
||
if e.response.status_code == 401:
|
||
return {
|
||
"error": "unauthorized",
|
||
"message": "Authentication required. Please login again."
|
||
}
|
||
elif e.response.status_code == 500:
|
||
return {
|
||
"error": "server_error",
|
||
"message": "Service temporarily unavailable. Please try again in a few minutes.",
|
||
"status_code": 500
|
||
}
|
||
|
||
frappe.log_error(f"HTTP error in choose_taxpayer: {str(e)}", "API Error")
|
||
return {"error": "http_error", "message": "An unknown error occurred, please try again in a few minutes."}
|
||
except Exception as e:
|
||
frappe.log_error(f"Error choosing taxpayer: {str(e)}", "API Error")
|
||
return {"error": "unknown_error", "message": "An unknown error occurred, please try again in a few minutes."}
|
||
|
||
@frappe.whitelist()
|
||
def get_default_asan_login():
|
||
"""Get default authorization settings"""
|
||
try:
|
||
# Look for setting with is_default = 1
|
||
default_login = frappe.get_list(
|
||
"Asan Login",
|
||
filters={"is_default": 1},
|
||
fields=["name"]
|
||
)
|
||
|
||
if default_login:
|
||
# If default setting is found, return its data
|
||
asan_login = frappe.get_doc("Asan Login", default_login[0].name)
|
||
return {
|
||
"found": True,
|
||
"name": asan_login.name,
|
||
"phone": asan_login.phone,
|
||
"user_id": asan_login.user_id,
|
||
"auth_status": asan_login.auth_status,
|
||
"bearer_token": asan_login.bearer_token,
|
||
"main_token": asan_login.main_token,
|
||
"selected_certificate": asan_login.selected_certificate
|
||
}
|
||
else:
|
||
# If no default setting is found, look for any first setting
|
||
any_login = frappe.get_list(
|
||
"Asan Login",
|
||
fields=["name"]
|
||
)
|
||
|
||
if any_login:
|
||
asan_login = frappe.get_doc("Asan Login", any_login[0].name)
|
||
return {
|
||
"found": True,
|
||
"name": asan_login.name,
|
||
"phone": asan_login.phone,
|
||
"user_id": asan_login.user_id,
|
||
"auth_status": asan_login.auth_status,
|
||
"bearer_token": asan_login.bearer_token,
|
||
"main_token": asan_login.main_token,
|
||
"selected_certificate": asan_login.selected_certificate
|
||
}
|
||
else:
|
||
return {"found": False, "message": "No Asan Login settings found."}
|
||
|
||
except Exception as e:
|
||
return {"found": False, "error": str(e)}
|
||
|
||
@frappe.whitelist()
|
||
def handle_authentication(asan_login_name=None):
|
||
"""Handles the authentication process for the specified or default setting"""
|
||
# Записываем активность
|
||
record_etaxes_activity()
|
||
|
||
try:
|
||
# If setting name is not specified, get default setting name
|
||
if not asan_login_name:
|
||
default_settings = get_default_asan_login()
|
||
if not default_settings.get("found", False):
|
||
return {"success": False, "message": "No Asan Login settings found."}
|
||
asan_login_name = default_settings.get("name")
|
||
|
||
# Get Asan Login document
|
||
asan_login = frappe.get_doc("Asan Login", asan_login_name)
|
||
|
||
# Check that there is a phone number and user ID
|
||
if not asan_login.phone or not asan_login.user_id:
|
||
return {"success": False, "message": "Phone and User ID are required."}
|
||
|
||
# Start the authorization process
|
||
auth_result = get_auth_token(asan_login.phone, asan_login.user_id)
|
||
|
||
# Check for errors
|
||
if auth_result and auth_result.get("error"):
|
||
return {"success": False, "message": auth_result.get("message")}
|
||
|
||
if auth_result and auth_result.get("bearer_token"):
|
||
# Save token and change status
|
||
asan_login.bearer_token = auth_result.get("bearer_token")
|
||
asan_login.auth_status = "Waiting for confirmation"
|
||
asan_login.save()
|
||
|
||
# Extract verification code from response if it exists
|
||
verification_code = None
|
||
if auth_result.get("response_data") and isinstance(auth_result.get("response_data"), dict):
|
||
verification_code = auth_result.get("response_data").get("verificationCode", None)
|
||
|
||
return {
|
||
"success": True,
|
||
"message": "Authentication initiated. Please confirm on your phone.",
|
||
"bearer_token": auth_result.get("bearer_token"),
|
||
"asan_login_name": asan_login_name,
|
||
"verification_code": verification_code
|
||
}
|
||
else:
|
||
return {"success": False, "message": "Failed to get authentication token."}
|
||
|
||
except Exception as e:
|
||
frappe.log_error(f"Error in handle_authentication: {str(e)}\n{frappe.get_traceback()}", "Authentication Error")
|
||
return {"success": False, "message": "An unknown error occurred, please try again in a few minutes."}
|
||
|
||
@frappe.whitelist()
|
||
def poll_auth_status(asan_login_name, bearer_token):
|
||
"""Checks authentication status"""
|
||
# Записываем активность
|
||
record_etaxes_activity()
|
||
|
||
try:
|
||
# Check status
|
||
status_result = check_auth_status(bearer_token)
|
||
|
||
# Check for errors
|
||
if status_result and status_result.get("error"):
|
||
return {"success": False, "message": status_result.get("message")}
|
||
|
||
# Get Asan Login document
|
||
asan_login = frappe.get_doc("Asan Login", asan_login_name)
|
||
|
||
if status_result and status_result.get("successful") == True:
|
||
# Update status and token
|
||
asan_login.auth_status = "Authenticated"
|
||
if status_result.get("bearer_token"):
|
||
asan_login.bearer_token = status_result.get("bearer_token")
|
||
asan_login.save()
|
||
|
||
return {
|
||
"success": True,
|
||
"authenticated": True,
|
||
"message": "Authentication successful."
|
||
}
|
||
else:
|
||
return {
|
||
"success": True,
|
||
"authenticated": False,
|
||
"message": "Waiting for confirmation."
|
||
}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "authenticated": False, "message": "An unknown error occurred, please try again in a few minutes."}
|
||
|
||
@frappe.whitelist()
|
||
def get_auth_certificates(asan_login_name):
|
||
"""Gets certificates for the specified Asan Login setting"""
|
||
# Записываем активность
|
||
record_etaxes_activity()
|
||
|
||
try:
|
||
# Get Asan Login document
|
||
asan_login = frappe.get_doc("Asan Login", asan_login_name)
|
||
|
||
# Check authorization status
|
||
if asan_login.auth_status != "Authenticated" and asan_login.auth_status != "Fully Authenticated":
|
||
return {"success": False, "message": "Not authenticated. Please login first."}
|
||
|
||
# Get certificates
|
||
certificates_result = get_certificates(asan_login.bearer_token)
|
||
|
||
# Check for authorization error
|
||
if certificates_result.get("error") == "unauthorized":
|
||
return {
|
||
"success": False,
|
||
"unauthorized": True,
|
||
"message": "Authentication required. Please login again."
|
||
}
|
||
|
||
# Check for server error
|
||
if certificates_result.get("error") == "server_error":
|
||
return {
|
||
"success": False,
|
||
"message": "Service temporarily unavailable. Please try again in a few minutes."
|
||
}
|
||
|
||
# Check for other errors
|
||
if certificates_result.get("error"):
|
||
return {
|
||
"success": False,
|
||
"message": certificates_result.get("message", "An unknown error occurred, please try again in a few minutes.")
|
||
}
|
||
|
||
# Check for certificates in the response
|
||
if certificates_result and certificates_result.get("certificates"):
|
||
# Save certificates in the document
|
||
asan_login.certificates_json = json.dumps(certificates_result.get("certificates"))
|
||
asan_login.save()
|
||
|
||
return {
|
||
"success": True,
|
||
"message": "Certificates retrieved successfully.",
|
||
"certificates": certificates_result.get("certificates")
|
||
}
|
||
else:
|
||
return {"success": False, "message": "No certificates found."}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "message": "An unknown error occurred, please try again in a few minutes."}
|
||
|
||
@frappe.whitelist()
|
||
def select_certificate(asan_login_name, certificate_data, certificate_name):
|
||
"""Selects certificate for the specified Asan Login setting"""
|
||
# Записываем активность
|
||
record_etaxes_activity()
|
||
|
||
try:
|
||
# Get Asan Login document
|
||
asan_login = frappe.get_doc("Asan Login", asan_login_name)
|
||
|
||
# Check authorization status
|
||
if asan_login.auth_status != "Authenticated" and asan_login.auth_status != "Fully Authenticated":
|
||
return {"success": False, "message": "Not authenticated. Please login first."}
|
||
|
||
# Convert JSON string to object if necessary
|
||
if isinstance(certificate_data, str):
|
||
certificate_data = json.loads(certificate_data)
|
||
|
||
# Save selected certificate
|
||
asan_login.selected_certificate = certificate_name
|
||
asan_login.selected_certificate_json = json.dumps(certificate_data)
|
||
asan_login.save()
|
||
|
||
return {
|
||
"success": True,
|
||
"message": "Certificate selected successfully."
|
||
}
|
||
|
||
except Exception as e:
|
||
frappe.log_error(f"Error in select_certificate: {str(e)}\n{frappe.get_traceback()}", "Select Certificate Error")
|
||
return {"success": False, "message": "An unknown error occurred, please try again in a few minutes."}
|
||
|
||
@frappe.whitelist()
|
||
def select_taxpayer(asan_login_name):
|
||
"""Selects taxpayer for the specified Asan Login setting"""
|
||
# Записываем активность
|
||
record_etaxes_activity()
|
||
|
||
try:
|
||
# Get Asan Login document
|
||
asan_login = frappe.get_doc("Asan Login", asan_login_name)
|
||
|
||
# Check authorization status
|
||
if asan_login.auth_status != "Authenticated" and asan_login.auth_status != "Fully Authenticated":
|
||
return {"success": False, "message": "Not authenticated. Please login first."}
|
||
|
||
# Check for selected certificate
|
||
if not asan_login.selected_certificate_json:
|
||
return {"success": False, "message": "No certificate selected."}
|
||
|
||
# Parse certificate data
|
||
cert_data = json.loads(asan_login.selected_certificate_json)
|
||
|
||
# Determine taxpayer type and ID
|
||
owner_type = cert_data.get("taxpayerType", "")
|
||
tin = ""
|
||
|
||
if owner_type == "individual" and cert_data.get("individualInfo"):
|
||
tin = cert_data.get("individualInfo", {}).get("fin", "")
|
||
owner_type = "individual"
|
||
elif cert_data.get("legalInfo"):
|
||
owner_type = "legal"
|
||
tin = cert_data.get("legalInfo", {}).get("tin", "") or cert_data.get("legalInfo", {}).get("voen", "")
|
||
|
||
if not owner_type or not tin:
|
||
return {"success": False, "message": "Invalid certificate data. Cannot determine taxpayer type or ID."}
|
||
|
||
# Log taxpayer selection attempt
|
||
frappe.log_error(
|
||
f"Attempting to choose taxpayer: type={owner_type}, tin={tin}, bearer_token={asan_login.bearer_token[:10]}...",
|
||
"Choose Taxpayer"
|
||
)
|
||
|
||
# Select taxpayer
|
||
taxpayer_result = choose_taxpayer(asan_login.bearer_token, owner_type, tin)
|
||
|
||
# Check for authorization error
|
||
if taxpayer_result.get("error") == "unauthorized":
|
||
return {
|
||
"success": False,
|
||
"unauthorized": True,
|
||
"message": "Authentication required. Please login again."
|
||
}
|
||
|
||
# Check for server error
|
||
if taxpayer_result.get("error") == "server_error":
|
||
return {
|
||
"success": False,
|
||
"message": "Service temporarily unavailable. Please try again in a few minutes."
|
||
}
|
||
|
||
# Check for other errors
|
||
if taxpayer_result.get("error"):
|
||
return {
|
||
"success": False,
|
||
"message": taxpayer_result.get("message", "An unknown error occurred, please try again in a few minutes.")
|
||
}
|
||
|
||
# Check selection success
|
||
if taxpayer_result and taxpayer_result.get("main_token"):
|
||
# Save main token and response
|
||
asan_login.main_token = taxpayer_result.get("main_token")
|
||
asan_login.choose_taxpayer_response = json.dumps(taxpayer_result.get("response_data", {}))
|
||
asan_login.auth_status = "Fully Authenticated"
|
||
asan_login.save()
|
||
|
||
# Log successful taxpayer selection
|
||
frappe.log_error(
|
||
f"Successfully chose taxpayer: type={owner_type}, tin={tin}, main_token={asan_login.main_token[:10]}...",
|
||
"Choose Taxpayer Success"
|
||
)
|
||
|
||
return {
|
||
"success": True,
|
||
"message": "Taxpayer selected successfully."
|
||
}
|
||
else:
|
||
# Log taxpayer selection error
|
||
frappe.log_error(
|
||
f"Failed to choose taxpayer: type={owner_type}, tin={tin}, response={taxpayer_result}",
|
||
"Choose Taxpayer Error"
|
||
)
|
||
|
||
return {"success": False, "message": "Failed to select taxpayer."}
|
||
|
||
except Exception as e:
|
||
frappe.log_error(f"Error in select_taxpayer: {str(e)}\n{frappe.get_traceback()}", "Choose Taxpayer Exception")
|
||
return {"success": False, "message": "An unknown error occurred, please try again in a few minutes."}
|
||
|
||
@frappe.whitelist()
|
||
def check_token_validity(asan_login_name=None):
|
||
"""Проверяет валидность main_token для указанных настроек Asan Login"""
|
||
# Записываем активность
|
||
record_etaxes_activity()
|
||
|
||
try:
|
||
# Если имя не указано, получаем настройки по умолчанию
|
||
if not asan_login_name:
|
||
default_settings = get_default_asan_login()
|
||
if not default_settings.get("found", False):
|
||
return {"valid": False, "message": "No Asan Login settings found."}
|
||
asan_login_name = default_settings.get("name")
|
||
|
||
# Получаем документ Asan Login
|
||
asan_login = frappe.get_doc("Asan Login", asan_login_name)
|
||
|
||
# Проверяем наличие main_token
|
||
if not asan_login.main_token:
|
||
return {"valid": False, "message": "Main token not found."}
|
||
|
||
# Создаем минимальный фильтр для запроса инвойсов (импортируем функцию из api)
|
||
from invoice_az.api import get_invoices
|
||
|
||
minimal_filter = {
|
||
"maxCount": 1, # Запрашиваем только один инвойс для минимальной нагрузки
|
||
"offset": 0
|
||
}
|
||
|
||
# Пробуем получить список инвойсов для проверки main_token
|
||
invoices_result = get_invoices(asan_login.main_token, json.dumps(minimal_filter))
|
||
|
||
# Проверяем на ошибку авторизации
|
||
if invoices_result.get("error") == "unauthorized":
|
||
# Обновляем статус авторизации в документе
|
||
asan_login.auth_status = "Not Authenticated"
|
||
asan_login.save(ignore_permissions=True)
|
||
frappe.db.commit()
|
||
|
||
return {
|
||
"valid": False,
|
||
"message": "Authentication required. Main token is expired."
|
||
}
|
||
|
||
# Проверяем на ошибку сервера
|
||
if invoices_result.get("error") == "server_error":
|
||
return {
|
||
"valid": False,
|
||
"message": "Service temporarily unavailable. Please try again in a few minutes."
|
||
}
|
||
|
||
# Проверяем на другие ошибки
|
||
if invoices_result.get("error"):
|
||
return {
|
||
"valid": False,
|
||
"message": invoices_result.get("message", "An unknown error occurred.")
|
||
}
|
||
|
||
# Если нет ошибок, токен валиден
|
||
return {"valid": True, "message": "Main token is valid."}
|
||
|
||
except Exception as e:
|
||
frappe.log_error(f"Error in check_token_validity: {str(e)}\n{frappe.get_traceback()}", "Token Validity Check")
|
||
return {"valid": False, "message": "An unknown error occurred, please try again in a few minutes."}
|
||
|
||
@frappe.whitelist()
|
||
def handle_unauthorized_request(asan_login_name, operation_name="API request"):
|
||
"""Common function for handling unauthorized requests"""
|
||
# Log event
|
||
frappe.log_error(
|
||
f"Unauthorized request for operation: {operation_name}, asan_login_name: {asan_login_name}",
|
||
"Unauthorized API Request"
|
||
)
|
||
|
||
return {
|
||
"error": "unauthorized",
|
||
"message": "Authentication required. Please login again.",
|
||
"operation": operation_name
|
||
} |