norets/knd_klinteplo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: norets:ba742f7d2129976f0ada1dc8c7968874ff9c407a
Branches Tags
norets:main
norets:master
..
compare: norets:151a915911f0c316011e6d746570b5f796e744a3
Branches Tags
norets:main
norets:master

No commits in common. "ba742f7d2129976f0ada1dc8c7968874ff9c407a" and "151a915911f0c316011e6d746570b5f796e744a3" have entirely different histories.
ba742f7d21 ... 151a915911

37 changed files with 138 additions and 1844 deletions
Show Stats Expand all files Collapse all files

2
classes/desinfection.py
View File

@ -36,7 +36,7 @@ def complete_task(session, headers, task, address):
session_t.headers.update( session_t.headers.update(
{ {
'referer': 'https://knd.mosreg.ru/executions'+str(task['id'])+'/solve', 'referer': 'https://knd.mosreg.ru/executions'+str(task['id'])+'/solve',
'x-platform': 'android', 'x-platform': 'wb',
'accept': '*/*' 'accept': '*/*'
} }
) )

135
classes/dip.py
View File

@ -1,135 +0,0 @@
import os
import datetime
import time
import json
import pprint
from natsort import natsorted
def assign_task(session, headers, task, address):
# print('assign_task')
url = 'https://knd.mosreg.ru//api/v1/executions/'+str(task['id'])+'/assign'
task_assign = session.put(url, headers=headers)
if task_assign.status_code == 200:
print(f'Получили задание по адресу: {address}')
complete_task(session, headers, task, address)
else:
print(f"Не смогли получить задание {address} {task['id']}!")
def complete_task(session, headers, task, address):
photos = get_photo(address, headers['uid'])
if len(photos) > 0:
# photos.sort()
photos = natsorted(photos)
else:
print(len(photos))
"""В папке лежит фото графика, которое отправляется из модуля des, поэтому убираем отсюда"""
# if len(photos) == 11 or len(photos) == 13:
# photos.pop()
id_task = task['id']
quest = get_questions(id_task, session, headers)
# print(quest)
# length_question = get_length_questions(quest)
prepared_answers = []
if True:
prepared_answers = prepare_answer(quest, photos, session, task, headers)
# print(prepared_answers)
ans = {"question_chains": prepared_answers}
# print(ans)
# print(task['id'])
session_t = session
session_t.headers.update(
{
'referer': 'https://knd.mosreg.ru/executions'+str(task['id'])+'/solve',
'x-platform': 'android',
'accept': '*/*'
}
)
complete = session_t.post('https://knd.mosreg.ru//api/v1/executions/'+str(task['id'])+'/answers', headers=headers, json=ans)
if complete.status_code == 200:
print(f'Задание по адресу {address} выполнено успешно!\n\n')
else:
print(f'Задание по адресу {address} не выполненно! Статус ошибки {complete.status_code}\n\n')
else:
print(f"\n\n\nНесоответствие количества вопросов и фотографий по адресу {address}!!! \
Необходимо проверить!!!\n\n\n")
# print(len(quest)) 9 and 13
def get_photo(address, email):
photos = []
month = datetime.datetime.now().month
main_path = os.path.join(os.path.abspath('.'), 'data_dip', email, address)
if not os.path.exists(main_path):
return photos
else:
for images in os.listdir(main_path):
if images.endswith(".jpg") or images.endswith(".jpeg") or images.endswith(".JPG"):
photos.append(main_path + '/' + images)
return photos
def get_questions(id_task, session, headers):
url = 'https://knd.mosreg.ru//api/v1/executions/'+str(id_task)+'/questions'
questions = session.get(url, headers=headers)
json_quest = questions.json()
return json_quest['selected_chains']
def prepare_answer(quest, photos, session, task, headers):
questions = []
question_components= []
question_chains = []
for main_chain in quest:
for second_chain in main_chain['questions']:
for third_chain in second_chain['question_components']:
question_type = third_chain['type']
if question_type == 'list':
data = {"id": third_chain['id'], "answer": 77, "answer_status": None, "start_time": int(time.time())}
time.sleep(2)
data.update({"end_time": int(time.time())})
question_components.append(data)
elif question_type == "geo":
data = {"id": third_chain['id'], "answer": {"location": task['coord']}, "answer_status": None, \
"start_time": int(time.time())}
time.sleep(2)
data.update({"end_time": int(time.time())})
question_components.append(data)
elif question_type == "gallery":
img = send_image(session, headers, task['id'], photos.pop(0))#photos.pop(0))
data = {"id": third_chain['id'], "answer": [img['id']], "answer_status": None, \
"start_time": int(time.time())}
time.sleep(2)
data.update({"end_time": int(time.time())})
question_components.append(data)
if len(question_components) > 0:
main_data = {
"id": second_chain['id'],
"question_components": question_components
}
questions.append(main_data)
question_components = []
if len(questions) > 0:
q_dict = {
"id": main_chain['id'],
"questions" : questions
}
question_chains.append(q_dict)
questions = []
return question_chains
def send_image(session, headers, url, photo):
"""Отправляем фото на сервер. Сервер должен вернуть id и url."""
files = {
'image': (photo[-5:], open(photo, 'rb')),
}
img = session.post('https://knd.mosreg.ru//api/v1/executions/'+str(url)+'/images', headers=headers, files=files)
if img.status_code == 200:
print(f"Фото {photo} успешно отправлено")
return img.json()
else:
print(f"Не удалось загрузить фото на сервер по заданию {url}, статус ошибки {img.status_code}! Повторная попытка!")
second_img = send_image(session, headers, url, photo)
return second_img

138
classes/kp.py
View File

@ -1,138 +0,0 @@
import os
import datetime
import time
import json
import pprint
from natsort import natsorted
def assign_task(session, headers, task):
# print('assign_task')
address = task['address']
url = 'https://knd.mosreg.ru//api/v1/executions/'+str(task['id'])+'/assign'
task_assign = session.put(url, headers=headers)
if task_assign.status_code == 200:
print(f'Получили задание по адресу: {address}')
complete_task(session, headers, task, address)
else:
print(f"Не смогли получить задание {address} {task['id']}!")
def complete_task(session, headers, task, address):
photos = get_photo(address, headers['uid'])
if len(photos) > 0:
# photos.sort()
photos = natsorted(photos)
else:
print(len(photos))
"""В папке лежит фото графика, которое отправляется из модуля des, поэтому убираем отсюда"""
# if len(photos) == 11 or len(photos) == 13:
# photos.pop()
id_task = task['id']
quest = get_questions(id_task, session, headers)
# print(quest)
# length_question = get_length_questions(quest)
prepared_answers = []
if True:
prepared_answers = prepare_answer(quest, photos, session, task, headers)
# print(prepared_answers)
ans = {"question_chains": prepared_answers}
# print(ans)
# print(task['id'])
session_t = session
session_t.headers.update(
{
'referer': 'https://knd.mosreg.ru/executions'+str(task['id'])+'/solve',
'x-platform': 'android',
'accept': '*/*'
}
)
complete = session_t.post('https://knd.mosreg.ru//api/v1/executions/'+str(task['id'])+'/answers', headers=headers, json=ans)
if complete.status_code == 200:
print(f'Задание по адресу {address} выполнено успешно!\n\n')
else:
print(f'Задание по адресу {address} не выполненно! Статус ошибки {complete.status_code}\n\n')
else:
print(f"\n\n\nНесоответствие количества вопросов и фотографий по адресу {address}!!! \
Необходимо проверить!!!\n\n\n")
# print(len(quest)) 9 and 13
def get_photo(address, email):
photos = []
# month = datetime.datetime.now().month
main_path = os.path.join(os.path.abspath('.'), 'data_kp', email)
if not os.path.exists(main_path):
return photos
else:
for images in os.listdir(main_path):
if images.endswith(".jpg") or images.endswith(".jpeg") or images.endswith(".JPG"):
photos.append(main_path + '/' + images)
return photos
def get_questions(id_task, session, headers):
url = 'https://knd.mosreg.ru//api/v1/executions/'+str(id_task)+'/questions'
questions = session.get(url, headers=headers)
json_quest = questions.json()
return json_quest['question_chains']
def prepare_answer(quest, photos, session, task, headers):
questions = []
question_components= []
question_chains = []
for main_chain in quest:
for second_chain in main_chain['questions']:
for third_chain in second_chain['question_components']:
question_type = third_chain['type']
if question_type == 'list':
data = {"id": third_chain['id'], "answer": 888, "start_time": int(time.time())}
time.sleep(2)
data.update({"end_time": int(time.time())})
question_components.append(data)
elif question_type == "geo":
data = {"id": third_chain['id'], "answer": {"location": task['coord']}, "start_time": int(time.time())}
time.sleep(2)
data.update({"end_time": int(time.time())})
question_components.append(data)
elif question_type == "gallery":
images = []
for image in photos:
img = send_image(session, headers, task['id'], image)
images.append(img['id'])
data = {"id": third_chain['id'], "answer": images,
"start_time": int(time.time())}
time.sleep(2)
data.update({"end_time": int(time.time())})
question_components.append(data)
if len(question_components) > 0:
main_data = {
"id": second_chain['id'],
"question_components": question_components
}
questions.append(main_data)
question_components = []
if len(questions) > 0:
q_dict = {
"id": main_chain['id'],
"questions": questions
}
question_chains.append(q_dict)
questions = []
return question_chains
def send_image(session, headers, url, photo):
"""Отправляем фото на сервер. Сервер должен вернуть id и url."""
files = {
'image': (photo[-5:], open(photo, 'rb')),
}
img = session.post('https://knd.mosreg.ru//api/v1/executions/'+str(url)+'/images', headers=headers, files=files)
if img.status_code == 200:
print(f"Фото {photo} успешно отправлено")
return img.json()
else:
print(f"Не удалось загрузить фото на сервер по заданию {url}, статус ошибки {img.status_code}! Повторная попытка!")
second_img = send_image(session, headers, url, photo)
return second_img

130
classes/led.py
View File

@ -1,130 +0,0 @@
import os
import datetime
import time
from natsort import natsorted
def assign_task(session, headers, task, address):
# print('assign_task')
url = 'https://knd.mosreg.ru//api/v1/executions/'+str(task['id'])+'/assign'
task_assign = session.put(url, headers=headers)
if task_assign.status_code == 200:
print(f'Получили задание по адресу: {address}')
complete_task(session, headers, task, address)
else:
print(f"Не смогли получить задание {address} {task['id']}!")
def complete_task(session, headers, task, address):
photos = get_photo(address, headers['uid'])
if len(photos) > 0:
# photos.sort()
photos = natsorted(photos)
"""В папке лежит фото графика, которое отправляется из модуля des, поэтому убираем отсюда"""
# if len(photos) == 11 or len(photos) == 13:
# photos.pop()
id_task = task['id']
quest = get_questions(id_task, session, headers)
# print(quest)
# length_question = get_length_questions(quest)
prepared_answers = []
if True:
prepared_answers = prepare_answer(quest, photos, session, task, headers)
# print(prepared_answers)
ans = {"question_chains": prepared_answers}
# print(ans)
session_t = session
session_t.headers.update(
{
'referer': 'https://knd.mosreg.ru/executions'+str(task['id'])+'/solve',
'x-platform': 'android',
'accept': '*/*'
}
)
complete = session_t.post('https://knd.mosreg.ru//api/v1/executions/'+str(task['id'])+'/answers', headers=headers, json=ans)
if complete.status_code == 200:
print(f'Задание по адресу {address} выполнено успешно!\n\n')
else:
print(f'Задание по адресу {address} не выполненно! Статус ошибки {complete.status_code}\n\n')
else:
print(f"\n\n\nНесоответствие количества вопросов и фотографий по адресу {address}!!! \
Необходимо проверить!!!\n\n\n")
# print(len(quest)) 9 and 13
def get_photo(address, email):
photos = []
month = datetime.datetime.now().month
main_path = os.path.join(os.path.abspath('.'), 'data_led', email, address)
if not os.path.exists(main_path):
return photos
else:
for images in os.listdir(main_path):
if images.endswith(".jpg") or images.endswith(".jpeg") or images.endswith(".JPG"):
photos.append(main_path + '/' + images)
return photos
def get_questions(id_task, session, headers):
url = 'https://knd.mosreg.ru//api/v1/executions/'+str(id_task)+'/questions'
questions = session.get(url, headers=headers)
json_quest = questions.json()
return json_quest['selected_chains']
def prepare_answer(quest, photos, session, task, headers):
questions = []
question_components= []
question_chains = []
for main_chain in quest:
for second_chain in main_chain['questions']:
for third_chain in second_chain['question_components']:
question_type = third_chain['type']
if question_type == 'list':
data = {"id": third_chain['id'], "answer": 77, "answer_status": None, "start_time": int(time.time())}
time.sleep(2)
data.update({"end_time": int(time.time())})
question_components.append(data)
elif question_type == "geo":
data = {"id": third_chain['id'], "answer": {"location": task['coord']}, "answer_status": None, \
"start_time": int(time.time())}
time.sleep(2)
data.update({"end_time": int(time.time())})
question_components.append(data)
elif question_type == "gallery":
img = send_image(session, headers, task['id'], photos.pop(0))#photos.pop(0))
data = {"id": third_chain['id'], "answer": [img['id']], "answer_status": None, \
"start_time": int(time.time())}
time.sleep(2)
data.update({"end_time": int(time.time())})
question_components.append(data)
if len(question_components) > 0:
main_data = {
"id": second_chain['id'],
"question_components": question_components
}
questions.append(main_data)
question_components = []
if len(questions) > 0:
q_dict = {
"id": main_chain['id'],
"questions" : questions
}
question_chains.append(q_dict)
questions = []
return question_chains
def send_image(session, headers, url, photo):
"""Отправляем фото на сервер. Сервер должен вернуть id и url."""
files = {
'image': (photo[-5:], open(photo, 'rb')),
}
img = session.post('https://knd.mosreg.ru//api/v1/executions/'+str(url)+'/images', headers=headers, files=files)
if img.status_code == 200:
print(f"Фото {photo} успешно отправлено")
return img.json()
else:
print(f"Не удалось загрузить фото на сервер по заданию {url}, статус ошибки {img.status_code}! Повторная попытка!")
second_img = send_image(session, headers, url, photo)
return second_img

9
classes/proverka.py
View File

@ -1,7 +1,6 @@
import os import os
import datetime import datetime
import time import time
from natsort import natsorted
def assign_task(session, headers, task, address): def assign_task(session, headers, task, address):
@ -18,11 +17,9 @@ def assign_task(session, headers, task, address):
def complete_task(session, headers, task, address): def complete_task(session, headers, task, address):
photos = get_photo(address, headers['uid']) photos = get_photo(address, headers['uid'])
if len(photos) > 0: if len(photos) > 0:
# photos.sort() photos.sort()
photos = natsorted(photos)
"""В папке лежит фото графика, которое отправляется из модуля des, поэтому убираем отсюда""" """В папке лежит фото графика, которое отправляется из модуля des, поэтому убираем отсюда"""
# if len(photos) == 11 or len(photos) == 14: if len(photos) == 7 or len(photos) == 9:
if len(photos) == 3 or len(photos) == 4:
photos.pop() photos.pop()
id_task = task['id'] id_task = task['id']
quest = get_questions(id_task, session, headers) quest = get_questions(id_task, session, headers)
@ -36,7 +33,7 @@ def complete_task(session, headers, task, address):
session_t.headers.update( session_t.headers.update(
{ {
'referer': 'https://knd.mosreg.ru/executions'+str(task['id'])+'/solve', 'referer': 'https://knd.mosreg.ru/executions'+str(task['id'])+'/solve',
'x-platform': 'android', 'x-platform': 'wb',
'accept': '*/*' 'accept': '*/*'
} }
) )

142
classes/proverka_monthly.py
View File

@ -1,142 +0,0 @@
import os
import datetime
import time
from natsort import natsorted
def assign_task(session, headers, task, address):
# print('assign_task')
# print("Start monthly")
url = 'https://knd.mosreg.ru//api/v1/executions/'+str(task['id'])+'/assign'
task_assign = session.put(url, headers=headers)
if task_assign.status_code == 200:
print(f'Получили задание по адресу: {address}')
complete_task(session, headers, task, address)
else:
print(f"Не смогли получить задание {address} {task['id']}!")
def complete_task(session, headers, task, address):
photos = get_photo(address, headers['uid'])
if len(photos) > 0:
# photos.sort()
photos = natsorted(photos)
"""В папке лежит фото графика, которое отправляется из модуля des, поэтому убираем отсюда"""
# print(len(photos))
if len(photos) == 11 or len(photos) == 14:
# if len(photos) == 3 or len(photos) == 4:
photos.pop()
id_task = task['id']
quest = get_questions(id_task, session, headers)
length_question = get_length_questions(quest)
prepared_answers = []
if len(photos) == length_question:
prepared_answers = prepare_answer(quest, photos, session, task, headers)
ans = {"question_chains": prepared_answers}
# print(ans)
session_t = session
session_t.headers.update(
{
'referer': 'https://knd.mosreg.ru/executions'+str(task['id'])+'/solve',
'x-platform': 'android',
'accept': '*/*'
}
)
complete = session_t.post('https://knd.mosreg.ru//api/v1/executions/'+str(task['id'])+'/answers', headers=headers, json=ans)
if complete.status_code == 200:
print(f'Задание по адресу {address} выполнено успешно!\n\n')
else:
print(f'Задание по адресу {address} не выполненно! Статус ошибки {complete.status_code}\n\n')
else:
print(f"\n\n\nНесоответствие количества вопросов и фотографий по адресу {address}!!! \
Необходимо проверить!!!\n\n\n")
# print(len(quest)) 9 and 13
def get_length_questions(quest):
count = 0
for main_chain in quest:
for second_chain in main_chain['questions']:
for third_chain in second_chain['question_components']:
question_type = third_chain['type']
if question_type == "gallery":
count = count + 1
return count
def prepare_answer(quest, photos, session, task, headers):
questions = []
question_components= []
question_chains = []
for main_chain in quest:
for second_chain in main_chain['questions']:
for third_chain in second_chain['question_components']:
question_type = third_chain['type']
if question_type == 'list':
data = {"id": third_chain['id'], "answer": 77, "answer_status": None, "start_time": int(time.time())}
time.sleep(2)
data.update({"end_time": int(time.time())})
question_components.append(data)
elif question_type == "geo":
data = {"id": third_chain['id'], "answer": {"location": task['coord']}, "answer_status": None, \
"start_time": int(time.time())}
time.sleep(2)
data.update({"end_time": int(time.time())})
question_components.append(data)
elif question_type == "gallery":
img = send_image(session, headers, task['id'], photos.pop(0))#photos.pop(0))
data = {"id": third_chain['id'], "answer": [img['id']], "answer_status": None, \
"start_time": int(time.time())}
time.sleep(2)
data.update({"end_time": int(time.time())})
question_components.append(data)
if len(question_components) > 0:
main_data = {
"id": second_chain['id'],
"question_components": question_components
}
questions.append(main_data)
question_components = []
if len(questions) > 0:
q_dict = {
"id": main_chain['id'],
"questions" : questions
}
question_chains.append(q_dict)
questions = []
return question_chains
def get_photo(address, email):
photos = []
month = datetime.datetime.now().month
main_path = os.path.join(os.path.abspath('.'), 'data1', email, str(month), address)
if not os.path.exists(main_path):
return photos
else:
for images in os.listdir(main_path):
if images.endswith(".jpg") or images.endswith(".jpeg") or images.endswith(".JPG"):
photos.append(main_path + '/' + images)
return photos
def get_questions(id_task, session, headers):
url = 'https://knd.mosreg.ru//api/v1/executions/'+str(id_task)+'/questions'
questions = session.get(url, headers=headers)
json_quest = questions.json()
return json_quest['selected_chains']
def send_image(session, headers, url, photo):
"""Отправляем фото на сервер. Сервер должен вернуть id и url."""
files = {
'image': (photo[-5:], open(photo, 'rb')),
}
img = session.post('https://knd.mosreg.ru//api/v1/executions/'+str(url)+'/images', headers=headers, files=files)
if img.status_code == 200:
print(f"Фото {photo} успешно отправлено")
return img.json()
else:
print(f"Не удалось загрузить фото на сервер по заданию {url}, статус ошибки {img.status_code}! Повторная попытка!")
second_img = send_image(session, headers, url, photo)
return second_img

28
classes/users_dip.py
View File

@ -1,28 +0,0 @@
import os
class Users_dip:
def __init__(self):
self.data = []
self.path = os.path.join(os.path.abspath('.'), 'data_dip')
def get_users(self):
"""Get names from names of folders."""
self.logins = []
main_dir = os.walk(self.path)
for d, dirs, files in main_dir:
self.logins.append(dirs)
return self.logins[0]
def get_passwords(self):
if not os.path.exists(self.path):
return "Папка data_led не существует. Создайте папку с необходимыми параметрами и запустите программу снова!"
self.emails = self.get_users()
if len(self.emails) > 0:
for user in self.emails:
with open(os.path.join(self.path, user, 'pass.txt'), 'r') as f:
log_pass = {'email': user, 'password': f.readline().split('\n')[0]}
self.data.append(log_pass)
else:
return "Нет ни одного пользователя! Заполните папку и запустите программу снова!"
return self.data

28
classes/users_kp.py
View File

@ -1,28 +0,0 @@
import os
class Users_kp:
def __init__(self):
self.data = []
self.path = os.path.join(os.path.abspath('.'), 'data_kp')
def get_users(self):
"""Get names from names of folders."""
self.logins = []
main_dir = os.walk(self.path)
for d, dirs, files in main_dir:
self.logins.append(dirs)
return self.logins[0]
def get_passwords(self):
if not os.path.exists(self.path):
return "Папка data_led не существует. Создайте папку с необходимыми параметрами и запустите программу снова!"
self.emails = self.get_users()
if len(self.emails) > 0:
for user in self.emails:
with open(os.path.join(self.path, user, 'pass.txt'), 'r') as f:
log_pass = {'email': user, 'password': f.readline().split('\n')[0]}
self.data.append(log_pass)
else:
return "Нет ни одного пользователя! Заполните папку и запустите программу снова!"
return self.data

27
classes/users_led.py
View File

@ -1,27 +0,0 @@
import os
class Users_led:
def __init__(self):
self.data = []
self.path = os.path.join(os.path.abspath('.'), 'data_led')
def get_users(self):
self.logins = []
main_dir = os.walk(self.path)
for d, dirs, files in main_dir:
self.logins.append(dirs)
return self.logins[0]
def get_passwords(self):
if not os.path.exists(self.path):
return "Папка data_led не существует. Создайте папку с необходимыми параметрами и запустите программу снова!"
self.emails = self.get_users()
if len(self.emails) > 0:
for user in self.emails:
with open(os.path.join(self.path, user, 'pass.txt'), 'r') as f:
log_pass = {'email': user, 'password': f.readline().split('\n')[0]}
self.data.append(log_pass)
else:
return "Нет ни одного пользователя! Заполните папку и запустите программу снова!"
return self.data

27
classes/users_month.py
View File

@ -1,27 +0,0 @@
import os
class Users:
def __init__(self):
self.data = []
self.path = os.path.join(os.path.abspath('.'), 'data1')
def get_users(self):
self.logins = []
main_dir = os.walk(self.path)
for d, dirs, files in main_dir:
self.logins.append(dirs)
return self.logins[0]
def get_passwords(self):
if not os.path.exists(self.path):
return "Папка data не существует. Создайте папку с необходимыми параметрами и запустите программу снова!"
self.emails = self.get_users()
if len(self.emails) > 0:
for user in self.emails:
with open(os.path.join(self.path, user, 'pass.txt'), 'r') as f:
log_pass = {'email': user, 'password': f.readline().split('\n')[0]}
self.data.append(log_pass)
else:
return "Нет ни одного пользователя! Заполните папку и запустите программу снова!"
return self.data

3
data/5020051404_17@check.mo/img_5.txt
View File

@ -1,3 +0,0 @@
Чернов Александр Алексеевич
мастер РЭУ
8(910)484-22-20

BIN
data/5020051404_17@check.mo/new_template.docx
View File

Binary file not shown.

1
data/5020051404_17@check.mo/pass.txt
View File

@ -1 +0,0 @@
kn7687

BIN
data/5020051404_17@check.mo/template.docx
View File

Binary file not shown.

BIN
data/5020051404_17@check.mo/template_lift.docx
View File

Binary file not shown.

4
data/5020051404_1@check.mo/img_5.txt
View File

@ -1,4 +0,0 @@
Курынин Александр Иванович
начальник РЭУ
8(985)574-78-37

BIN
data/5020051404_1@check.mo/new_template.docx
View File

Binary file not shown.

1
data/5020051404_1@check.mo/pass.txt
View File

@ -1 +0,0 @@
kn8746

BIN
data/5020051404_1@check.mo/template.docx
View File

Binary file not shown.

BIN
data/5020051404_1@check.mo/template_lift.docx
View File

Binary file not shown.

3
data/domreshetnikovo@gmail.com/img_5.txt
View File

@ -1,3 +0,0 @@
Калякина Людмила Николаевна
начальник РЭУ
8(985)574-78-30

BIN
data/domreshetnikovo@gmail.com/new_template.docx
View File

Binary file not shown.

1
data/domreshetnikovo@gmail.com/pass.txt
View File

@ -1 +0,0 @@
kn2001

BIN
data/domreshetnikovo@gmail.com/template.docx
View File

Binary file not shown.

BIN
data/domreshetnikovo@gmail.com/template_lift.docx
View File

Binary file not shown.

1
data/saunkina2020@mail.ru/pass.txt
View File

@ -1 +0,0 @@
kn4221

4
ddd.txt
View File

File diff suppressed because one or more lines are too long

32
des.py
View File

@ -1,15 +1,13 @@
# import json import requests
import datetime
import os
import random import random
import string import string
# import json
import datetime
import time import time
import os
import requests
import classes.desinfection as des
# import classes.proverka as prov
from classes.users import Users from classes.users import Users
import classes.desinfection as des
import classes.proverka as prov
def get_questions(id_task, session, headers): def get_questions(id_task, session, headers):
@ -89,7 +87,7 @@ def get_task(session, headers):
session_t = session session_t = session
session_t.headers.update( session_t.headers.update(
{'referer': 'https://knd.mosreg.ru/executions', {'referer': 'https://knd.mosreg.ru/executions',
'x-platform': 'android', 'x-platform': 'wb',
'accept': '*/*' 'accept': '*/*'
} }
) )
@ -101,7 +99,7 @@ def get_task(session, headers):
search = addr.replace("_", "/")[:-4] search = addr.replace("_", "/")[:-4]
# print(search) # print(search)
data = {'search': search} data = {'search': search}
tasks = session_t.post('https://knd.mosreg.ru//api/v1/actor/executions', tasks = session_t.post('https://knd.mosreg.ru//api/v1/executions',
headers=headers, headers=headers,
data=data data=data
) )
@ -111,22 +109,18 @@ def get_task(session, headers):
# #
# Уже полученные задания # Уже полученные задания
exec_assigned = json_tasks['executions_assigned'] exec_assigned = json_tasks['executions_assigned']
print(exec_assigned)
# Еще не принятые задания # Еще не принятые задания
exec_available = json_tasks['executions_available'] exec_available = json_tasks['executions_available']
print(exec_available)
# print(address) # print(address)
if len(exec_available) > 0: if len(exec_available) > 0:
for n in exec_available: for n in exec_available:
coord = n['coord'] coord = n['coord']
enterances = n['tasks'] enterances = n['tasks']
for execution in enterances: for execution in enterances:
address_from_tasks = execution['dimensions'][0].get( address_from_tasks = execution['dimensions'][0].get('entity_value_name')
'entity_value_name')
name_from_tasks = execution['name'] name_from_tasks = execution['name']
if name_from_tasks == desinfection: if name_from_tasks == desinfection:
enterance = execution['dimensions'][2].get( enterance = execution['dimensions'][2].get('entity_value_name')
'entity_value_name')
if address_from_tasks == 'Клин г, ' + addr.replace("_", "/")[:-4] and enterance[-1] == address[-2]: if address_from_tasks == 'Клин г, ' + addr.replace("_", "/")[:-4] and enterance[-1] == address[-2]:
print('\n') print('\n')
print(name_from_tasks) print(name_from_tasks)
@ -138,16 +132,14 @@ def get_task(session, headers):
'coord': coord 'coord': coord
} }
if name_from_tasks == desinfection: if name_from_tasks == desinfection:
des.assign_task( des.assign_task(session, headers, task, address)
session, headers, task, address)
elif name_from_tasks == proverka: elif name_from_tasks == proverka:
pass pass
# prov.assign_task(session, headers, task, address) # prov.assign_task(session, headers, task, address)
else: else:
pass pass
else: else:
print( print(f"Нет доступных заданий для пользователя {headers['uid']} по адресу: {address}")
f"Нет доступных заданий для пользователя {headers['uid']} по адресу: {address}")
# Если адреса не получили выводим ошибку # Если адреса не получили выводим ошибку
else: else:
print(addresses) print(addresses)

4
des_assigned.py
View File

@ -172,7 +172,7 @@ def get_task(session, headers):
session_t = session session_t = session
session_t.headers.update( session_t.headers.update(
{'referer': 'https://knd.mosreg.ru/executions', {'referer': 'https://knd.mosreg.ru/executions',
'x-platform': 'android', 'x-platform': 'wb',
'accept': '*/*' 'accept': '*/*'
} }
) )
@ -184,7 +184,7 @@ def get_task(session, headers):
search = addr.replace("_", "/")[:-4] search = addr.replace("_", "/")[:-4]
# print(search) # print(search)
data = {'search': search} data = {'search': search}
tasks = session_t.post('https://knd.mosreg.ru//api/v1/actor/executions', tasks = session_t.post('https://knd.mosreg.ru//api/v1/executions',
headers=headers, headers=headers,
data=data data=data
) )

253
knd_bot.py
View File

@ -1,178 +1,137 @@
# import json import requests
import datetime
import os
import random import random
import string import string
# import json
import datetime
import time import time
import os
import requests from classes.users import Users
import classes.desinfection as des import classes.desinfection as des
import classes.proverka as prov import classes.proverka as prov
from classes.users import Users
def get_questions(id_task, session, headers): def get_questions(id_task, session, headers):
url = "https://knd.mosreg.ru//api/v1/executions/" + str(id_task) + "/questions" url = 'https://knd.mosreg.ru//api/v1/executions/'+str(id_task)+'/questions'
questions = session.get(url, headers=headers) questions = session.get(url, headers=headers)
json_quest = questions.json() json_quest = questions.json()
return json_quest["selected_chains"] return json_quest['selected_chains']
def send_image(session, headers, url, photo): def send_image(session, headers, url, photo):
"""Отправляем фото на сервер. Сервер должен вернуть id и url.""" """Отправляем фото на сервер. Сервер должен вернуть id и url."""
files = { files = {
"image": (photo[-5:], open(photo, "rb")), 'image': (photo[-5:], open(photo, 'rb')),
} }
# print(files) # print(files)
img = session.post( img = session.post('https://knd.mosreg.ru//api/v1/executions/'+str(url)+'/images', headers=headers, files=files)
"https://knd.mosreg.ru//api/v1/executions/" + str(url) + "/images",
headers=headers,
files=files,
)
if img.status_code == 200: if img.status_code == 200:
print(f"Фото {photo} успешно отправлено") print(f"Фото {photo} успешно отправлено")
return img.json() return img.json()
else: else:
print( print(f"Не удалось загрузить фото на сервер по заданию {url}, статус ошибки {img.status_code}! Повторная попытка!")
f"Не удалось загрузить фото на сервер по заданию {url}, статус ошибки {img.status_code}! Повторная попытка!"
)
second_img = send_image(session, headers, url, photo) second_img = send_image(session, headers, url, photo)
return second_img return second_img
def complete_task(session, headers, task, address): def complete_task(session, headers, task, address):
photos = get_photo(address, headers["uid"]) photos = get_photo(address, headers['uid'])
if len(photos) > 0: if len(photos) > 0:
photos.sort() photos.sort()
print(task["id"]) print(task['id'])
id_task = task["id"] id_task = task['id']
questions = get_questions(id_task, session, headers) questions = get_questions(id_task, session, headers)
data = [] data = []
for answers in questions: for answers in questions:
answer_chain = {"id": answers["id"], "questions": []} # answers['questions'] answer_chain = {
if answer_chain["id"] == 6146: # Цепочка без ответов 'id': answers['id'],
'questions': [] # answers['questions']
}
if answer_chain['id'] == 6146: # Цепочка без ответов
continue continue
for components in answers["questions"]: for components in answers['questions']:
comp = [] comp = []
q_c = {"id": components["id"], "question_components": []} q_c = {
if q_c["id"] == 6560: # Вопрос без ответа 'id': components['id'],
'question_components': []
}
if q_c['id'] == 6560: # Вопрос без ответа
continue continue
comp.append(q_c) comp.append(q_c)
answer_chain.update({"questions": comp}) answer_chain.update({'questions': comp})
data.append(answer_chain) data.append(answer_chain)
if len(data) == len(photos): if len(data) == len(photos):
for chain in data: for chain in data:
for item in chain["questions"]: for item in chain['questions']:
if len(data) == len(photos): if len(data) == len(photos):
photo_from_photos = photos.pop(0) photo_from_photos = photos.pop(0)
img = send_image( img = send_image(session, headers, task['id'], photo_from_photos) # photos.pop(0))
session, headers, task["id"], photo_from_photos
) # photos.pop(0))
# count_send_photo = 0 # count_send_photo = 0
# while img['id'] is None: # while img['id'] is None:
# count_send_photo += 1 # count_send_photo += 1
# print("Попытка № " + str(count_send_photo) + "отправить фото" + str(photo_from_photos)) # print("Попытка № " + str(count_send_photo) + "отправить фото" + str(photo_from_photos))
# img = send_image(session, headers, task['id'], photo_from_photos)#photos.pop(0)) # img = send_image(session, headers, task['id'], photo_from_photos)#photos.pop(0))
first = { first = {"id": 18850, "answer": [img['id']], 'answer_status': None, 'start_time': int(time.time())}
"id": 18850,
"answer": [img["id"]],
"answer_status": None,
"start_time": int(time.time()),
}
time.sleep(2) time.sleep(2)
first.update({"end_time": int(time.time())}) first.update({'end_time': int(time.time())})
second = { second = {"id": 18851, "answer": {"location": task['coord']}, 'answer_status': None, 'start_time': int(time.time())}
"id": 18851,
"answer": {"location": task["coord"]},
"answer_status": None,
"start_time": int(time.time()),
}
time.sleep(2) time.sleep(2)
second.update({"end_time": int(time.time())}) second.update({'end_time': int(time.time())})
item["question_components"].append(first) item['question_components'].append(first)
item["question_components"].append(second) item['question_components'].append(second)
else: else:
# img = send_image(session, headers, task['id'], photos.pop(0)) #img = send_image(session, headers, task['id'], photos.pop(0))
photo_from_photos = photos.pop(0) photo_from_photos = photos.pop(0)
img = send_image( img = send_image(session, headers, task['id'], photo_from_photos)#photos.pop(0))
session, headers, task["id"], photo_from_photos
) # photos.pop(0))
# count_send_photo = 0 # count_send_photo = 0
# while img['id'] is None: # while img['id'] is None:
# count_send_photo += 1 # count_send_photo += 1
# print("Попытка № " + str(count_send_photo) + " отправить фото" + str(photo_from_photos)) # print("Попытка № " + str(count_send_photo) + " отправить фото" + str(photo_from_photos))
# img = send_image(session, headers, task['id'], photo_from_photos)#photos.pop(0)) # img = send_image(session, headers, task['id'], photo_from_photos)#photos.pop(0))
first = { first = {"id": 18856, "answer": 77, 'answer_status': None, 'start_time': int(time.time())}
"id": 18856,
"answer": 77,
"answer_status": None,
"start_time": int(time.time()),
}
time.sleep(2) time.sleep(2)
first.update({"end_time": int(time.time())}) first.update({'end_time': int(time.time())})
second = { second = {"id": 18857, "answer": [img['id']], 'answer_status': None, 'start_time': int(time.time())}
"id": 18857,
"answer": [img["id"]],
"answer_status": None,
"start_time": int(time.time()),
}
time.sleep(2) time.sleep(2)
second.update({"end_time": int(time.time())}) second.update({'end_time': int(time.time())})
item["question_components"].append(first) item['question_components'].append(first)
item["question_components"].append(second) item['question_components'].append(second)
ans = {"question_chains": data} ans = {"question_chains": data}
# print(ans) # print(ans)
session_t = session session_t = session
session_t.headers.update( session_t.headers.update(
{ {
"referer": "https://knd.mosreg.ru/executions" + str(task["id"]) 'referer': 'https://knd.mosreg.ru/executions'+str(task['id'])+'/solve',
+ "/solve", 'x-platform': 'wb',
"x-platform": "wb", 'accept': '*/*'
"accept": "*/*", }
}
)
complete = session_t.post(
"https://knd.mosreg.ru//api/v1/executions/" + str(task["id"]) + "/answers",
headers=headers,
json=ans,
) )
complete = session_t.post('https://knd.mosreg.ru//api/v1/executions/'+str(task['id'])+'/answers', headers=headers, json=ans)
if complete.status_code == 200: if complete.status_code == 200:
print(f"Задание по адресу {address} выполнено успешно!") print(f'Задание по адресу {address} выполнено успешно!')
else: else:
print( print(f'Задание по адресу {address} не выполненно! Статус ошибки {complete.status_code}')
f"Задание по адресу {address} не выполненно! Статус ошибки {complete.status_code}"
)
else: else:
print( print(f"Несоответствие количества вопросов и количества фотографий по адресу: {address}")
f"Несоответствие количества вопросов и количества фотографий по адресу: {address}"
)
# send_execution(task,session, headers) # send_execution(task,session, headers)
def get_photo(address, email): def get_photo(address, email):
photos = [] photos = []
month = datetime.datetime.now().month month = datetime.datetime.now().month
main_path = os.path.join(os.path.abspath("."), "data", email, str(month), address) main_path = os.path.join(os.path.abspath('.'), 'data', email, str(month), address)
if not os.path.exists(main_path): if not os.path.exists(main_path):
return photos return photos
else: else:
for images in os.listdir(main_path): for images in os.listdir(main_path):
if ( if images.endswith(".jpg") or images.endswith(".jpeg") or images.endswith(".JPG"):
images.endswith(".jpg") photos.append(main_path + '/' + images)
or images.endswith(".jpeg")
or images.endswith(".JPG")
):
photos.append(main_path + "/" + images)
return photos return photos
def get_addresses(email): def get_addresses(email):
addresses = [] addresses = []
month = datetime.datetime.now().month month = datetime.datetime.now().month
main_path = os.path.join(os.path.abspath("."), "data") main_path = os.path.join(os.path.abspath('.'), 'data')
path = os.walk(os.path.join(main_path, email, str(month))) path = os.walk(os.path.join(main_path, email, str(month)))
if not os.path.exists(os.path.join(main_path, email, str(month))): if not os.path.exists(os.path.join(main_path, email, str(month))):
return f"Не найдена папка с месяцем у пользователя {email}! Проверьте структуру каталога!" return f"Не найдена папка с месяцем у пользователя {email}! Проверьте структуру каталога!"
@ -204,19 +163,18 @@ def get_task(session, headers):
2 адрес 2 адрес
3 подъезд 3 подъезд
4 id_задания""" 4 id_задания"""
desinfection = "Дезинфекция подъездов (МОП)" desinfection = 'Дезинфекция подъездов (МОП)'
proverka = "Систематическая проверка подъездов МКД" proverka = 'Систематическая проверка подъездов МКД'
# Получаем адреса # Получаем адреса
addresses = get_addresses(headers["uid"]) addresses = get_addresses(headers['uid'])
# Если адреса получили # Если адреса получили
if isinstance(addresses, list): if isinstance(addresses, list):
session_t = session session_t = session
session_t.headers.update( session_t.headers.update(
{ {'referer': 'https://knd.mosreg.ru/executions',
"referer": "https://knd.mosreg.ru/executions", 'x-platform': 'wb',
"x-platform": "android", 'accept': '*/*'
"accept": "*/*", }
}
) )
for address in addresses: for address in addresses:
# change '-' for '/' # change '-' for '/'
@ -225,61 +183,45 @@ def get_task(session, headers):
addr = address.replace("Маиданово", "Майданово") addr = address.replace("Маиданово", "Майданово")
search = addr.replace("_", "/")[:-4] search = addr.replace("_", "/")[:-4]
# print(search) # print(search)
# search = "" data = {'search': search}
data = {"search": search} tasks = session_t.post('https://knd.mosreg.ru//api/v1/executions',
tasks = session_t.post( headers=headers,
"https://knd.mosreg.ru//api/v1/executions", headers=headers, data=data data=data
) )
json_tasks = tasks.json() json_tasks = tasks.json()
get_t = False get_t = False
# Если есть доступное задание с таким адресом # Если есть доступное задание с таким адресом
exec_assigned = json_tasks["executions_assigned"] # Уже полученные задания exec_assigned = json_tasks['executions_assigned'] # Уже полученные задания
exec_available = json_tasks[ exec_available = json_tasks['executions_available'] # Еще не принятые задания
"executions_available"
] # Еще не принятые задания
# print(address) # print(address)
if len(exec_available) > 0: if len(exec_available) > 0:
for n in exec_available: for n in exec_available:
coord = n["coord"] coord = n['coord']
enterances = n["tasks"] enterances = n['tasks']
for execution in enterances: for execution in enterances:
address_from_tasks = execution["dimensions"][0].get( address_from_tasks = execution['dimensions'][0].get('entity_value_name')
"entity_value_name" name_from_tasks = execution['name']
) enterance = execution['dimensions'][2].get('entity_value_name')
name_from_tasks = execution["name"] if address_from_tasks == 'Клин г, '+ addr.replace("_", "/")[:-4] and \
if name_from_tasks == proverka: enterance[-1] == address[-2]:
enterance = execution["dimensions"][2].get( print('\n')
"entity_value_name" print(name_from_tasks)
) print('\n')
if ( task = {
address_from_tasks 'id': execution['execution_id'],
== "Клин г, " + addr.replace("_", "/")[:-4] 'address': execution['dimensions'][0].get('entity_value_name'),
and enterance[-1] == address[-2] 'enterance': execution['dimensions'][2].get('entity_value_name'),
): 'coord': coord
print("\n") }
print(name_from_tasks) if name_from_tasks == desinfection:
print("\n") pass
task = {
"id": execution["execution_id"],
"address": execution["dimensions"][0].get(
"entity_value_name"
),
"enterance": execution["dimensions"][2].get(
"entity_value_name"
),
"coord": coord,
}
if name_from_tasks == desinfection:
pass
# des.assign_task(session, headers, task, address) # des.assign_task(session, headers, task, address)
elif name_from_tasks == proverka: elif name_from_tasks == proverka:
prov.assign_task(session, headers, task, address) prov.assign_task(session, headers, task, address)
else: else:
pass pass
else: else:
print( print(f"Нет доступных заданий для пользователя {headers['uid']} по адресу: {address}")
f"Нет доступных заданий для пользователя {headers['uid']} по адресу: {address}"
)
# Если адреса не получили выводим ошибку # Если адреса не получили выводим ошибку
else: else:
print(addresses) print(addresses)
@ -289,15 +231,18 @@ def main():
users = Users() users = Users()
logins = users.get_passwords() logins = users.get_passwords()
if isinstance(logins, list): if isinstance(logins, list):
url = "https://knd.mosreg.ru//api/v1/auth/sign_in" url = 'https://knd.mosreg.ru//api/v1/auth/sign_in'
for user in logins: for user in logins:
session = requests.Session() session = requests.Session()
login = {"email": user["email"], "password": user["password"]} login = {
'email': user['email'],
'password': user['password']
}
response = session.post(url, data=login) response = session.post(url, data=login)
headers = { headers = {
"client": response.headers.get("client"), 'client': response.headers.get('client'),
"Access-token": response.headers.get("Access-token"), 'Access-token': response.headers.get('Access-token'),
"uid": response.headers.get("uid"), 'uid': response.headers.get('uid'),
} }
if response.status_code == 200: if response.status_code == 200:
get_task(session, headers) get_task(session, headers)
@ -316,4 +261,4 @@ if __name__ == "__main__":
def random_string(stringLength): def random_string(stringLength):
letters = string.ascii_letters letters = string.ascii_letters
digits = string.digits digits = string.digits
return "".join(random.choice(letters + digits) for i in range(stringLength)) return ''.join(random.choice(letters+digits) for i in range(stringLength))

58
knd_bot_assigned.py
View File

@ -165,16 +165,14 @@ def get_task(session, headers):
4 id_задания""" 4 id_задания"""
desinfection = 'Дезинфекция подъездов (МОП)' desinfection = 'Дезинфекция подъездов (МОП)'
proverka = 'Систематическая проверка подъездов МКД' proverka = 'Систематическая проверка подъездов МКД'
proverka_monthly = 'Ежемесячная проверка подъездов МКД'
# Получаем адреса # Получаем адреса
addresses = get_addresses(headers['uid']) addresses = get_addresses(headers['uid'])
# Если адреса получили # Если адреса получили
# {"filters":[{"value":{"distance":"50000","lat":55.81999969482422,"lon":37.380001068115234},"filter_attribute":10,"id":6,"name":"Локация","type":"distance"}]}
if isinstance(addresses, list): if isinstance(addresses, list):
session_t = session session_t = session
session_t.headers.update( session_t.headers.update(
{'referer': 'https://knd.mosreg.ru/executions', {'referer': 'https://knd.mosreg.ru/executions',
'x-platform': 'android', 'x-platform': 'wb',
'accept': '*/*' 'accept': '*/*'
} }
) )
@ -185,18 +183,8 @@ def get_task(session, headers):
addr = address.replace("Маиданово", "Майданово") addr = address.replace("Маиданово", "Майданово")
search = addr.replace("_", "/")[:-4] search = addr.replace("_", "/")[:-4]
# print(search) # print(search)
# data = {'search': search,
# 'filters': [{'value':{
# 'distance': 25000,
# 'lat': 55.81999969482422,
# 'lon': 37.380001068115234
# },
# 'filter_attribute': 10,
# 'id': 6,
# 'name': 'Локация',
# 'type': 'distance'}]}
data = {'search': search} data = {'search': search}
tasks = session_t.post('https://knd.mosreg.ru//api/v1/actor/executions', tasks = session_t.post('https://knd.mosreg.ru//api/v1/executions',
headers=headers, headers=headers,
data=data data=data
) )
@ -204,9 +192,7 @@ def get_task(session, headers):
get_t = False get_t = False
# Если есть доступное задание с таким адресом # Если есть доступное задание с таким адресом
exec_assigned = json_tasks['executions_assigned'] # Уже полученные задания exec_assigned = json_tasks['executions_assigned'] # Уже полученные задания
# print(exec_assigned)
exec_available = json_tasks['executions_available'] # Еще не принятые задания exec_available = json_tasks['executions_available'] # Еще не принятые задания
# print(exec_available)
# print(address) # print(address)
if len(exec_assigned) > 0: if len(exec_assigned) > 0:
for n in exec_assigned: for n in exec_assigned:
@ -215,27 +201,25 @@ def get_task(session, headers):
for execution in enterances: for execution in enterances:
address_from_tasks = execution['dimensions'][0].get('entity_value_name') address_from_tasks = execution['dimensions'][0].get('entity_value_name')
name_from_tasks = execution['name'] name_from_tasks = execution['name']
# print(name_from_tasks) enterance = execution['dimensions'][2].get('entity_value_name')
if name_from_tasks == proverka or name_from_tasks == proverka_monthly: if address_from_tasks == 'Клин г, '+ addr.replace("_", "/")[:-4] and \
enterance = execution['dimensions'][2].get('entity_value_name') enterance[-1] == address[-2]:
if address_from_tasks == 'Клин г, '+ addr.replace("_", "/")[:-4] and \ print('\n')
enterance[-1] == address[-2]: print(name_from_tasks)
print('\n') print('\n')
print(name_from_tasks) task = {
print('\n') 'id': execution['execution_id'],
task = { 'address': execution['dimensions'][0].get('entity_value_name'),
'id': execution['execution_id'], 'enterance': execution['dimensions'][2].get('entity_value_name'),
'address': execution['dimensions'][0].get('entity_value_name'), 'coord': coord
'enterance': execution['dimensions'][2].get('entity_value_name'), }
'coord': coord if name_from_tasks == desinfection:
} pass
if name_from_tasks == desinfection:
pass
# des.assign_task(session, headers, task, address) # des.assign_task(session, headers, task, address)
elif name_from_tasks == proverka or name_from_tasks == proverka_monthly: elif name_from_tasks == proverka:
prov.assign_task(session, headers, task, address) prov.assign_task(session, headers, task, address)
else: else:
pass pass
else: else:
print(f"Нет доступных заданий для пользователя {headers['uid']} по адресу: {address}") print(f"Нет доступных заданий для пользователя {headers['uid']} по адресу: {address}")
# Если адреса не получили выводим ошибку # Если адреса не получили выводим ошибку
@ -264,7 +248,7 @@ def main():
get_task(session, headers) get_task(session, headers)
else: else:
print(f"Отказ в авторизации для пользователя {user['email']}!") print(f"Отказ в авторизации для пользователя {user['email']}!")
time.sleep(20) time.sleep(1)
else: else:
print(users.get_passwords()) print(users.get_passwords())

282
knd_bot_monthly.py
View File

@ -1,282 +0,0 @@
import requests
import random
import string
# import json
import datetime
import time
import os
# from classes.users import Users
from classes.users_month import Users
import classes.desinfection as des
import classes.proverka as prov
import classes.proverka_monthly as monthly_prov
def get_questions(id_task, session, headers):
url = 'https://knd.mosreg.ru//api/v1/executions/'+str(id_task)+'/questions'
questions = session.get(url, headers=headers)
json_quest = questions.json()
return json_quest['selected_chains']
def send_image(session, headers, url, photo):
"""Отправляем фото на сервер. Сервер должен вернуть id и url."""
files = {
'image': (photo[-5:], open(photo, 'rb')),
}
# print(files)
img = session.post('https://knd.mosreg.ru//api/v1/executions/'+str(url)+'/images', headers=headers, files=files)
if img.status_code == 200:
print(f"Фото {photo} успешно отправлено")
return img.json()
else:
print(f"Не удалось загрузить фото на сервер по заданию {url}, статус ошибки {img.status_code}! Повторная попытка!")
second_img = send_image(session, headers, url, photo)
return second_img
def complete_task(session, headers, task, address):
photos = get_photo(address, headers['uid'])
if len(photos) > 0:
photos.sort()
print(task['id'])
id_task = task['id']
questions = get_questions(id_task, session, headers)
data = []
for answers in questions:
answer_chain = {
'id': answers['id'],
'questions': [] # answers['questions']
}
if answer_chain['id'] == 6146: # Цепочка без ответов
continue
for components in answers['questions']:
comp = []
q_c = {
'id': components['id'],
'question_components': []
}
if q_c['id'] == 6560: # Вопрос без ответа
continue
comp.append(q_c)
answer_chain.update({'questions': comp})
data.append(answer_chain)
if len(data) == len(photos):
for chain in data:
for item in chain['questions']:
if len(data) == len(photos):
photo_from_photos = photos.pop(0)
img = send_image(session, headers, task['id'], photo_from_photos) # photos.pop(0))
# count_send_photo = 0
# while img['id'] is None:
# count_send_photo += 1
# print("Попытка № " + str(count_send_photo) + "отправить фото" + str(photo_from_photos))
# img = send_image(session, headers, task['id'], photo_from_photos)#photos.pop(0))
first = {"id": 18850, "answer": [img['id']], 'answer_status': None, 'start_time': int(time.time())}
time.sleep(2)
first.update({'end_time': int(time.time())})
second = {"id": 18851, "answer": {"location": task['coord']}, 'answer_status': None, 'start_time': int(time.time())}
time.sleep(2)
second.update({'end_time': int(time.time())})
item['question_components'].append(first)
item['question_components'].append(second)
else:
#img = send_image(session, headers, task['id'], photos.pop(0))
photo_from_photos = photos.pop(0)
img = send_image(session, headers, task['id'], photo_from_photos)#photos.pop(0))
# count_send_photo = 0
# while img['id'] is None:
# count_send_photo += 1
# print("Попытка № " + str(count_send_photo) + " отправить фото" + str(photo_from_photos))
# img = send_image(session, headers, task['id'], photo_from_photos)#photos.pop(0))
first = {"id": 18856, "answer": 77, 'answer_status': None, 'start_time': int(time.time())}
time.sleep(2)
first.update({'end_time': int(time.time())})
second = {"id": 18857, "answer": [img['id']], 'answer_status': None, 'start_time': int(time.time())}
time.sleep(2)
second.update({'end_time': int(time.time())})
item['question_components'].append(first)
item['question_components'].append(second)
ans = {"question_chains": data}
# print(ans)
session_t = session
session_t.headers.update(
{
'referer': 'https://knd.mosreg.ru/executions'+str(task['id'])+'/solve',
'x-platform': 'wb',
'accept': '*/*'
}
)
complete = session_t.post('https://knd.mosreg.ru//api/v1/executions/'+str(task['id'])+'/answers', headers=headers, json=ans)
if complete.status_code == 200:
print(f'Задание по адресу {address} выполнено успешно!')
else:
print(f'Задание по адресу {address} не выполненно! Статус ошибки {complete.status_code}')
else:
print(f"Несоответствие количества вопросов и количества фотографий по адресу: {address}")
# send_execution(task,session, headers)
def get_photo(address, email):
photos = []
month = datetime.datetime.now().month
main_path = os.path.join(os.path.abspath('.'), 'data', email, str(month), address)
if not os.path.exists(main_path):
return photos
else:
for images in os.listdir(main_path):
if images.endswith(".jpg") or images.endswith(".jpeg") or images.endswith(".JPG"):
photos.append(main_path + '/' + images)
return photos
def get_addresses(email):
addresses = []
month = datetime.datetime.now().month
main_path = os.path.join(os.path.abspath('.'), 'data')
path = os.walk(os.path.join(main_path, email, str(month)))
if not os.path.exists(os.path.join(main_path, email, str(month))):
return f"Не найдена папка с месяцем у пользователя {email}! Проверьте структуру каталога!"
for d, dirs, files in path:
addresses.append(dirs)
if len(addresses[0]) > 0:
return addresses[0]
else:
return f"Ни одного адреса для пользователя {email} не найдено! Добавьте фото и запустите программу снова!"
# def assign_task(session, headers, task, address):
# url = 'https://knd.mosreg.ru//api/v1/executions/'+str(task['id'])+'/assign'
# task_assign = session.put(url, headers=headers)
# if task_assign.status_code == 200:
# print(f'Получили задани по адресу: {address}')
# complete_task(session, headers, task, address)
# else:
# print(f"Не смогли получить задание {address} {task['id']}!")
def get_task(session, headers):
"""Получаем все доступные задания. Если в переменную search добавить адрес,
то будет искать этот адрес. Он попадет в
json_tasks['executions_available'],
так что надо понять откуда берем задания. Отсюда можно возвращать
все параметры задания:
1 координаты
2 адрес
3 подъезд
4 id_задания"""
desinfection = 'Дезинфекция подъездов (МОП)'
proverka = 'Систематическая проверка подъездов МКД'
proverka_monthly = 'Ежемесячная проверка подъездов МКД'
# Получаем адреса
addresses = get_addresses(headers['uid'])
# Если адреса получили
# {"filters":[{"value":{"distance":"50000","lat":55.81999969482422,"lon":37.380001068115234},"filter_attribute":10,"id":6,"name":"Локация","type":"distance"}]}
if isinstance(addresses, list):
session_t = session
session_t.headers.update(
{'referer': 'https://knd.mosreg.ru/executions',
'x-platform': 'android',
'accept': '*/*'
}
)
for address in addresses:
# change '-' for '/'
# need split enterance from address
addr = address.replace("Чаиковского", "Чайковского")
addr = address.replace("Маиданово", "Майданово")
search = addr.replace("_", "/")[:-4]
# print(search)
# data = {'search': search,
# 'filters': [{'value':{
# 'distance': 25000,
# 'lat': 55.81999969482422,
# 'lon': 37.380001068115234
# },
# 'filter_attribute': 10,
# 'id': 6,
# 'name': 'Локация',
# 'type': 'distance'}]}
data = {'search': search}
tasks = session_t.post('https://knd.mosreg.ru//api/v1/actor/executions',
headers=headers,
data=data
)
json_tasks = tasks.json()
get_t = False
# Если есть доступное задание с таким адресом
exec_assigned = json_tasks['executions_assigned'] # Уже полученные задания
# print(exec_assigned)
exec_available = json_tasks['executions_available'] # Еще не принятые задания
# print(exec_available)
# print(address)
if len(exec_assigned) > 0:
for n in exec_assigned:
coord = n['coord']
enterances = n['tasks']
for execution in enterances:
address_from_tasks = execution['dimensions'][0].get('entity_value_name')
name_from_tasks = execution['name']
# print(name_from_tasks)
if name_from_tasks == proverka_monthly:
enterance = execution['dimensions'][2].get('entity_value_name')
if address_from_tasks == 'Клин г, '+ addr.replace("_", "/")[:-4] and \
enterance[-1] == address[-2]:
print('\n')
print(name_from_tasks)
print('\n')
task = {
'id': execution['execution_id'],
'address': execution['dimensions'][0].get('entity_value_name'),
'enterance': execution['dimensions'][2].get('entity_value_name'),
'coord': coord
}
if name_from_tasks == desinfection:
pass
# des.assign_task(session, headers, task, address)
elif name_from_tasks == proverka_monthly:
monthly_prov.assign_task(session, headers, task, address)
else:
pass
else:
print(f"Нет доступных заданий для пользователя {headers['uid']} по адресу: {address}")
# Если адреса не получили выводим ошибку
else:
print(addresses)
def main():
users = Users()
logins = users.get_passwords()
if isinstance(logins, list):
url = 'https://knd.mosreg.ru//api/v1/auth/sign_in'
for user in logins:
session = requests.Session()
login = {
'email': user['email'],
'password': user['password']
}
response = session.post(url, data=login)
headers = {
'client': response.headers.get('client'),
'Access-token': response.headers.get('Access-token'),
'uid': response.headers.get('uid'),
}
if response.status_code == 200:
get_task(session, headers)
else:
print(f"Отказ в авторизации для пользователя {user['email']}!")
time.sleep(20)
else:
print(users.get_passwords())
if __name__ == "__main__":
main()
k = input("Press ENTER for exit")
def random_string(stringLength):
letters = string.ascii_letters
digits = string.digits
return ''.join(random.choice(letters+digits) for i in range(stringLength))

162
knd_dip.py
View File

@ -1,162 +0,0 @@
import requests
import random
import string
# import json
import datetime
import time
import os
from classes.users import Users
from classes.users_led import Users_led
from classes.users_dip import Users_dip
import classes.desinfection as des
import classes.proverka as prov
import classes.led as led
import classes.dip as osmotr_dip
def get_addresses(email):
addresses = []
# month = datetime.datetime.now().month
main_path = os.path.join(os.path.abspath('.'), 'data_dip')
path = os.walk(os.path.join(main_path, email))
# if not os.path.exists(os.path.join(main_path, email, str(month))):
# return f"Не найдена папка с месяцем у пользователя {email}! Проверьте \
# структуру каталога!"
# print(path)
for d, dirs, files in path:
addresses.append(dirs)
if len(addresses[0]) > 0:
return addresses[0]
else:
return f"Ни одного адреса для пользователя {email} не найдено! \
Добавьте фото и запустите программу снова!"
def get_task(session, headers):
"""Получаем все доступные задания. Если в переменную search добавить адрес,
то будет искать этот адрес. Он попадет в
json_tasks['executions_available'],
так что надо понять откуда берем задания. Отсюда можно возвращать
все параметры задания:
1 координаты
2 адрес
3 подъезд
4 id_задания"""
desinfection = 'Дезинфекция подъездов (МОП)'
proverka = 'Систематическая проверка подъездов МКД'
naled = 'Проверка образования наледи на кровле'
dip = 'Осмотр ДИП/ФОП'
s = 'ГЖИ. Системная проверка МКД '
# Получаем адреса
addresses = get_addresses(headers['uid'])
# Если адреса получили
if isinstance(addresses, list):
session_t = session
session_t.headers.update(
{'referer': 'https://knd.mosreg.ru/executions',
'x-platform': 'android',
'accept': '*/*'
}
)
for address in addresses:
# change '-' for '/'
# need split enterance from address
addr = address.replace("Чаиковского", "Чайковского")
addr = address.replace("Маиданово", "Майданово")
addr = address.replace("Молодежныи", "Молодежный")
addr = addr.replace("Ломоносовскии", "Ломоносовский")
search = addr.replace("_", "/")
# print(search)
data = {'search': search}
tasks = session_t.post('https://knd.mosreg.ru//api/v1/actor/executions',
headers=headers,
data=data
)
json_tasks = tasks.json()
# print(json_tasks)
# get_t = False
# Если есть доступное задание с таким адресом
#
# Уже полученные задания
print(json_tasks)
exec_assigned = json_tasks['executions_assigned']
print(exec_assigned)
# Еще не принятые задания
exec_available = json_tasks['executions_available']
# print(exec_available)
# print(address)
if len(exec_assigned) > 0:
# print(len(exec_assigned))
for n in exec_assigned:
coord = n['coord']
enterances = n['tasks']
for execution in enterances:
address_from_tasks = execution['dimensions'][1].get('entity_value_name')
print(address_from_tasks)
name_from_tasks = execution['name']
# if name_from_tasks == dip:
if name_from_tasks == s:
# enterance = execution['dimensions'][2].get('entity_value_name')
if address_from_tasks == 'г. Клин, ' + addr.replace("_", "/"):
print('\n')
print(name_from_tasks)
print('\n')
task = {
'id': execution['execution_id'],
'address': execution['dimensions'][1].get('entity_value_name'),
'coord': coord
}
if name_from_tasks == desinfection:
# des.assign_task(session, headers, task, address)
pass
elif name_from_tasks == proverka:
pass
# prov.assign_task(session, headers, task, address)
elif name_from_tasks == naled:
pass
# led.assign_task(session, headers, task, address)
elif name_from_tasks == dip:
# osmotr_dip.assign_task(session, headers, task, address)
pass
elif name_from_tasks == s:
osmotr_dip.assign_task(session, headers, task, address)
else:
pass
else:
print(f"Нет доступных заданий для пользователя {headers['uid']} по адресу: {address}")
# Если адреса не получили выводим ошибку
else:
print(addresses)
def main():
users = Users_dip()
logins = users.get_passwords()
if isinstance(logins, list):
url = 'https://knd.mosreg.ru//api/v1/auth/sign_in'
for user in logins:
session = requests.Session()
login = {
'email': user['email'],
'password': user['password']
}
response = session.post(url, data=login)
headers = {
'client': response.headers.get('client'),
'Access-token': response.headers.get('Access-token'),
'uid': response.headers.get('uid'),
}
if response.status_code == 200:
print(f"Пользователь {user['email']} успешно авторизовался!")
# print(response.headers)
get_task(session, headers)
else:
print(f"Отказ в авторизации для пользователя {user['email']}!")
time.sleep(1)
else:
print(users.get_passwords())
if __name__ == "__main__":
main()
k = input("Press ENTER for exit")

162
knd_dip2.py
View File

@ -1,162 +0,0 @@
import requests
import random
import string
# import json
import datetime
import time
import os
from classes.users import Users
from classes.users_led import Users_led
from classes.users_dip import Users_dip
import classes.desinfection as des
import classes.proverka as prov
import classes.led as led
import classes.dip as osmotr_dip
def get_addresses(email):
addresses = []
# month = datetime.datetime.now().month
main_path = os.path.join(os.path.abspath('.'), 'data_dip')
path = os.walk(os.path.join(main_path, email))
# if not os.path.exists(os.path.join(main_path, email, str(month))):
# return f"Не найдена папка с месяцем у пользователя {email}! Проверьте \
# структуру каталога!"
# print(path)
for d, dirs, files in path:
addresses.append(dirs)
if len(addresses[0]) > 0:
return addresses[0]
else:
return f"Ни одного адреса для пользователя {email} не найдено! \
Добавьте фото и запустите программу снова!"
def get_task(session, headers):
"""Получаем все доступные задания. Если в переменную search добавить адрес,
то будет искать этот адрес. Он попадет в
json_tasks['executions_available'],
так что надо понять откуда берем задания. Отсюда можно возвращать
все параметры задания:
1 координаты
2 адрес
3 подъезд
4 id_задания"""
desinfection = 'Дезинфекция подъездов (МОП)'
proverka = 'Систематическая проверка подъездов МКД'
naled = 'Проверка образования наледи на кровле'
dip = 'Осмотр ДИП/ФОП'
s = 'ГЖИ. Системная проверка МКД'
# Получаем адреса
addresses = get_addresses(headers['uid'])
# Если адреса получили
# if isinstance(addresses, list):
session_t = session
session_t.headers.update(
{'referer': 'https://knd.mosreg.ru/executions',
'x-platform': 'android',
'accept': '*/*'
}
)
for address in addresses:
# change '-' for '/'
# need split enterance from address
addr = address.replace("Чаиковского", "Чайковского")
addr = address.replace("Маиданово", "Майданово")
addr = address.replace("Молодежныи", "Молодежный")
addr = addr.replace("Ломоносовскии", "Ломоносовский")
search = addr.replace("_", "/")
# print(search)
data = {'search': search}
tasks = session_t.post('https://knd.mosreg.ru//api/v1/actor/executions',
headers=headers
# data=data
)
json_tasks = tasks.json()
# print(json_tasks)
# get_t = False
# Если есть доступное задание с таким адресом
#
# Уже полученные задания
# print(json_tasks)
exec_assigned = json_tasks['executions_assigned']
# print(exec_assigned)
# Еще не принятые задания
exec_available = json_tasks['executions_available']
# print(exec_available)
# print(address)
if len(exec_assigned) > 0:
# print(len(exec_assigned))
for n in exec_assigned:
coord = n['coord']
enterances = n['tasks']
for execution in enterances:
address_from_tasks = execution['dimensions'][1].get('entity_value_name')
# print(address_from_tasks)
name_from_tasks = execution['name']
# if name_from_tasks == dip:
# if name_from_tasks == s:
# # enterance = execution['dimensions'][2].get('entity_value_name')
# if address_from_tasks == 'г. Клин, ' + addr.replace("_", "/"):
# print('\n')
# print(name_from_tasks)
# print('\n')
task = {
'id': execution['execution_id'],
'address': execution['dimensions'][1].get('entity_value_name'),
'coord': coord
}
# if name_from_tasks == desinfection:
# # des.assign_task(session, headers, task, address)
# pass
# elif name_from_tasks == proverka:
# pass
# # prov.assign_task(session, headers, task, address)
# elif name_from_tasks == naled:
# pass
# # led.assign_task(session, headers, task, address)
# elif name_from_tasks == dip:
osmotr_dip.assign_task(session, headers, task, address)
# pass
# # elif name_from_tasks == s:
# # osmotr_dip.assign_task(session, headers, task, address)
# else:
# pass
# else:
# print(f"Нет доступных заданий для пользователя {headers['uid']} по адресу: {address}")
# Если адреса не получили выводим ошибку
# else:
# print(addresses)
def main():
users = Users_dip()
logins = users.get_passwords()
if isinstance(logins, list):
url = 'https://knd.mosreg.ru//api/v1/auth/sign_in'
for user in logins:
session = requests.Session()
login = {
'email': user['email'],
'password': user['password']
}
response = session.post(url, data=login)
headers = {
'client': response.headers.get('client'),
'Access-token': response.headers.get('Access-token'),
'uid': response.headers.get('uid'),
}
if response.status_code == 200:
print(f"Пользователь {user['email']} успешно авторизовался!")
# print(response.headers)
get_task(session, headers)
else:
print(f"Отказ в авторизации для пользователя {user['email']}!")
time.sleep(1)
else:
print(users.get_passwords())
if __name__ == "__main__":
main()
k = input("Press ENTER for exit")

195
knd_kp.py
View File

@ -1,195 +0,0 @@
import requests
import random
import string
# import json
import datetime
import time
import os
# from classes.users import Users
# from classes.users_led import Users_led
# from classes.users_dip import Users_dip
from classes.users_kp import Users_kp
import classes.kp as kp
# import classes.desinfection as des
# import classes.proverka as prov
# import classes.led as led
# import classes.dip as osmotr_dip
def get_addresses(email):
addresses = []
# month = datetime.datetime.now().month
main_path = os.path.join(os.path.abspath('.'), 'data_dip')
path = os.walk(os.path.join(main_path, email))
# if not os.path.exists(os.path.join(main_path, email, str(month))):
# return f"Не найдена папка с месяцем у пользователя {email}! Проверьте \
# структуру каталога!"
# print(path)
for d, dirs, files in path:
addresses.append(dirs)
if len(addresses[0]) > 0:
return addresses[0]
else:
return f"Ни одного адреса для пользователя {email} не найдено! \
Добавьте фото и запустите программу снова!"
def get_task(session, headers):
"""Получаем все доступные задания. Если в переменную search добавить адрес,
то будет искать этот адрес. Он попадет в
json_tasks['executions_available'],
так что надо понять откуда берем задания. Отсюда можно возвращать
все параметры задания:
1 координаты
2 адрес
3 подъезд
4 id_задания"""
# desinfection = 'Дезинфекция подъездов (МОП)'
# proverka = 'Систематическая проверка подъездов МКД'
# naled = 'Проверка образования наледи на кровле'
# dip = 'Осмотр ДИП/ФОП'
# s = 'ГЖИ. Системная проверка МКД '
session_t = session
session_t.headers.update(
{'referer': 'https://knd.mosreg.ru/executions',
'x-platform': 'android',
'accept': '*/*'
}
)
tasks = session_t.post('https://knd.mosreg.ru//api/v1/executions',
headers=headers
)
json_tasks = tasks.json()
# print(json_tasks)
exec_assigned = json_tasks['executions_assigned']
if len(exec_assigned) > 0:
for execution in exec_assigned:
coord = execution['coord']
tasks = execution['tasks']
for task in tasks:
task_name = task['name']
print(task_name)
access_task = {
'id': task['execution_id'],
'address': task['dimensions'][1].get('entity_value_name'),
'coord': coord
}
kp.assign_task(session, headers, access_task)
else:
print(f"Нет доступных заданий для пользователя {headers['uid']}!")
# Получаем адреса
# addresses = get_addresses(headers['uid'])
# # Если адреса получили
# if isinstance(addresses, list):
# session_t = session
# session_t.headers.update(
# {'referer': 'https://knd.mosreg.ru/executions',
# 'x-platform': 'android',
# 'accept': '*/*'
# }
# )
# for address in addresses:
# # change '-' for '/'
# # need split enterance from address
# addr = address.replace("Чаиковского", "Чайковского")
# addr = address.replace("Маиданово", "Майданово")
# addr = address.replace("Молодежныи", "Молодежный")
# addr = addr.replace("Ломоносовскии", "Ломоносовский")
# search = addr.replace("_", "/")
# # print(search)
# data = {'search': search}
# tasks = session_t.post('https://knd.mosreg.ru//api/v1/actor/executions',
# headers=headers,
# data=data
# )
# json_tasks = tasks.json()
# # print(json_tasks)
# # get_t = False
# # Если есть доступное задание с таким адресом
# #
# # Уже полученные задания
# print(json_tasks)
# exec_assigned = json_tasks['executions_assigned']
# print(exec_assigned)
# # Еще не принятые задания
# exec_available = json_tasks['executions_available']
# # print(exec_available)
# # print(address)
# if len(exec_assigned) > 0:
# # print(len(exec_assigned))
# for n in exec_assigned:
# coord = n['coord']
# enterances = n['tasks']
# for execution in enterances:
# address_from_tasks = execution['dimensions'][1].get('entity_value_name')
# print(address_from_tasks)
# name_from_tasks = execution['name']
# # if name_from_tasks == dip:
# if name_from_tasks == s:
# # enterance = execution['dimensions'][2].get('entity_value_name')
# if address_from_tasks == 'г. Клин, ' + addr.replace("_", "/"):
# print('\n')
# print(name_from_tasks)
# print('\n')
# task = {
# 'id': execution['execution_id'],
# 'address': execution['dimensions'][1].get('entity_value_name'),
# 'coord': coord
# }
# if name_from_tasks == desinfection:
# # des.assign_task(session, headers, task, address)
# pass
# elif name_from_tasks == proverka:
# pass
# # prov.assign_task(session, headers, task, address)
# elif name_from_tasks == naled:
# pass
# # led.assign_task(session, headers, task, address)
# elif name_from_tasks == dip:
# # osmotr_dip.assign_task(session, headers, task, address)
# pass
# elif name_from_tasks == s:
# osmotr_dip.assign_task(session, headers, task, address)
# else:
# pass
# else:
# print(f"Нет доступных заданий для пользователя {headers['uid']} по адресу: {address}")
# # Если адреса не получили выводим ошибку
# else:
# print(addresses)
def main():
users = Users_kp()
logins = users.get_passwords()
if isinstance(logins, list):
url = 'https://knd.mosreg.ru//api/v1/auth/sign_in'
for user in logins:
session = requests.Session()
login = {
'email': user['email'],
'password': user['password']
}
response = session.post(url, data=login)
headers = {
'client': response.headers.get('client'),
'Access-token': response.headers.get('Access-token'),
'uid': response.headers.get('uid'),
}
if response.status_code == 200:
print(f"Пользователь {user['email']} успешно авторизовался!")
# print(response.headers)
get_task(session, headers)
else:
print(f"Отказ в авторизации для пользователя {user['email']}!")
time.sleep(1)
else:
print(users.get_passwords())
if __name__ == "__main__":
main()
k = input("Press ENTER for exit")

149
naled.py
View File

@ -1,149 +0,0 @@
# import random
# import string
# import json
# import datetime
# import time
import os
import requests
# import classes.desinfection as des
# import classes.led as led
# import classes.proverka as prov
# from classes.users import Users
from classes.users_led import Users_led
def get_addresses(email):
addresses = []
# month = datetime.datetime.now().month
main_path = os.path.join(os.path.abspath('.'), 'data_led')
path = os.walk(os.path.join(main_path, email))
# if not os.path.exists(os.path.join(main_path, email, str(month))):
# return f"Не найдена папка с месяцем у пользователя {email}! Проверьте \
# структуру каталога!"
# print(path)
for d, dirs, files in path:
addresses.append(dirs)
if len(addresses[0]) > 0:
return addresses[0]
else:
return f"Ни одного адреса для пользователя {email} не найдено! \
Добавьте фото и запустите программу снова!"
def get_task(session, headers):
"""Получаем все доступные задания. Если в переменную search добавить адрес,
то будет искать этот адрес. Он попадет в
json_tasks['executions_available'],
так что надо понять откуда берем задания. Отсюда можно возвращать
все параметры задания:
1 координаты
2 адрес
3 подъезд
4 id_задания"""
desinfection = 'Дезинфекция подъездов (МОП)'
proverka = 'Систематическая проверка подъездов МКД'
naled = 'Проверка образования наледи на кровле'
dip = 'Осмотр ДИП/ФОП'
# Получаем адреса
addresses = get_addresses(headers['uid'])
# Если адреса получили
if isinstance(addresses, list):
session_t = session
session_t.headers.update(
{'referer': 'https://knd.mosreg.ru/executions',
'x-platform': 'android',
'accept': '*/*'
}
)
for address in addresses:
# change '-' for '/'
# need split enterance from address
addr = address.replace("Чаиковского", "Чайковского")
addr = address.replace("Маиданово", "Майданово")
search = addr.replace("_", "/")
# print(search)
data = {'search': search}
tasks = session_t.post('https://knd.mosreg.ru//api/v1/actor/executions',
headers=headers,
data=data
)
json_tasks = tasks.json()
# get_t = False
# Если есть доступное задание с таким адресом
#
# Уже полученные задания
exec_assigned = json_tasks['executions_assigned']
# print(exec_assigned)
# Еще не принятые задания
exec_available = json_tasks['executions_available']
# print(exec_available)
# print(address)
if len(exec_assigned) > 0:
for n in exec_assigned:
coord = n['coord']
enterances = n['tasks']
for execution in enterances:
address_from_tasks = execution['dimensions'][0].get('entity_value_name')
# print(address_from_tasks)
name_from_tasks = execution['name']
if name_from_tasks == naled:
enterance = execution['dimensions'][2].get('entity_value_name')
if address_from_tasks == 'Клин г, ' + addr.replace("_", "/"):
print('\n')
print(name_from_tasks)
print('\n')
task = {
'id': execution['execution_id'],
'address': execution['dimensions'][0].get('entity_value_name'),
'enterance': execution['dimensions'][2].get('entity_value_name'),
'coord': coord
}
if name_from_tasks == desinfection:
# des.assign_task(session, headers, task, address)
pass
elif name_from_tasks == proverka:
pass
# prov.assign_task(session, headers, task, address)
elif name_from_tasks == naled:
pass
# led.assign_task(session, headers, task, address)
else:
pass
else:
print(f"Нет доступных заданий для пользователя {headers['uid']} по адресу: {address}")
# Если адреса не получили выводим ошибку
else:
print(addresses)
def main():
users = Users_led()
logins = users.get_passwords()
if isinstance(logins, list):
url = 'https://knd.mosreg.ru//api/v1/auth/sign_in'
for user in logins:
session = requests.Session()
login = {
'email': user['email'],
'password': user['password']
}
response = session.post(url, data=login)
headers = {
'client': response.headers.get('client'),
'Access-token': response.headers.get('Access-token'),
'uid': response.headers.get('uid'),
}
if response.status_code == 200:
get_task(session, headers)
else:
print(f"Отказ в авторизации для пользователя {user['email']}!")
time.sleep(1)
else:
print(users.get_passwords())
if __name__ == "__main__":
main()
k = input("Press ENTER for exit")

1
some.py
View File

@ -1 +0,0 @@
print("some")