| import os |
| import zipfile |
| import tempfile |
| import json |
| import re |
| import subprocess |
| import time |
| import uuid |
| import UnityPy |
| from UnityPy.enums import CompressionFlags |
| from fastapi import FastAPI, UploadFile, File, BackgroundTasks |
| from fastapi.responses import FileResponse, JSONResponse |
| from fastapi.middleware.cors import CORSMiddleware |
| from typing import Optional |
|
|
| app = FastAPI() |
|
|
| app.add_middleware( |
| CORSMiddleware, |
| allow_origins=["*"], |
| allow_credentials=True, |
| allow_methods=["*"], |
| allow_headers=["*"], |
| ) |
|
|
| jobs = {} |
|
|
| def process_repack(job_id: str, asset_path: str, zip_path: str, original_filename: str): |
| try: |
| jobs[job_id] = {"status": "processing", "progress": 5, "message": "Oyun dosyaları çözümleniyor..."} |
| env = UnityPy.load(asset_path) |
| |
| filename_to_translations = {} |
| with zipfile.ZipFile(zip_path, 'r') as zipf: |
| for file_info in zipf.infolist(): |
| filename = file_info.filename |
| if "__MACOSX" in filename or filename.split("/")[-1].startswith("._"): continue |
| if filename.endswith(".json"): |
| try: |
| content = zipf.read(filename) |
| filename_to_translations[filename.split("/")[-1]] = json.loads(content.decode('utf-8')) |
| except json.JSONDecodeError: pass |
|
|
| total_objects = len(env.objects) |
| processed = 0 |
| start_time = time.time() |
|
|
| for obj in env.objects: |
| processed += 1 |
| if processed % 50 == 0: |
| elapsed = time.time() - start_time |
| rate = processed / elapsed if elapsed > 0 else 1 |
| remaining_seconds = int((total_objects - processed) / rate) |
| prog = 5 + int((processed / total_objects) * 45) |
| |
| jobs[job_id].update({ |
| "progress": prog, |
| "message": f"Diyaloglar yamalanıyor... (Kalan Süre: ~{remaining_seconds} sn)" |
| }) |
|
|
| if obj.type.name in ["TextAsset", "MonoBehaviour"]: |
| asset_name = getattr(obj.assets_file, 'name', 'assets') |
| asset_name = "".join([c if c.isalnum() else "_" for c in asset_name]) |
| obj_name = f"{asset_name}_{obj.type.name}_{obj.path_id}" |
| target_filename = f"{obj_name}_raw.json" |
| |
| if target_filename in filename_to_translations: |
| translations = filename_to_translations[target_filename] |
| if obj.type.name == "MonoBehaviour": |
| raw_data = obj.get_raw_data() |
| patched_data = bytearray(raw_data) |
| for orig_str, trans_str in translations.items(): |
| if orig_str == trans_str: continue |
| orig_bytes = orig_str.encode('utf-8') |
| orig_pad = (4 - (len(orig_bytes) % 4)) % 4 |
| orig_block = len(orig_bytes).to_bytes(4, 'little') + orig_bytes + (b'\x00' * orig_pad) |
| trans_bytes = trans_str.encode('utf-8') |
| trans_pad = (4 - (len(trans_bytes) % 4)) % 4 |
| trans_block = len(trans_bytes).to_bytes(4, 'little') + trans_bytes + (b'\x00' * trans_pad) |
| patched_data = patched_data.replace(orig_block, trans_block) |
| obj.set_raw_data(bytes(patched_data)) |
|
|
| jobs[job_id].update({ |
| "progress": 60, |
| "message": "Sıkıştırma algoritmaları deneniyor... (Sunucu hızına göre 1-3 dakika sürebilir)" |
| }) |
| |
| output_path = os.path.join(tempfile.gettempdir(), f"repacked_{job_id}_{original_filename}") |
| |
| compression_error = "" |
| try: |
| |
| packed_data = env.file.save(packer="lzma") |
| except Exception as e1: |
| try: |
| |
| packed_data = env.file.save(packer="lz4") |
| except Exception as e2: |
| try: |
| |
| packed_data = env.file.save(packer=CompressionFlags.LZMA) |
| except Exception as e3: |
| |
| compression_error = f"LZMA_Str: {str(e1)} | LZ4_Str: {str(e2)} | Enum: {str(e3)}" |
| packed_data = env.file.save() |
| |
| with open(output_path, "wb") as f: |
| f.write(packed_data) |
|
|
| if compression_error: |
| final_msg = f"Dosya hazırlandı AMA Sıkıştırılamadı! Hata Logu: {compression_error}" |
| else: |
| final_msg = "İşlem %100 Tamamlandı ve Orijinal Boyuta Sıkıştırıldı! Dosya İndiriliyor..." |
|
|
| jobs[job_id].update({ |
| "status": "completed", |
| "progress": 100, |
| "message": final_msg, |
| "file_path": output_path, |
| "filename": original_filename |
| }) |
|
|
| except Exception as e: |
| jobs[job_id].update({"status": "error", "message": f"Kritik Hata: {str(e)}"}) |
|
|
|
|
| |
|
|
| @app.post("/generate_dll") |
| async def generate_dll(binary_file: UploadFile = File(...), metadata_file: UploadFile = File(...)): |
| temp_dir = tempfile.mkdtemp() |
| binary_path = os.path.join(temp_dir, binary_file.filename) |
| metadata_path = os.path.join(temp_dir, metadata_file.filename) |
| with open(binary_path, "wb") as f: f.write(await binary_file.read()) |
| with open(metadata_path, "wb") as f: f.write(await metadata_file.read()) |
| output_dir = os.path.join(temp_dir, "output") |
| os.makedirs(output_dir, exist_ok=True) |
| subprocess.run(["/app/dumper/Il2CppDumper", binary_path, metadata_path, output_dir], capture_output=True) |
| zip_path = os.path.join(temp_dir, "DummyDlls.zip") |
| with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf: |
| for root, _, files in os.walk(output_dir): |
| for file in files: |
| if file.endswith(".dll"): zipf.write(os.path.join(root, file), arcname=file) |
| return FileResponse(zip_path, media_type="application/zip", filename="DummyDlls.zip") |
|
|
| @app.post("/extract") |
| async def extract_dialogues(file: UploadFile = File(...), assembly_file: Optional[UploadFile] = File(None)): |
| temp_dir = tempfile.mkdtemp() |
| asset_path = os.path.join(temp_dir, file.filename) |
| with open(asset_path, "wb") as f: f.write(await file.read()) |
| env = UnityPy.load(asset_path) |
| zip_filename = os.path.join(temp_dir, f"{file.filename}_dialogues.zip") |
| with zipfile.ZipFile(zip_filename, 'w', zipfile.ZIP_DEFLATED) as zipf: |
| for obj in env.objects: |
| if obj.type.name in ["TextAsset", "MonoBehaviour"]: |
| asset_name = getattr(obj.assets_file, 'name', 'assets') |
| asset_name = "".join([c if c.isalnum() else "_" for c in asset_name]) |
| obj_name = f"{asset_name}_{obj.type.name}_{obj.path_id}" |
| if obj.type.name == "MonoBehaviour": |
| raw_data = obj.get_raw_data() |
| found_strings = {} |
| i = 28 |
| while i <= len(raw_data) - 4: |
| length = int.from_bytes(raw_data[i:i+4], 'little') |
| if 2 <= length <= 5000 and (i + 4 + length) <= len(raw_data): |
| str_bytes = raw_data[i+4 : i+4+length] |
| pad_len = (4 - (length % 4)) % 4 |
| if (i + 4 + length + pad_len) <= len(raw_data): |
| padding = raw_data[i+4+length : i+4+length+pad_len] |
| if all(b == 0 for b in padding): |
| try: |
| text = str_bytes.decode('utf-8') |
| if len(text.strip()) > 1 and any(c.isalpha() for c in text): found_strings[text] = text |
| i += 4 + length + pad_len |
| continue |
| except UnicodeDecodeError: pass |
| i += 4 |
| if found_strings: zipf.writestr(f"{obj_name}_raw.json", json.dumps(found_strings, indent=4, ensure_ascii=False).encode('utf-8')) |
| elif obj.type.name == "TextAsset": |
| try: |
| content = obj.read().text.encode('utf-8') |
| zipf.writestr(f"{obj_name}.txt", content) |
| except: pass |
| return FileResponse(zip_filename, media_type="application/zip", filename=f"{file.filename}_dialogues.zip") |
|
|
| @app.post("/repack_async") |
| async def repack_async(background_tasks: BackgroundTasks, asset_file: UploadFile = File(...), mod_zip: UploadFile = File(...)): |
| job_id = str(uuid.uuid4()) |
| temp_dir = tempfile.mkdtemp() |
| asset_path = os.path.join(temp_dir, asset_file.filename) |
| zip_path = os.path.join(temp_dir, mod_zip.filename) |
| |
| with open(asset_path, "wb") as f: f.write(await asset_file.read()) |
| with open(zip_path, "wb") as f: f.write(await mod_zip.read()) |
| |
| background_tasks.add_task(process_repack, job_id, asset_path, zip_path, asset_file.filename) |
| jobs[job_id] = {"status": "queued", "progress": 0, "message": "İşlem sıraya alındı, başlıyor..."} |
| return {"job_id": job_id} |
|
|
| @app.get("/status/{job_id}") |
| async def get_status(job_id: str): |
| if job_id not in jobs: return JSONResponse(status_code=404, content={"error": "Görev bulunamadı"}) |
| return jobs[job_id] |
|
|
| @app.get("/download/{job_id}") |
| async def download_file(job_id: str): |
| job = jobs.get(job_id) |
| if not job or job.get("status") != "completed": return JSONResponse(status_code=400, content={"error": "Dosya hazır değil"}) |
| return FileResponse(job["file_path"], media_type="application/octet-stream", filename=job["filename"]) |
|
|
