Spaces:
Sleeping
Sleeping
| 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": "世界教育者大会" | |
| # } | |
| def read_main(): | |
| return {"message": "This is your main app"} | |
| # @app.post("/getInvitationPDF/{name}/{title}/{conference}") | |
| 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 | |
| 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 | |
| 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}" | |
| 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} | |
| 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'] if 'rank' in x else 0) | |
| 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 | |
| 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" | |
| 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/1H9o2vvkIBMB83oeVQd1q" | |
| MOCK_URL = "https://mock.jsont.run/c3oQ2XLm1ZIHoVWZa18Ui" | |
| res = requests.get(MOCK_URL) | |
| mock_data = res.json() | |
| print(mock_data) | |
| mock_data = ConferencePosterInfo(**mock_data) | |
| getConferencePoster(mock_data) | |
| # mock_data = ConferenceInfo(**mock_data["conference_info"]) | |
| # getGuestComposePoster(mock_data) | |
| # mock_data = GuestEScreenInfo(**mock_data) | |
| # getGuestEScreen(mock_data) | |