app.py

import json
import os
from datetime import datetime
from zoneinfo import ZoneInfo

import gradio as gr
from extract import extract
import call_logger
from pgsoft.pgdate.date_utils import beijing
import call_pgai
from utils import normalize_text

#######################
# proxy version
#######################
proxy_version = "1.0.0-2025-05-26-a"

t = datetime.now()
t = t.astimezone(ZoneInfo("Asia/Shanghai"))
print(f"[Beijing]: {t.replace(microsecond=0)}")
t = t.astimezone(ZoneInfo("America/Los_Angeles"))
print(f"[Seattle]: {t.replace(microsecond=0)}")


identity = os.environ.get("identity", "local")
print(f"identity: {identity}")
db_token = os.environ.get("db_token")
if db_token:
    print(db_token[:5])

game_list = [
    "matchn",
    "house",
    "watermelon",
    "snake",
]


def run(info, game, nlp_command):
    """event handler"""
    # reuse hf_token field as json string
    user, source, username, _ = extract(info)
    if nlp_command is None:
        return "command is required"
    nlp_command = normalize_text(nlp_command)
    if nlp_command == "":
        return "invalid command"

    service_start = beijing()
    print(f"[{service_start}] service starts >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")
    print(f"[{user}] [{game}] [{nlp_command}]")

    call_pgai.from_cache = True
    outp = call_pgai.call_pgai(nlp_command, game)
    if outp is None:
        return "no output"
    if isinstance(outp, str):
        return outp
    # add proxy version info to the output
    outp["timestamp"] = beijing().__str__()
    outp["proxy-version"] = proxy_version
    outp["user"] = user
    outp["username"] = username
    outp["game"] = game
    outp["source"] = source
    outp["cache"] = call_pgai.from_cache
    call_logger.call_logger(outp, identity, db_token)
    service_end = beijing()
    timecost = service_end.timestamp() - service_start.timestamp()
    print(
        f"[{service_end}] service ends, costs {timecost:.2f}s "
        + "<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<\n\n"
    )
    return json.dumps(outp, indent=4)


demo = gr.Interface(
    fn=run,
    inputs=[
        "text",
        gr.Radio(
            game_list,
            value=game_list[0],
            info="Which game you want the AI to support?",
        ),
        "text",
    ],
    outputs="text",
    title="Pgai",
    allow_flagging="never",
)

demo.launch()