AI_python_yoooger/Ai_tottle/map_cut.py

import os
import tempfile
import uuid
import aiohttp
import yaml
from minio import Minio
from minio.error import S3Error
import subprocess
import sys
import json
from osgeo import gdal, osr
from miniohelp import load_config,upload_file

# 异步下载文件函数
async def download_file(url: str, output_path: str) -> bool:
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                if response.status == 200:
                    with open(output_path, 'wb') as f:
                        f.write(await response.read())
                    return True
                else:
                    print(f"下载失败: {url}")
                    return False
    except Exception as e:
        print(f"下载错误: {e}")
        return False

# 上传文件到 MinIO
def upload_to_minio(local_file: str, bucket: str, object_name: str,minio_client: Minio):
    try:
        minio_client.fput_object(bucket, object_name, local_file)
    except S3Error as e:
        print(f"上传失败: {local_file} -> {object_name} 错误: {e}")

def generate_tiles_from_tif_cli(
    input_tif: str,
    output_dir: str,
    zoom_levels: str = "0-18",
    processes: int = 4,
    verbose: bool = True
):
    # Step 1: 调用 gdal2tiles
    args = [
        sys.executable,
        "-m", "osgeo_utils.gdal2tiles",
        "-z", zoom_levels,
        "--processes", str(processes)
    ]
    if verbose:
        args.append("-v")
    args += [input_tif, output_dir]
    print(f"Running: {' '.join(args)}")
    subprocess.run(args, check=True)

    # Step 2: 提取地理信息
    ds = gdal.Open(input_tif)
    gt = ds.GetGeoTransform()
    width = ds.RasterXSize
    height = ds.RasterYSize

    minx = gt[0]
    maxy = gt[3]
    pixel_width = gt[1]
    pixel_height = gt[5]
    maxx = minx + width * pixel_width
    miny = maxy + height * pixel_height

    # 尝试解析坐标系 EPSG 编码
    srs = osr.SpatialReference(wkt=ds.GetProjection())
    try:
        proj_epsg = int(srs.GetAttrValue("AUTHORITY", 1))
    except:
        proj_epsg = 4326  # fallback 默认值

    # Step 3: 构造 JSON 数据
    bounds = {
        "west": minx,
        "south": miny,
        "east": maxx,
        "north": maxy
    }

    json_data = {
        "bounds": bounds,
        "contentType": "image/png",
        "generatetool": "custom_script@yourdomain.com/gdal2tiles",
        "latLonBounds": bounds,  # 和 bounds 相同
        "maxzoom": int(zoom_levels.split("-")[-1]),
        "minzoom": int(zoom_levels.split("-")[0]),
        "proj": proj_epsg,
        "tilesize": 256,
        "tiletrans": "google",
        "type": "image",
        "ziped": False
    }

    # Step 4: 保存为 JSON 文件
    json_path = output_dir.rstrip("\\/") + ".json"
    with open(json_path, 'w', encoding='utf-8') as f:
        json.dump(json_data, f, indent=2)

    print(f"瓦片生成成功，元数据已保存到: {json_path}")

# 主处理函数
async def process_tiling(tif_url: str = None, prj_url: str = None, zoom_levels: str = "1-18", bucket: str = "300",config: str = "112.44.103.230") -> dict:
    # 创建唯一任务ID
    task_id = str(uuid.uuid4())
    temp_dir = tempfile.mkdtemp()
    tif_path = os.path.join(temp_dir, f"{task_id}.tif")
    prj_path = os.path.join(temp_dir, f"{task_id}.prj") if prj_url else None
    output_dir = os.path.join(temp_dir, f"tiles_{task_id}")
    os.makedirs(output_dir, exist_ok=True)

    # 下载 TIF
    if tif_url:
        tif_ok = await download_file(tif_url, tif_path)
        if not tif_ok:
            raise Exception("TIF 下载失败")

    # 下载 PRJ（如果有）
    if prj_url:
        prj_ok = await download_file(prj_url, prj_path)
        if prj_ok:
            # GDAL 要求 PRJ 文件与 TIF 同名（不同后缀）
            fixed_prj_path = tif_path.replace('.tif', '.prj')
            os.rename(prj_path, fixed_prj_path)
        else:
            print("PRJ 下载失败，继续处理...")

    # 切片
    generate_tiles_from_tif_cli(tif_path, output_dir, zoom_levels=zoom_levels, processes=4, verbose=True)
    
    # 上传结果到 MinIO
    minio_base_path = f"maps/{task_id}/"
    for root, _, files in os.walk(output_dir):
        for file in files:
            local_path = os.path.join(root, file)
            rel_path = os.path.relpath(local_path, output_dir)
            object_name = os.path.join(minio_base_path, rel_path).replace("\\", "/")
            upload_to_minio(local_path, bucket, object_name)

    minio_info = load_config(config)
    # 清洗 endpoint，去掉 http:// 或 https:// 前缀
    endpoint = minio_info["MinIOEndpoint"].replace("http://", "").replace("https://", "")
    
    # 返回构建的瓦片URL信息
    base_url = f"http://{endpoint}//{bucket}/{minio_base_path}"
    min_zoom, max_zoom = zoom_levels.split("-")

    return {
        "tile_url": {
            "base_url": base_url,
            "format": "{z}/{x}/{y}.png",
            "preview_url": f"{base_url}openlayers.html",
            "min_zoom": min_zoom,
            "max_zoom": max_zoom,
            "tilejson_url": f"{base_url}tilemapresource.xml"
        }
    }

# 示例调用
if __name__ == "__main__":
    # tif_url = "https://your-server.com/sample.tif"
    # prj_url = "https://your-server.com/sample.prj"  # 可选
    # result = asyncio.run(process_tiling(tif_url, prj_url, zoom_levels="5-15"))
    # print(result)
    
    input_tif = r"e:/new.tif"
    output_dir = r"D:/work/AI_Python/cut"
    generate_tiles_from_tif_cli(input_tif, output_dir, zoom_levels="5-15", processes=4, verbose=True)