from fastapi import FastAPI, Path, Query, Response, BackgroundTasks from fastapi.responses import FileResponse, HTMLResponse import gradio as gr import sys import re import os project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')) sys.path.append(project_root) from src.pyscripts import file_process, oss, img_process from datetime import datetime from collections import namedtuple from pydantic import BaseModel import requests # from easydict import EasyDict Person = namedtuple('Person', ['name', 'title', 'category', 'conference', 'respond_person']) CUSTOM_PATH = "/gradio" app = FastAPI() pdf_path = "src/assets/InvitationTemplate.pdf" output_path = 'output/output.pdf' person = Person(name="雷军", title="先生", category="观礼", conference="世界教育者大会", respond_person="Eazy") # data = { # 'name': '雷军', # 'title': '先生', # "conference": "世界教育者大会" # } @app.get("/") def read_main(): return {"message": "This is your main app"} # @app.post("/getInvitationPDF/{name}/{title}/{conference}") @app.get("/getInvitationPDF") def getInvitationPDF(name, title, conference, category, respond_person): person = Person(name=name, title=title, category=category, conference=conference, respond_person=respond_person) formatted_date = datetime.now().strftime('%Y-%m-%d') file_name = f"致{name+title}的WWEC2024邀请函.pdf" file_dir = f"invitation/{category}" object_name = file_dir + "/" + file_name pdfStream = file_process.writeInvitationPDF(person) # with open(f"output/{file_name}", 'wb') as f: # f.write(pdfStream.getvalue()) oss.upload_file_stream_to_s3(pdfStream, "host", object_name) file_url = f"https://isidoresong.us.kg/{object_name}" return {"url": file_url} class GuestPosterInfo(BaseModel): exhibitor: str logo_link: str slogan: str content: str @app.post("/getExhibitorPoster") def getExhibitorPoster(item: GuestPosterInfo): logo_link = item.logo_link exhibitor = item.exhibitor slogan = item.slogan content = item.content img_stream = img_process.make_exhibitor_poster(logo_link, slogan, content) file_dir = "ExhibitorPoster" object_name = file_dir + "/" + exhibitor + ".jpg" # with open(f"output/{exhibitor + '.jpg'}", 'wb') as f: # f.write(img_stream.getvalue()) oss.upload_file_stream_to_s3(img_stream, "host", object_name) file_url = f"https://isidoresong.us.kg/{object_name}" return {"url": file_url} class GuestPosterInfo(BaseModel): name: str name_pinyin: str photo_link: str original_photo_link: str titles: str @app.post("/getGuestPoster") def getGuestPoster(item: GuestPosterInfo): photo_link = item.photo_link original_photo_link = item.original_photo_link name = item.name name_pinyin = item.name_pinyin titles = item.titles img_stream = img_process.make_guest_poster(name, name_pinyin, photo_link, original_photo_link, titles) file_dir = "GuestPoster" object_name = file_dir + "/" + name + ".jpg" # f.write(img_stream.getvalue()) oss.upload_file_stream_to_s3(img_stream, "host", object_name) file_url = f"https://pub-429c75a96a8f4597984dd7ebc525d652.r2.dev/{object_name}" return {"url": file_url} class ConferenceInfo(BaseModel): name: str eng_name: str full_name: str start_time: str end_time: str member: list address: str = None guest: list = None class ConferencePosterInfo(BaseModel): conference_info: ConferenceInfo conference_rundown:list def format_time(start_time, end_time): start_datetime = datetime.strptime(start_time, "%Y/%m/%d %H:%M:%S") end_datetime = datetime.strptime(end_time, "%Y/%m/%d %H:%M:%S") start_period = "上午" if start_datetime.hour < 12 else "下午" end_period = "上午" if end_datetime.hour < 12 else "下午" formatted_start_time = start_datetime.strftime(f"%Y年%m月%d日{start_period}%H:%M") formatted_end_time = end_datetime.strftime(f"{end_period}%H:%M") return f"{formatted_start_time}-{formatted_end_time}" @app.post("/getConferencePoster") def getConferencePoster(item: ConferencePosterInfo): # item.conference_rundown = [list(i) for i in item.conference_rundown] sorted_conference_rundown = sorted(item.conference_rundown, key=lambda x: x['end_time']) img_stream = img_process.make_conference_poster( item.conference_info.name, item.conference_info.eng_name, item.conference_info.full_name, [ *item.conference_info.member, ["论坛时间", format_time(item.conference_info.start_time, item.conference_info.end_time)], ["论坛地点", "国家会展中心4.2号馆 " + item.conference_info.address] ], sorted_conference_rundown) file_dir = "ConferencePoster" object_name = file_dir + "/" + item.conference_info.name + ".jpg" if os.path.exists("src/assets/output"): with open(f"src/assets/output/{item.conference_info.name + '.jpg'}", 'wb') as f: f.write(img_stream.getvalue()) else: oss.upload_file_stream_to_s3(img_stream, "host", object_name) file_url = f"https://pub-429c75a96a8f4597984dd7ebc525d652.r2.dev/{object_name}" return {"url": file_url} @app.post("/getGuestComposePoster") def getGuestComposePoster(item:ConferenceInfo): # item.conference_rundown = [list(i) for i in item.conference_rundown] sorted_conference_guest = sorted(item.guest, key=lambda x: x['rank']) img_stream = img_process.getGuestComposePoster( item.name, item.eng_name, sorted_conference_guest) file_dir = "ConferenceGuestPoster" object_name = file_dir + "/" + item.name + ".jpg" if os.path.exists("src/assets/output"): with open(f"src/assets/output/{item.name + '.jpg'}", 'wb') as f: f.write(img_stream.getvalue()) else: oss.upload_file_stream_to_s3(img_stream, "host", object_name) file_url = f"https://pub-429c75a96a8f4597984dd7ebc525d652.r2.dev/{object_name}" return {"url": file_url} class GuestEScreenInfo(BaseModel): room_types: list person: object @app.post("/getGuestEScreen") def getGuestEScreen(item: GuestEScreenInfo): file_dir = "GuestEScreen" file_urls = [] for room_type in item.room_types: rt = "" if room_type.startswith("M"): rt = "M" elif re.match(r"\d+", room_type): rt = "half" else: rt = "full" print(item.person) img_stream = img_process.getGuestEScreen(rt, item.person) object_name = file_dir + "/" + rt + "_" + item.person["name"] + ".jpg" if os.path.exists("src/assets/output"): with open(f"src/assets/output/{item.person['name'] + '.jpg'}", 'wb') as f: f.write(img_stream.getvalue()) else: oss.upload_file_stream_to_s3(img_stream, "host", object_name) file_url = f"https://pub-429c75a96a8f4597984dd7ebc525d652.r2.dev/{object_name}" file_urls.append(file_url) return {"url": "\n".join(file_urls)} interface = gr.Interface(lambda x: "Hello, " + x + "!", "textbox", "textbox") # app = gr.(app, interface, path=CUSTOM_PATH) if __name__ == "__main__": MOCK_URL = "https://mock.jsont.run/tjTCU1IWOGoIunWRovNJt" res = requests.get(MOCK_URL) mock_data = res.json() # mock_data = ConferencePosterInfo(**mock_data) # getConferencePoster(mock_data) print(mock_data) # mock_data = ConferenceInfo(**mock_data["conference_info"]) # getGuestComposePoster(mock_data) mock_data = GuestEScreenInfo(**mock_data) getGuestEScreen(mock_data)