import json import os from datetime import datetime from xml.dom.minidom import Document from django.conf import settings import requests from PIL import Image from django.core import paginator from django.http import HttpResponse, JsonResponse from DCbackend.settings import MEDIA_ROOT, oss_key_id, oss_key_secret, oss_endPoint, oss_bucket, oss_url import uuid import os import oss2 from DCbackend.utils.MyEncoder import MyEncoder # def success(data): # result = {"status": 200, "message": "success", "data": data} # return HttpResponse(json.dumps(result, cls=MyEncoder, indent=4), content_type="application/json") def success(data, message="success"): result = { "status": 200, "message": message, "data": data } return HttpResponse( json.dumps(result, cls=MyEncoder, indent=4), content_type="application/json" ) def success_image_upload(url, filename, message="success"): result = { "uploaded": 1, "fileName": filename, "url": url } return HttpResponse( json.dumps(result, cls=MyEncoder, indent=4), content_type="application/json" ) def fail(message): return HttpResponse(JsonResponse({"status": 999, "message": message}), content_type="application/json") def fail(message=None): if message is None: message = [] return HttpResponse(JsonResponse({"status": 999, "message": message}), content_type="application/json") def pageData(page, pageSize, totalPage, totalRecord, dataList): return {"page": page, "pageSize": pageSize, "totalPage": totalPage, "totalRecord": totalRecord, "dataList": dataList} def handle_uploaded_file(file): date = datetime.now().strftime("%Y%m%d%H%M%S") filename = f"{date}_{file.name}" if not os.path.exists(MEDIA_ROOT): os.mkdir(MEDIA_ROOT) path = os.path.join(MEDIA_ROOT, filename) with open(path, 'wb+') as destination: for chunk in file.chunks(): destination.write(chunk) return path def handle_uploaded_image(image_file): # 确保 MEDIA_ROOT 存在 if not os.path.exists(settings.MEDIA_ROOT): os.makedirs(settings.MEDIA_ROOT) # 检查文件是否为图像 try: img = Image.open(image_file) img.verify() # 验证图像文件 except: raise ValueError("Uploaded file is not a valid image") # 生成唯一的文件名 file_extension = os.path.splitext(image_file.name)[1].lower() unique_filename = f"{uuid.uuid4().hex}{file_extension}" # 构建完整的文件路径 relative_path = os.path.join('images', unique_filename) full_path = os.path.join(settings.MEDIA_ROOT, relative_path) # 确保 'images' 子目录存在 os.makedirs(os.path.dirname(full_path), exist_ok=True) # 保存图像文件 with open(full_path, 'wb+') as destination: for chunk in image_file.chunks(): destination.write(chunk) # 可选:处理图像(例如,调整大小) # with Image.open(full_path) as img: # img.thumbnail((800, 800)) # 调整图像大小到最大 800x800 # img.save(full_path) # 返回相对路径,可以用于数据库存储或构建URL return relative_path # 上传文件到 OSS # 这里的file_path就是上面我们保存到local_url容器里面的dst def oss_upload_file(file_path): # 通过basename方法得到路径最后的文件名,也就是图片名(火影忍者.jpg) file_name = os.path.basename(file_path) # 构造oss中的路径,我这里就是:anime/火影忍者.jpg # 创建 OSS 链接 auth = oss2.Auth(oss_key_id, oss_key_secret) bucket = oss2.Bucket(auth, oss_endPoint, oss_bucket) oss_path = "document/" + file_name # 上传文件,以二进制方式(rb)打开图片上传 with open(file_path, 'rb') as file_obj: result = bucket.put_object(oss_path, file_obj) return oss_url + "/" + oss_path # return result def oss_upload_image(image_file): timestamp = datetime.now().strftime("%Y%m%d%H%M%S") random_uuid = uuid.uuid4().hex[:8] # 使用UUID的前8个字符 if isinstance(image_file, str): # 如果 image_file 是字符串(文件路径) _, file_extension = os.path.splitext(image_file) local_path = os.path.join(settings.MEDIA_ROOT, image_file) # 使用完整路径 else: # 如果 image_file 是文件对象 _, file_extension = os.path.splitext(image_file.name) local_path = handle_uploaded_file(image_file) # 假设这个函数返回完整路径 # 确保文件存在 if not os.path.exists(local_path): raise FileNotFoundError(f"File not found: {local_path}") # 生成新的文件名 new_file_name = f"{timestamp}_{random_uuid}{file_extension}" # 创建 OSS 连接 auth = oss2.Auth(oss_key_id, oss_key_secret) bucket = oss2.Bucket(auth, oss_endPoint, oss_bucket) # 构造 OSS 路径 oss_path = f"images/{new_file_name}" # 上传文件到 OSS with open(local_path, 'rb') as file_obj: result = bucket.put_object(oss_path, file_obj) # 如果 image_file 是文件对象,可能需要删除临时文件 if not isinstance(image_file, str): os.remove(local_path) # 返回完整的 URL return f"{oss_url}/{oss_path}" def getHttpPost(): url = "https://www.h3yun.com/OpenApi/Invoke" payload = { "ActionName": "LoadBizObjects", "SchemaCode": "isaxp2no91kwgggtzmnfsmf90", "Filter": "{\"FromRowNum\": 0,\"RequireCount\": false,\"ReturnItems\": [], \"SortByCollection\": [],\"ToRowNum\": 500, \"Matcher\": { \"Type\": \"And\", \"Matchers\": []}}" } headers = { "EngineCode": "bddua5a6od64p2nu", "EngineSecret": "Az24wdw8f1dgZw5j7UeGivgcJuRLuaxqQ0banHM6YLkD/oa2uNIH8g==", "Content-Type": "application/json", 'content-type': "application/json" } response = requests.request("POST", url, json=payload, headers=headers) if response.status_code == 200: data = response.json() content = json.dumps(data, cls=MyEncoder, indent=4) filename = "customer.txt" if not os.path.exists(MEDIA_ROOT): os.mkdir(MEDIA_ROOT) path = os.path.join(MEDIA_ROOT, filename) with open(path, "w", encoding='utf-8') as file: file.write(content) with open(path, "r", encoding='utf-8') as openFile: fileContent = openFile.read() fileContent = fileContent.replace('\'', '\"') fileContent = fileContent.replace('true', 'True') fileContent = fileContent.replace('false', 'False') fileContent = fileContent.replace('null', '""') item = eval(fileContent) result = [] if item['Successful']: for it in item['ReturnData']['BizObjectArray']: result.append({ "objectId": it["ObjectId"], "Name": it['Name'], "creator": it["OwnerId"], "SeqNo": it['SeqNo'], "mobile": it["F0000005"], 'customerName': it['F0000007'], 'enName': it['F0000002'], 'CreatedTime': it['CreatedTime'], 'ModifiedTime': it['ModifiedTime'], 'step': it['F0000010'], 'address': it['F0000003'], 'contact': it['F0000008'], }) # js = json.loads(json.dumps(temp,cls=MyEncoder, indent=4)) return result return else: return False def formatDateTime(time): if time == "": return datetime.now() date_object = datetime.datetime.strptime(time, "%Y/%m/%d %H:%M:%S") formatted_date = date_object.strftime("%Y-%m-%d %H:%M:%S") return formatted_date def saveDocFile(fileName, htmlText): date = datetime.now().strftime("%Y%m%d%H%M%S") filename = f"{date}_{fileName}" # if not os.path.exists(MEDIA_ROOT): # os.mkdir(MEDIA_ROOT) path = os.path.join(MEDIA_ROOT, filename) # output = pypandoc.convert_text(htmlText, 'docx', format='html') # doc = Document() # doc.add_paragraph(output) # doc.save(path) return path # 获取文档文件名称 def getDocxFileName(ext="docx"): date = datetime.now().strftime("%Y%m%d%H%M%S") fileName = "output." + ext filename = f"{date}_{fileName}" path = os.path.join(MEDIA_ROOT, filename) return path def downloadFile(id, fileUrl): local_filename = os.path.join(MEDIA_ROOT, str(id)+".xlsx") if os.path.exists(local_filename): return local_filename response = requests.get(fileUrl) # Check if the request was successful if response.status_code == 200: # Open the local file in binary write mode and write the content of the remote file to it with open(local_filename, 'wb') as file: file.write(response.content) return local_filename else: return ""