import asyncio import os import logging import random import re from datetime import datetime from playwright.async_api import async_playwright # Настройка логов logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[logging.FileHandler("logs/contracts.log"), logging.StreamHandler()] ) logger = logging.getLogger("GIS-Titanium-Harvester") USER_DATA_DIR = "/app/browser_context" OBSIDIAN_BASE = "/app/obsidian/Gemini/Projects/GIS_ZHKH" NAS_DOWNLOADS = "/app/downloads/nas/contracts" processed_contracts = set() 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 automation_mission(): async with async_playwright() as p: logger.info("Launching Titanium Harvester (Text-Based State Machine)...") 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}) os.makedirs(f"{NAS_DOWNLOADS}", exist_ok=True) os.makedirs(f"{OBSIDIAN_BASE}/screenshots", exist_ok=True) os.makedirs(f"{OBSIDIAN_BASE}/contracts_data/evidence", exist_ok=True) await page.goto("https://dom.gosuslugi.ru/") await send_tg_report("🤖 Титановый Жнец запущен.\nАрхитектура перестроена. Теперь я не теряю ссылки из памяти. Заходи в список договоров.") while True: try: await asyncio.sleep(5) # Глазок для Обсидиана await page.screenshot(path=f"{OBSIDIAN_BASE}/screenshots/contracts_view.png") frames = [page] + page.frames full_text = "" for f in frames: try: full_text += await f.content() except: continue # --- 1. ПРОВЕРКА: МЫ ВНУТРИ КАРТОЧКИ? --- if "Карточка договора" in full_text or "Документы договора" in full_text: logger.info("State: INSIDE CONTRACT") # Пытаемся понять, в каком мы договоре contract_name = "Unknown_Contract" for f in frames: try: h1 = await f.query_selector("h1, .page-header, .document-title") if h1: contract_name = (await h1.inner_text()).strip().split("\n")[0] break except: continue safe_name = "".join([c for c in contract_name if c.isalnum() or c in (' ', '_', '-')]).strip().replace(" ", "_") if safe_name not in processed_contracts: await send_tg_report(f"🎣 Начинаю выкачку: {contract_name[:50]}...") # Скриншот scr_name = f"evidence_{safe_name}.png" await page.screenshot(path=f"{OBSIDIAN_BASE}/contracts_data/evidence/{scr_name}", full_page=True) # Агрессивный поиск файлов downloaded_files = [] for f in frames: selectors = [ "a[title*='Скачать']", "button:has-text('Скачать')", "a:has-text('Скачать все')", "a[href*='.pdf']", "a[href*='.zip']", "a[href*='.docx']" ] for s in selectors: try: elements = await f.locator(s).all() for btn in elements: if not await btn.is_visible(): continue try: async with page.expect_download(timeout=30000) as di: await btn.click() download = await di.value fname = download.suggested_filename dest = f"{NAS_DOWNLOADS}/{safe_name}/{fname}" os.makedirs(os.path.dirname(dest), exist_ok=True) await download.save_as(dest) downloaded_files.append(f"- `{fname}`") logger.info(f"Downloaded: {fname}") except Exception as dl_e: logger.error(f"Download err: {dl_e}") except: continue # Пишем отчет with open(f"{OBSIDIAN_BASE}/contracts_data/{safe_name}.md", "w", encoding="utf-8") as of: of.write(f"# {contract_name}\n\n![[Gemini/Projects/GIS_ZHKH/contracts_data/evidence/{scr_name}]]\n\n") if downloaded_files: of.write("## 📄 Скачанные документы\n" + "\n".join(downloaded_files)) else: of.write("## 📄 Скачанные документы\nФайлы не обнаружены. Скриншот зафиксирован.") processed_contracts.add(safe_name) if downloaded_files: await send_tg_report(f"✅ Успешно: Файлов собрано: {len(downloaded_files)}. Возвращаюсь к списку.") else: await send_tg_report(f"⚠️ Внимание: В договоре нет файлов. Скриншот сделан. Иду дальше.") # Возвращаемся в список logger.info("Returning to list...") found_back = False for f in frames: try: back_btn = f.locator("text='Информация о договорах по дому'").first if await back_btn.is_visible(timeout=1000): await back_btn.click() found_back = True break except: continue if not found_back: await page.go_back() await asyncio.sleep(8) # Ждем прогрузки списка continue # --- 2. ПРОВЕРКА: МЫ В СПИСКЕ? --- if "Результаты поиска" in full_text: logger.info("State: IN LIST") # Ищем тексты всех договоров на экране available_contracts_texts = [] for f in frames: try: # Ищем все элементы, содержащие слово Договор elements = await f.locator("text=/Договор.*/i").all() for el in elements: if await el.is_visible(): txt = (await el.inner_text()).strip() # Жесткий фильтр мусора if txt.startswith("Договор") and "№" in txt and "Закон" not in txt and len(txt) > 15: if txt not in available_contracts_texts: available_contracts_texts.append(txt) except: continue if not available_contracts_texts: logger.info("List detected, but no valid contract titles found yet.") continue # Ищем первый необработанный target_text = None for txt in available_contracts_texts: s_name = "".join([c for c in txt if c.isalnum() or c in (' ', '_', '-')]).strip().replace(" ", "_") if s_name not in processed_contracts: target_text = txt break if target_text: await send_tg_report(f"🚀 Цель найдена: {target_text[:40]}...\nКликаю!") # Кликаем именно по тексту, заново его находя clicked = False for f in frames: try: btn = f.locator(f"text='{target_text}'").first if await btn.is_visible(timeout=1000): await btn.scroll_into_view_if_needed() await btn.click() clicked = True break except: continue if clicked: await asyncio.sleep(8) # Ждем прогрузки карточки else: logger.error(f"Failed to click target text: {target_text}") else: logger.info("All contracts on this page are processed.") # --- 3. НЕИЗВЕСТНОЕ СОСТОЯНИЕ --- else: logger.info("State: UNKNOWN. Waiting for user action.") except Exception as e: logger.error(f"Main Loop Error: {e}") await asyncio.sleep(10) if __name__ == "__main__": asyncio.run(automation_mission())