Update utils.py

utils.py

@@ -1,156 +1,123 @@
-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)
-    msg = msg[:7000]
-    
-    response = client.chat.completions.create(
-        messages=[
-            {
-                "role": "user",
-                "content": msg,
-            }
-        ],
-        model="azureai",
-        ) 
-    
-    return response.choices[0].message.content
-    
-def call(prompt, text):
-  client = Groq(api_key=os.getenv("key"),)
-
-  prompt = prompt + " Answer only the summary, no instructions"
-  chat_completion = client.chat.completions.create(
-  messages=[
-      {
-          "role": "user",
-          "content": "{} {}".format(prompt, text),
-      }
-  ],
-  model=model,
-  )
-
-  return chat_completion.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 = callAzure(promptShort[id], long_text)
-    try:
-        idx = answer.index("\n")
-    except:
-        idx = -2
-    output['Short Summary'] = answer[idx+2:]
-
-    answer = callAzure(promptLong[id], long_text)
-    try:
-        idx = answer.index("\n")
-    except:
-        idx = -2
-    output['Long summary'] = answer[idx+2:]
-
-    prompt = "1 word Financial SEO tag for this news article"
-    answer = callAzure(prompt, output['Short Summary'])
-    try:
-        idx = answer.index("\n")
-    except:
-        idx = -2
-    output['Tag'] = answer[idx+2:]
-
-    prompt = "Give a single headline for this News Article"
-    answer = callAzure(prompt, output['Short Summary'])
-    try:
-        idx = answer.index("\n")
-    except:
-        idx = -2
-    output['Headline'] = answer[idx+2:]
-
-    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
-    
+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",
+        ) 
+    
+    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 = callAzure(promptShort[id], long_text)
+    output['Short Summary'] = answer
+
+    answer = callAzure(promptLong[id], long_text)
+    output['Long summary'] = answer
+
+    prompt = "1 word 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