df_ai_int / classes /response_off.py
Ronio Jerico Roque
Refactor: Clean up code structure and improve payload handling in SEO and Social Media analysis
7dfa00b
raw
history blame
3.31 kB
from io import StringIO
from urllib.parse import urlparse
import streamlit as st
import requests
from dotenv import load_dotenv
import os
import time
from helper.upload_response import upload_response
from helper.upload_File import uploadFile
from helper.button_behaviour import hide_button, unhide_button
from helper.initialize_analyze_session import initialize_analyze_session
import pandas as pd
import asyncio
import json
from pymongo import MongoClient
from helper.data_field import data_field
class SeoOffPageAnalyst:
def __init__(self, model_url):
self.uploaded_files = []
self.file_dict = {}
self.model_url = model_url
#self.analyst_name = analyst_name
#self.data_src = data_src
#self.analyst_description = analyst_description
self.initialize()
self.row1()
def initialize(self):
# FOR ENV
load_dotenv()
# AGENT NAME
#st.header(self.analyst_name)
def request_model(self, payload_txt, headers):
response = requests.post(self.model_url, json=payload_txt, headers=headers)
response.raise_for_status()
output = response.json()
text = output["outputs"][0]["outputs"][0]["results"]["text"]["data"]["text"]
text = json.loads(text)
backlinks = text[0]
referring_domains = text[1]
return text
def process(self):
session = st.session_state['off_page_file_uploaded']
if session == 'uploaded':
with st.spinner('SEO Off Page Analyst...', show_time=True):
st.write('')
# OUTPUT FOR SEO ANALYST
payload_txt = {"input_value": data_field("Backlinks"),
"output_type": "text",
"input_type": "chat"}
headers = {
"Content-Type": "application/json",
"x-api-key": f"{os.getenv('x-api-key')}"
}
result = self.request_model(payload_txt, headers)
#end_time = time.time()
#time_lapsed = end_time - start_time
debug_info = {'data_field' : 'Off Page Analyst', 'result': result}
#debug_info = {'url_uuid': self.model_url.split("-")[-1],'time_lapsed' : time_lapsed, 'files': [*st.session_state['uploaded_files']],'payload': payload_txt, 'result': result}
upload_response(debug_info)
#with st.expander("Debug information", icon="⚙"):
# st.write(debug_info)
st.session_state["off_page_file_uploaded"] = ''
st.session_state['analyzing'] = False
def row1(self):
st.session_state['analyzing'] = False
session = st.session_state['off_page_file_uploaded']
if session == 'uploaded':
self.process()
if __name__ == "__main__":
st.set_page_config(layout="wide")
upload = uploadFile()