app.py

import os
import gradio as gr
import pymongo
import pandas as pd
from bson import ObjectId
from dotenv import load_dotenv
import requests

load_dotenv()

video_status = {
    "Not Started": "NOT_STARTED",
    "Dubbing Started": "DUBBING_STARTED",
    "Waiting QA": "WAITING_QA",
    "Sent to Client": "SENT_TO_CLIENT",
    "Approved": "APPROVED",
    "Rejected": "REJECTED",
    "Churned": "CHURNED",
    "Re-review Requested": "RE_REVIEW_REQUESTED"
}

ALL_LANGUAGES = ['Hindi', 'English', 'Bengali', 'Kannada', 'Spanish', 'French', 'German', 'Italian', 'Tamil', 'Telugu']

language_maps = {
    "Hindi": "hi",
    "English": "en",
    "Spanish": "es",
    "French": "fr",
    "German": "de",
    "Italian": "it",
    "Tamil": "ta",
    "Telugu": "te",
    "Bengali": "bn",
    "Kannada": "kn",
    "hi": "Hindi",
    "en": "English",
    "es": "Spanish",
    "fr": "French",
    "de": "German",
    "it": "Italian",
    "ta": "Tamil",
    "te": "Telugu",
    "bn": "Bengali",
    "kn": "Kannada"
}

REVIEW_LIST_API = "https://api.dev-env.deepsync.co/api/v1/dashboard/dubbed/videos/review/requested"

def get_translators(platform):
    if platform == "dev":
        client = pymongo.MongoClient(os.environ.get("MONGO_DEV_URI"))
        source_db = client.deepsync_stage
    else:
        client = pymongo.MongoClient(os.environ.get("MONGO_URI"))
        source_db = client.deepsync_prod
    translators = source_db.translators.find({})
    transaltors_data = []
    for translator in translators:
        try:
            data = source_db.companies.find_one({"_id": translator["translator"]})
            transaltors_data.append({
                "_id": str(translator["_id"]),
                "Name": data["name"],
                "Email": data["email"],
                "Source Language": ", ".join(list(map(lambda x: language_maps[x], translator["sourceAccent"]))),
                "Target Language": ", ".join(list(map(lambda x: language_maps[x], translator["targetAccent"])))
            })
        except:
            pass
    return pd.DataFrame(transaltors_data)


def add_translator(platform, email_id, source_langs, target_langs, contact_num, pricing):
    if platform == "dev":
        client = pymongo.MongoClient(os.environ.get("MONGO_DEV_URI"))
        source_db = client.deepsync_stage
    else:
        client = pymongo.MongoClient(os.environ.get("MONGO_URI"))
        source_db = client.deepsync_prod
    
    try:
        company_id_validity = source_db.companies.find_one({"email": email_id})
    except:
        company_id_validity = None
    if company_id_validity is None:
        return "Invalid Company ID. Unable to add translator."
    
    if source_langs is None:
        source_accents = []
    else:
        source_accents = [language_maps[l] for l in source_langs]
    if target_langs is None:
        target_accents = []
    else:
        target_accents = [language_maps[l] for l in target_langs]

    response = source_db.translators.insert_one({
        "translator": company_id_validity["_id"],
        "targetAccent": target_accents,
        "sourceAccent": source_accents,
        "contactNum": None if len(contact_num.strip()) == 0 else contact_num.strip(),
        "pricing": None if pricing is None or pricing == -1 else pricing
    })
    return f"Translator added succesfully, ID : {str(response.inserted_id)}"


def get_translator_names(platform):
    if platform == "dev":
        client = pymongo.MongoClient(os.environ.get("MONGO_DEV_URI"))
        source_db = client.deepsync_stage
    else:
        client = pymongo.MongoClient(os.environ.get("MONGO_URI"))
        source_db = client.deepsync_prod
    translators = source_db.translators.find({})
    translators_names = []
    for translator in translators:
        try:
            data = source_db.companies.find_one({"_id": translator["translator"]})
            translators_names.append(data["name"])
        except:
            pass
    return gr.Dropdown.update(choices=translators_names)


def select_current_data(platform, translator_name):
    if translator_name is None:
        return "Please select a translator."
    if platform == "dev":
        client = pymongo.MongoClient(os.environ.get("MONGO_DEV_URI"))
        source_db = client.deepsync_stage
    else:
        client = pymongo.MongoClient(os.environ.get("MONGO_URI"))
        source_db = client.deepsync_prod
    companies = source_db.companies.find({"name": translator_name})
    translators = list(source_db.translators.find({}))
    translator_company_ids = {tr["translator"]: tr["_id"] for tr in translators if "translator" in tr}
    company_id = None
    for company in companies:
        if company["_id"] in translator_company_ids.keys():
            company_id = company["_id"]
            break
    for tr in translators:
        if tr["_id"] == translator_company_ids[company_id]:
            translator = tr
            break
    return gr.CheckboxGroup.update(value=[language_maps[l] for l in translator["sourceAccent"]]), gr.CheckboxGroup.update(value=[language_maps[l] for l in translator["targetAccent"]]), gr.Textbox.update(value="" if translator["contactNum"] is None else translator["contactNum"]), gr.Textbox.update(value=-1 if translator["pricing"] is None else translator["pricing"])


def update_translator(platform, translator_name, source_languages, target_languages, contact_num, pricing):
    if translator_name is None:
        return "Please select a translator."
    if platform == "dev":
        client = pymongo.MongoClient(os.environ.get("MONGO_DEV_URI"))
        source_db = client.deepsync_stage
    else:
        client = pymongo.MongoClient(os.environ.get("MONGO_URI"))
        source_db = client.deepsync_prod
    companies = source_db.companies.find({"name": translator_name})
    translators = list(source_db.translators.find({}))
    translator_company_ids = {tr["translator"]: tr["_id"] for tr in translators if "translator" in tr}
    company_id = None
    for company in companies:
        if company["_id"] in translator_company_ids.keys():
            company_id = company["_id"]
            break
    source_accents = [language_maps[l] for l in source_languages]
    target_accents = [language_maps[l] for l in target_languages]

    updateable_data = {
        "sourceAccent": source_accents,
        "targetAccent": target_accents,
        "contactNum": None if len(contact_num.strip()) == 0 else contact_num.strip(),
        "pricing": None if pricing is None or pricing == -1 else pricing
    }

    response = source_db.translators.update_one(
        {"_id": translator_company_ids[company_id]}, 
        {"$set": updateable_data}
    )
    return f"Updated translator accent succesfully."


def get_videos_under_review(status_keys):
    headers = {
        "x-api-key": os.environ.get('X-API-KEY'),
        "Content-Type": "application/json"
    }
    if status_keys is None or len(status_keys) == 0:
        API_URL = "https://api.dev-env.deepsync.co/api/v1/dashboard/dubbed/videos/review/requested"
        response = requests.get(API_URL, headers=headers)
        if response.status_code == 200:
            options = [f"{str(v['_id'])} / {v['title']}" for v in response.json()['data'] if 'title' in v]
            return gr.Dropdown.update(choices=options, value=None)
        else:
            return gr.Dropdown.update(choices=[], value=None)
    else:
        options = []
        for status in status_keys:
            API_URL = f"https://api.dev-env.deepsync.co/api/v1/dashboard/dubbed/videos/review/requested/{video_status[status]}"
            response = requests.get(API_URL, headers=headers)
            if response.status_code == 200:
                current_options = [f"{str(v['_id'])} / {v['title']}" for v in response.json()['data'] if 'title' in v]
                options.extend(current_options)
        return gr.Dropdown.update(choices=options, value=None)


def get_traslators_assign(video):
    video_id = video.split("/", maxsplit=1)[0].strip()
    client = pymongo.MongoClient(os.environ.get("MONGO_DEV_URI"))
    source_db = client.deepsync_stage
    video = source_db.dubbedvideos.find_one({"_id": ObjectId(video_id)})
    source_language = video["sourceLanguage"].split("-")[0]
    target_language = video["targetLanguage"].split("-")[0]
    translators = source_db.translators.find({"sourceAccent": {"$elemMatch": {"$eq": source_language}}, "targetAccent": {"$elemMatch": {"$eq": target_language}}})
    translators_list = []
    for translator in translators:
        company = source_db.companies.find_one({"_id": translator["translator"]})
        translators_list.append(f"{company['name']} / {str(translator['_id'])}")
    
    button = gr.Button.update(interactive=False)
    if "videoReviewInformation" in video:
        if video["videoReviewInformation"]["status"] in {"NOT_STARTED", "DUBBING_STARTED", "WAITING_QA"}:
            button = gr.Button.update(interactive=True)
    return gr.Dropdown.update(choices=translators_list), language_maps[source_language], language_maps[target_language], button


def assign_translator(translator, video, role):
    translator_id = translator.rsplit("/", maxsplit=1)[1].strip()
    video_id = video.split("/", maxsplit=1)[0].strip()
    headers = {
        "x-api-key": os.environ.get('X-API-KEY'),
        "Content-Type": "application/json"
    }
    role = "QA" if role == "QA" else "translator"
    API_URL = f"https://api.dev-env.deepsync.co/api/v1/dashboard/dubbed/video/assign/{role}"
    data = {
        "videoId": video_id,
        "userId": translator_id
    }
    response = requests.post(API_URL, json=data, headers=headers)
    return str(response.status_code)


with gr.Blocks() as demo:
    gr.Markdown("# Translators Admin Dashboard")
    with gr.Tab("List"):
        source_platform = gr.Radio(["dev", "prod"], value="prod", label="Source Platform")
        refresh_list = gr.Button("Load")
        translators_list = gr.Dataframe(label="Translators", interactive=False, max_rows=20, overflow_row_behaviour="paginate")
    with gr.Tab("Add"):
        gr.Markdown("Add a new translator")
        source_platform_add = gr.Radio(["dev", "prod"], value="dev", label="Source Platform")
        email_id = gr.Textbox(label="Email ID", lines=1)
        source_languages = gr.CheckboxGroup(ALL_LANGUAGES, label="Source Languages")
        target_languages = gr.CheckboxGroup(ALL_LANGUAGES, label="Target Languages")
        contact_num_box = gr.Textbox(label="Contact Number", lines=1)
        pricing_box = gr.Number(label="Pricing", value=-1)
        submit = gr.Button("Submit")
        result = gr.Textbox(label="Logs")
    with gr.Tab("Update"):
        gr.Markdown("Update Translator")
        source_platform_update = gr.Radio(["dev", "prod"], value="dev", label="Source Platform")
        refresh_update_platform = gr.Button("Load")
        translator_name = gr.Dropdown([], label="Translators")
        source_languages_update = gr.CheckboxGroup(ALL_LANGUAGES, label="Source Languages")
        target_languages_update = gr.CheckboxGroup(ALL_LANGUAGES, label="Target Languages")
        contact_num_update = gr.Textbox(label="Contact Number", lines=1)
        pricing_update = gr.Number(label="Pricing")
        update_language = gr.Button("Update")
        result_update = gr.Textbox(label="Logs")
    with gr.Tab("Assign"):
        gr.Markdown("Assign Translator / QA")
        filter_videos_assign = gr.CheckboxGroup(['Not Started', 'Dubbing Started', 'Waiting QA', 'Sent to Client', 'Approved', 'Rejected', 'Churned', 'Re-review Requested'], label="Filter videos. Do not select any of them for videos without request info.")
        refresh_assign = gr.Button("Load Videos")
        under_review_videos_list = gr.Dropdown([], label="Video IDs")
        with gr.Row():
            source_language_box = gr.Textbox(label="Source Language")
            target_language_box = gr.Textbox(label="Target Language")
        translator_name_assign = gr.Dropdown([], label="Translators")
        role_assign = gr.Radio(["Translator", "QA"], label="Role", value="Translator")
        submit_assign_button = gr.Button("Assign Role")
        assign_logs = gr.Textbox(label="Logs")

    
    # For list
    refresh_list.click(get_translators, source_platform, translators_list)
    source_platform.change(get_translators, source_platform, translators_list)

    # For add
    submit.click(add_translator, [source_platform_add, email_id, source_languages, target_languages, contact_num_box, pricing_box], result)

    # For Update
    refresh_update_platform.click(get_translator_names, source_platform_update, translator_name)
    source_platform_update.change(get_translator_names, source_platform_update, translator_name)
    translator_name.change(select_current_data, [source_platform_update, translator_name], [source_languages_update, target_languages_update, contact_num_update, pricing_update])
    update_language.click(update_translator, [source_platform_update, translator_name, source_languages_update, target_languages_update, contact_num_update, pricing_update], result_update)

    # For Assign
    refresh_assign.click(get_videos_under_review, filter_videos_assign, under_review_videos_list)
    under_review_videos_list.change(get_traslators_assign, under_review_videos_list, [translator_name_assign, source_language_box, target_language_box, submit_assign_button])
    submit_assign_button.click(assign_translator, [translator_name_assign, under_review_videos_list, role_assign], assign_logs)


if __name__=="__main__":
    demo.launch(auth=(os.environ.get("GRADIO_USERNAME"), os.environ.get("GRADIO_PASSWORD")))