110 lines
4.5 KiB
Python
110 lines
4.5 KiB
Python
import asyncio
|
||
import os
|
||
import logging
|
||
import random
|
||
from playwright.async_api import async_playwright
|
||
|
||
# Настройка логов
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(levelname)s - %(message)s',
|
||
handlers=[logging.FileHandler("logs/recon.log"), logging.StreamHandler()]
|
||
)
|
||
logger = logging.getLogger("GIS-Aristoteles-Recon")
|
||
|
||
USER_DATA_DIR = "/app/browser_context"
|
||
TARGET_URL = "https://dom.gosuslugi.ru/"
|
||
OBSIDIAN_BASE = "/app/obsidian/Gemini/Projects/GIS_ZHKH"
|
||
|
||
async def send_tg_report(message):
|
||
token = os.getenv("TELEGRAM_BOT_TOKEN")
|
||
chat_id = os.getenv("TELEGRAM_CHAT_ID")
|
||
if not token or not chat_id: return
|
||
import aiohttp
|
||
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
await session.post(url, json={"chat_id": chat_id, "text": message, "parse_mode": "HTML"})
|
||
except: pass
|
||
|
||
async def deep_scan():
|
||
async with async_playwright() as p:
|
||
logger.info("Launching browser (Aristoteles Mode)...")
|
||
context = await p.chromium.launch_persistent_context(
|
||
user_data_dir=USER_DATA_DIR,
|
||
headless=False,
|
||
args=["--no-sandbox", "--disable-setuid-sandbox", "--start-maximized"]
|
||
)
|
||
|
||
page = await context.new_page()
|
||
await page.set_viewport_size({"width": 1600, "height": 1024})
|
||
|
||
logger.info(f"Navigating to {TARGET_URL}")
|
||
await page.goto(TARGET_URL)
|
||
|
||
os.makedirs(f"{OBSIDIAN_BASE}/screenshots", exist_ok=True)
|
||
|
||
markers = [
|
||
"Информация о договорах по дому",
|
||
"Капитальный ремонт",
|
||
"Итоги общих собраний собственников"
|
||
]
|
||
|
||
found_marker = None
|
||
attempt = 0
|
||
|
||
while not found_marker:
|
||
attempt += 1
|
||
logger.info(f"Scan attempt {attempt}...")
|
||
|
||
# Сохраняем "глазок" в Обсидиан для отладки
|
||
debug_scr = f"{OBSIDIAN_BASE}/screenshots/current_view.png"
|
||
await page.screenshot(path=debug_scr)
|
||
|
||
# Проверяем наличие маркеров
|
||
content = await page.content()
|
||
for m in markers:
|
||
if m in content:
|
||
found_marker = m
|
||
break
|
||
|
||
if found_marker: break
|
||
|
||
if attempt % 10 == 0:
|
||
await send_tg_report(f"⏳ <b>Жду входа...</b> (Попытка {attempt})\nЯ смотрю на экран и не вижу разделов жильца. Проверь файл <code>current_view.png</code> в Обсидиане.")
|
||
|
||
await asyncio.sleep(10)
|
||
|
||
await send_tg_report(f"🎯 <b>Цель захвачена!</b> Увидел: <code>{found_marker}</code>.\nНачинаю тотальное картирование всех кнопок.")
|
||
|
||
# 1. Снимаем финальный вид
|
||
await page.screenshot(path=f"{OBSIDIAN_BASE}/screenshots/dashboard_full.png", full_page=True)
|
||
|
||
# 2. Собираем элементы (Исправленный CSS селектор!)
|
||
elements = await page.query_selector_all("a, button, [role='button']")
|
||
|
||
report_lines = ["# 🏛️ ТОТАЛЬНАЯ КАРТА ЛК: ТЫРИН АЛЕКСАНДР\n"]
|
||
report_lines.append(f"Дата сканирования: {asyncio.get_event_loop().time()}\n")
|
||
report_lines.append("![[Gemini/Projects/GIS_ZHKH/screenshots/dashboard_full.png]]\n")
|
||
report_lines.append("## 🔍 Все найденные интерактивные элементы:\n")
|
||
|
||
seen = set()
|
||
for el in elements:
|
||
try:
|
||
txt = await el.inner_text()
|
||
txt = txt.strip().replace("\n", " ")
|
||
if txt and len(txt) > 2 and txt not in seen:
|
||
seen.add(txt)
|
||
report_lines.append(f"- `{txt}`")
|
||
except: continue
|
||
|
||
with open(f"{OBSIDIAN_BASE}/TOTAL_MAP.md", "w", encoding="utf-8") as f:
|
||
f.write("\n".join(report_lines))
|
||
|
||
await send_tg_report("🏆 <b>Миссия выполнена!</b>\n\nФайл <code>TOTAL_MAP.md</code> и скриншоты в папке проекта в Obsidian.")
|
||
|
||
while True:
|
||
await asyncio.sleep(3600)
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(deep_scan())
|