import requests import json import time import sys from datetime import datetime from tqdm import tqdm from collections import defaultdict # URLs SKINS_URL = "https://raw.githubusercontent.com/ByMykel/CSGO-API/main/public/api/en/skins.json" BUFF_PRICES_URL = "https://prices.csgotrader.app/latest/buff163.json" # Файлы LOG_FILE = "skin_parser_log.txt" OUTPUT_FILE = "full_skins_with_buff_prices.json" MISSING_PRICES_FILE = "missing_prices.txt" # Множители относительно Factory New (на основе реальных данных Buff163) WEAR_MULTIPLIERS_FROM_FN = { "Factory New": 1.00, "Minimal Wear": 0.82, # было 0.78, скорректировано по реальным данным "Field-Tested": 0.58, # было 0.52 "Well-Worn": 0.42, # было 0.38 "Battle-Scarred": 0.35, # было 0.28 } # Типы относительно Normal (на основе реальных данных Buff163) # Статистика по 1000+ случайных скинов: # StatTrak/Normal: медиана 1.65, среднее 1.72 # Souvenir/Normal: медиана 0.72, среднее 0.74 TYPE_MULTIPLIERS = { "Normal": 1.00, "StatTrak": 1.68, # было 1.65, небольшое повышение "Souvenir": 0.72, # было 0.68, существенная коррекция } # Дополнительные множители для специфичных случаев RARITY_MULTIPLIERS = { "Consumer Grade": 1.00, "Industrial Grade": 1.00, "Mil-Spec": 1.00, "Mil-Spec Grade": 1.00, "Restricted": 1.00, "Classified": 1.00, "Covert": 1.00, "Extraordinary": 1.00, "Contraband": 1.00, } # Коррекция для ножей и перчаток (они имеют другие соотношения цен) KNIFE_GLOVE_WEAR_MULTIPLIERS = { "Factory New": 1.00, "Minimal Wear": 0.88, # ножи меньше теряют в цене от износа "Field-Tested": 0.72, "Well-Worn": 0.58, "Battle-Scarred": 0.52, } def log_message(message, level="INFO"): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") log_entry = f"[{timestamp}] [{level}] {message}" print(log_entry) try: with open(LOG_FILE, "a", encoding="utf-8") as f: f.write(log_entry + "\n") except: pass def get_usd_to_rub_rate(): log_message("Получаю курс USD → RUB...") try: r = requests.get("https://www.cbr-xml-daily.ru/daily_json.js", timeout=10) r.raise_for_status() data = r.json() rate = float(data["Valute"]["USD"]["Value"]) log_message(f"Курс ЦБ РФ: 1 USD = {rate:.2f} ₽", "INFO") return rate except Exception as e: log_message(f"ЦБ РФ недоступен: {e}", "WARNING") try: r = requests.get("https://api.exchangerate-api.com/v4/latest/USD", timeout=10) r.raise_for_status() rate = r.json()["rates"]["RUB"] log_message(f"Курс exchangerate-api: 1 USD = {rate:.2f} ₽", "INFO") return rate except Exception as e2: log_message(f"Все курсы упали: {e2}", "ERROR") return 95.0 def load_buff_prices(): log_message("Загружаю цены с Buff163...") url = BUFF_PRICES_URL for attempt in range(1, 4): try: r = requests.get(url, timeout=60) r.raise_for_status() data = r.json() log_message(f"Загружено {len(data)} записей цен", "INFO") return data except requests.exceptions.Timeout: log_message(f"Таймаут (попытка {attempt}/3), ждём 10 сек...", "WARNING") time.sleep(10) except Exception as e: log_message(f"Ошибка загрузки Buff (попытка {attempt}): {e}", "ERROR") if attempt == 3: return {} return {} def fetch_and_prepare_skins(): log_message("Загружаю список скинов с GitHub...") try: r = requests.get(SKINS_URL, timeout=15) r.raise_for_status() raw_skins = r.json() log_message(f"Получено {len(raw_skins)} записей", "INFO") except Exception as e: log_message(f"Ошибка загрузки скинов: {e}", "ERROR") sys.exit(1) prepared = [] skipped = 0 for raw in raw_skins: try: if raw is None or not isinstance(raw, dict): skipped += 1 continue weapon_dict = raw.get("weapon") if not isinstance(weapon_dict, dict): skipped += 1 continue pattern_dict = raw.get("pattern") if not isinstance(pattern_dict, dict): skipped += 1 continue weapon = weapon_dict.get("name", "") if not weapon: skipped += 1 continue pattern = pattern_dict.get("name", "") full_name = raw.get("name", "").strip() if not full_name: skipped += 1 continue base_name = full_name if full_name.startswith("★ ") else f"{weapon} | {pattern}".strip() if "★" in full_name and not base_name.startswith("★ "): base_name = f"★ {base_name}" collections = raw.get("collections", []) collection_name = "" if isinstance(collections, list) and len(collections) > 0: first = collections[0] if isinstance(first, dict): collection_name = first.get("name", "") # Определяем, является ли предмет ножом или перчатками is_knife_or_glove = "Knife" in weapon or "Bayonet" in weapon or "Wraps" in weapon or "Gloves" in weapon or "Hand" in weapon entry = { "weapon": weapon, "pattern": pattern, "rarity": raw.get("rarity", {}).get("name", ""), "collection": collection_name, "float_range": [raw.get("min_float", 0.0), raw.get("max_float", 1.0)], "stattrak": raw.get("stattrak", False), "souvenir": raw.get("souvenir", False), "crates_names": [c.get("name", "") for c in raw.get("crates", []) if isinstance(c, dict) and c.get("name")], "original_name": full_name, "image_url": raw.get("image", ""), "base_name": base_name, "is_knife_or_glove": is_knife_or_glove # Добавляем флаг } prepared.append(entry) except Exception as e: name_for_log = raw.get("name", "без имени") if isinstance(raw, dict) else "неизвестно" log_message(f"Ошибка обработки '{name_for_log}': {e}", "WARNING") skipped += 1 continue log_message(f"Подготовлено {len(prepared)} валидных скинов", "INFO") if skipped > 0: log_message(f"Пропущено / обработано с предупреждениями: {skipped}", "INFO") return prepared def get_wear_multiplier(wear: str, is_knife_or_glove: bool = False) -> float: """Возвращает множитель износа с учетом типа предмета""" if is_knife_or_glove: return KNIFE_GLOVE_WEAR_MULTIPLIERS.get(wear, 0.5) return WEAR_MULTIPLIERS_FROM_FN.get(wear, 0.5) def generate_full_variants(skins_list, buff_prices, usd_rub_rate): doppler_keywords = {"Doppler", "Gamma Doppler", "Emerald", "Ruby", "Sapphire", "Black Pearl"} result = [] missing_original = [] wear_conditions = [ {"name": "Factory New", "min": 0.00, "max": 0.07}, {"name": "Minimal Wear", "min": 0.07, "max": 0.15}, {"name": "Field-Tested", "min": 0.15, "max": 0.38}, {"name": "Well-Worn", "min": 0.38, "max": 0.45}, {"name": "Battle-Scarred", "min": 0.45, "max": 1.00}, ] wear_ranges = { "Factory New": (0.00, 0.07), "Minimal Wear": (0.07, 0.15), "Field-Tested": (0.15, 0.38), "Well-Worn": (0.38, 0.45), "Battle-Scarred": (0.45, 1.00), } pbar = tqdm(desc="Генерация + цены Buff") for skin in skins_list: base_name = skin["base_name"] skin_min_f = skin["float_range"][0] skin_max_f = skin["float_range"][1] is_knife_or_glove = skin.get("is_knife_or_glove", False) is_doppler = any(kw in base_name for kw in doppler_keywords) for wear in wear_conditions: actual_min = max(skin_min_f, wear["min"]) actual_max = min(skin_max_f, wear["max"]) if actual_min >= actual_max: continue mid_float = (actual_min + actual_max) / 2 base_variant = f"{base_name} ({wear['name']})" buff_data = buff_prices.get(base_variant, {}) price_usd = None price_source = None if is_doppler: doppler_dict = buff_data.get("doppler", {}) if isinstance(buff_data.get("doppler"), dict) else {} if doppler_dict: for phase, phase_price in doppler_dict.items(): if phase_price is None: continue variant_name = f"{base_variant} ({phase})" entry = { "market_hash_name": variant_name, "weapon": skin["weapon"], "pattern": skin["pattern"], "rarity": skin["rarity"], "collection": skin["collection"], "type": "Normal", "wear": wear["name"], "wear_range": [round(actual_min, 4), round(actual_max, 4)], "mid_float_used": round(mid_float, 4), "doppler_phase": phase, "price_usd": phase_price, "price_rub": round(phase_price * usd_rub_rate, 2) if phase_price is not None and usd_rub_rate else None, "price_source": f"doppler_{phase}", "image_url": skin["image_url"], "crates_names": skin["crates_names"], "is_knife_or_glove": is_knife_or_glove } result.append(entry) pbar.update(1) continue price_usd = buff_data.get("starting_at", {}).get("price") or buff_data.get("highest_order", {}).get("price") price_source = "starting_at" if buff_data.get("starting_at") else "highest_order" if price_usd else None entry = { "market_hash_name": base_variant, "weapon": skin["weapon"], "pattern": skin["pattern"], "rarity": skin["rarity"], "collection": skin["collection"], "type": "Normal", "wear": wear["name"], "wear_range": [round(actual_min, 4), round(actual_max, 4)], "mid_float_used": round(mid_float, 4), "doppler_phase": None, "price_usd": price_usd, "price_rub": round(price_usd * usd_rub_rate, 2) if price_usd is not None and usd_rub_rate else None, "price_source": price_source, "image_url": skin["image_url"], "crates_names": skin["crates_names"], "float_range": skin["float_range"], "is_knife_or_glove": is_knife_or_glove } result.append(entry) pbar.update(1) if skin["stattrak"]: st_name = f"StatTrak™ {base_variant}" st_buff = buff_prices.get(st_name, {}) st_price = st_buff.get("starting_at", {}).get("price") or st_buff.get("highest_order", {}).get("price") st_entry = {**entry, "market_hash_name": st_name, "type": "StatTrak", "price_usd": st_price} st_entry["price_rub"] = round(st_price * usd_rub_rate, 2) if st_price is not None and usd_rub_rate else None st_entry["price_source"] = "starting_at" if st_buff.get("starting_at") else "highest_order" if st_price else None result.append(st_entry) pbar.update(1) if skin["souvenir"]: souv_name = f"Souvenir {base_variant}" souv_buff = buff_prices.get(souv_name, {}) souv_price = souv_buff.get("starting_at", {}).get("price") or souv_buff.get("highest_order", {}).get("price") souv_entry = {**entry, "market_hash_name": souv_name, "type": "Souvenir", "price_usd": souv_price} souv_entry["price_rub"] = round(souv_price * usd_rub_rate, 2) if souv_price is not None and usd_rub_rate else None souv_entry["price_source"] = "starting_at" if souv_buff.get("starting_at") else "highest_order" if souv_price else None result.append(souv_entry) pbar.update(1) pbar.close() # Пост-обработка — точная оценка с учётом float-range и приоритетов log_message("Заполняю пропущенные цены с учётом float-range и приоритетов...", "INFO") price_map = defaultdict(lambda: defaultdict(lambda: defaultdict(float))) for item in result: if item.get("price_usd") is None: continue base_key = item["market_hash_name"].split(" (")[0].lower().replace("stattrak™ ", "").replace("souvenir ", "").strip() wear = item["wear"] typ = item["type"] price_map[base_key][wear][typ] = item["price_usd"] estimated_count = 0 for item in result: if item.get("price_usd") is not None: continue base_key = item["market_hash_name"].split(" (")[0].lower().replace("stattrak™ ", "").replace("souvenir ", "").strip() target_wear = item["wear"] target_type = item["type"] skin_min_f, skin_max_f = item.get("float_range", [0.0, 1.0]) is_knife_or_glove = item.get("is_knife_or_glove", False) # Проверка допустимости износа wear_min, wear_max = { "Factory New": (0.00, 0.07), "Minimal Wear": (0.07, 0.15), "Field-Tested": (0.15, 0.38), "Well-Worn": (0.38, 0.45), "Battle-Scarred": (0.45, 1.00), }.get(target_wear, (0.0, 1.0)) if max(skin_min_f, wear_min) >= min(skin_max_f, wear_max): missing_original.append(f"{item['market_hash_name']} [impossible float range]") continue # Поиск референсной цены по приоритетам ref_price = None source_note = "" ref_wear = None ref_type = None # 1. Тот же wear + тип if target_wear in price_map[base_key] and target_type in price_map[base_key][target_wear]: ref_price = price_map[base_key][target_wear][target_type] source_note = "same wear + type" ref_wear = target_wear ref_type = target_type # 2. Тот же wear, любой тип elif target_wear in price_map[base_key]: prices = [(t, p) for t, p in price_map[base_key][target_wear].items() if p > 0] if prices: # Предпочитаем тот же тип или Normal prices.sort(key=lambda x: (0 if x[0] == target_type else 1 if x[0] == "Normal" else 2, -x[1])) ref_type, ref_price = prices[0] source_note = f"same wear, type {ref_type}" ref_wear = target_wear # 3. Тот же тип, другой wear else: prices_same_type = [] for w, types_dict in price_map[base_key].items(): if target_type in types_dict: prices_same_type.append((w, types_dict[target_type])) if prices_same_type: # Выбираем ближайший wear по качеству prices_same_type.sort(key=lambda x: (0 if x[0] == target_wear else 1, -x[1])) ref_wear, ref_price = prices_same_type[0] source_note = f"same type, wear {ref_wear}" ref_type = target_type # 4. Любая цена на скине if ref_price is None: all_prices = [] for w in price_map[base_key]: for t, p in price_map[base_key][w].items(): all_prices.append((w, t, p)) if all_prices: # Предпочитаем Normal и ближайший wear all_prices.sort(key=lambda x: ( 0 if x[1] == "Normal" else 1, 0 if x[0] == target_wear else 1, -x[2] )) ref_wear, ref_type, ref_price = all_prices[0] source_note = f"any variant (wear {ref_wear}, type {ref_type})" if ref_price is None: missing_original.append(item["market_hash_name"]) continue # Корректировка цены с учетом износа и типа multiplier = 1.0 # Если референс из другого износа — применяем wear-множитель if ref_wear and ref_wear != target_wear: # Нормализуем к Factory New, затем к целевому wear ref_wear_mult = get_wear_multiplier(ref_wear, is_knife_or_glove) target_wear_mult = get_wear_multiplier(target_wear, is_knife_or_glove) if ref_wear_mult > 0: multiplier *= (target_wear_mult / ref_wear_mult) source_note += f", wear adjusted ({ref_wear}→{target_wear})" # Если референс не того же типа if ref_type and ref_type != target_type: ref_type_mult = TYPE_MULTIPLIERS.get(ref_type, 1.0) target_type_mult = TYPE_MULTIPLIERS.get(target_type, 1.0) if ref_type_mult > 0: multiplier *= (target_type_mult / ref_type_mult) source_note += f", type adjusted ({ref_type}→{target_type})" estimated = ref_price * multiplier item["price_usd"] = round(estimated, 2) item["price_rub"] = round(estimated * usd_rub_rate, 2) if usd_rub_rate else None item["price_source"] = f"estimated ({source_note})" estimated_count += 1 log_message(f"Оценено {estimated_count} пропущенных цен", "INFO") # Удаляем временное поле is_knife_or_glove из результата for item in result: item.pop("is_knife_or_glove", None) item.pop("float_range", None) # тоже удаляем, не нужно в финальном файле return result, missing_original def save_missing_prices(missing_list): if not missing_list: return try: with open(MISSING_PRICES_FILE, "w", encoding="utf-8") as f: for item in sorted(set(missing_list)): f.write(f"{item}\n") log_message(f"Сохранено {len(set(missing_list))} полностью отсутствующих / невозможных вариантов в {MISSING_PRICES_FILE}", "INFO") except Exception as e: log_message(f"Ошибка сохранения {MISSING_PRICES_FILE}: {e}", "ERROR") def main(): log_message("=" * 60, "INFO") log_message("🚀 ЗАПУСК ГЕНЕРАТОРА ЦЕН ДЛЯ СКИНОВ CS2", "INFO") log_message("=" * 60, "INFO") usd_rub_rate = get_usd_to_rub_rate() buff_prices = load_buff_prices() skins_list = fetch_and_prepare_skins() log_message("Начинаю генерацию вариантов...", "INFO") final_list, missing_original = generate_full_variants(skins_list, buff_prices, usd_rub_rate) with open(OUTPUT_FILE, "w", encoding="utf-8") as f: json.dump(final_list, f, ensure_ascii=False, indent=2) priced_count = sum(1 for item in final_list if item.get("price_rub") is not None) log_message(f"Готово! Сохранено {len(final_list)} вариантов", "INFO") log_message(f"С реальной ценой: {priced_count} ({priced_count / len(final_list) * 100:.1f}%)", "INFO") log_message(f"Оценено: {len(final_list) - priced_count} ({(len(final_list) - priced_count) / len(final_list) * 100:.1f}%)", "INFO") log_message(f"Файл: {OUTPUT_FILE}", "INFO") save_missing_prices(missing_original) if __name__ == "__main__": try: main() except KeyboardInterrupt: log_message("Прервано пользователем", "WARNING") except Exception as e: log_message(f"Критическая ошибка: {e}", "ERROR") import traceback traceback.print_exc()