service-finder/backend/app/workers/vehicle/vehicle_robot_2_researcher.py.old

# /opt/docker/dev/service_finder/backend/app/workers/researcher_v2_1.py
import asyncio
import logging
import warnings
import os
from datetime import datetime, timezone
from typing import Optional, List
from sqlalchemy import select, update, and_, func, or_, case
from app.db.session import AsyncSessionLocal
from app.models.vehicle_definitions import VehicleModelDefinition

# DuckDuckGo search API hiba-elnyomás és import
warnings.filterwarnings("ignore", category=RuntimeWarning, module='duckduckgo_search')
from duckduckgo_search import DDGS

# Logolás beállítása
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(name)s: %(message)s')
logger = logging.getLogger("Robot-Researcher-v2.1")

class ResearcherBot:
    """ 
    Robot 2.1: Az internet porszívója. 
    Technikai adatokat gyűjt (DuckDuckGo), hogy előkészítse az AI dúsítást.
    Kihasználja a motorkódot és a gyártási évet a pontosabb találatokért.
    """
    def __init__(self):
        self.batch_size = 5 # Egyszerre 5 járművet vesz ki
        self.max_parallel_queries = 3 # Párhuzamos keresések száma

    async def fetch_source(self, label: str, query: str) -> str:
        """ Egyedi forrás lekérése szálbiztos módon. """
        try:
            def search():
                with DDGS() as ddgs:
                    # Az első 3 találat body részét gyűjtjük be kontextusnak
                    results = ddgs.text(query, max_results=3)
                    return [f"[{r.get('title', 'No Title')}] {r.get('body', '')}" for r in results] if results else []
            
            results = await asyncio.to_thread(search)
            
            if not results:
                return f"=== SOURCE: {label} | STATUS: EMPTY ===\n\n"

            content = f"=== SOURCE: {label} | QUERY: {query} ===\n"
            content += "\n---\n".join(results)
            content += "\n=== END SOURCE ===\n\n"
            return content
        except Exception as e:
            logger.error(f"❌ Keresési hiba ({label}): {str(e)}")
            return f"=== SOURCE: {label} | ERROR: {str(e)} ===\n\n"

    async def research_vehicle(self, vehicle_id: int):
        """ Egyetlen jármű teljes körű átvilágítása. """
        async with AsyncSessionLocal() as db:
            res = await db.execute(select(VehicleModelDefinition).where(VehicleModelDefinition.id == vehicle_id))
            v = res.scalar_one_or_none()
            if not v: return
            
            make = v.make
            model = v.marketing_name
            engine = v.engine_code or ""
            year = f"{v.year_from}" if v.year_from else ""

            # Státusz zárolása
            v.status = 'research_in_progress'
            await db.commit()

        logger.info(f"🔎 Kutatás indul: {make} {model} (Motor: {engine}, Év: {year})")
        
        # Célzott keresési kulcsszavak (Multi-Channel stratégia)
        queries = [
            ("TECH_SPECS", f"{make} {model} {engine} {year} technical specifications engine power kw torque"),
            ("MAINTENANCE", f"{make} {model} {engine} oil capacity coolant transmission fluid type capacity"),
            ("TIRES_PROD", f"{make} {model} {year} tire size load index production years status")
        ]
        
        # Párhuzamos forrásgyűjtés
        tasks = [self.fetch_source(label, q) for label, q in queries]
        search_results = await asyncio.gather(*tasks)
        full_context = "".join(search_results)
        
        async with AsyncSessionLocal() as db:
            if len(full_context.strip()) > 200: # Ha van elegendő kontextus
                await db.execute(
                    update(VehicleModelDefinition)
                    .where(VehicleModelDefinition.id == vehicle_id)
                    .values(
                        raw_search_context=full_context,
                        status='awaiting_ai_synthesis', # Átadás a Robot 2.2-nek
                        last_research_at=func.now(),
                        attempts=VehicleModelDefinition.attempts + 1
                    )
                )
                logger.info(f"✅ Kontextus rögzítve: {make} {model}")
            else:
                # Sikertelen keresés, visszatesszük később
                await db.execute(
                    update(VehicleModelDefinition)
                    .where(VehicleModelDefinition.id == vehicle_id)
                    .values(
                        status='unverified', 
                        attempts=VehicleModelDefinition.attempts + 1,
                        last_research_at=func.now()
                    )
                )
                logger.warning(f"⚠️ Kevés adat: {make} {model} - Újrapróbálkozás később")
            await db.commit()

    async def run(self):
        logger.info("🚀 Robot 2.1 (Researcher) ONLINE - Cél: 407 Toyota feldolgozása")
        while True:
            async with AsyncSessionLocal() as db:
                # Prioritás: unverified autók előre
                priorities = case(
                    (VehicleModelDefinition.make == 'TOYOTA', 1),
                    else_=2
                )

                stmt = select(VehicleModelDefinition.id).where(
                    or_(VehicleModelDefinition.status == 'unverified', 
                        VehicleModelDefinition.status == 'awaiting_research')
                ).order_by(priorities, VehicleModelDefinition.attempts.asc()).limit(self.batch_size)
                
                res = await db.execute(stmt)
                ids = [r[0] for r in res.fetchall()]

            if not ids:
                await asyncio.sleep(30)
                continue

            # Szekvenciális feldolgozás a rate-limit miatt
            for rid in ids:
                await self.research_vehicle(rid)
                await asyncio.sleep(5) # 5 másodperc szünet a keresések között

if __name__ == "__main__":
    asyncio.run(ResearcherBot().run())