root
971 lines
37 KiB
Python
971 lines
37 KiB
Python
# main.py
|
|
from fastapi import FastAPI, HTTPException, Query
|
|
from pydantic import BaseModel, field_validator, validator
|
|
from typing import List, Optional, Dict, Any
|
|
import json
|
|
from pathlib import Path
|
|
from collections import Counter, defaultdict
|
|
import random
|
|
import sys
|
|
from tqdm import tqdm
|
|
|
|
app = FastAPI(
|
|
title="CS2 Trade-Up Contract Simulator API",
|
|
description="Симуляция контрактов 10→1 по механике CS2",
|
|
version="1.0.0"
|
|
)
|
|
|
|
DATA_FILE = Path("skins.json")
|
|
CUSTOM_DATA_FILE = Path("custom_items.json")
|
|
|
|
# ─── Загрузка кастомных предметов ─────────────────────────────────────────
|
|
def load_custom_items() -> List[dict]:
|
|
"""Загружает кастомные предметы из отдельного файла"""
|
|
custom_items = []
|
|
if CUSTOM_DATA_FILE.exists():
|
|
try:
|
|
with CUSTOM_DATA_FILE.open("r", encoding="utf-8") as f:
|
|
custom_items = json.load(f)
|
|
print(f"Загружено {len(custom_items)} кастомных предметов")
|
|
except Exception as e:
|
|
print(f"Ошибка загрузки кастомных предметов: {e}")
|
|
else:
|
|
# Создаем пример файла с кастомными предметами
|
|
example_custom = [
|
|
{
|
|
"market_hash_name": "★ Dildo | Fade (Factory New)",
|
|
"weapon": "Dildo",
|
|
"pattern": "Fade",
|
|
"rarity": "Covert",
|
|
"collection": "",
|
|
"type": "Normal",
|
|
"wear": "Factory New",
|
|
"wear_range": [0.00, 0.07],
|
|
"mid_float_used": 0.01,
|
|
"doppler_phase": None,
|
|
"price_usd": 1000.0,
|
|
"price_rub": 100000.0,
|
|
"price_source": "custom",
|
|
"image_url": "",
|
|
"crates_names": ["create: dildo_case"],
|
|
"float_range": [0.00, 0.07],
|
|
"can_be_assigned_from": ["create: dildo_case"],
|
|
"is_craftable": True
|
|
},
|
|
{
|
|
"market_hash_name": "★ Dildo | Crimson Web (Minimal Wear)",
|
|
"weapon": "Dildo",
|
|
"pattern": "Crimson Web",
|
|
"rarity": "Covert",
|
|
"collection": "",
|
|
"type": "Normal",
|
|
"wear": "Minimal Wear",
|
|
"wear_range": [0.07, 0.15],
|
|
"mid_float_used": 0.10,
|
|
"doppler_phase": None,
|
|
"price_usd": 800.0,
|
|
"price_rub": 80000.0,
|
|
"price_source": "custom",
|
|
"image_url": "",
|
|
"crates_names": ["create: dildo_case"],
|
|
"float_range": [0.07, 0.15],
|
|
"can_be_assigned_from": ["create: dildo_case"],
|
|
"is_craftable": True
|
|
},
|
|
{
|
|
"market_hash_name": "Dildo | Blue Gem (Factory New)",
|
|
"weapon": "Dildo",
|
|
"pattern": "Blue Gem",
|
|
"rarity": "Classified",
|
|
"collection": "",
|
|
"type": "Normal",
|
|
"wear": "Factory New",
|
|
"wear_range": [0.00, 0.07],
|
|
"mid_float_used": 0.01,
|
|
"doppler_phase": None,
|
|
"price_usd": 500.0,
|
|
"price_rub": 50000.0,
|
|
"price_source": "custom",
|
|
"image_url": "",
|
|
"crates_names": ["create: dildo_case"],
|
|
"float_range": [0.00, 0.07],
|
|
"can_be_assigned_from": ["create: dildo_case"],
|
|
"is_craftable": True
|
|
}
|
|
]
|
|
try:
|
|
with CUSTOM_DATA_FILE.open("w", encoding="utf-8") as f:
|
|
json.dump(example_custom, f, ensure_ascii=False, indent=2)
|
|
print(f"Создан пример файла кастомных предметов: {CUSTOM_DATA_FILE}")
|
|
custom_items = example_custom
|
|
except Exception as e:
|
|
print(f"Не удалось создать файл кастомных предметов: {e}")
|
|
|
|
return custom_items
|
|
|
|
# ─── Генерация sub-вариаций цен и float ───────────────────────────────────
|
|
def generate_item_variants(num_variants: int = 6):
|
|
"""Для каждого предмета генерирует sub-вариации с разными ценами и float"""
|
|
global ALL_ITEMS
|
|
new_all = []
|
|
for item in ALL_ITEMS:
|
|
base_price = item.get("price_rub", 0)
|
|
wr = item.get("wear_range", [0, 1])
|
|
mid = item.get("mid_float_used", 0.5)
|
|
wr_min, wr_max = wr
|
|
f_range = max(wr_max - wr_min, 0.001)
|
|
|
|
for v in range(num_variants):
|
|
variant = dict(item)
|
|
t = v / (num_variants - 1) if num_variants > 1 else 0
|
|
# price factor: 0.55 to 1.40 across variants
|
|
factor = 0.55 + t * 0.85
|
|
# add small random jitter
|
|
factor += random.uniform(-0.05, 0.05)
|
|
factor = max(0.30, min(1.50, factor))
|
|
|
|
p = int(base_price * factor)
|
|
# scramble last digits
|
|
if p >= 1_000_000:
|
|
prefix = p // 1000
|
|
suffix = random.randint(100, 999)
|
|
p = prefix * 1000 + suffix
|
|
elif p >= 100_000:
|
|
prefix = p // 100
|
|
suffix = random.randint(10, 99)
|
|
p = prefix * 100 + suffix
|
|
elif p >= 10_000:
|
|
prefix = p // 10
|
|
suffix = random.randint(1, 9)
|
|
p = prefix * 10 + suffix
|
|
variant["price_rub"] = max(p, 100)
|
|
variant["price_usd"] = round(variant["price_rub"] / 100, 2)
|
|
|
|
# float spreads across the wear range
|
|
float_offset = -0.4 + t * 0.8
|
|
float_offset += random.uniform(-0.1, 0.1)
|
|
new_mid = mid + float_offset * f_range
|
|
new_mid = max(wr_min + 0.001, min(wr_max - 0.001, new_mid))
|
|
variant["mid_float_used"] = round(new_mid, 4)
|
|
|
|
new_all.append(variant)
|
|
|
|
count_before = len(ALL_ITEMS)
|
|
ALL_ITEMS = new_all
|
|
print(f"Сгенерировано {len(ALL_ITEMS)} sub-вариаций из {count_before} предметов (x{num_variants})")
|
|
|
|
# ─── Загрузка основных данных ─────────────────────────────────────────────
|
|
def load_base_items() -> List[dict]:
|
|
"""Загружает основные предметы из skins.json"""
|
|
try:
|
|
with DATA_FILE.open("r", encoding="utf-8") as f:
|
|
items = json.load(f)
|
|
print(f"Загружено {len(items):,} основных записей")
|
|
return items
|
|
except FileNotFoundError:
|
|
print(f"Критическая ошибка: файл {DATA_FILE} не найден")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"Ошибка чтения файла данных: {e}")
|
|
sys.exit(1)
|
|
|
|
# Загружаем все предметы (основные + кастомные)
|
|
BASE_ITEMS = load_base_items()
|
|
CUSTOM_ITEMS = load_custom_items()
|
|
|
|
# Объединяем предметы: сначала основные, потом кастомные
|
|
ALL_ITEMS = BASE_ITEMS + CUSTOM_ITEMS
|
|
print(f"Всего предметов после объединения: {len(ALL_ITEMS):,}")
|
|
|
|
# Генерируем sub-вариации цен и float для каждого предмета
|
|
generate_item_variants(num_variants=6)
|
|
print(f"Всего предметов после генерации вариаций: {len(ALL_ITEMS):,}")
|
|
|
|
ITEM_BY_ID: Dict[int, dict] = {i: item for i, item in enumerate(ALL_ITEMS)}
|
|
|
|
# ─── Глобальные кэши / индексы ──────────────────────────────────────────────
|
|
ALL_ITEMS_WITH_ID: List[dict] = []
|
|
ITEMS_BY_TYPE: Dict[str, List[dict]] = defaultdict(list)
|
|
ITEMS_BY_RARITY: Dict[str, List[dict]] = defaultdict(list)
|
|
NAME_LOWER_TO_IDS: Dict[str, List[int]] = defaultdict(list)
|
|
COLLECTIONS_INDEX: Dict[str, List[int]] = {}
|
|
CAN_BE_ASSIGNED_FROM_INDEX: Dict[str, List[int]] = defaultdict(list)
|
|
|
|
# ─── Проверка типа предмета ───────────────────────────────────────────────
|
|
def is_knife(item: dict) -> bool:
|
|
"""Проверяет, является ли предмет ножом"""
|
|
weapon = item.get("weapon", "")
|
|
return "Knife" in weapon or "Bayonet" in weapon
|
|
|
|
def is_gloves(item: dict) -> bool:
|
|
"""Проверяет, являются ли предмет перчатками"""
|
|
weapon = item.get("weapon", "")
|
|
return "Wraps" in weapon or "Gloves" in weapon or "Hand" in weapon
|
|
|
|
def is_rare_special_item(item: dict) -> bool:
|
|
"""Проверяет, является ли предмет Rare Special Item (ножи и перчатки)"""
|
|
return is_knife(item) or is_gloves(item)
|
|
|
|
# ─── Редкости ──────────────────────────────────────────────────────────────
|
|
RARITY_ORDER = {
|
|
"Consumer Grade": 0,
|
|
"Industrial Grade": 1,
|
|
"Mil-Spec": 2,
|
|
"Mil-Spec Grade": 2,
|
|
"Restricted": 3,
|
|
"Classified": 4,
|
|
"Covert": 5, # Базовый уровень для Covert
|
|
"Rare Special Item": 6, # Уровень для ножей/перчаток (конвертируется из Covert)
|
|
"Extraordinary": 7, # Для некоторых перчаток
|
|
"Contraband": 8,
|
|
"Unknown": -1,
|
|
}
|
|
|
|
def get_rarity_level(rarity: str | None, item: Optional[dict] = None) -> int:
|
|
"""
|
|
Возвращает числовой уровень редкости.
|
|
Для ножей и перчаток с редкостью Covert возвращает уровень Rare Special Item (6)
|
|
"""
|
|
if not rarity:
|
|
return -1
|
|
|
|
rarity = rarity.strip()
|
|
|
|
# Если это Extraordinary - это уже правильный уровень
|
|
if rarity == "Extraordinary":
|
|
return RARITY_ORDER["Extraordinary"]
|
|
|
|
# Если передан предмет и это Covert
|
|
if item and rarity == "Covert":
|
|
# Проверяем, является ли предмет ножом или перчатками
|
|
if is_rare_special_item(item):
|
|
return RARITY_ORDER["Rare Special Item"]
|
|
else:
|
|
return RARITY_ORDER["Covert"]
|
|
|
|
return RARITY_ORDER.get(rarity, -1)
|
|
|
|
# ─── Модели ────────────────────────────────────────────────────────────────
|
|
class ItemShort(BaseModel):
|
|
id: int
|
|
market_hash_name: str
|
|
rarity: Optional[str] = None
|
|
collection: Optional[str] = None
|
|
type: Optional[str] = None
|
|
price_rub: Optional[float] = None
|
|
price_usd: Optional[float] = None
|
|
|
|
class ContractSubmitRequest(BaseModel):
|
|
item_ids: List[int]
|
|
|
|
@field_validator("item_ids")
|
|
@classmethod
|
|
def check_length(cls, v):
|
|
if len(v) != 10:
|
|
raise ValueError("Нужно ровно 10 предметов")
|
|
return v
|
|
|
|
class TradeUpSimulationResult(BaseModel):
|
|
success: bool
|
|
input_items: List[ItemShort]
|
|
received_item: Optional[ItemShort] = None
|
|
received_float: Optional[float] = None
|
|
probability: Optional[float] = None
|
|
message: str
|
|
|
|
class CollectionItemBase(BaseModel):
|
|
base_name: str
|
|
rarity: str
|
|
item_ids_normal: List[int]
|
|
item_ids_stattrak: List[int]
|
|
item_ids_souvenir: List[int]
|
|
count_variants: int
|
|
|
|
class CollectionDetail(BaseModel):
|
|
collection: str
|
|
total_unique_skins: int
|
|
items: List[CollectionItemBase]
|
|
|
|
# ─── Вспомогательные ───────────────────────────────────────────────────────
|
|
def get_item(idx: int) -> Optional[dict]:
|
|
return ITEM_BY_ID.get(idx)
|
|
|
|
def is_final_in_collection(item: dict) -> bool:
|
|
"""Проверяет, есть ли предметы выше редкостью в той же коллекции/источнике"""
|
|
my_level = get_rarity_level(item.get("rarity"), item)
|
|
|
|
# Собираем все источники, откуда мог выпасть предмет
|
|
sources = []
|
|
|
|
# Из collection
|
|
coll = item.get("collection", "").strip()
|
|
if coll and coll != "Без коллекции":
|
|
sources.append(coll)
|
|
|
|
# Из can_be_assigned_from
|
|
assigned = item.get("can_be_assigned_from", [])
|
|
if isinstance(assigned, list):
|
|
sources.extend(assigned)
|
|
elif isinstance(assigned, str) and assigned:
|
|
sources.append(assigned)
|
|
|
|
# Из crates_names (кейсы)
|
|
crates = item.get("crates_names", [])
|
|
if crates and isinstance(crates, list):
|
|
for crate in crates:
|
|
sources.append(f"Case: {crate}")
|
|
|
|
# Проверяем каждый источник
|
|
for source in sources:
|
|
# Проверяем в COLLECTIONS_INDEX
|
|
for idx in COLLECTIONS_INDEX.get(source, []):
|
|
other = get_item(idx)
|
|
if other:
|
|
other_level = get_rarity_level(other.get("rarity"), other)
|
|
if other_level > my_level:
|
|
return False
|
|
|
|
# Проверяем в CAN_BE_ASSIGNED_FROM_INDEX
|
|
for idx in CAN_BE_ASSIGNED_FROM_INDEX.get(source, []):
|
|
other = get_item(idx)
|
|
if other:
|
|
other_level = get_rarity_level(other.get("rarity"), other)
|
|
if other_level > my_level:
|
|
return False
|
|
|
|
return True
|
|
|
|
def can_be_upgraded(items: List[dict]) -> tuple[bool, str]:
|
|
"""Проверяет, можно ли использовать эти предметы в контракте"""
|
|
if len(items) != 10:
|
|
return False, "Нужно ровно 10 предметов"
|
|
|
|
# Проверяем is_craftable
|
|
for item in items:
|
|
if not item.get("is_craftable", True):
|
|
return False, f"Предмет нельзя использовать в крафте: {item.get('market_hash_name', '???')}"
|
|
|
|
# Проверяем, что ни один предмет не является финальным
|
|
for item in items:
|
|
if is_final_in_collection(item):
|
|
return False, f"Нельзя использовать финальный предмет коллекции: {item.get('market_hash_name', '???')}"
|
|
|
|
# Проверяем тип (Normal/StatTrak/Souvenir)
|
|
types = {it.get("type") for it in items if it.get("type")}
|
|
if len(types) != 1:
|
|
return False, "Все предметы должны быть одного типа"
|
|
|
|
# Проверяем редкость с учетом типа предмета
|
|
rarities = set()
|
|
for it in items:
|
|
level = get_rarity_level(it.get("rarity"), it)
|
|
rarities.add(level)
|
|
|
|
if len(rarities) != 1:
|
|
return False, "Все предметы должны быть одной редкости"
|
|
|
|
return True, ""
|
|
|
|
def extract_base_name(market_hash_name: str) -> str:
|
|
"""Извлекает базовое имя скина без StatTrak, Souvenir и качества"""
|
|
name = market_hash_name.strip()
|
|
|
|
# Убираем префиксы
|
|
for prefix in ["StatTrak™ ", "Souvenir "]:
|
|
if name.startswith(prefix):
|
|
name = name[len(prefix):].strip()
|
|
|
|
# Убираем качество в скобках
|
|
if " (" in name:
|
|
name = name.split(" (", 1)[0].strip()
|
|
|
|
return name
|
|
|
|
def build_indexes():
|
|
"""Строит все индексы"""
|
|
global COLLECTIONS_INDEX, CAN_BE_ASSIGNED_FROM_INDEX
|
|
|
|
coll = defaultdict(list)
|
|
assigned_from = defaultdict(list)
|
|
|
|
for idx, item in ITEM_BY_ID.items():
|
|
# Индекс коллекций
|
|
c = item.get("collection", "").strip()
|
|
if c and c != "Без коллекции":
|
|
coll[c].append(idx)
|
|
|
|
# Индекс can_be_assigned_from
|
|
assigned = item.get("can_be_assigned_from", [])
|
|
if isinstance(assigned, list):
|
|
for source in assigned:
|
|
assigned_from[source].append(idx)
|
|
elif isinstance(assigned, str) and assigned:
|
|
assigned_from[assigned].append(idx)
|
|
|
|
# Индекс кейсов (crates_names)
|
|
crates = item.get("crates_names", [])
|
|
if crates and isinstance(crates, list):
|
|
for crate in crates:
|
|
collection_name = f"Case: {crate}"
|
|
coll[collection_name].append(idx)
|
|
|
|
COLLECTIONS_INDEX = dict(coll)
|
|
CAN_BE_ASSIGNED_FROM_INDEX = dict(assigned_from)
|
|
|
|
def load_and_index_data():
|
|
"""Индексирует все данные для быстрого поиска"""
|
|
global ALL_ITEMS_WITH_ID, ITEMS_BY_TYPE, ITEMS_BY_RARITY, NAME_LOWER_TO_IDS
|
|
|
|
ALL_ITEMS_WITH_ID = []
|
|
ITEMS_BY_TYPE.clear()
|
|
ITEMS_BY_RARITY.clear()
|
|
NAME_LOWER_TO_IDS.clear()
|
|
|
|
print("Создание индексов...")
|
|
for i in tqdm(range(len(ALL_ITEMS)), desc="Индексация"):
|
|
item = ALL_ITEMS[i]
|
|
item_with_id = {**item, "id": i}
|
|
ALL_ITEMS_WITH_ID.append(item_with_id)
|
|
ITEM_BY_ID[i] = item
|
|
|
|
# По типу
|
|
t = item.get("type", "Normal")
|
|
ITEMS_BY_TYPE[t].append(item_with_id)
|
|
|
|
# По редкости
|
|
r = item.get("rarity", "Unknown")
|
|
ITEMS_BY_RARITY[r].append(item_with_id)
|
|
|
|
# Поиск по имени
|
|
name = item.get("market_hash_name", "").lower().strip()
|
|
if name:
|
|
NAME_LOWER_TO_IDS[name].append(i)
|
|
|
|
base_name = extract_base_name(item.get("market_hash_name", "")).lower().strip()
|
|
if base_name and base_name != name:
|
|
NAME_LOWER_TO_IDS[base_name].append(i)
|
|
|
|
# Строим индексы коллекций
|
|
build_indexes()
|
|
|
|
print(f"Готово. Уникальных имён для поиска: {len(NAME_LOWER_TO_IDS):,}")
|
|
print(f"Типов: {len(ITEMS_BY_TYPE)}")
|
|
print(f"Редкостей: {len(ITEMS_BY_RARITY)}")
|
|
print(f"Коллекций/кейсов: {len(COLLECTIONS_INDEX)}")
|
|
print(f"Источников (can_be_assigned_from): {len(CAN_BE_ASSIGNED_FROM_INDEX)}")
|
|
|
|
def warmup_and_debug_data():
|
|
"""Диагностика данных при старте"""
|
|
print("\n" + "="*80)
|
|
print("ПРОГРЕВ ДАННЫХ И ДИАГНОСТИКА")
|
|
print("="*80)
|
|
|
|
rarities = Counter()
|
|
types_cnt = Counter()
|
|
knife_count = 0
|
|
glove_count = 0
|
|
|
|
for item in ALL_ITEMS:
|
|
rarities[item.get("rarity", "Без редкости")] += 1
|
|
types_cnt[item.get("type", "Без типа")] += 1
|
|
|
|
if is_knife(item):
|
|
knife_count += 1
|
|
elif is_gloves(item):
|
|
glove_count += 1
|
|
|
|
print(f"Ножей: {knife_count:,}")
|
|
print(f"Перчаток: {glove_count:,}")
|
|
print(f"Rare Special Items всего: {knife_count + glove_count:,}")
|
|
|
|
print("\nРедкости:")
|
|
for r, c in sorted(rarities.items(), key=lambda x: x[1], reverse=True):
|
|
print(f" {r:24} → {c:6,}")
|
|
|
|
print("\nТипы:")
|
|
for t, c in types_cnt.most_common(5):
|
|
print(f" {t:24} → {c:6,}")
|
|
|
|
# Проверяем Covert предметы
|
|
covert_items = [it for it in ALL_ITEMS if it.get("rarity") == "Covert"]
|
|
covert_knives = [it for it in covert_items if is_knife(it)]
|
|
covert_skins = [it for it in covert_items if not is_knife(it) and not is_gloves(it)]
|
|
|
|
print(f"\nCovert предметы:")
|
|
print(f" Всего: {len(covert_items):,}")
|
|
print(f" Ножи: {len(covert_knives):,}")
|
|
print(f" Обычные скины: {len(covert_skins):,}")
|
|
|
|
# Проверяем кейсы
|
|
print(f"\nКейсов (Case: *): {sum(1 for k in COLLECTIONS_INDEX.keys() if k.startswith('Case:')):,}")
|
|
|
|
print("="*80 + "\n")
|
|
|
|
# Загружаем и индексируем данные
|
|
load_and_index_data()
|
|
warmup_and_debug_data()
|
|
|
|
def calculate_average_float(items: List[dict]) -> float:
|
|
"""Вычисляет средний float из входных предметов"""
|
|
if not items:
|
|
return 0.5
|
|
return sum(it.get("mid_float_used", 0.5) for it in items) / len(items)
|
|
|
|
def get_possible_outcomes(items: List[dict]) -> tuple[List[tuple], Dict[str, float]]:
|
|
"""
|
|
Определяет возможные исходы контракта и их вероятности.
|
|
Возвращает список (item_id, probability) и словарь весов источников.
|
|
"""
|
|
input_type = items[0].get("type", "Normal")
|
|
first_item = items[0]
|
|
|
|
# Определяем текущий уровень редкости
|
|
current_level = get_rarity_level(first_item.get("rarity"), first_item)
|
|
target_level = current_level + 1
|
|
|
|
# Собираем источники из входных предметов
|
|
source_counter = Counter()
|
|
|
|
for it in items:
|
|
# Из can_be_assigned_from
|
|
assigned = it.get("can_be_assigned_from", [])
|
|
if isinstance(assigned, list):
|
|
for source in assigned:
|
|
source_counter[source] += 1
|
|
elif isinstance(assigned, str) and assigned:
|
|
source_counter[assigned] += 1
|
|
|
|
# Из collection
|
|
coll = it.get("collection", "").strip()
|
|
if coll and coll != "Без коллекции":
|
|
source_counter[coll] += 1
|
|
|
|
# Из crates_names
|
|
crates = it.get("crates_names", [])
|
|
if crates and isinstance(crates, list):
|
|
for crate in crates:
|
|
source_counter[f"Case: {crate}"] += 1
|
|
|
|
total = sum(source_counter.values())
|
|
if total == 0:
|
|
return [], {}
|
|
|
|
# Веса источников пропорциональны количеству предметов из них
|
|
source_weights = {s: cnt / total for s, cnt in source_counter.items()}
|
|
|
|
# Группируем возможные исходы по базовому имени
|
|
base_to_ids = defaultdict(list)
|
|
|
|
for source_name in source_weights.keys():
|
|
# Проверяем в COLLECTIONS_INDEX
|
|
for idx in COLLECTIONS_INDEX.get(source_name, []):
|
|
cand = get_item(idx)
|
|
if not cand:
|
|
continue
|
|
|
|
# Проверяем уровень редкости
|
|
cand_level = get_rarity_level(cand.get("rarity"), cand)
|
|
if cand_level != target_level:
|
|
continue
|
|
|
|
# Проверяем тип
|
|
if cand.get("type") != input_type:
|
|
continue
|
|
|
|
base = extract_base_name(cand["market_hash_name"])
|
|
base_to_ids[base].append(idx)
|
|
|
|
# Проверяем в CAN_BE_ASSIGNED_FROM_INDEX
|
|
for idx in CAN_BE_ASSIGNED_FROM_INDEX.get(source_name, []):
|
|
cand = get_item(idx)
|
|
if not cand:
|
|
continue
|
|
|
|
cand_level = get_rarity_level(cand.get("rarity"), cand)
|
|
if cand_level != target_level:
|
|
continue
|
|
|
|
if cand.get("type") != input_type:
|
|
continue
|
|
|
|
base = extract_base_name(cand["market_hash_name"])
|
|
base_to_ids[base].append(idx)
|
|
|
|
if not base_to_ids:
|
|
return [], source_weights
|
|
|
|
# Равномерный шанс на каждый уникальный базовый скин
|
|
num_unique = len(base_to_ids)
|
|
prob_per_base = 1.0 / num_unique
|
|
|
|
possible_outcomes = []
|
|
for base, ids_list in base_to_ids.items():
|
|
prob_per_id = prob_per_base / len(ids_list)
|
|
for idx in ids_list:
|
|
possible_outcomes.append((idx, prob_per_id))
|
|
|
|
return possible_outcomes, source_weights
|
|
|
|
def simulate_trade_up(items: List[dict]) -> TradeUpSimulationResult:
|
|
"""Симулирует контракт обмена 10→1"""
|
|
# Подготовка input_items
|
|
input_short = []
|
|
for it in items:
|
|
if isinstance(it, dict):
|
|
enriched = it.copy()
|
|
else:
|
|
enriched = {}
|
|
|
|
if "id" not in enriched:
|
|
enriched["id"] = -1
|
|
|
|
try:
|
|
short = ItemShort(**enriched)
|
|
input_short.append(short)
|
|
except Exception:
|
|
input_short.append(ItemShort(
|
|
id=enriched.get("id", -1),
|
|
market_hash_name=enriched.get("market_hash_name", "???")
|
|
))
|
|
|
|
ok, msg = can_be_upgraded(items)
|
|
if not ok:
|
|
return TradeUpSimulationResult(
|
|
success=False,
|
|
input_items=input_short,
|
|
received_item=None,
|
|
received_float=None,
|
|
probability=None,
|
|
message=msg
|
|
)
|
|
|
|
possible_outcomes, source_weights = get_possible_outcomes(items)
|
|
|
|
if not possible_outcomes:
|
|
input_type = items[0].get("type", "Normal")
|
|
first_item = items[0]
|
|
current_level = get_rarity_level(first_item.get("rarity"), first_item)
|
|
target_level = current_level + 1
|
|
|
|
# Определяем название целевой редкости
|
|
rarity_names = {v: k for k, v in RARITY_ORDER.items()}
|
|
target_rarity_name = rarity_names.get(target_level, f"уровень {target_level}")
|
|
|
|
return TradeUpSimulationResult(
|
|
success=False,
|
|
input_items=input_short,
|
|
received_item=None,
|
|
received_float=None,
|
|
probability=None,
|
|
message=f"Нет {input_type} предметов редкостью {target_rarity_name} в выбранных источниках"
|
|
)
|
|
|
|
# Выбираем случайный предмет
|
|
indices = [o[0] for o in possible_outcomes]
|
|
weights = [o[1] for o in possible_outcomes]
|
|
|
|
chosen_idx = random.choices(indices, weights=weights, k=1)[0]
|
|
received = get_item(chosen_idx)
|
|
|
|
# Вычисляем float (среднее арифметическое входных float)
|
|
avg_float = calculate_average_float(items)
|
|
|
|
# Находим вероятность выпавшего предмета
|
|
probability = next((p for idx, p in possible_outcomes if idx == chosen_idx), 0)
|
|
|
|
return TradeUpSimulationResult(
|
|
success=True,
|
|
input_items=input_short,
|
|
received_item=ItemShort(**{**received, "id": chosen_idx}),
|
|
received_float=round(avg_float, 4),
|
|
probability=round(probability * 100, 2),
|
|
message="Контракт успешен"
|
|
)
|
|
|
|
# ─── Эндпоинты ─────────────────────────────────────────────────────────────
|
|
@app.get("/items")
|
|
def get_all_items(
|
|
skip: int = Query(0, ge=0),
|
|
limit: int = Query(100, ge=1, le=1000),
|
|
search: Optional[str] = Query(None),
|
|
id: Optional[int] = Query(None),
|
|
item_type: Optional[str] = Query(None),
|
|
rarity: Optional[str] = Query(None),
|
|
min_price: Optional[float] = Query(None, ge=0),
|
|
max_price: Optional[float] = Query(None, ge=0),
|
|
wear: Optional[str] = Query(None),
|
|
):
|
|
"""Возвращает список предметов с фильтрацией и поиском"""
|
|
if id is not None:
|
|
item = get_item(id)
|
|
if item is None:
|
|
return []
|
|
return [ItemShort(**{**item, "id": id})]
|
|
|
|
# Базовый набор кандидатов
|
|
if item_type and item_type.lower() != "all":
|
|
candidates = ITEMS_BY_TYPE.get(item_type, [])
|
|
else:
|
|
candidates = ALL_ITEMS_WITH_ID
|
|
|
|
# Фильтр по редкости
|
|
if rarity:
|
|
r_norm = rarity.strip()
|
|
candidates = [it for it in candidates if it.get("rarity", "").strip() == r_norm]
|
|
|
|
# Поиск по имени
|
|
if search:
|
|
query = search.lower().strip()
|
|
found_ids = set()
|
|
|
|
# Точное совпадение
|
|
if query in NAME_LOWER_TO_IDS:
|
|
found_ids.update(NAME_LOWER_TO_IDS[query])
|
|
|
|
# Частичное совпадение по словам
|
|
search_words = query.split()
|
|
for name_lower, ids in NAME_LOWER_TO_IDS.items():
|
|
if all(word in name_lower for word in search_words):
|
|
found_ids.update(ids)
|
|
elif name_lower.startswith(query):
|
|
found_ids.update(ids)
|
|
|
|
if not found_ids:
|
|
return []
|
|
candidates = [it for it in candidates if it["id"] in found_ids]
|
|
|
|
# Фильтры по цене и wear
|
|
filtered = candidates
|
|
if min_price is not None or max_price is not None:
|
|
filtered = [
|
|
it for it in filtered
|
|
if it.get("price_rub") is not None and
|
|
(min_price is None or it["price_rub"] >= min_price) and
|
|
(max_price is None or it["price_rub"] <= max_price)
|
|
]
|
|
|
|
if wear:
|
|
wear_norm = wear.strip()
|
|
filtered = [it for it in filtered if it.get("wear", "").strip() == wear_norm]
|
|
|
|
# Пагинация
|
|
sliced = filtered[skip:skip + limit]
|
|
return [ItemShort(**it) for it in sliced]
|
|
|
|
@app.get("/filters")
|
|
def get_filters():
|
|
"""Возвращает доступные значения для фильтров"""
|
|
rarities = set()
|
|
types = set()
|
|
wears = set()
|
|
price_range = {"min": float('inf'), "max": float('-inf')}
|
|
|
|
knife_count = 0
|
|
glove_count = 0
|
|
normal_count = 0
|
|
|
|
for item in ALL_ITEMS:
|
|
if item.get("rarity"):
|
|
rarities.add(item["rarity"])
|
|
if item.get("type"):
|
|
types.add(item["type"])
|
|
if item.get("wear"):
|
|
wears.add(item["wear"])
|
|
if item.get("price_rub"):
|
|
price = item["price_rub"]
|
|
if price < price_range["min"]:
|
|
price_range["min"] = price
|
|
if price > price_range["max"]:
|
|
price_range["max"] = price
|
|
|
|
if is_knife(item):
|
|
knife_count += 1
|
|
elif is_gloves(item):
|
|
glove_count += 1
|
|
else:
|
|
normal_count += 1
|
|
|
|
return {
|
|
"rarities": sorted(list(rarities)),
|
|
"types": sorted(list(types)),
|
|
"wears": sorted(list(wears)),
|
|
"price_range": {
|
|
"min_rub": price_range["min"] if price_range["min"] != float('inf') else 0,
|
|
"max_rub": price_range["max"] if price_range["max"] != float('-inf') else 0
|
|
},
|
|
"statistics": {
|
|
"total_items": len(ALL_ITEMS),
|
|
"knives": knife_count,
|
|
"gloves": glove_count,
|
|
"normal_skins": normal_count,
|
|
"note": "Ножи и перчатки нельзя использовать в trade-up контрактах"
|
|
}
|
|
}
|
|
|
|
@app.post("/contract/submit", response_model=TradeUpSimulationResult)
|
|
def submit_contract(req: ContractSubmitRequest):
|
|
"""Выполняет симуляцию контракта обмена"""
|
|
try:
|
|
selected = []
|
|
for idx in req.item_ids:
|
|
item = get_item(idx)
|
|
if item is None:
|
|
raise HTTPException(400, detail=f"id {idx} не найден")
|
|
item_with_id = item.copy()
|
|
item_with_id["id"] = idx
|
|
selected.append(item_with_id)
|
|
return simulate_trade_up(selected)
|
|
except Exception as e:
|
|
import traceback
|
|
traceback.print_exc()
|
|
raise HTTPException(500, detail=str(e))
|
|
|
|
@app.get("/collections/")
|
|
def get_collections_list():
|
|
"""Возвращает список всех коллекций и источников"""
|
|
result = []
|
|
|
|
all_sources = set(COLLECTIONS_INDEX.keys()) | set(CAN_BE_ASSIGNED_FROM_INDEX.keys())
|
|
|
|
for name in sorted(all_sources):
|
|
ids = set()
|
|
if name in COLLECTIONS_INDEX:
|
|
ids.update(COLLECTIONS_INDEX[name])
|
|
if name in CAN_BE_ASSIGNED_FROM_INDEX:
|
|
ids.update(CAN_BE_ASSIGNED_FROM_INDEX[name])
|
|
|
|
unique_bases = set()
|
|
for idx in ids:
|
|
item = get_item(idx)
|
|
if item:
|
|
base = extract_base_name(item.get("market_hash_name", ""))
|
|
if base:
|
|
unique_bases.add(base)
|
|
|
|
result.append({
|
|
"name": name,
|
|
"total_items": len(ids),
|
|
"unique_skins": len(unique_bases),
|
|
"is_case": name.startswith("Case: ")
|
|
})
|
|
|
|
return {
|
|
"total_collections": len(result),
|
|
"collections": result
|
|
}
|
|
|
|
@app.get("/collections/all")
|
|
def get_collection_names():
|
|
"""Возвращает подробную информацию о всех коллекциях"""
|
|
def rarity_key(idx: int) -> int:
|
|
item = get_item(idx)
|
|
return get_rarity_level(item.get("rarity"), item) if item else -1
|
|
|
|
result = []
|
|
all_sources = set(COLLECTIONS_INDEX.keys()) | set(CAN_BE_ASSIGNED_FROM_INDEX.keys())
|
|
|
|
for source_name in sorted(all_sources):
|
|
ids = set()
|
|
if source_name in COLLECTIONS_INDEX:
|
|
ids.update(COLLECTIONS_INDEX[source_name])
|
|
if source_name in CAN_BE_ASSIGNED_FROM_INDEX:
|
|
ids.update(CAN_BE_ASSIGNED_FROM_INDEX[source_name])
|
|
|
|
base_names = set()
|
|
items_info = []
|
|
|
|
for i in ids:
|
|
item = get_item(i)
|
|
if item:
|
|
base = extract_base_name(item["market_hash_name"])
|
|
if base:
|
|
base_names.add(base)
|
|
|
|
items_info.append({
|
|
"id": i,
|
|
"name": item.get("market_hash_name", "Unknown"),
|
|
"rarity": item.get("rarity", "Unknown"),
|
|
"type": item.get("type", "Normal"),
|
|
"wear": item.get("wear", "Unknown"),
|
|
"price_rub": item.get("price_rub"),
|
|
"price_usd": item.get("price_usd"),
|
|
"is_craftable": item.get("is_craftable", True)
|
|
})
|
|
|
|
sorted_ids = sorted(ids, key=rarity_key)
|
|
items_info.sort(key=lambda x: (
|
|
get_rarity_level(x["rarity"], get_item(x["id"])),
|
|
x["name"]
|
|
))
|
|
|
|
result.append({
|
|
"name": source_name,
|
|
"unique_skin_count": len(base_names),
|
|
"total_variants": len(ids),
|
|
"item_ids_by_rarity": sorted_ids,
|
|
"items": items_info
|
|
})
|
|
|
|
return result
|
|
|
|
@app.get("/collection/{name}", response_model=CollectionDetail)
|
|
def get_one_collection(name: str):
|
|
"""Возвращает детальную информацию о конкретной коллекции/источнике"""
|
|
ids = set()
|
|
if name in COLLECTIONS_INDEX:
|
|
ids.update(COLLECTIONS_INDEX[name])
|
|
if name in CAN_BE_ASSIGNED_FROM_INDEX:
|
|
ids.update(CAN_BE_ASSIGNED_FROM_INDEX[name])
|
|
|
|
if not ids:
|
|
raise HTTPException(404, detail="Коллекция или источник не найдены")
|
|
|
|
groups = defaultdict(list)
|
|
for idx in ids:
|
|
item = get_item(idx)
|
|
if not item:
|
|
continue
|
|
base = extract_base_name(item["market_hash_name"])
|
|
if not base:
|
|
continue
|
|
groups[base].append((
|
|
item.get("mid_float_used", 999.9),
|
|
idx,
|
|
item.get("type", "Normal"),
|
|
item.get("rarity", "Unknown")
|
|
))
|
|
|
|
result_items = []
|
|
for base_name, variants in sorted(groups.items()):
|
|
rarity = next((v[3] for v in variants if v[3] != "Unknown"), "Unknown")
|
|
normal = [v[1] for v in sorted(variants) if v[2] == "Normal"]
|
|
stattrak = [v[1] for v in sorted(variants) if v[2] == "StatTrak"]
|
|
souvenir = [v[1] for v in sorted(variants) if v[2] == "Souvenir"]
|
|
result_items.append(CollectionItemBase(
|
|
base_name=base_name,
|
|
rarity=rarity,
|
|
item_ids_normal=normal,
|
|
item_ids_stattrak=stattrak,
|
|
item_ids_souvenir=souvenir,
|
|
count_variants=len(variants)
|
|
))
|
|
|
|
return CollectionDetail(
|
|
collection=name,
|
|
total_unique_skins=len(result_items),
|
|
items=result_items
|
|
)
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
"""Проверка работоспособности сервера"""
|
|
return {
|
|
"status": "ok",
|
|
"items_loaded": len(ALL_ITEMS),
|
|
"base_items": len(BASE_ITEMS),
|
|
"custom_items": len(CUSTOM_ITEMS)
|
|
}
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=13668) |