src/pyscripts/run.py

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'] 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



@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"
        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)