|
|
import os
|
|
|
import uuid
|
|
|
import pandas as pd
|
|
|
|
|
|
from flask import render_template, request, send_file
|
|
|
from common.logger import system_logger
|
|
|
|
|
|
|
|
|
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
|
|
UPLOAD_FOLDER = os.path.join(BASE_DIR, "uploads")
|
|
|
OUTPUT_FOLDER = os.path.join(BASE_DIR, "output")
|
|
|
|
|
|
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
|
|
|
os.makedirs(OUTPUT_FOLDER, exist_ok=True)
|
|
|
|
|
|
|
|
|
# =========================
|
|
|
# 1. 读取文件（CSV/TXT/Excel）
|
|
|
# =========================
|
|
|
def read_any(file_path):
|
|
|
|
|
|
ext = file_path.lower().split(".")[-1]
|
|
|
|
|
|
# Excel
|
|
|
if ext in ["xls", "xlsx"]:
|
|
|
df = pd.read_excel(file_path, dtype=str)
|
|
|
return df
|
|
|
|
|
|
# CSV / TXT
|
|
|
if ext in ["csv", "txt"]:
|
|
|
|
|
|
seps = [",", "|", "\t", " "]
|
|
|
|
|
|
for sep in seps:
|
|
|
try:
|
|
|
df = pd.read_csv(
|
|
|
file_path,
|
|
|
sep=sep,
|
|
|
engine="python",
|
|
|
dtype=str,
|
|
|
encoding="utf-8"
|
|
|
)
|
|
|
|
|
|
if df.shape[1] > 1:
|
|
|
return df
|
|
|
|
|
|
except:
|
|
|
continue
|
|
|
|
|
|
# fallback GBK
|
|
|
for sep in seps:
|
|
|
try:
|
|
|
df = pd.read_csv(
|
|
|
file_path,
|
|
|
sep=sep,
|
|
|
engine="python",
|
|
|
dtype=str,
|
|
|
encoding="gbk"
|
|
|
)
|
|
|
|
|
|
if df.shape[1] > 1:
|
|
|
return df
|
|
|
|
|
|
except:
|
|
|
continue
|
|
|
|
|
|
raise Exception("无法解析TXT/CSV结构")
|
|
|
|
|
|
raise Exception("不支持的文件类型")
|
|
|
|
|
|
|
|
|
# =========================
|
|
|
# 2. 转换核心
|
|
|
# =========================
|
|
|
def convert_file(df, target):
|
|
|
|
|
|
out_name = f"convert_{uuid.uuid4().hex}"
|
|
|
|
|
|
# Excel
|
|
|
if target == "excel":
|
|
|
path = os.path.join(OUTPUT_FOLDER, out_name + ".xlsx")
|
|
|
df.to_excel(path, index=False)
|
|
|
|
|
|
# CSV
|
|
|
elif target == "csv":
|
|
|
path = os.path.join(OUTPUT_FOLDER, out_name + ".csv")
|
|
|
df.to_csv(path, index=False, encoding="utf-8-sig")
|
|
|
|
|
|
# TXT（默认 tab）
|
|
|
elif target == "txt":
|
|
|
path = os.path.join(OUTPUT_FOLDER, out_name + ".txt")
|
|
|
df.to_csv(path, index=False, sep="\t", encoding="utf-8")
|
|
|
|
|
|
else:
|
|
|
raise Exception("未知转换类型")
|
|
|
|
|
|
return path, os.path.basename(path)
|
|
|
|
|
|
|
|
|
# =========================
|
|
|
# 3. 页面
|
|
|
# =========================
|
|
|
def file_convert_page():
|
|
|
|
|
|
preview = None
|
|
|
download_file = None
|
|
|
message = None
|
|
|
|
|
|
if request.method == "POST":
|
|
|
|
|
|
try:
|
|
|
file = request.files.get("file")
|
|
|
target = request.form.get("target")
|
|
|
|
|
|
if not file or not file.filename:
|
|
|
raise Exception("请选择文件")
|
|
|
|
|
|
filename = f"{uuid.uuid4()}_{file.filename}"
|
|
|
path = os.path.join(UPLOAD_FOLDER, filename)
|
|
|
file.save(path)
|
|
|
|
|
|
df = read_any(path)
|
|
|
|
|
|
preview = df.head(100).fillna("").to_html(
|
|
|
classes="table table-dark table-hover",
|
|
|
index=False
|
|
|
)
|
|
|
|
|
|
out_path, download_file = convert_file(df, target)
|
|
|
|
|
|
message = f"转换成功：{len(df)} 行"
|
|
|
|
|
|
system_logger.info(
|
|
|
f"[FileConvert] {file.filename} -> {target}"
|
|
|
)
|
|
|
|
|
|
except Exception as e:
|
|
|
message = str(e)
|
|
|
system_logger.exception("[FileConvert ERROR]")
|
|
|
|
|
|
return render_template(
|
|
|
"file_convert.html",
|
|
|
preview=preview,
|
|
|
download_file=download_file,
|
|
|
message=message
|
|
|
)
|
|
|
|
|
|
|
|
|
# =========================
|
|
|
# download
|
|
|
# =========================
|
|
|
def file_convert_download(filename):
|
|
|
|
|
|
path = os.path.join(OUTPUT_FOLDER, filename)
|
|
|
|
|
|
system_logger.info(f"[FileConvertDownload] {filename}")
|
|
|
|
|
|
return send_file(path, as_attachment=True) |
