How to use stop_app method in Airtest
Best Python code snippet using Airtest
views.py
Source:views.py 
1import decimal2import json3import logging4import time5from logging.handlers import RotatingFileHandler6from django.core.exceptions import ObjectDoesNotExist7from django.http import JsonResponse, HttpResponse8from django.shortcuts import get_object_or_4049from django.views.decorators.csrf import csrf_exempt10from django.core.serializers.json import DjangoJSONEncoder11from pybitrix24 import Bitrix2412from core.bitrix24.bitrix24 import (ActivityB24, DealB24, ProductB24,13 CompanyB24, SmartProcessB24, ListsB24)14from activities.discount import InvoiceDiscount, PartnerDiscount, \15 AccumulativeDiscount16from core.models import Portals17from volumes.models import Volume18from settings.models import SettingsPortal19from .models import Activity20from .messages import MESSAGES_FOR_BP, MESSAGES_FOR_LOG21loggers = {}22@csrf_exempt23def install(request):24 """View-ÑÑнкÑÐ¸Ñ ÑÑÑановки акÑивиÑи на поÑÑал."""25 member_id = request.POST.get('member_id')26 activity_code = request.POST.get('code')27 portal: Portals = get_object_or_404(Portals, member_id=member_id)28 portal.check_auth()29 activity = get_object_or_404(Activity, code=activity_code)30 try:31 activity_b24 = ActivityB24(portal)32 result = activity_b24.install(activity.build_params())33 except RuntimeError as ex:34 return JsonResponse({35 'result': 'False',36 'error_name': ex.args[0],37 'error_description': ex.args[1]})38 return JsonResponse({'result': result})39@csrf_exempt40def uninstall(request):41 """View-ÑÑнкÑÐ¸Ñ ÑÐ´Ð°Ð»ÐµÐ½Ð¸Ñ Ð°ÐºÑивиÑи на поÑÑале."""42 member_id = request.POST.get('member_id')43 activity_code = request.POST.get('code')44 portal: Portals = get_object_or_404(Portals, member_id=member_id)45 portal.check_auth()46 try:47 activity_b24 = ActivityB24(portal, code=activity_code)48 result = activity_b24.uninstall()49 except RuntimeError as ex:50 return JsonResponse({51 'result': 'False',52 'error_name': ex.args[0],53 'error_description': ex.args[1]})54 return JsonResponse({'result': result})55@csrf_exempt56def send_to_db(request):57 """View-ÑÑнкÑÐ¸Ñ Ð´Ð»Ñ ÑабоÑÑ Ð°ÐºÑивиÑи 'ÐеÑедаÑа обÑемов в ÐÐ'."""58 # УÑÑановки логиÑованиÑ59 logger = logging.getLogger(__name__)60 logger.setLevel(logging.DEBUG)61 handler = RotatingFileHandler(62 '/home/bitrix/ext_www/skidkipril.plazma-t.ru/logs/send_to_db.log',63 maxBytes=5000000,64 backupCount=565 )66 formatter = logging.Formatter(67 "%(asctime)s [%(levelname)s] %(message)s")68 handler.setFormatter(formatter)69 logger.addHandler(handler)70 # ÐапÑÑк пÑиложениÑ71 logger.info(MESSAGES_FOR_LOG['start_app'])72 if request.method != 'POST':73 logger.error(MESSAGES_FOR_LOG['request_not_post'])74 logger.info(MESSAGES_FOR_LOG['stop_app'])75 return76 initial_data = {77 'member_id': request.POST.get('auth[member_id]'),78 'event_token': request.POST.get('event_token'),79 'deal_id': request.POST.get('properties[deal_id]') or 0,80 'company_id': request.POST.get('properties[company_id]') or 0,81 }82 # Создаем поÑÑал83 try:84 portal: Portals = Portals.objects.get(85 member_id=initial_data['member_id'])86 portal.check_auth()87 settings_portal = SettingsPortal.objects.get(portal=portal)88 except ObjectDoesNotExist:89 logger.error(MESSAGES_FOR_LOG['portal_not_found'].format(90 initial_data['member_id']))91 logger.info(MESSAGES_FOR_LOG['stop_app'])92 return93 # ÐÑовеÑÑем наÑалÑнÑе даннÑе94 try:95 deal_id = int(initial_data['deal_id'])96 company_id = int(initial_data['company_id'])97 except Exception as ex:98 logger.error(MESSAGES_FOR_LOG['error_start_data'].format(99 initial_data['deal_id'], initial_data['company_id']100 ))101 logger.info(MESSAGES_FOR_LOG['stop_app'])102 response_for_bp(103 portal,104 initial_data['event_token'],105 '{} {}'.format(MESSAGES_FOR_BP['main_error'], ex.args[0]),106 )107 return108 # ÐолÑÑаем вÑе пÑодÑкÑÑ Ñделки109 try:110 deal = DealB24(deal_id, portal)111 deal.get_all_products()112 except RuntimeError:113 logger.error(MESSAGES_FOR_LOG['impossible_get_products'])114 logger.info(MESSAGES_FOR_LOG['stop_app'])115 response_for_bp(portal, initial_data['event_token'],116 MESSAGES_FOR_BP['impossible_get_products'])117 return118 nomenclatures = dict()119 for product in deal.products:120 try:121 prod: ProductB24 = ProductB24(portal, product["PRODUCT_ID"])122 prod.get_properties()123 except RuntimeError:124 logger.error(125 MESSAGES_FOR_LOG['impossible_get_product_props'].format(126 product['ID']127 ))128 logger.info(MESSAGES_FOR_LOG['stop_app'])129 response_for_bp(130 portal, initial_data['event_token'],131 MESSAGES_FOR_BP['impossible_get_product_props'].format(132 product['ID']133 ))134 return135 if not prod.props[settings_portal.code_nomenclature_group_id]:136 logger.info(MESSAGES_FOR_LOG['skip_product'].format(137 product['ID']138 ))139 continue140 price = round(decimal.Decimal(product['PRICE_ACCOUNT']), 2)141 quantity = round(decimal.Decimal(product['QUANTITY']), 2)142 volume = quantity * price143 nomenclature_group_id = int(144 prod.props[settings_portal.code_nomenclature_group_id]['value']145 )146 if nomenclature_group_id not in nomenclatures:147 nomenclatures[nomenclature_group_id] = volume148 else:149 nomenclatures[nomenclature_group_id] += volume150 logger.info(MESSAGES_FOR_LOG['volume_add'].format(151 nomenclature_group_id, company_id, str(volume)))152 for key, value in nomenclatures.items():153 volume, created = Volume.objects.get_or_create(154 nomenclature_group_id=key, company_id=company_id, portal=portal,155 defaults={156 'volume': value,157 }158 )159 if not created:160 volume.volume += value161 volume.save()162 logger.info(MESSAGES_FOR_LOG['send_data_to_db_ok'])163 logger.info(MESSAGES_FOR_LOG['stop_app'])164 response_for_bp(portal, initial_data['event_token'],165 MESSAGES_FOR_BP['send_data_to_db_ok'])166 handler.close()167 logger.removeHandler(handler)168 return HttpResponse(status=200)169@csrf_exempt170def get_from_db(request):171 """View-ÑÑнкÑÐ¸Ñ Ð´Ð»Ñ ÑабоÑÑ Ð°ÐºÑивиÑи 'ÐолÑÑение обÑемов из ÐÐ'."""172 # УÑÑановки логиÑованиÑ173 logger = logging.getLogger(__name__)174 logger.setLevel(logging.DEBUG)175 handler = RotatingFileHandler(176 '/home/bitrix/ext_www/skidkipril.plazma-t.ru/logs/get_from_db.log',177 maxBytes=5000000,178 backupCount=5179 )180 formatter = logging.Formatter(181 "%(asctime)s [%(levelname)s] %(message)s")182 handler.setFormatter(formatter)183 logger.addHandler(handler)184 # ÐапÑÑк пÑиложениÑ185 logger.info(MESSAGES_FOR_LOG['start_app'])186 if request.method != 'POST':187 logger.error(MESSAGES_FOR_LOG['request_not_post'])188 logger.info(MESSAGES_FOR_LOG['stop_app'])189 return190 initial_data = {191 'member_id': request.POST.get('auth[member_id]'),192 'event_token': request.POST.get('event_token'),193 'nomenclature_group_id': request.POST.get(194 'properties[nomenclature_group_id]') or 0,195 'company_id': request.POST.get('properties[company_id]') or 0,196 'is_all': (True if request.POST.get('properties[is_all]') == 'Y'197 else False)198 }199 # Создаем поÑÑал200 try:201 portal: Portals = Portals.objects.get(202 member_id=initial_data['member_id'])203 portal.check_auth()204 # settings_portal = SettingsPortal.objects.get(portal=portal)205 except ObjectDoesNotExist:206 logger.error(MESSAGES_FOR_LOG['portal_not_found'].format(207 initial_data['member_id']))208 logger.info(MESSAGES_FOR_LOG['stop_app'])209 return210 # ÐÑовеÑÑем наÑалÑнÑе даннÑе211 try:212 nomenclature_group_id = int(initial_data['nomenclature_group_id'])213 company_id = int(initial_data['company_id'])214 is_all = initial_data['is_all']215 except Exception as ex:216 logger.error(MESSAGES_FOR_LOG['error_start_data'].format(217 initial_data['deal_id'], initial_data['company_id']218 ))219 logger.info(MESSAGES_FOR_LOG['stop_app'])220 response_for_bp(221 portal,222 initial_data['event_token'],223 '{} {}'.format(MESSAGES_FOR_BP['main_error'], ex.args[0]),224 )225 return226 # ÐапÑÐ¾Ñ Ð² ÐРна полÑÑение накопленного обÑема227 if is_all:228 volumes = Volume.objects.filter(229 company_id=company_id,230 portal=portal231 )232 if not volumes:233 logger.info(MESSAGES_FOR_LOG['volumes_no_db'].format(company_id))234 logger.info(MESSAGES_FOR_LOG['stop_app'])235 response_for_bp(portal, initial_data['event_token'],236 MESSAGES_FOR_BP['volume_no_db'],237 return_values={'result': 'no_data'})238 return239 result_volume = {}240 for volume in volumes:241 try:242 nomenclature_name = ListsB24(portal, 18).get_element_by_id(243 volume.nomenclature_group_id)[0]['NAME']244 result_volume[nomenclature_name] = str(volume.volume)245 except RuntimeError:246 logger.error(247 MESSAGES_FOR_LOG['impossible_get_nomenclature_name'])248 continue249 except Exception as ex:250 response_for_bp(251 portal,252 initial_data['event_token'],253 '{} {}'.format(MESSAGES_FOR_BP['main_error'], ex.args[0]),254 )255 return256 logger.info(MESSAGES_FOR_LOG['get_volumes'].format(257 str(result_volume), company_id))258 response_for_bp(portal, initial_data['event_token'],259 MESSAGES_FOR_BP['get_from_db_ok'],260 return_values={261 'volume': str(result_volume),262 'result': 'ok',263 })264 logger.info(MESSAGES_FOR_LOG['stop_app'])265 else:266 try:267 volume = Volume.objects.get(268 nomenclature_group_id=nomenclature_group_id,269 company_id=company_id,270 portal=portal271 )272 nomenclature_name = ListsB24(portal, 18).get_element_by_id(273 volume.nomenclature_group_id)[0]['NAME']274 except ObjectDoesNotExist:275 logger.info(MESSAGES_FOR_LOG['volume_no_db'].format(276 nomenclature_group_id, company_id))277 logger.info(MESSAGES_FOR_LOG['stop_app'])278 response_for_bp(portal, initial_data['event_token'],279 MESSAGES_FOR_BP['volume_no_db'],280 return_values={'result': 'no_data'})281 return282 logger.info(MESSAGES_FOR_LOG['get_volume'].format(283 str(volume.volume), nomenclature_name, company_id))284 response_for_bp(portal, initial_data['event_token'],285 MESSAGES_FOR_BP['get_from_db_ok'],286 return_values={287 'volume': f'{nomenclature_name}: '288 f'{str(volume.volume)}',289 'result': 'ok',290 })291 logger.info(MESSAGES_FOR_LOG['stop_app'])292 handler.close()293 logger.removeHandler(handler)294 return HttpResponse(status=200)295@csrf_exempt296def calculation(request):297 """View-ÑÑнкÑÐ¸Ñ Ð´Ð»Ñ ÑабоÑÑ Ð°ÐºÑивиÑи 'РаÑÑÐµÑ Ñкидок'."""298 # УÑÑановки логиÑованиÑ299 logger = logging.getLogger(__name__)300 logger.setLevel(logging.DEBUG)301 handler = RotatingFileHandler(302 '/home/bitrix/ext_www/skidkipril.plazma-t.ru/logs/calculation.log',303 maxBytes=5000000,304 backupCount=5305 )306 formatter = logging.Formatter(307 "%(asctime)s [%(levelname)s] %(message)s")308 handler.setFormatter(formatter)309 logger.addHandler(handler)310 # ÐапÑÑк пÑиложениÑ311 logger.info(MESSAGES_FOR_LOG['start_app'])312 logger.info('{} {}'.format(MESSAGES_FOR_LOG['start_block'],313 'ÐаÑалÑнÑе даннÑе'))314 if request.method != 'POST':315 logger.error(MESSAGES_FOR_LOG['request_not_post'])316 logger.info(MESSAGES_FOR_LOG['stop_app'])317 return318 initial_data = {319 'member_id': request.POST.get('auth[member_id]'),320 'event_token': request.POST.get('event_token'),321 'deal_id': request.POST.get('properties[deal_id]') or 0,322 'company_id': request.POST.get('properties[company_id]') or 0,323 }324 # Создаем поÑÑал325 try:326 portal: Portals = Portals.objects.get(327 member_id=initial_data['member_id'])328 portal.check_auth()329 settings_portal = SettingsPortal.objects.get(portal=portal)330 except ObjectDoesNotExist:331 logger.error(MESSAGES_FOR_LOG['portal_not_found'].format(332 initial_data['member_id']))333 logger.info(MESSAGES_FOR_LOG['stop_app'])334 return335 # ÐÑовеÑÑем наÑалÑнÑе даннÑе336 try:337 deal_id = int(initial_data['deal_id'])338 company_id = int(initial_data['company_id'])339 except Exception as ex:340 logger.error(MESSAGES_FOR_LOG['error_start_data'].format(341 initial_data['deal_id'], initial_data['company_id']342 ))343 logger.info(MESSAGES_FOR_LOG['stop_app'])344 response_for_bp(345 portal,346 initial_data['event_token'],347 '{} {}'.format(MESSAGES_FOR_BP['main_error'], ex.args[0]),348 )349 return350 # ÐолÑÑаем вÑе пÑодÑкÑÑ Ñделки351 try:352 deal = DealB24(deal_id, portal)353 deal.get_all_products()354 print(deal.products)355 except RuntimeError:356 logger.error(MESSAGES_FOR_LOG['impossible_get_products'])357 logger.info(MESSAGES_FOR_LOG['stop_app'])358 response_for_bp(portal, initial_data['event_token'],359 MESSAGES_FOR_BP['impossible_get_products'])360 return361 except Exception as ex:362 logger.error('test')363 logger.info(MESSAGES_FOR_LOG['stop_app'])364 response_for_bp(portal, initial_data['event_token'],365 ex.args[0])366 return367 # ÐобавлÑем в ÑпиÑок ÑоваÑов в каждÑй ÑловаÑÑ Ð´Ð¾Ð¿Ð¾Ð»Ð½Ð¸ÑелÑное поле368 # nomenclature_group_id369 for product in deal.products:370 try:371 prod: ProductB24 = ProductB24(portal, product["PRODUCT_ID"])372 prod.get_properties()373 except RuntimeError:374 logger.error(375 MESSAGES_FOR_LOG['impossible_get_product_props'].format(376 product['id']377 ))378 logger.info(MESSAGES_FOR_LOG['stop_app'])379 response_for_bp(380 portal, initial_data['event_token'],381 MESSAGES_FOR_BP['impossible_get_product_props'].format(382 product['id']383 ))384 return385 if not prod.props[settings_portal.code_nomenclature_group_id]:386 product['nomenclature_group_id'] = 0387 continue388 product['nomenclature_group_id'] = int(389 prod.props[settings_portal.code_nomenclature_group_id]['value']390 )391 logger.debug('{}{}'.format(392 MESSAGES_FOR_LOG['get_products'],393 json.dumps(deal.products, indent=2, ensure_ascii=False)394 ))395 # СÑоÑмиÑÑем ÑловаÑÑ Ð½Ð¾Ð¼ÐµÐ½ÐºÐ»Ð°ÑÑÑнÑÑ
гÑÑпп396 nomenclatures_groups: dict[str, decimal.Decimal] = dict()397 for product in deal.products:398 nomenclature_group_id = product['nomenclature_group_id']399 price = round(decimal.Decimal(product['PRICE_BRUTTO']), 2)400 quantity = round(decimal.Decimal(product['QUANTITY']), 2)401 product_sum: decimal.Decimal = quantity * price402 if nomenclature_group_id not in nomenclatures_groups:403 nomenclatures_groups[nomenclature_group_id] = product_sum404 else:405 nomenclatures_groups[nomenclature_group_id] += product_sum406 logger.info('{}{}'.format(407 MESSAGES_FOR_LOG['get_nomenclature_groups'],408 json.dumps(nomenclatures_groups, indent=2, ensure_ascii=False,409 cls=DjangoJSONEncoder)410 ))411 # ÐолÑÑаем Ñип компании412 try:413 company: CompanyB24 = CompanyB24(portal, company_id)414 company.get_type()415 except RuntimeError:416 logger.error(MESSAGES_FOR_LOG['impossible_get_company_type'])417 logger.info(MESSAGES_FOR_LOG['stop_app'])418 response_for_bp(portal, initial_data['event_token'],419 MESSAGES_FOR_BP['impossible_get_company_type'])420 return421 # ÐÑновной ÑловаÑÑ Ñкидок по номенклаÑÑÑе422 discounts: dict[str, int] = dict()423 logger.info(MESSAGES_FOR_LOG['stop_block'])424 # #######################Скидки Ð´Ð»Ñ ÐаÑÑнеÑов#############################425 logger.info('{} {}'.format(MESSAGES_FOR_LOG['start_block'],426 'Скидки Ð´Ð»Ñ Ð¿Ð°ÑÑнеÑов'))427 if settings_portal.is_active_partner:428 partner_discounts = PartnerDiscount(429 settings_portal.code_discount_smart_partner,430 settings_portal.code_company_type_smart_partner,431 settings_portal.code_nomenclature_group_id_smart_partner,432 settings_portal.id_smart_process_partner,433 nomenclatures_groups,434 discounts,435 portal436 )437 try:438 partner_discounts.get_all_elements_smart_process()439 except RuntimeError:440 logger.error(MESSAGES_FOR_LOG['impossible_get_smart_partner'])441 logger.info(MESSAGES_FOR_LOG['stop_app'])442 response_for_bp(portal, initial_data['event_token'],443 MESSAGES_FOR_BP['impossible_get_smart_partner'])444 return445 logger.debug('{}{}'.format(446 MESSAGES_FOR_LOG['get_elements_discounts_partners'],447 json.dumps(partner_discounts.smart_process_elements,448 indent=2, ensure_ascii=False)449 ))450 partner_discounts.check_input_date()451 partner_discounts.check_company_type(company.type)452 partner_discounts.calculate_discounts()453 partner_discounts.compare_discounts()454 logger.info('{}{}'.format(455 MESSAGES_FOR_LOG['discounts_partner'],456 json.dumps(discounts, indent=2, ensure_ascii=False)))457 else:458 logger.info(MESSAGES_FOR_LOG['partner_off'])459 logger.info(MESSAGES_FOR_LOG['stop_block'])460 # #######################Ð Ð°Ð·Ð¾Ð²Ð°Ñ Ð¾Ñ ÑÑÐ¼Ð¼Ñ ÑÑеÑа############################461 logger.info('{} {}'.format(MESSAGES_FOR_LOG['start_block'],462 'Ð Ð°Ð·Ð¾Ð²Ð°Ñ Ð¾Ñ ÑÑÐ¼Ð¼Ñ ÑÑеÑа'))463 if settings_portal.is_active_sum_invoice:464 invoice_discounts: InvoiceDiscount = InvoiceDiscount(465 settings_portal.code_discount_smart_sum_invoice,466 settings_portal.id_smart_process_sum_invoice,467 nomenclatures_groups,468 discounts,469 portal470 )471 try:472 invoice_discounts.get_all_elements_smart_process()473 except RuntimeError:474 logger.error(MESSAGES_FOR_LOG['impossible_get_smart_sum_invoice'])475 logger.info(MESSAGES_FOR_LOG['stop_app'])476 response_for_bp(portal, initial_data['event_token'],477 MESSAGES_FOR_BP[478 'impossible_get_smart_sum_invoice'])479 return480 logger.debug('{}{}'.format(481 MESSAGES_FOR_LOG['get_elements_sum_invoice'],482 json.dumps(invoice_discounts.smart_process_elements, indent=2,483 ensure_ascii=False)))484 invoice_discounts.check_input_date()485 invoice_discounts.check_is_active_nomenclature_group(18, 'PROPERTY_75')486 invoice_discounts.set_limits()487 invoice_discounts.calculate_discounts()488 invoice_discounts.compare_discounts()489 logger.info('{}{}'.format(490 MESSAGES_FOR_LOG['discounts_sum_invoice'],491 json.dumps(discounts, indent=2, ensure_ascii=False)))492 else:493 logger.info(MESSAGES_FOR_LOG['sum_invoice_off'])494 logger.info(MESSAGES_FOR_LOG['stop_block'])495 # #######################ÐакопиÑелÑнаÑ#############################496 logger.info('{} {}'.format(MESSAGES_FOR_LOG['start_block'],497 'ÐакопиÑелÑÐ½Ð°Ñ Ñкидка'))498 if settings_portal.is_active_accumulative:499 accumulative_discounts: AccumulativeDiscount = AccumulativeDiscount(500 settings_portal.code_nomenclature_group_accumulative,501 settings_portal.code_upper_one_accumulative,502 settings_portal.code_discount_upper_one_accumulative,503 settings_portal.code_upper_two_accumulative,504 settings_portal.code_discount_upper_two_accumulative,505 settings_portal.code_upper_three_accumulative,506 settings_portal.code_discount_upper_three_accumulative,507 settings_portal.id_smart_process_accumulative,508 nomenclatures_groups,509 discounts,510 company_id,511 portal512 )513 try:514 accumulative_discounts.get_all_elements_smart_process()515 except RuntimeError:516 logger.error(MESSAGES_FOR_LOG['impossible_get_smart_accumulative'])517 logger.info(MESSAGES_FOR_LOG['stop_app'])518 response_for_bp(portal, initial_data['event_token'],519 MESSAGES_FOR_BP[520 'impossible_get_smart_accumulative'])521 return522 logger.debug('{}{}'.format(523 MESSAGES_FOR_LOG['get_elements_accumulative'],524 json.dumps(accumulative_discounts.smart_process_elements, indent=2,525 ensure_ascii=False)))526 accumulative_discounts.check_input_date()527 accumulative_discounts.calculate_discounts()528 accumulative_discounts.compare_discounts()529 logger.info('{}{}'.format(530 MESSAGES_FOR_LOG['discounts_accumulative'],531 json.dumps(discounts, indent=2, ensure_ascii=False)))532 else:533 logger.info(MESSAGES_FOR_LOG['accumulative_off'])534 logger.info(MESSAGES_FOR_LOG['stop_block'])535 # #######################Скидки на ÑоваÑ#############################536 logger.info('{} {}'.format(MESSAGES_FOR_LOG['start_block'],537 'Скидки на конкÑеÑнÑй ÑоваÑ'))538 all_discounts_products = dict()539 if settings_portal.is_active_discount_product:540 # ÐолÑÑим вÑе ÑлеменÑÑ ÑмаÑÑ Ð¿ÑоÑеÑÑа Скидки на ÑоваÑ541 try:542 discounts_product = SmartProcessB24(543 portal, settings_portal.id_smart_process_discount_product)544 discounts_product.get_all_elements()545 except RuntimeError:546 logger.error(MESSAGES_FOR_LOG['impossible_get_smart_one_product'])547 logger.info(MESSAGES_FOR_LOG['stop_app'])548 response_for_bp(portal, initial_data['event_token'],549 MESSAGES_FOR_BP[550 'impossible_get_smart_one_product'])551 return552 logger.debug('{}{}'.format(553 MESSAGES_FOR_LOG['get_elements_discounts_product'],554 json.dumps(discounts_product.elements, indent=2,555 ensure_ascii=False)))556 # ÐеÑÐµÐ±Ð¾Ñ Ð²ÑеÑ
ÑлеменÑов ÑмаÑÑ Ð¿ÑоÑеÑÑа Скидки на ÑоваÑ557 for element in discounts_product.elements:558 # ÐÑовеÑÑем вÑ
однÑе даннÑе ÑлеменÑов ÑмаÑÑ Ð¿ÑоÑеÑÑа559 if (not element[560 settings_portal.code_discount_smart_discount_product]):561 logger.error(562 MESSAGES_FOR_LOG['wrong_input_data_smart'].format(563 element['title'], discounts_product.id564 ))565 continue566 logger.info('{} {}'.format(567 MESSAGES_FOR_LOG['algorithm_for_smart'],568 element['title']569 ))570 # ÐÑовеÑÑем id компании в ÑлеменÑе ÑмаÑÑ Ð¿ÑоÑеÑÑа и Ñделке571 smart_company_id = element['companyId']572 # ÐÑовеÑÑем ÑÐ¾Ð²Ð¿Ð°Ð´Ð°ÐµÑ Ð»Ð¸ id_company Ñделки и ÑлеменÑа573 if smart_company_id != company.id:574 logger.info(575 MESSAGES_FOR_LOG['company_deal_not_company_smart'].format(576 smart_company_id, company.id577 ))578 continue579 # ÐолÑÑаем вÑе ÑоваÑÑ ÑлеменÑа ÑмаÑÑ Ð¿ÑоÑеÑÑа580 discounts_product.get_all_products(581 settings_portal.code_smart_process_discount_product,582 element['id'])583 # ФоÑмиÑÑем ÑловаÑÑ Ð²ÑеÑ
Ñкидок на пÑодÑкÑ584 for product in discounts_product.products:585 all_discounts_products[product['productId']] = element[586 settings_portal.code_discount_smart_discount_product]587 logger.debug('{} {}'.format(588 MESSAGES_FOR_LOG['get_all_discounts_products'],589 json.dumps(all_discounts_products, indent=2,590 ensure_ascii=False)))591 else:592 logger.info(MESSAGES_FOR_LOG['discount_product_off'])593 logger.info(MESSAGES_FOR_LOG['stop_block'])594 # #######################ÐÑименÑем Ñкидки#############################595 logger.info('{} {}'.format(MESSAGES_FOR_LOG['start_block'],596 'ÐÑименение Ñкидок'))597 for product in deal.products:598 nomenclature_group_id = product['nomenclature_group_id']599 # price_acc = decimal.Decimal(product['PRICE_ACCOUNT'])600 price_brutto = decimal.Decimal(product['PRICE_BRUTTO'])601 product_id = product['PRODUCT_ID']602 keys_for_del = ['ID', 'OWNER_ID', 'OWNER_TYPE', 'PRODUCT_NAME',603 'ORIGINAL_PRODUCT_NAME', 'PRODUCT_DESCRIPTION',604 'PRICE_EXCLUSIVE',605 'PRICE_BRUTTO', 'PRICE_ACCOUNT', 'DISCOUNT_SUM',606 'nomenclature_group_id']607 for key in keys_for_del:608 del product[key]609 # ÐÑименÑем Ñкидки по номенклаÑÑÑнÑм гÑÑппам610 if nomenclature_group_id in discounts:611 discount_rate = discounts[nomenclature_group_id]612 product['DISCOUNT_RATE'] = discount_rate613 price = price_brutto * (100 - discount_rate) / 100614 product['PRICE'] = str(price)615 logger.info(MESSAGES_FOR_LOG['discount_ok_product'].format(616 product_id, discount_rate617 ))618 else:619 product['DISCOUNT_RATE'] = 0620 # ÐÑименÑем Ñкидки на конкÑеÑнÑй ÑоваÑ621 if settings_portal.is_active_discount_product:622 if product_id not in all_discounts_products:623 logger.info(MESSAGES_FOR_LOG['no_discount_one_product'].format(624 product_id625 ))626 continue627 discount_rate = all_discounts_products[product_id]628 product['DISCOUNT_RATE'] = discount_rate629 price = price_brutto * (100 - discount_rate) / 100630 product['PRICE'] = str(price)631 logger.info(MESSAGES_FOR_LOG['discount_ok_product'].format(632 product_id, discount_rate633 ))634 logger.debug('{}{}'.format(635 MESSAGES_FOR_LOG['all_products_send_bp'],636 json.dumps(deal.products, indent=2, ensure_ascii=False)))637 try:638 deal.set_products(deal.products)639 except RuntimeError:640 logger.error(MESSAGES_FOR_LOG['impossible_send_to_deal'])641 logger.info(MESSAGES_FOR_LOG['stop_app'])642 response_for_bp(643 portal, initial_data['event_token'],644 MESSAGES_FOR_BP['impossible_send_to_deal'],645 )646 return647 # ÐозвÑаÑаем ÑезÑлÑÑаÑ648 response_for_bp(portal, initial_data['event_token'],649 MESSAGES_FOR_BP['calculation_ok'])650 logger.info(MESSAGES_FOR_LOG['stop_block'])651 logger.info(MESSAGES_FOR_LOG['stop_app'])652 handler.close()653 logger.removeHandler(handler)654 return HttpResponse(status=200)655def response_for_bp(portal, event_token, log_message, return_values=None):656 """ÐеÑод оÑпÑавки паÑамеÑÑов оÑвеÑа в ÐÐ."""657 bx24 = Bitrix24(portal.name)658 bx24._access_token = portal.auth_id659 method_rest = 'bizproc.event.send'660 params = {661 'event_token': event_token,662 'log_message': log_message,663 'return_values': return_values,664 }...
schedule.py
Source:schedule.py
...11 checkin.fanqie(device, w, h)12 # [x] é
读çªèå°è¯´13 phone.tap(device, w / 3, h / 3)14 app.read_novel(device, w, h, num=1)15 phone.stop_app(device, info.packages['fanqie'])16def fanchang(device, w, h):17 checkin.fanchang(device, w, h)18 return None19def kuchang(device, w, h):20 return None21def shuqi(device, w, h):22 def watch_advert():23 print('书æç广åèµéå¸ ' + datetime.now().time().__str__())24 phone.tap(device, 900, 2150, gap=10)25 time.sleep(30)26 if datetime.now().minute < info.SCHEDULE_TIME:27 checkin.shuqi(device, w, h)28 phone.tap(device, w / 2, 2300)29 # [x] ç广åèµéå¸30 watch_advert()31 phone.stop_app(device, info.packages['shuqi'])32def yingke(device, w, h):33 return None34def kugou(device, w, h):35 return None36def zhongqing(device, w, h):37 checkin.zhongqing(device, w, h)38 # [x] é
读ä¸éçç¹æç« 39 app.read_article(device, w, h, num=1)40 phone.stop_app(device, info.packages['zhongqing'])41def kuaiyin(device, w, h):42 # [x] ææ¾43 if datetime.now().hour == 1:44 checkin.kuaiyin(device, w, h)45 if datetime.now().hour == 4:46 phone.stop_app(device, info.packages['kuaiyin'])47def kuge(device, w, h):48 return None49def momo(device, w, h):50 return None51def qingtuanshe(device, w, h):52 return None53def eleme(device, w, h):54 return None55def changdou(device, w, h):56 return None57def kuaikandian(device, w, h):58 checkin.kuaikandian(device, w, h)59 # [x] é
读快çç¹æç« 60 app.read_article(device, w, h, num=1)61 phone.stop_app(device, info.packages['kuaikandian'])62def zhaoshang(device, w, h):63 return None64def toutiao(device, w, h):65 def open_treasure():66 print('ä»æ¥å¤´æ¡å¼å®ç®± ' + datetime.now().time().__str__())67 checkin.toutiao(device)68 # [x] é
读头æ¡æç« 69 app.read_article(device, w, h, num=1)70 # [x] å¼å®ç®±71 phone.tap(device, w / 2, 2330)72 phone.tap(device, 930, 2120, gap=3)73 phone.stop_app(device, info.packages['toutiao'])74def kuaishou(device, w, h):75 checkin.kuaishou(device)76 # [x] çå¿«æè§é¢77 app.watch_video(device, w, h, num=20)78 phone.stop_app(device, info.packages['kuaishou'])79def douyin(device, w, h):80 def open_treasure():81 print('æé³æéçå¼å®ç®± ' + datetime.now().time().__str__())82 phone.tap(device, w / 2, h - 100) # modify83 phone.tap(device, w - 160, h - 200) # modify84 checkin.douyin(device)85 # [x] çæé³è§é¢86 app.watch_video(device, w, h, num=10)87 # [x] å¼å®ç®±88 open_treasure()89 phone.stop_app(device, info.packages['douyin'])90def qutoutiao(device, w, h):91 if datetime.now().hour > 8:92 checkin.qutoutiao(device)93 # [x] é
读趣头æ¡æç« 94 app.read_article(device, w, h, num=1)95 phone.stop_app(device, info.packages['qutoutiao'])96def baidu(device, w, h):97 checkin.baidu(device)98 # [x] é
读ç¾åº¦æç« 99 app.read_article(device, w, h, num=1)100 phone.stop_app(device, info.packages['baidu'])101def weishi(device, w, h):102 checkin.weishi(device)103 # [x] çå¾®è§è§é¢104 app.watch_video(device, w, h, num=10)105 phone.stop_app(device, info.packages['weishi'])106def douhuo(device, w, h):107 checkin.douhuo(device)108 app.watch_video(device, w, h, num=10)109 phone.stop_app(device, info.packages['douhuo'])110def chejia(device, w, h):111 return None112def uc(device, w, h):113 return None114def diantao(device, w, h):115 if datetime.now().hour > 8:116 checkin.diantao(device)117 # [x] ç¹å»è¿å
¥ç´æ页é¢118 phone.tap(device, w / 3, h / 3, gap=5)119 app.watch_video(device, w, h, num=10)120 phone.stop_app(device, info.packages['diantao'])121def huitoutiao(device, w, h):122 return None123def touda(device, w, h):124 checkin.touda(device)125 # [x] å¼å®ç®±126 phone.tap(device, 850, 2290)127 phone.tap(device, 890, 2010, gap=3)128 phone.stop_app(device, info.packages['touda'])129def shuabao(device, w, h):130 checkin.shuabao(device)131 app.watch_video(device, w, h, num=10)...
test_yttv.py
Source:test_yttv.py
...9def test_get_settings_tv_code(yttv):10 yttv.load_app()11 tv_code = yttv.get_settings_tv_code()12 assert len(tv_code) == 1213 yttv.stop_app()14def test_get_settings_tv_code_multiple_times(yttv):15 yttv.load_app()16 tv_code = yttv.get_settings_tv_code()17 assert len(tv_code) == 1218 yttv.reset_app()19 tv_code = yttv.get_settings_tv_code()20 assert len(tv_code) == 1221 yttv.stop_app()22# TODO: Write Test23def test_install_no_script(yttv):24 yttv.stop_app()25# TODO: Write Test26def test_load_app(yttv):27 yttv.stop_app()28# TODO: Write Test29def test_reset_app(yttv):30 yttv.reset_app()31 yttv.stop_app()32# TODO: Write Test33def test_stop_app(yttv):...
tasks.py
Source:tasks.py
...9 logger.debug('start_app')10 app.state.db = db_tasks.connect_to_db(config.DATABASE_URL)11 return start_app12def create_stop_app_handler(app: FastAPI) -> Callable:13 async def stop_app() -> None:14 logger.debug('stop_app')15 db_tasks.close_db_connection(app.state.db)16 app.state.app_context = None...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run Airtest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
