utils.py

import datetime
from urllib.request import Request, urlopen
from pypdf import PdfReader
from io import StringIO
import io
import pandas as pd
import os
import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from transformers import pipeline
from openai import OpenAI
from groq import Groq
import time 
from openai import OpenAI

#openai_key = "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"
#client = OpenAI(api_key = openai_key)
desc = pd.read_excel('Descriptor.xlsx',header = None)
desc_list  = desc.iloc[:,0].to_list()

def callAzure(prompt,text):
    
    url = "https://Mistral-large-tmhcg-serverless.eastus2.inference.ai.azure.com"
    api_key = "LB0ha1R4k3pNpHl68P3VtUZ3sMLr3wT7"
    client = OpenAI(base_url=url, api_key=api_key)
    msg = "{} {}".format(prompt, text)
    
    response = client.chat.completions.create(
        messages=[
            {
                "role": "user",
                "content": msg,
            }
        ],
        model="azureai",
        max_tokens = 20000
        ) 
    
    return response.choices[0].message.content
    
def filter(input_json):

  sym = pd.read_excel('symbol.xlsx',header = None)
  sym_list  = sym.iloc[:,0].to_list()
  
  if input_json['FileURL']==None or input_json['FileURL'].lower()=='null':
    return [0,"File_URL"]
  if input_json['symbol']== 'null' or input_json['symbol'] not in sym_list:
    return [0,"symbol"]
  if input_json['TypeofAnnouncement'] not in ['General_Announcements','Outcome','General']:
    return [0,"Annoucement"]
  if input_json['Descriptor'] not in desc_list:
    return [0,"Desc"]
      
  url = 'https://www.bseindia.com/xml-data/corpfiling/AttachLive/'+ input_json['FileURL'].split('Pname=')[-1]
  req = Request(url, headers={'User-Agent': 'Mozilla/5.0'})
  html = urlopen(req)
  cont = html.read()
  reader = PdfReader(io.BytesIO(cont))
  content = ''
  for i in range(len(reader.pages)):
    content+= reader.pages[i].extract_text()
  document = content

  return [1, document]

def summary(input_json):
    
    prompt = pd.read_excel('DescriptorPrompt.xlsx')
    promptShort  = prompt.iloc[:,1].to_list()
    promptLong  = prompt.iloc[:,2].to_list() 
    
    output = {}
    filtering_results = filter(input_json)
    if filtering_results[0] == 0:
        #return 0
        return filtering_results[1]

    id = desc_list.index(input_json['Descriptor'])
    long_text = filtering_results[1]

    url = 'https://www.bseindia.com/xml-data/corpfiling/AttachLive/'+ input_json['FileURL'].split('Pname=')[-1]

    output["Link to BSE website"] = url

    output["Date of time of receiving data from BSE"] = input_json["newsdate"] + "Z"
    
    output["Stock Ticker"] = input_json['symbol']

    answer = "You are an financial expert" + callAzure(promptShort[id], long_text)
    output['Short Summary'] = answer

    answer = "You are an financial expert" + callAzure(promptLong[id] + " The Summary should be around 2000 characters. Please make sure it is not smaller than that. Do not truncate the final answers", long_text)
    output['Long summary'] = answer

    prompt = "Answer in 1 word only. Financial SEO tag for this news article"
    answer = callAzure(prompt, output['Short Summary'])
    output['Tag'] = answer

    prompt = "Give a single headline for this News Article"
    answer = callAzure(prompt, output['Short Summary'])
    output['Headline'] = answer

    utc_now = datetime.datetime.utcnow()
    ist_now = utc_now.astimezone(datetime.timezone(datetime.timedelta(hours=5, minutes=30)))
    
    Date = ist_now.strftime("%Y-%m-%d")
    time = ist_now.strftime("%X")
    output['Date and time of data delivery from Skylark'] = Date+"T"+time+"Z"
    
    prompt = "Answer in one word the sentiment of this News out of Positive, Negative or Neutral {}"
    output['Sentiment'] = callAzure(prompt, output['Short Summary'])

    #time.sleep(60)
    # response = client.images.generate(
    # model="dall-e-3",
    # prompt=headline.text,
    # size="1024x1024",
    # quality="standard",
    # n=1
    # ) 
    # output["Link to Infographic (data visualization only)] = response.data[0].url
    
    return output