import os import shutil 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 middleware.minio_util import downFile, upload_file, upload_folder from miniohelp import load_config # 异步下载文件函数 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() print(f"temp_dir {os.path.abspath(temp_dir)}") 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) tif_ok = downFile(tif_url,) tif_path = os.path.abspath(tif_ok) 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}/" tiles_dir="" tiles_json="" for entry in os.listdir(temp_dir): full_path = os.path.join(temp_dir, entry) # 获取完整路径 if os.path.isdir(full_path): tiles_dir=upload_folder(full_path,None) elif os.path.isfile(full_path): tiles_json,pic_type= upload_file(full_path,None) else: print(f"其他类型: {entry}") # 可能是符号链接等 try: shutil.rmtree(temp_dir) print(f"成功删除文件夹: {temp_dir}") except FileNotFoundError: print(f"文件夹不存在: {temp_dir}") except PermissionError: print(f"权限不足，无法删除: {temp_dir}") except Exception as e: print(f"删除失败: {e}") # 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) # print(f"filefile {os.path.abspath(file)}") # upload_file(local_path,None) # minio_info = load_config(config) # # 清洗 endpoint，去掉 http:// 或 https:// 前缀 # endpoint = minio_info["MinIOEndpoint"].replace("http://", "").replace("https://", "") # # # 返回构建的瓦片URL信息 # base_url = f"http://{endpoint}o = 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", "tiles_dir":tiles_dir, "tiles_json":tiles_json, # "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" input_tif = r"D:\sharefile\甘江社区-s1-align.tif\tile_0_0.tif" output_dir = r"D:\project\AI-PYTHON\Ai_tottle\touying" generate_tiles_from_tif_cli(input_tif, output_dir, zoom_levels="5-18", processes=4, verbose=True)