from fastapi import FastAPI from fastapi.responses import PlainTextResponse, FileResponse from collections import OrderedDict import threading import time import os from datetime import datetime, timedelta app = FastAPI() import requests from fastapi import Request PAGE_SIZE = 50 DATA_URL = "https://script.google.com/macros/s/AKfycbxPYAG3xM-ju3V8DockaI2UbOIpUd4w5XO2FZsdOagsUzfNKjX_wph1DA-sqeUPhVmJ/exec" def format_time(ms): try: dt = datetime.utcfromtimestamp(ms / 1000) + timedelta(hours=3) return dt.strftime("%Y/%m/%d %H:%M:%S") except: return "unknown" def format_user(user, info): lines = [] lines.append(f"Наказуемый: {user}") # МУТЫ if info.get("mutes"): lines.append("-Муты:") for m in info["mutes"]: lines.append(f"&f- &fПричина: &c{m.get('r')}&f") lines.append(f"&f- &fДлительность: {m.get('l')} сек") lines.append(f"&f- &fВыдал: &a{m.get('w')}&f") lines.append(f"&f- &fДата: &e{format_time(m.get('s'))}&f") else: lines.append("Муты: нет") # БАНЫ if info.get("bans"): lines.append("Баны:") for b in info["bans"]: lines.append(f"&f- &fПричина: &c{b.get('r')}&f") lines.append(f"&f- &fДлительность: {b.get('l')} сек&f") lines.append(f"&f- &fВыдал: &c{b.get('w')}&f") lines.append(f"&f- &fДата: &e{format_time(b.get('s'))}&f") else: lines.append("Баны: нет") return "\n".join(lines) @app.get("/players") async def players(request: Request): query = request.query_params.get("get") start_raw = request.query_params.get("start") start = 0 if start_raw and start_raw.isdigit(): start = int(start_raw) if not query: return PlainTextResponse("use ?get=all or ?get=nickname") try: data = requests.get(DATA_URL).json() except Exception as e: return PlainTextResponse(f"error loading json: {e}") # 🔥 ITEM MODE is_item = query.endswith("_item") if is_item: query = query.replace("_item", "") def format_item(user, info): lore = [] lore.append(f'{{color:"white",italic:0b,text:"Наказуемый: {user}"}}') if info.get("mutes"): lore.append('{color:"white",italic:0b,text:"- Муты:"}') for m in info["mutes"]: lore.append(f'{{color:"red",italic:0b,text:"- Причина: {m.get("r")}"}}') lore.append(f'{{color:"white",italic:0b,text:"- Длительность: {m.get("l")}"}}') lore.append(f'{{color:"green",italic:0b,text:"- Выдал: {m.get("w")}"}}') lore.append(f'{{color:"yellow",italic:0b,text:"- Дата: {format_time(m.get("s"))}"}}') else: lore.append('{color:"white",italic:0b,text:"- Муты: нет"}') if info.get("bans"): lore.append('{color:"white",italic:0b,text:"- Баны:"}') for b in info["bans"]: lore.append(f'{{color:"red",italic:0b,text:"- Причина: {b.get("r")}"}}') lore.append(f'{{color:"white",italic:0b,text:"- Длительность: {b.get("l")}"}}') lore.append(f'{{color:"green",italic:0b,text:"- Выдал: {b.get("w")}"}}') lore.append(f'{{color:"yellow",italic:0b,text:"- Дата: {format_time(b.get("s"))}"}}') else: lore.append('{color:"white",italic:0b,text:"Баны: нет"}') lore_str = ",".join(lore) return f'{{components:{{"minecraft:lore":[{lore_str}]}},count:1,id:"minecraft:writable_book"}}' # 📦 ALL if query == "all": result = [] items = list(data.items()) # 🔥 pagination chunk = items[start:start + PAGE_SIZE] for user, info in chunk: if is_item: result.append(format_item(user, info)) else: result.append(format_user(user, info)) if not chunk: return PlainTextResponse("end") return PlainTextResponse("%newline%".join(result)) # 🔍 ONE USER if query not in data: return PlainTextResponse("not found") if is_item: return PlainTextResponse(format_item(query, data[query])) return PlainTextResponse(format_user(query, data[query])) import os import io import uuid import time from typing import List, Dict import requests from fastapi import FastAPI, HTTPException, Request from fastapi.responses import PlainTextResponse from pydantic import BaseModel, Field from PIL import Image # ---------- Хранилище блоков (в памяти) ---------- blocks_storage: Dict[str, List[str]] = {} # request_id → список текстовых блоков storage_timestamps: Dict[str, float] = {} # request_id → время создания def clean_old_entries(max_age_seconds: int = 600): """Удаляет блоки старше max_age_seconds секунд (10 минут).""" now = time.time() for rid in list(storage_timestamps.keys()): if now - storage_timestamps[rid] > max_age_seconds: del blocks_storage[rid] del storage_timestamps[rid] # ---------- Вспомогательные функции ---------- def rgb_to_hex(rgb_tuple): return '#{:02X}{:02X}{:02X}'.format(rgb_tuple[0], rgb_tuple[1], rgb_tuple[2]) def process_image(image_bytes: bytes, chars_limit: int = 15000): """Кодирует изображение в блоки цветов.""" try: raw_img = Image.open(io.BytesIO(image_bytes)) except Exception as e: raise ValueError(f"Не удалось открыть изображение: {e}") # Прозрачность → белый фон if raw_img.mode in ('RGBA', 'LA') or (raw_img.mode == 'P' and 'transparency' in raw_img.info): background = Image.new("RGBA", raw_img.size, (255, 255, 255, 255)) img = Image.alpha_composite(background, raw_img.convert('RGBA')).convert('RGB') else: img = raw_img.convert('RGB') width, height = img.size encoded_rows = [ ",".join(rgb_to_hex(img.getpixel((x, y))) for x in range(width)) for y in range(height) ] # Разбивка на блоки с ограничением длины blocks = [] current_rows = [] current_len = 0 for row in encoded_rows: rlen = len(row) if not current_rows: current_rows.append(row) current_len = rlen else: sep = 2 # ";\n" if current_len + sep + rlen <= chars_limit: current_rows.append(row) current_len += sep + rlen else: blocks.append(";\n".join(current_rows)) current_rows = [row] current_len = rlen if current_rows: blocks.append(";\n".join(current_rows)) return blocks, width, height # ---------- Модели запроса/ответа ---------- class ProcessRequest(BaseModel): url: str = Field(..., description="Прямая ссылка на PNG или JPG изображение") chars_limit: int = Field(15000, description="Максимальное число символов в одном блоке", ge=100) class ProcessResponse(BaseModel): request_id: str width: int height: int blocks_count: int chars_limit: int block_urls: List[str] # полные URL для скачивания каждого блока # ---------- FastAPI ---------- app = FastAPI( title="Minecraft Image Encoder API", description="Принимает URL изображения (POST) и возвращает ссылки на текстовые блоки с цветами пикселей", version="1.1.0" ) @app.get("/") def root(): return {"message": "Minecraft Image Encoder API. Используйте POST /process с JSON: {\"url\": \"...\", \"chars_limit\": 15000}"} @app.post("/process", response_model=ProcessResponse) async def process_image_url(req_data: ProcessRequest, request: Request): """Загружает изображение по URL, кодирует и создаёт блоки с отдельными ссылками.""" # Проверка расширения if not req_data.url.lower().endswith(('.png', '.jpg', '.jpeg')): raise HTTPException(status_code=400, detail="Поддерживаются только PNG и JPEG изображения") # Загрузка изображения try: resp = requests.get(req_data.url, timeout=15, stream=True) resp.raise_for_status() content = b'' max_size = 10 * 1024 * 1024 for chunk in resp.iter_content(chunk_size=8192): content += chunk if len(content) > max_size: raise HTTPException(status_code=400, detail="Изображение слишком большое (макс. 10 МБ)") except requests.RequestException as e: raise HTTPException(status_code=400, detail=f"Ошибка загрузки: {e}") # Обработка try: blocks, width, height = process_image(content, req_data.chars_limit) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception: raise HTTPException(status_code=500, detail="Внутренняя ошибка обработки") # Сохраняем блоки и генерируем ссылки request_id = str(uuid.uuid4()) blocks_storage[request_id] = blocks storage_timestamps[request_id] = time.time() # Формируем полные URL для каждого блока block_urls = [] for i in range(len(blocks)): # request.url_for даёт относительный путь, но можно собрать абсолютный url_path = request.url_for("get_block", request_id=request_id, index=i) block_urls.append(str(url_path)) # в Hugging Face это будет полный URL # Очистка старых записей clean_old_entries() return ProcessResponse( request_id=request_id, width=width, height=height, blocks_count=len(blocks), chars_limit=req_data.chars_limit, block_urls=block_urls ) @app.get("/block/{request_id}/{index}", response_class=PlainTextResponse) async def get_block(request_id: str, index: int): """Отдаёт текстовое содержимое одного блока.""" clean_old_entries() if request_id not in blocks_storage: raise HTTPException(status_code=404, detail="Блоки не найдены или устарели") blocks = blocks_storage[request_id] if index < 0 or index >= len(blocks): raise HTTPException(status_code=404, detail="Неверный индекс блока") return blocks[index] if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860) # ========================= # 🚀 RUN # ========================= if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)