dddennnisss/trueconf_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

photo_bot: remove AI validator, restore prepare_ad_photo workflow

Browse Source
This commit is contained in:
Денис Кривоченко
2026-08-14 00:54:30 +07:00
parent 95803b2c3c
commit 7b129e0836
15 changed files with 1051 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service_desk/__pycache__/handlers.cpython-311.pyc
BIN
View File
Binary file not shown.
service_desk/test_basic.py
+33
View File
@@ -0,0 +1,33 @@
#!/usr/bin/env python3
import asyncio
import httpx
import json
import urllib3
BASE_URL = "https://192.168.1.237:8080"
API_TOKEN = "062CE056-C1F5-490C-A6DF-FD2C653A71DD"
async def main():
url = f"{BASE_URL}/api/v3/requests"
headers = {
"authtoken": API_TOKEN,
"Accept": "application/vnd.manageengine.sdp.v3+json"
}
async with httpx.AsyncClient(verify=False, timeout=30.0) as client:
resp = await client.get(url, headers=headers, params={
"input_data": json.dumps({
"list_info": {"row_count": 100, "start_index": 1}
})
})
data = resp.json()
print(f"Status: {resp.status_code}")
print(f"Has more: {data.get('list_info', {}).get('has_more_rows')}")
requests = data.get("requests", [])
print(f"Total: {len(requests)}")
for r in requests:
print(f" ID={r.get('id')} subject={r.get('subject')} created={r.get('created_time')}")
if __name__ == "__main__":
urllib3.disable_warnings()
asyncio.run(main())
service_desk/test_date_format.py
+40
View File
@@ -0,0 +1,40 @@
#!/usr/bin/env python3
"""Проверка формата даты и вариантов фильтрации."""
import asyncio
import httpx
import json
import urllib3
BASE_URL = "https://192.168.1.237:8080"
API_TOKEN = "062CE056-C1F5-490C-A6DF-FD2C653A71DD"
async def debug():
url = f"{BASE_URL}/api/v3/requests"
headers = {
"authtoken": API_TOKEN,
"Accept": "application/vnd.manageengine.sdp.v3+json"
}
# Вариант 1: без фильтра, посмотреть все поля
print("=== Все заявки (без фильтра) ===")
async with httpx.AsyncClient(verify=False, timeout=30.0) as client:
resp = await client.get(url, headers=headers, params={
"input_data": json.dumps({
"list_info": {"row_count": 100, "start_index": 1}
})
})
data = resp.json()
requests = data.get("requests", [])
for req in requests:
print(f"\n--- Заявка {req.get('request_id', '?')} ---")
print(f" subject: {req.get('subject')}")
print(f" created_time: {req.get('created_time')}")
print(f" created_time type: {type(req.get('created_time')).__name__}")
# Показать все ключи
print(f" Ключи: {list(req.keys())}")
if __name__ == "__main__":
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
asyncio.run(debug())
service_desk/test_filter.py
+88
View File
@@ -0,0 +1,88 @@
#!/usr/bin/env python3
"""Поиск рабочего формата фильтрации."""
import asyncio
import httpx
import json
import urllib3
BASE_URL = "https://192.168.1.237:8080"
API_TOKEN = "062CE056-C1F5-490C-A6DF-FD2C653A71DD"
async def test_filters():
url = f"{BASE_URL}/api/v3/requests"
headers = {
"authtoken": API_TOKEN,
"Accept": "application/vnd.manageengine.sdp.v3+json"
}
# Сначала посмотрим все заявки с их ID и датами
print("=== Все заявки ===")
async with httpx.AsyncClient(verify=False, timeout=30.0) as client:
resp = await client.get(url, headers=headers, params={
"input_data": json.dumps({
"list_info": {"row_count": 100, "start_index": 1}
})
})
data = resp.json()
requests = data.get("requests", [])
for req in requests:
req_id = req.get("request_id", req.get("id", "?"))
subject = req.get("subject", "?")
created = req.get("created_time", {})
if isinstance(created, dict):
created_str = created.get("display_value", "?")
ts = created.get("value", "?")
else:
created_str = str(created)
ts = "?"
print(f" ID={req_id} created={created_str} ts={ts}")
# Попробуем разные форматы фильтрации
print("\n=== Попробуем фильтры ===")
filters = [
# Вариант 1: ISO строка
{
"search_criteria": {
"column_name": "created_time",
"condition": "between",
"values": ["Jul 1, 2026 00:00:00", "Jul 31, 2026 23:59:59"]
}
},
# Вариант 2: без search_criteria, с custom_field
{
"search_criteria": {
"column_name": "created_time",
"condition": "between",
"values": ["2026-07-01", "2026-07-31"]
}
},
# Вариант 3: timestamp как строка
{
"search_criteria": {
"column_name": "created_time",
"condition": "between",
"values": ["1782864000000", "1785542399000"]
}
},
]
for i, criteria in enumerate(filters):
params = {
"input_data": json.dumps({
"list_info": {"row_count": 100, "start_index": 1},
"search_criteria": criteria.get("search_criteria")
})
}
resp = await client.get(url, headers=headers, params=params)
data = resp.json()
reqs = data.get("requests", [])
print(f"\nФильтр {i+1}: {criteria['search_criteria']['values']} -> {len(reqs)} заявок")
for r in reqs:
print(f" {r.get('id')}: {r.get('subject')}")
if __name__ == "__main__":
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
asyncio.run(test_filters())
service_desk/test_list_tickets.py
+80
View File
@@ -0,0 +1,80 @@
#!/usr/bin/env python3
"""Получение списка заявок из ServiceDesk Plus через API v3."""
import asyncio
import httpx
import json
import urllib3
BASE_URL = "https://192.168.1.237:8080"
API_TOKEN = "062CE056-C1F5-490C-A6DF-FD2C653A71DD"
async def list_tickets():
url = f"{BASE_URL}/api/v3/requests"
headers = {
"authtoken": API_TOKEN,
"Accept": "application/vnd.manageengine.sdp.v3+json"
}
params = {
"input_data": json.dumps({
"list_info": {
"row_count": 50,
"start_index": 1
}
})
}
async with httpx.AsyncClient(verify=False, timeout=30.0) as client:
resp = await client.get(url, headers=headers, params=params)
data = resp.json()
requests = data.get("requests", [])
print(f"Всего заявок: {len(requests)}\n")
for req in requests:
req_id = req.get("request_id", req.get("id", "?"))
subject = req.get("subject", req.get("short_description", "?"))
# Группа
group = req.get("group", {})
if isinstance(group, dict):
group_name = group.get("name", "?")
else:
group_name = str(group)
# Статус
status = req.get("status", {})
if isinstance(status, dict):
status_name = status.get("name", "?")
else:
status_name = str(status)
# Приоритет
priority = req.get("priority", {})
if isinstance(priority, dict):
priority_name = priority.get("name", "?")
else:
priority_name = str(priority)
# Описание (убираем HTML)
desc = req.get("description", "")
desc_clean = desc.replace("<p>", "").replace("</p>", "").replace("<br>", "\n").replace("<br/>", "\n").replace("<br />", "\n")
desc_clean = desc_clean[:300]
# Создатель
requester = req.get("requester", {})
req_name = requester.get("name", "?") if isinstance(requester, dict) else "?"
print(f"--- Заявка #{req_id} ---")
print(f" Тема: {subject}")
print(f" Группа: {group_name}")
print(f" Статус: {status_name}")
print(f" Приоритет: {priority_name}")
print(f" Автор: {req_name}")
print(f" Описание:\n {desc_clean}")
print()
if __name__ == "__main__":
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
asyncio.run(list_tickets())
service_desk/test_list_tickets_debug.py
+57
View File
@@ -0,0 +1,57 @@
#!/usr/bin/env python3
"""Дебаг запроса к ServiceDesk Plus API v3."""
import asyncio
import httpx
import json
import urllib3
BASE_URL = "https://192.168.1.237:8080"
API_TOKEN = "062CE056-C1F5-490C-A6DF-FD2C653A71DD"
async def debug():
url = f"{BASE_URL}/api/v3/requests"
headers = {
"authtoken": API_TOKEN,
"Accept": "application/vnd.manageengine.sdp.v3+json"
}
# Разные варианты формата
variants = [
# Вариант 1: input_data как JSON
{
"input_data": json.dumps({
"list_info": {
"row_count": 5,
"start_index": 1
}
})
},
# Вариант 2: без input_data, просто параметры
None,
# Вариант 3: input_data без json.dumps
{
"input_data": {
"list_info": {
"row_count": 5,
"start_index": 1
}
}
}
]
for i, params in enumerate(variants):
print(f"\n=== Вариант {i+1} ===")
print(f"Params: {params}")
async with httpx.AsyncClient(verify=False, timeout=15.0) as client:
try:
resp = await client.get(url, headers=headers, params=params)
print(f"HTTP {resp.status_code}")
print(f"Response: {resp.text[:500]}")
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
asyncio.run(debug())
service_desk/test_list_tickets_july.py
+97
View File
@@ -0,0 +1,97 @@
#!/usr/bin/env python3
"""Получение списка заявок из ServiceDesk Plus с 20 июля 2026. Экспорт: Группа, Описание, Назначено, Категория в JSONL."""
import asyncio
import httpx
import json
import urllib3
from datetime import datetime, timezone
from html import unescape
import re
BASE_URL = "https://192.168.1.237:8080"
API_TOKEN = "062CE056-C1F5-490C-A6DF-FD2C653A71DD"
FROM_DATE_MS = int(datetime(2026, 7, 20, 0, 0, 0, tzinfo=timezone.utc).timestamp() * 1000)
def get_ts(req):
ct = req.get("created_time", {})
if isinstance(ct, dict):
v = ct.get("value", "0")
return int(v) if v else 0
return 0
def clean_html(text):
if not text:
return ""
text = re.sub(r'<[^>]+>', '', text)
text = unescape(text)
return text.strip()
def get_name(obj):
"""Извлечь имя из объекта группы/техника/категории."""
if not obj:
return ""
if isinstance(obj, dict):
return obj.get("name", "") or obj.get("display_value", "")
return str(obj)
async def fetch_ticket_detail(client, req_id):
try:
resp = await client.get(
f"{BASE_URL}/api/v3/requests/{req_id}",
headers={"authtoken": API_TOKEN, "Accept": "application/vnd.manageengine.sdp.v3+json"},
timeout=15.0
)
data = resp.json()
return data.get("request", data)
except Exception as e:
return {}
async def list_tickets():
url = f"{BASE_URL}/api/v3/requests"
headers = {"authtoken": API_TOKEN, "Accept": "application/vnd.manageengine.sdp.v3+json"}
async with httpx.AsyncClient(verify=False, timeout=30.0) as client:
resp = await client.get(url, headers=headers)
data = resp.json()
all_reqs = data.get("requests", [])
from_20 = [r for r in all_reqs if get_ts(r) >= FROM_DATE_MS]
# Получаем детали для каждой заявки
tasks = [fetch_ticket_detail(client, r.get("request_id", r.get("id", "?"))) for r in from_20]
details = await asyncio.gather(*tasks)
rows = []
for req, detail in zip(from_20, details):
# Группа — из списка (есть там)
group = req.get("group", {})
group_name = get_name(group)
# Описание — из детального запроса
desc = detail.get("description", "")
desc_clean = clean_html(desc)
# Назначено (специалист) — из детального запроса
tech = detail.get("technician", {})
assignee = get_name(tech)
# Категория — из детального запроса
cat = detail.get("category", {})
category = get_name(cat)
rows.append({
"group": group_name,
"description": desc_clean,
"assignee": assignee,
"category": category
})
# Выводим JSONL
for row in rows:
print(json.dumps(row, ensure_ascii=False))
if __name__ == "__main__":
urllib3.disable_warnings()
asyncio.run(list_tickets())