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

small visual changes

Browse Source
This commit is contained in:
norets 2024-12-22 07:42:15 +03:00
parent ba742f7d21
commit b2fed9f2f8
2 changed files with 74 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

125
des.py
View File

@ -8,31 +8,38 @@ import time
import requests import requests
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 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('https://knd.mosreg.ru//api/v1/executions/'+str(url) +
'/images', headers=headers, files=files) img = session.post(
"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(f"Не удалось загрузить фото на сервер по заданию {url}, \ print(
статус ошибки {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
@ -40,23 +47,24 @@ def send_image(session, headers, url, photo):
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), main_path = os.path.join(os.path.abspath("."), "data", email, str(month), address)
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 (images.endswith(".jpg") or if (
images.endswith(".jpeg") or images.endswith(".jpg")
images.endswith(".JPG")): or images.endswith(".jpeg")
photos.append(main_path + '/' + images) 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}! Проверьте \
@ -80,18 +88,19 @@ 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', {
'x-platform': 'android', "referer": "https://knd.mosreg.ru/executions",
'accept': '*/*' "x-platform": "android",
} "accept": "*/*",
}
) )
for address in addresses: for address in addresses:
# change '-' for '/' # change '-' for '/'
@ -100,46 +109,56 @@ 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} data = {"search": search}
tasks = session_t.post('https://knd.mosreg.ru//api/v1/actor/executions', tasks = session_t.post(
headers=headers, "https://knd.mosreg.ru//api/v1/actor/executions",
data=data headers=headers,
) 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"]
print(exec_assigned) print(exec_assigned)
# Еще не принятые задания # Еще не принятые задания
exec_available = json_tasks['executions_available'] exec_available = json_tasks["executions_available"]
print(exec_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]: )
print('\n') if (
address_from_tasks
== "Клин г, " + addr.replace("_", "/")[:-4]
and enterance[-1] == address[-2]
):
print("\n")
print(name_from_tasks) print(name_from_tasks)
print('\n') print("\n")
task = { task = {
'id': execution['execution_id'], "id": execution["execution_id"],
'address': execution['dimensions'][0].get('entity_value_name'), "address": execution["dimensions"][0].get(
'enterance': execution['dimensions'][2].get('entity_value_name'), "entity_value_name"
'coord': coord ),
"enterance": execution["dimensions"][2].get(
"entity_value_name"
),
"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)
@ -147,7 +166,8 @@ def get_task(session, headers):
pass pass
else: else:
print( print(
f"Нет доступных заданий для пользователя {headers['uid']} по адресу: {address}") f"Нет доступных заданий для пользователя {headers['uid']} по адресу: {address}"
)
# Если адреса не получили выводим ошибку # Если адреса не получили выводим ошибку
else: else:
print(addresses) print(addresses)
@ -157,18 +177,15 @@ 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 = { login = {"email": user["email"], "password": user["password"]}
'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)
@ -187,4 +204,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))

5
knd_bot.py
View File

@ -49,12 +49,12 @@ def complete_task(session, headers, task, address):
id_task = task["id"] id_task = task["id"]
questions = get_questions(id_task, session, headers) questions = get_questions(id_task, session, headers)
data = [] data = []
comp = []
for answers in questions: for answers in questions:
answer_chain = {"id": answers["id"], "questions": []} # answers['questions'] answer_chain = {"id": answers["id"], "questions": []} # answers['questions']
if answer_chain["id"] == 6146: # Цепочка без ответов if answer_chain["id"] == 6146: # Цепочка без ответов
continue continue
for components in answers["questions"]: for components in answers["questions"]:
comp = []
q_c = {"id": components["id"], "question_components": []} q_c = {"id": components["id"], "question_components": []}
if q_c["id"] == 6560: # Вопрос без ответа if q_c["id"] == 6560: # Вопрос без ответа
continue continue
@ -126,7 +126,8 @@ def complete_task(session, headers, task, address):
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": "*/*",