#!/usr/bin/python3 # -*- coding: utf-8 -*- # prefilter.cgi """ Первая ступень (pre-filter) для воронки определения объявлений о продаже авто. Задача: ДЁШЕВО и с ВЫСОКИМ RECALL отсеять явный мусор и пропустить кандидатов "похоже на продажу авто" дальше — на дорогой LLM-анализ. Принцип: recall важнее precision. - Пропустить лишнее (мусор) — дёшево (LLM отсеет на 2-й ступени). - Потерять настоящую продажу — дорого (объявление потеряно навсегда). Поэтому логика "ИЛИ": набралось >= PASS_THRESHOLD очков сигналов -> пропускаем. Использование как CGI (как ваш autosale.cgi): prefilter.cgi?user_text=...&threshold=2 Использование из консоли: echo "текст" | python3 prefilter.cgi python3 prefilter.cgi "текст объявления" """ import os import re import sys import json import urllib.parse import datetime # ============================================================ # НАСТРОЙКИ # ============================================================ # Сколько очков сигналов нужно набрать, чтобы пропустить в LLM. # Ниже порог -> выше recall (больше пропускаем), больше нагрузка на LLM. # 1 — максимальный recall (пропускаем по любому одному сигналу). # 2 — разумный баланс по умолчанию. PASS_THRESHOLD = 2 # Текущий год для проверки диапазона года выпуска CURRENT_YEAR = datetime.datetime.now().year # ============================================================ # НОРМАЛИЗАЦИЯ ИСКАЖЁННЫХ БУКВ (homoglyphs) # Авторынки маскируют латинские марки кириллицей: "chеvrоlеt", "Тoyota". # Чтобы словарь марок сматчился, приводим кириллические двойники к латинице. # Таблица покрывает все визуально совпадающие кирилл->лат пары. # ============================================================ # Кириллический/греческий символ -> визуально идентичная латинская буква CYR_TO_LAT = { 'а': 'a', 'А': 'A', 'в': 'b', 'В': 'B', 'е': 'e', 'Е': 'E', 'ё': 'e', 'Ё': 'E', 'к': 'k', 'К': 'K', 'м': 'm', 'М': 'M', 'н': 'h', 'Н': 'H', 'о': 'o', 'О': 'O', 'р': 'p', 'Р': 'P', 'с': 'c', 'С': 'C', 'т': 't', 'Т': 'T', 'у': 'y', 'У': 'Y', 'х': 'x', 'Х': 'X', 'і': 'i', 'І': 'I', 'ѕ': 's', 'Ѕ': 'S', 'ј': 'j', 'Ј': 'J', 'ԁ': 'd', 'џ': 'u', 'г': 'r', # греческие двойники (маскировка вида "φoльцвaген", "πaсaт") 'φ': 'f', 'Φ': 'F', 'π': 'p', 'Π': 'P', 'ρ': 'p', 'Ρ': 'P', 'ο': 'o', 'Ο': 'O', 'α': 'a', 'Α': 'A', 'ε': 'e', 'Ε': 'E', 'τ': 't', 'Τ': 'T', 'υ': 'y', 'κ': 'k', 'Κ': 'K', 'ν': 'v', 'β': 'b', 'Β': 'B', 'χ': 'x', 'Χ': 'X', ' μ': 'm', 'Μ': 'M', 'η': 'n', 'ι': 'i', 'Ι': 'I', 'ω': 'w', } # и обратная таблица — латиница, замаскированная под кириллицу в русских словах LAT_TO_CYR = { 'a': 'а', 'e': 'е', 'o': 'о', 'p': 'р', 'c': 'с', 'x': 'х', 'y': 'у', 'k': 'к', 'm': 'м', 'b': 'в', 'h': 'н', 't': 'т', } def normalize_homoglyphs(text): """ Возвращает версию текста, где кириллические двойники латинских букв заменены на латиницу. Используется ТОЛЬКО для матчинга латинских марок. Исходный текст не трогаем — нормализуем копию. """ return ''.join(CYR_TO_LAT.get(ch, ch) for ch in text) # ============================================================ # СЛОВАРЬ МАРОК (самый сильный сигнал) # Включает кириллические и латинские написания + частые модели. # Расширяйте под свой поток. Ключи — то, что ищем в нижнем регистре. # ============================================================ CAR_BRANDS = [ # массовые иностранные "volkswagen", "vw", "фольксваген", "тойота", "toyota", "kia", "киа", "hyundai", "хендай", "хёндай", "ниссан", "nissan", "honda", "хонда", "mazda", "мазда", "mitsubishi", "митсубиси", "мицубиси", "ford", "форд", "opel", "опель", "renault", "рено", "peugeot", "пежо", "citroen", "ситроен", "skoda", "шкода", "chevrolet", "шевроле", "subaru", "субару", "suzuki", "сузуки", "сузуки", # премиум "bmw", "бмв", "mercedes", "мерседес", "мерс", "audi", "ауди", "lexus", "лексус", "infiniti", "инфинити", "volvo", "вольво", "land rover", "ленд ровер", "range rover", "рендж ровер", "jaguar", "ягуар", "porsche", "порше", "porshe", "cadillac", "кадиллак", "jeep", "джип", "mini", # отечественные / снг "lada", "лада", "ваз", "ваз", "газ", "уаз", "нива", "niva", "приора", "калина", "kalina", "гранта", "granta", "веста", "vesta", "ларгус", "largus", "datsun", "датсун", "chery", "чери", "geely", "джили", "haval", "хавал", "exeed", "эксид", "tank", "omoda", "омода", "changan", "чанган", "lifan", "лифан", "byd", "бид", "chevrolet niva", "шевроле нива", # частые модели (когда марка не написана явно) "octavia", "октавия", "rapid", "рапид", "solaris", "солярис", "rio", "рио", "polo", "поло", "logan", "логан", "duster", "дастер", "qashqai", "кашкай", "x-trail", "икстрейл", "camry", "камри", "corolla", "королла", "rav4", "рав4", "tiguan", "тигуан", "passat", "пассат", "focus", "фокус", "cruze", "круз", "lacetti", "лачетти", "matiz", "матиз", "soul", "соул", "sportage", "спортейдж", "ceed", "сид", "optima", "оптима", "x5", "x6", "x3", "e-class", "e-klass", "e klass", "c-class", ] # Числовые модели ВАЗ/ГАЗ — матчим отдельно, со словесным контекстом, # чтобы не путать с годом/ценой. Срабатывает, если рядом авто-маркеры. VAZ_MODEL_RE = re.compile( r'\b(210[1-9]|2110|2111|2112|2113|2114|2115|21099|2121|2123|' r'217[0-2]|219[0-2])\b' ) VAZ_CONTEXT_RE = re.compile( r'ваз|лада|жигул|классик|приор|калин|гранат|надо\s*позанимат|' r'на\s*ход|не\s*гнил|движок|двигат|короб|кпп|резин|штамп', re.IGNORECASE ) # Скомпилируем словарь марок в один regex с границами слова, где уместно. # Для латиницы и кириллицы используем простой поиск подстроки по \b там, где можно. _brand_alts = sorted({re.escape(b) for b in CAR_BRANDS}, key=len, reverse=True) BRAND_RE = re.compile(r'(?= min_len}, key=len, reverse=True) if not alts: return None return re.compile(r'(? штраф к score # Калибровано по реальному мусору: запчасти, разбор, услуги, # авто под заказ/пригон, новости, реклама, розыгрыши, вакансии. # Вес штрафа подбирается так, чтобы НЕ резать настоящие продажи. # ============================================================ STOP_PATTERNS = { # запчасти / разбор — частый мусор с марками и ценами "запчасти": re.compile(r'запчаст|авторазбор|на\s*разбор|по\s*запчаст|б/?у\s*запчаст|разбор\s*:|в\s*разбор', re.IGNORECASE), # услуги (подбор, осмотр, диагностика, детейлинг, автоподбор) "услуги": re.compile(r'помощь\s*в\s*подбор|подбор\s*авто|осмотр\s*перед\s*покупк|разовый\s*осмотр|диагностик|детейлинг|автоподбор|подбор\s*под\s*ключ|выездн\w*\s*осмотр|оценка\s*авто', re.IGNORECASE), # авто под заказ / пригон / доставка "под_заказ": re.compile(r'под\s*заказ|пригон\b|пригоня|привез[её]м|доставка\s*из|срок\s*доставк|таможн|растаможим|авто\s*из\s*(кита|кореи|японии|европ|оаэ|германии)', re.IGNORECASE), # кредит / лизинг / выкуп / скупка (не продажа конкретной машины) "финуслуги": re.compile(r'\bвыкуп|скупка\s*авто|куплю|купим|покупа|автоломбард|деньги\s*под\s*птс|залог\s*авто|лизинг|trade-?in\s*центр', re.IGNORECASE), # реклама/розыгрыш/конкурс/новости/вакансии "реклама_новости": re.compile(r'розыгрыш|конкурс\b|подпишись|подписку|приглашаем\s*на\s*работу|вакансия|зарплата|требуется\s*\w+|新|банкрот|обяжут|медиков', re.IGNORECASE), # шиномонтаж / резина / диски отдельно (не авто) "шины_диски": re.compile(r'шиномонтаж|комплект\s*резин|зимн\w*\s*резин\w*\s*\d|штамповк|литье\s*r?\d{2}\b', re.IGNORECASE), } # ============================================================ # СТОП-СИГНАЛЫ (украинский номер -> отдельный класс, banned) # ============================================================ def has_ukrainian_phone(text): t = text.lower() t = re.sub(r'^\+', '', t) if re.search(r'380\d{9}', t): return True sep = r'[\s\-\(\)]*' pattern = r'380' + sep + (r'\d' + sep) * 8 + r'\d' return re.search(pattern, t) is not None # ============================================================ # ОСНОВНАЯ ЛОГИКА: ПОДСЧЁТ СИГНАЛОВ # ============================================================ def evaluate(text, threshold=PASS_THRESHOLD): """ Возвращает dict с решением и сработавшими сигналами. score = сумма очков; pass = score >= threshold. """ if not text: return { "pass": False, "score": 0, "signals": [], "banned": False, "banned_reason": None } signals = [] score = 0 # 0. Бан по украинскому номеру (не идёт в LLM как продажа) if has_ukrainian_phone(text): return { "pass": False, "score": 0, "signals": ["ukrainian_phone"], "banned": True, "banned_reason": "ukrainian_phone" } # 1. Марка/модель — сильный сигнал (2 очка) # Используем полный справочник (688 марок), если загружен, иначе встроенный. active_brand_re = FULL_BRAND_RE if FULL_BRAND_RE is not None else BRAND_RE norm_text = normalize_homoglyphs(text) brand_hits = active_brand_re.findall(text) if norm_text != text: brand_hits += active_brand_re.findall(norm_text) if brand_hits: uniq = sorted(set(h.lower() for h in brand_hits)) signals.append({"type": "brand", "weight": 2, "hits": uniq[:10]}) score += 2 else: # числовые модели ВАЗ — только если рядом есть авто-контекст if VAZ_MODEL_RE.search(text) and VAZ_CONTEXT_RE.search(text): signals.append({"type": "brand", "weight": 2, "hits": ["ваз-модель"]}) score += 2 # 1b. Модель из справочника (5301 модель) — доп. сигнал (1 очко) if MODEL_RE is not None: mhits = MODEL_RE.findall(text) if norm_text != text: mhits += MODEL_RE.findall(norm_text) if mhits: signals.append({"type": "model", "weight": 1, "hits": sorted(set(h.lower() for h in mhits))[:6]}) score += 1 # 2. Год выпуска в авто-диапазоне (1 очко) year_hits = [int(y) for y in YEAR_RE.findall(text) if 1980 <= int(y) <= CURRENT_YEAR] if year_hits: signals.append({"type": "year", "weight": 1, "hits": sorted(set(year_hits))}) score += 1 # 3. Цена в любом формате (2 очка — сильный сигнал) if any(r.search(text) for r in PRICE_RES): signals.append({"type": "price", "weight": 2}) score += 2 # 4. Пробег (1 очко) if any(r.search(text) for r in MILEAGE_RES): signals.append({"type": "mileage", "weight": 1}) score += 1 # 5. Слова продажи (1 очко) if SALE_WORDS_RE.search(text): signals.append({"type": "sale_word", "weight": 1}) score += 1 # 6. Автолексика (1 очко, но не более 1 раза) lex_hits = AUTO_LEX_RE.findall(text) if lex_hits: uniq = sorted(set(h.lower() for h in lex_hits)) signals.append({"type": "auto_lex", "weight": 1, "hits": uniq[:10]}) score += 1 # 6b. Доп. зацепки из справочника: кузов / привод / КПП / цвет (по 1 очку суммарно) extra = [] if BODY_RE is not None and BODY_RE.search(text): extra.append("body") if DRIVE_RE is not None and DRIVE_RE.search(text): extra.append("drive") if TRANS_RE is not None and TRANS_RE.search(text): extra.append("trans") if COLOR_RE is not None and COLOR_RE.search(text): extra.append("color") if extra: signals.append({"type": "spec", "weight": 1, "hits": extra}) score += 1 # 7. СТОП-СИГНАЛЫ: штраф за маркеры "не продажа конкретного авто". # Каждый сработавший стоп-паттерн снимает очки. Подобрано так, чтобы # убрать запчасти/услуги/под-заказ, не задев настоящие объявления. stop_hits = [] for name, rx in STOP_PATTERNS.items(): if rx.search(text): stop_hits.append(name) if stop_hits: penalty = 2 * len(stop_hits) # вес штрафа за каждый стоп-сигнал signals.append({"type": "stop", "weight": -penalty, "hits": stop_hits}) score -= penalty return { "pass": score >= threshold, "score": score, "signals": signals, "banned": False, "banned_reason": None } # ============================================================ # CGI / CLI входная точка # ============================================================ def _get_input_text(): # CGI режим qs = os.environ.get('QUERY_STRING', '') if qs: params = urllib.parse.parse_qs(qs) text = params.get('user_text', [''])[0] thr = params.get('threshold', [''])[0] threshold = int(thr) if thr.isdigit() else PASS_THRESHOLD return text, threshold, True # CLI: аргумент или stdin if len(sys.argv) > 1: return sys.argv[1], PASS_THRESHOLD, False if not sys.stdin.isatty(): return sys.stdin.read(), PASS_THRESHOLD, False return "", PASS_THRESHOLD, False def main(): text, threshold, is_cgi = _get_input_text() result = evaluate(text, threshold) response = { "success": True, "pass": result["pass"], "score": result["score"], "threshold": threshold, "signals": result["signals"], "banned": result["banned"], "banned_reason": result["banned_reason"], "error": None if text else "Текст не получен", } if not text: response["success"] = False out = json.dumps(response, ensure_ascii=False, indent=2) # ВЫВОДИМ ТОЛЬКО JSON, без заголовков (их добавляет PHP) # if is_cgi: # sys.stdout.write("Content-Type: application/json; charset=utf-8\r\n\r\n") print(out) if __name__ == "__main__": main()