227 lines
7.2 KiB
Python
227 lines
7.2 KiB
Python
from fastapi import FastAPI, HTTPException
|
||
from pydantic import BaseModel
|
||
import os
|
||
from ftplib import FTP
|
||
import win32gui
|
||
import win32con
|
||
import logging
|
||
import time
|
||
import re
|
||
|
||
app = FastAPI()
|
||
|
||
|
||
class FileUploadRequest(BaseModel):
|
||
filename: str
|
||
|
||
|
||
# FTP 服务器配置信息
|
||
ftp_host = '47.119.33.43'
|
||
ftp_port = 21
|
||
ftp_user = 'ftpuser'
|
||
ftp_password = 'FTPftp123'
|
||
ftp_directory = 'ThreeTest/XT/Before/'
|
||
|
||
|
||
def find_child_window(parent_hwnd, class_name):
|
||
def enum_child_windows(hwnd, lparam):
|
||
class_name = win32gui.GetClassName(hwnd)
|
||
if class_name == lparam:
|
||
child_hwnds.append(hwnd)
|
||
return True
|
||
|
||
child_hwnds = []
|
||
win32gui.EnumChildWindows(parent_hwnd, enum_child_windows, class_name)
|
||
return child_hwnds
|
||
|
||
|
||
def find_child_window_by_partial_title(parent_hwnd, partial_title):
|
||
def enum_child_windows(hwnd, lparam):
|
||
# 获取窗口的标题
|
||
title = win32gui.GetWindowText(hwnd)
|
||
# 检查标题是否包含指定的部分标题
|
||
if partial_title in title:
|
||
child_hwnds.append(hwnd)
|
||
return True
|
||
|
||
child_hwnds = []
|
||
win32gui.EnumChildWindows(parent_hwnd, enum_child_windows, None)
|
||
return child_hwnds
|
||
|
||
|
||
def find_child_window_by_title(parent_hwnd, title):
|
||
def enum_child_windows(hwnd, lparam):
|
||
if win32gui.GetWindowText(hwnd) == lparam:
|
||
child_hwnds.append(hwnd)
|
||
return True
|
||
|
||
child_hwnds = []
|
||
win32gui.EnumChildWindows(parent_hwnd, enum_child_windows, title)
|
||
return child_hwnds
|
||
|
||
|
||
# 获取 ComboBox 的句柄
|
||
def get_combobox_handle(parent_handle):
|
||
combo_handle = win32gui.FindWindowEx(parent_handle, 0, "ComboBox", None)
|
||
if combo_handle == 0:
|
||
raise Exception("ComboBox not found")
|
||
return combo_handle
|
||
|
||
|
||
# 获取 ComboBox 中的所有选项
|
||
def get_combobox_items(combo_handle):
|
||
count = win32gui.SendMessage(combo_handle, win32con.CB_GETCOUNT, 0, 0)
|
||
items = []
|
||
for i in range(count):
|
||
length = win32gui.SendMessage(combo_handle, win32con.CB_GETLBTEXTLEN, i, 0)
|
||
buffer = win32gui.PyMakeBuffer(length + 1)
|
||
win32gui.SendMessage(combo_handle, win32con.CB_GETLBTEXT, i, buffer)
|
||
byte_data = buffer.tobytes()
|
||
|
||
# 尝试多种编码方式
|
||
text = None
|
||
# 尝试多种编码方式，包括utf-16le
|
||
for encoding in ['utf-16le', 'utf-8', 'gbk', 'latin-1']:
|
||
try:
|
||
decoded_text = byte_data.decode(encoding).rstrip('\x00')
|
||
# 如果解码后的文本看起来是合理的，就接受它
|
||
if any(char.isprintable() for char in decoded_text):
|
||
text = decoded_text
|
||
break
|
||
except UnicodeDecodeError:
|
||
continue
|
||
|
||
# 如果所有解码方式都失败，或者内容仍有大量乱码，显示为十六进制字符串
|
||
if text is None or not all(char.isprintable() or char.isspace() for char in text):
|
||
text = byte_data.hex()
|
||
|
||
items.append(text)
|
||
return items
|
||
|
||
|
||
# 获取当前选定项
|
||
def get_combobox_selected(combo_handle):
|
||
index = win32gui.SendMessage(combo_handle, win32con.CB_GETCURSEL, 0, 0)
|
||
if index == -1:
|
||
return None
|
||
length = win32gui.SendMessage(combo_handle, win32con.CB_GETLBTEXTLEN, index, 0)
|
||
buffer = win32gui.PyMakeBuffer(1024) # 调整缓冲区大小
|
||
win32gui.SendMessage(combo_handle, win32con.CB_GETLBTEXT, index, buffer)
|
||
|
||
# 尝试多种编码方式进行解码
|
||
for encoding in ['utf-16le', 'utf-8', 'latin-1']:
|
||
try:
|
||
# 解码
|
||
raw_text = buffer.tobytes().decode(encoding, errors='ignore').rstrip('\x00')
|
||
# 使用正则表达式查找 "数字 + 月" 格式的部分
|
||
match = re.search(r'\d+月', raw_text)
|
||
if match:
|
||
return match.group()
|
||
# 使用正则表达式提取有效字符（中文、数字、字母和常见标点）
|
||
filtered_text = re.findall(r'[\w\u4e00-\u9fa5]+', raw_text)
|
||
# 返回匹配到的第一个有效部分
|
||
if filtered_text:
|
||
return filtered_text[0].strip()
|
||
except UnicodeDecodeError:
|
||
continue # 尝试下一个编码方式
|
||
|
||
# 如果所有解码方式都失败，返回 None
|
||
return None
|
||
|
||
|
||
# 设置 ComboBox 的值
|
||
def set_combobox_value(combo_handle, value):
|
||
items = get_combobox_items(combo_handle)
|
||
try:
|
||
index = items.index(value)
|
||
win32gui.SendMessage(combo_handle, win32con.CB_SETCURSEL, index, 0)
|
||
except ValueError:
|
||
raise Exception("Value not found in ComboBox")
|
||
|
||
|
||
def set_path_and_save(filename):
|
||
parent_hwnd = win32gui.FindWindow(None, '保存Excel文件')
|
||
|
||
if parent_hwnd == 0:
|
||
raise HTTPException(status_code=404, detail="没有找到保存报告的窗口，请检查！")
|
||
|
||
combo_handle = get_combobox_handle(parent_hwnd)
|
||
#logging.info(f"ComboBox Items: {get_combobox_items(combo_handle)}")
|
||
logging.info(f"Current Selected: {get_combobox_selected(combo_handle)}")
|
||
local_file_path = "C:\\RationalDMIS64\\Output\\" + get_combobox_selected(combo_handle) + "\\" + filename
|
||
logging.info(f"设置路径: {local_file_path}")
|
||
|
||
path_hwnds = find_child_window(parent_hwnd, 'Edit')
|
||
|
||
if not path_hwnds:
|
||
raise HTTPException(status_code=404, detail="未找到路径框")
|
||
|
||
path_hwnd = path_hwnds[0]
|
||
win32gui.SendMessage(path_hwnd, win32con.WM_SETTEXT, 0, filename)
|
||
|
||
button_hwnds = find_child_window_by_title(parent_hwnd, '保存(&S)')
|
||
|
||
if not button_hwnds:
|
||
raise HTTPException(status_code=404, detail="未找到保存按钮")
|
||
|
||
save_button_hwnd = button_hwnds[0]
|
||
win32gui.PostMessage(save_button_hwnd, win32con.BM_CLICK, 0, 0)
|
||
|
||
return local_file_path
|
||
|
||
|
||
def wait_for_file_to_save(filepath, timeout=30):
|
||
start_time = time.time()
|
||
|
||
while time.time() - start_time < timeout:
|
||
if os.path.isfile(filepath):
|
||
return True
|
||
time.sleep(0.1)
|
||
|
||
return False
|
||
|
||
|
||
def upload_file_to_ftp(local_file):
|
||
if not os.path.isfile(local_file):
|
||
raise HTTPException(status_code=204, detail="文件未找到")
|
||
|
||
ftp = FTP()
|
||
try:
|
||
ftp.connect(ftp_host, ftp_port)
|
||
ftp.login(ftp_user, ftp_password)
|
||
ftp.cwd(ftp_directory)
|
||
with open(local_file, 'rb') as file:
|
||
ftp.storbinary(f'STOR {os.path.basename(local_file)}', file)
|
||
return True
|
||
except Exception as e:
|
||
print(f"文件上传失败: {e}")
|
||
return False
|
||
finally:
|
||
ftp.quit()
|
||
|
||
|
||
@app.post("/get/check/report")
|
||
async def upload_file(request: FileUploadRequest):
|
||
# 设置路径框并点击保存
|
||
local_file_path = set_path_and_save(request.filename)
|
||
logging.info(f"文件上传请求: {request.filename}")
|
||
logging.info(f"文件保存路径: {local_file_path}")
|
||
|
||
# 等待文件保存完成
|
||
if not wait_for_file_to_save(local_file_path):
|
||
raise HTTPException(status_code=500, detail="文件保存超时")
|
||
|
||
# 上传文件到 FTP
|
||
success = upload_file_to_ftp(local_file_path)
|
||
if success:
|
||
ftp_file_path = os.path.join(ftp_directory, request.filename)
|
||
return {"ftp_file_path": ftp_file_path}
|
||
else:
|
||
raise HTTPException(status_code=500, detail="文件上传失败")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import uvicorn
|
||
|
||
uvicorn.run(app, host="0.0.0.0", port=8999)
|