fixed bugs in company, and added agricultural_purpose_type

This commit is contained in:
Ali 2025-09-09 20:42:07 +04:00
parent 024a5f53ee
commit 0ce377eb11
11 changed files with 1870 additions and 264 deletions

View File

@ -25,7 +25,7 @@ ACTIVITY_TIMEOUT_MINUTES = 10 # Настраиваемый параметр в
@frappe.whitelist()
def record_etaxes_activity(asan_login_name=None):
"""Записывает время последней активности с e-taxes с задержкой 2 секунды"""
"""Записывает время последней активности с e-taxes напрямую"""
try:
# Если имя не указано, получаем дефолтный профиль
if not asan_login_name:
@ -41,13 +41,10 @@ def record_etaxes_activity(asan_login_name=None):
asan_login_name = default_settings[0].name
# Запускаем отложенную задачу для обновления поля last_activity_time
frappe.enqueue(
'invoice_az.auth.update_activity_time',
asan_login_name=asan_login_name,
queue='short',
timeout=300
)
# Обновляем поле 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")

File diff suppressed because it is too large Load Diff

View File

@ -170,7 +170,6 @@ def validate_gambling_activity(company_name, main_activity=None):
"message": f"Error validating gambling activity: {str(e)}"
}
# Остальные функции остаются без изменений
@frappe.whitelist()
def get_company_profile(asan_login_name=None):
"""Получает профиль компании из E-Taxes"""
@ -265,6 +264,284 @@ def get_company_profile(asan_login_name=None):
"message": "An unknown error occurred, please try again in a few minutes."
}
@frappe.whitelist()
def get_tax_authorities(asan_login_name=None):
"""Получает справочник налоговых органов из E-Taxes"""
record_etaxes_activity()
try:
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")
asan_login = frappe.get_doc("Asan Login", asan_login_name)
if not asan_login.main_token:
return {"success": False, "message": "Main token not found. Please authenticate first."}
if asan_login.auth_status != "Fully Authenticated":
return {"success": False, "message": "Not fully authenticated. Please complete authentication process."}
url = "https://new.e-taxes.gov.az/api/po/dictionary/public/v1/taxAuthorities"
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",
"x-authorization": f"Bearer {asan_login.main_token}"
}
response = requests.get(url, headers=headers, timeout=15)
if response.status_code == 500:
return {
"error": "server_error",
"message": "E-Taxes service temporarily unavailable. Please try again in a few minutes.",
"status_code": 500
}
if response.status_code == 401:
frappe.db.set_value("Asan Login", asan_login_name, "auth_status", "Not Authenticated")
frappe.db.commit()
return {
"error": "unauthorized",
"message": "Authentication required. Please login again.",
"status_code": 401
}
response.raise_for_status()
tax_authorities_data = response.json()
frappe.logger().info(f"Tax authorities dictionary retrieved successfully for {asan_login_name}")
return {
"success": True,
"data": tax_authorities_data,
"message": "Tax authorities dictionary retrieved successfully"
}
except requests.exceptions.HTTPError as e:
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": "E-Taxes service temporarily unavailable. Please try again in a few minutes.",
"status_code": 500
}
frappe.log_error(f"HTTP error in get_tax_authorities: {str(e)}", "Tax Authorities API Error")
return {
"error": "http_error",
"message": "An error occurred while fetching tax authorities dictionary"
}
except requests.exceptions.RequestException as e:
frappe.log_error(f"Request error in get_tax_authorities: {str(e)}", "Tax Authorities API Error")
return {
"error": "network_error",
"message": "Network error occurred. Please check your connection and try again."
}
except Exception as e:
frappe.log_error(f"Unexpected error in get_tax_authorities: {str(e)}\n{frappe.get_traceback()}", "Tax Authorities Error")
return {
"error": "unknown_error",
"message": "An unknown error occurred, please try again in a few minutes."
}
@frappe.whitelist()
def get_property_types(asan_login_name=None):
"""Получает справочник типов собственности из E-Taxes"""
record_etaxes_activity()
try:
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")
asan_login = frappe.get_doc("Asan Login", asan_login_name)
if not asan_login.main_token:
return {"success": False, "message": "Main token not found. Please authenticate first."}
if asan_login.auth_status != "Fully Authenticated":
return {"success": False, "message": "Not fully authenticated. Please complete authentication process."}
url = "https://new.e-taxes.gov.az/api/po/profile/public/v1/dictionary/propertyTypes"
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",
"x-authorization": f"Bearer {asan_login.main_token}"
}
response = requests.get(url, headers=headers, timeout=15)
if response.status_code == 500:
return {
"error": "server_error",
"message": "E-Taxes service temporarily unavailable. Please try again in a few minutes.",
"status_code": 500
}
if response.status_code == 401:
frappe.db.set_value("Asan Login", asan_login_name, "auth_status", "Not Authenticated")
frappe.db.commit()
return {
"error": "unauthorized",
"message": "Authentication required. Please login again.",
"status_code": 401
}
response.raise_for_status()
property_types_data = response.json()
frappe.logger().info(f"Property types dictionary retrieved successfully for {asan_login_name}")
return {
"success": True,
"data": property_types_data,
"message": "Property types dictionary retrieved successfully"
}
except requests.exceptions.HTTPError as e:
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": "E-Taxes service temporarily unavailable. Please try again in a few minutes.",
"status_code": 500
}
frappe.log_error(f"HTTP error in get_property_types: {str(e)}", "Property Types API Error")
return {
"error": "http_error",
"message": "An error occurred while fetching property types dictionary"
}
except requests.exceptions.RequestException as e:
frappe.log_error(f"Request error in get_property_types: {str(e)}", "Property Types API Error")
return {
"error": "network_error",
"message": "Network error occurred. Please check your connection and try again."
}
except Exception as e:
frappe.log_error(f"Unexpected error in get_property_types: {str(e)}\n{frappe.get_traceback()}", "Property Types Error")
return {
"error": "unknown_error",
"message": "An unknown error occurred, please try again in a few minutes."
}
@frappe.whitelist()
def get_tax_authority_name_by_code(code, asan_login_name=None):
"""Получает название налогового органа по коду"""
try:
tax_authorities_result = get_tax_authorities(asan_login_name)
if not tax_authorities_result.get("success"):
return {
"success": False,
"message": tax_authorities_result.get("message", "Failed to get tax authorities dictionary")
}
tax_authorities = tax_authorities_result.get("data", [])
if tax_authorities and isinstance(tax_authorities, list):
for authority in tax_authorities:
if authority.get("code") == code:
name = authority.get("name", {})
if isinstance(name, dict) and name.get("az"):
return {
"success": True,
"name": name.get("az"),
"code": code
}
elif isinstance(name, str):
return {
"success": True,
"name": name,
"code": code
}
return {
"success": True,
"name": code,
"code": code,
"message": "Tax authority name not found, returning code"
}
except Exception as e:
frappe.log_error(f"Error in get_tax_authority_name_by_code: {str(e)}", "Tax Authority Name Error")
return {
"success": False,
"message": "An error occurred while getting tax authority name"
}
@frappe.whitelist()
def get_property_type_name_by_code(code, asan_login_name=None):
"""Получает название типа собственности по коду"""
try:
property_types_result = get_property_types(asan_login_name)
if not property_types_result.get("success"):
return {
"success": False,
"message": property_types_result.get("message", "Failed to get property types dictionary")
}
property_types = property_types_result.get("data", {})
if property_types and isinstance(property_types, dict):
for property_type in property_types.values():
if isinstance(property_type, dict) and property_type.get("code") == code:
name = property_type.get("name", {})
if isinstance(name, dict) and name.get("az"):
return {
"success": True,
"name": name.get("az"),
"code": code
}
elif isinstance(name, str):
return {
"success": True,
"name": name,
"code": code
}
return {
"success": True,
"name": code,
"code": code,
"message": "Property type name not found, returning code"
}
except Exception as e:
frappe.log_error(f"Error in get_property_type_name_by_code: {str(e)}", "Property Type Name Error")
return {
"success": False,
"message": "An error occurred while getting property type name"
}
@frappe.whitelist()
def force_update_single_field(company_name, field_name, field_value):
"""Форсированное обновление одного поля компании"""
@ -313,13 +590,177 @@ def force_update_single_field(company_name, field_name, field_value):
"message": f"Error updating field: {str(e)}"
}
# Остальные функции (get_tax_authorities, get_property_types, force_update_company и т.д.) остаются без изменений
@frappe.whitelist()
def force_update_company(company_name, field_updates=None, additional_activities=None, affiliate_names=None, affiliate_tins=None):
"""Форсированное обновление полей компании через прямые SQL-запросы"""
import json
try:
# Парсим JSON если нужно
if isinstance(field_updates, str):
field_updates = json.loads(field_updates) if field_updates else {}
if isinstance(additional_activities, str):
additional_activities = json.loads(additional_activities) if additional_activities else []
if isinstance(affiliate_names, str):
affiliate_names = json.loads(affiliate_names) if affiliate_names else []
if isinstance(affiliate_tins, str):
affiliate_tins = json.loads(affiliate_tins) if affiliate_tins else []
# Проверяем существование компании
if not frappe.db.exists("Company", company_name):
return {"success": False, "message": f"Company {company_name} not found"}
# Обновляем основные поля через SQL
if field_updates:
set_clauses = []
values = []
for field, value in field_updates.items():
set_clauses.append(f"`{field}` = %s")
values.append(value)
if set_clauses:
sql = f"""
UPDATE `tabCompany`
SET {', '.join(set_clauses)}, `modified` = NOW(), `modified_by` = %s
WHERE `name` = %s
"""
values.append(frappe.session.user)
values.append(company_name)
frappe.db.sql(sql, tuple(values))
# Обработка additional activities
activities_added = 0
if additional_activities and len(additional_activities) > 0:
# ИСПРАВЛЕНО: Правильное имя таблицы
table_name = "Company Additional Activity" # БЕЗ "Types" в конце!
if not frappe.db.table_exists(table_name):
frappe.log_error(f"Table '{table_name}' does not exist", "Table Check Error")
return {"success": False, "message": f"Table '{table_name}' does not exist"}
# Удаляем старые записи
frappe.db.sql(f"""DELETE FROM `tab{table_name}` WHERE parent = %s""", company_name)
# Добавляем новые additional activities
for idx, activity_code in enumerate(additional_activities):
if activity_code and activity_code.strip():
# Проверяем существование activity в справочнике
activity_exists = frappe.db.exists("Main type of activity", activity_code.strip())
if activity_exists:
# Генерируем уникальное имя
record_name = frappe.generate_hash(length=10)
frappe.db.sql(f"""
INSERT INTO `tab{table_name}`
(name, parent, parenttype, parentfield, idx, activity_type, docstatus, creation, modified, modified_by, owner)
VALUES (%s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), %s, %s)
""", (
record_name,
company_name,
'Company',
'additional_activity_types', # fieldname остается тем же
idx + 1,
activity_code.strip(),
0,
frappe.session.user,
frappe.session.user
))
activities_added += 1
frappe.logger().info(f"Added additional activity: {activity_code} for company {company_name}")
else:
frappe.logger().warning(f"Activity code {activity_code} not found in Main type of activity")
# Проверяем, что записи действительно добавились
count = frappe.db.sql(f"""
SELECT COUNT(*) FROM `tab{table_name}`
WHERE parent = %s
""", company_name)[0][0]
frappe.logger().info(f"Total additional activities in DB after update: {count}")
# Обработка affiliate organizations
affiliates_added = 0
if (affiliate_names or affiliate_tins):
# ИСПРАВЛЕНО: Правильное имя таблицы
table_name = "Company Affiliate Organization" # В единственном числе!
if frappe.db.table_exists(table_name):
# Удаляем старые записи
frappe.db.sql(f"""DELETE FROM `tab{table_name}` WHERE parent = %s""", company_name)
# Добавляем новые affiliate organizations
max_length = max(len(affiliate_names or []), len(affiliate_tins or []))
for idx in range(max_length):
org_name = affiliate_names[idx] if affiliate_names and idx < len(affiliate_names) else ""
org_tin = affiliate_tins[idx] if affiliate_tins and idx < len(affiliate_tins) else ""
if org_name or org_tin:
if not org_name and org_tin:
org_name = f"Organization (TIN: {org_tin})"
record_name = frappe.generate_hash(length=10)
frappe.db.sql(f"""
INSERT INTO `tab{table_name}`
(name, parent, parenttype, parentfield, idx, organization_name, organization_tin, docstatus, creation, modified, modified_by, owner)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), %s, %s)
""", (
record_name,
company_name,
'Company',
'affiliate_organizations', # fieldname остается тем же
idx + 1,
org_name,
org_tin or None,
0,
frappe.session.user,
frappe.session.user
))
affiliates_added += 1
frappe.logger().info(f"Added affiliate organization: {org_name} (TIN: {org_tin}) for company {company_name}")
else:
frappe.logger().error(f"Table '{table_name}' does not exist")
# Коммитим изменения
frappe.db.commit()
# Очищаем весь кэш для компании
frappe.clear_cache(doctype="Company")
frappe.clear_document_cache("Company", company_name)
# Если обновляли main_activity, проверяем игорную деятельность
if field_updates and 'main_activity' in field_updates:
validate_gambling_activity(company_name, field_updates['main_activity'])
# Возвращаем детальную информацию об обновлении
return {
"success": True,
"message": "Company updated successfully",
"updated_fields": len(field_updates) if field_updates else 0,
"additional_activities_added": activities_added,
"affiliates_added": affiliates_added
}
except Exception as e:
frappe.log_error(f"Error in force_update_company: {str(e)}\n{frappe.get_traceback()}", "Company Update Error")
frappe.db.rollback()
return {
"success": False,
"message": f"Error updating company: {str(e)}"
}
# Helper functions
def record_etaxes_activity():
"""Placeholder for activity recording"""
pass
def get_default_asan_login():
"""Placeholder for getting default login"""
"""Получает настройки по умолчанию для Asan Login"""
try:
default_login = frappe.get_list(
"Asan Login",
@ -344,3 +785,61 @@ def get_default_asan_login():
except Exception as e:
return {"found": False, "error": str(e)}
@frappe.whitelist()
def check_company_structure(company_name):
"""Диагностическая функция для проверки структуры Company"""
try:
company = frappe.get_doc("Company", company_name)
# Проверяем наличие полей
has_additional_activities = hasattr(company, 'additional_activity_types')
has_affiliate_orgs = hasattr(company, 'affiliate_organizations')
# ИСПРАВЛЕНО: Правильные имена таблиц
table1_exists = frappe.db.table_exists("Company Additional Activity")
table2_exists = frappe.db.table_exists("Company Affiliate Organization")
# Получаем текущее количество записей
activities_count = 0
affiliates_count = 0
if table1_exists:
activities_count = frappe.db.sql("""
SELECT COUNT(*) FROM `tabCompany Additional Activity`
WHERE parent = %s
""", company_name)[0][0]
if table2_exists:
affiliates_count = frappe.db.sql("""
SELECT COUNT(*) FROM `tabCompany Affiliate Organization`
WHERE parent = %s
""", company_name)[0][0]
# Получаем метаданные о child tables
meta = frappe.get_meta("Company")
child_tables = []
for field in meta.fields:
if field.fieldtype == "Table":
child_tables.append({
"fieldname": field.fieldname,
"options": field.options,
"label": field.label
})
return {
"success": True,
"has_additional_activities_field": has_additional_activities,
"has_affiliate_orgs_field": has_affiliate_orgs,
"table1_exists": table1_exists,
"table2_exists": table2_exists,
"activities_count_in_db": activities_count,
"affiliates_count_in_db": affiliates_count,
"child_tables": child_tables
}
except Exception as e:
return {
"success": False,
"error": str(e)
}

View File

@ -0,0 +1,16 @@
[
{
"docstatus": 0,
"doctype": "Agricultural Purpose Type",
"modified": "2025-09-09 20:32:37.618522",
"name": "JE00002",
"purpose_name": "Təyinatı üzrə istifadə edilməyən kənd təsərrüfatı torpaqları üzrə hesablanmış vergi məbləği"
},
{
"docstatus": 0,
"doctype": "Agricultural Purpose Type",
"modified": "2025-09-09 20:32:53.438504",
"name": "JE00001",
"purpose_name": "Təyinatı üzrə istifadə edilən və ya irriqasiya, meliorasiya və digər aqrotexniki səbəblərdən təyinatı üzrə istifadə edilməsi mümkün olmayan kənd təsərrüfatı torpaqları üzrə verginin məbləği"
}
]

File diff suppressed because one or more lines are too long

View File

@ -12,18 +12,6 @@ fixtures = [
["module", "=", "Taxes Az"]
]
},
# {
# "doctype": "Item",
# "filters": [
# ["item_name", "in", ["Avtomobil benzini (Aİ-92)", "Avtomobil benzini (Aİ-95)", "Avtomobil benzini (Aİ-80)", "Dizel yanacağı", "Maye qaz"]]
# ]
# },
# {
# "doctype": "Item Group",
# "filters": [
# ["item_group_name", "in", ["Neft məhsulları", "Yanacaq növləri"]]
# ]
# },
{
"doctype": "Main type of activity",
"filters": []

View File

@ -0,0 +1,8 @@
// Copyright (c) 2025, Jey Soft and contributors
// For license information, please see license.txt
// frappe.ui.form.on("Agricultural Purpose Type", {
// refresh(frm) {
// },
// });

View File

@ -0,0 +1,48 @@
{
"actions": [],
"allow_copy": 1,
"allow_rename": 1,
"autoname": "prompt",
"creation": "2025-09-09 20:30:10.172311",
"doctype": "DocType",
"engine": "InnoDB",
"field_order": [
"purpose_name"
],
"fields": [
{
"fieldname": "purpose_name",
"fieldtype": "Text",
"label": "Purpose Name"
}
],
"hide_toolbar": 1,
"index_web_pages_for_search": 1,
"links": [],
"modified": "2025-09-09 20:38:45.745959",
"modified_by": "Administrator",
"module": "Taxes Az",
"name": "Agricultural Purpose Type",
"naming_rule": "Set by user",
"owner": "Administrator",
"permissions": [
{
"create": 1,
"delete": 1,
"email": 1,
"export": 1,
"print": 1,
"read": 1,
"report": 1,
"role": "System Manager",
"share": 1,
"write": 1
}
],
"search_fields": "purpose_name",
"show_title_field_in_link": 1,
"sort_field": "creation",
"sort_order": "DESC",
"states": [],
"title_field": "purpose_name"
}

View File

@ -0,0 +1,9 @@
# Copyright (c) 2025, Jey Soft and contributors
# For license information, please see license.txt
# import frappe
from frappe.model.document import Document
class AgriculturalPurposeType(Document):
pass

View File

@ -0,0 +1,30 @@
# Copyright (c) 2025, Jey Soft and Contributors
# See license.txt
# import frappe
from frappe.tests import IntegrationTestCase, UnitTestCase
# On IntegrationTestCase, the doctype test records and all
# link-field test record depdendencies are recursively loaded
# Use these module variables to add/remove to/from that list
EXTRA_TEST_RECORD_DEPENDENCIES = [] # eg. ["User"]
IGNORE_TEST_RECORD_DEPENDENCIES = [] # eg. ["User"]
class UnitTestAgriculturalPurposeType(UnitTestCase):
"""
Unit tests for AgriculturalPurposeType.
Use this class for testing individual functions and methods.
"""
pass
class IntegrationTestAgriculturalPurposeType(IntegrationTestCase):
"""
Integration tests for AgriculturalPurposeType.
Use this class for testing interactions between multiple components.
"""
pass