492 lines
21 KiB
Python
492 lines
21 KiB
Python
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() |