OxbridgeEconomics
commit
fba27b9
raw
history blame
16.9 kB
"""Utilis Functions"""
import os
import re
import json
import uuid
import time
import glob
import urllib.request
from urllib.parse import urlparse
from datetime import datetime, timedelta
from decimal import Decimal
import pandas as pd
import requests
import boto3
from lxml import etree
from googletrans import Translator
from transformers import pipeline
from PyPDF2 import PdfReader
AWS_ACCESS_KEY_ID = os.environ['AWS_ACCESS_KEY_ID']
AWS_SECRET_ACCESS_KEY = os.environ['AWS_SECRET_ACCESS_KEY']
# AWS_ACCESS_KEY_ID="AKIAQFXZMGHQYXKWUDWR"
# AWS_SECRET_ACCESS_KEY="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"
analyzer = pipeline("sentiment-analysis", model="ProsusAI/finbert")
translator = Translator()
with open('xpath.json', 'r', encoding='UTF-8') as f:
xpath_dict = json.load(f)
with open('xpath.json', 'r', encoding='UTF-8') as f:
patterns = json.load(f)
def get_client_connection():
"""Get dynamoDB connection"""
dynamodb = boto3.client(
service_name='dynamodb',
region_name='us-east-1',
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY
)
return dynamodb
def update_reference(report):
dynamodb = get_client_connection()
response = dynamodb.update_item(
TableName="reference_china",
Key={
'id': {'S': str(report['refID'])},
'sourceID': {'S': report['sourceID']}
},
UpdateExpression='SET link = :link, referenceID = :referenceID, LastModifiedDate = :LastModifiedDate',
ExpressionAttributeValues={
':link': {'S': report['link']},
':referenceID': {'S': report['referenceID']},
':LastModifiedDate': {'S': datetime.now().strftime("%Y-%m-%dT%H:%M:%S")},
}
)
print(response)
def download_files_from_s3(folder):
"""Download Data Files"""
if not os.path.exists(folder):
os.makedirs(folder)
client = boto3.client(
's3',
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
)
response = client.list_objects_v2(Bucket='china-securities-report', Prefix=f"{folder}/")
for obj in response['Contents']:
key = obj['Key']
if key.endswith('.parquet'):
client.download_file('china-securities-report', key, key)
file_paths = glob.glob(os.path.join(folder, '*.parquet'))
return pd.concat([pd.read_parquet(file_path) for file_path in file_paths], ignore_index=True)
def extract_from_pdf_by_pattern(url, pattern):
# Send a GET request to the URL and retrieve the PDF content
try:
response = requests.get(url, timeout=60)
pdf_content = response.content
# Save the PDF content to a local file
with open("downloaded_file.pdf", "wb") as file:
file.write(pdf_content)
# Open the downloaded PDF file and extract the text
with open("downloaded_file.pdf", "rb") as file:
pdf_reader = PdfReader(file)
extracted_text = ""
if 'pages' in pattern:
pages = pattern['pages']
else:
pages = len(pdf_reader.pages)
for page in pages:
text = pdf_reader.pages[page].extract_text()
if 'keyword' in pattern and pattern['keyword'] in text:
text = text.split(pattern['keyword'], 1)[1].strip()
else:
text = text.strip()
extracted_text += text
except:
extracted_text = ''
return extracted_text.replace('?\n', '?-\n').replace('!\n', '!-\n').replace('。\n', '。-\n').replace('\n',' ').replace('?-','?\n').replace('!-','!\n').replace('。-','。\n')
def get_reference_by_regex(pattern, text):
return re.findall(pattern, text)
def isnot_substring(list_a, string_to_check):
for s in list_a:
if s in string_to_check:
return False
return True
def extract_reference(row):
try:
pattern = next((elem for elem in patterns if elem['site'] == row['site']), None)
extracted_text = extract_from_pdf_by_pattern(row['attachment'],pattern)
reference_titles = re.findall(pattern['article_regex'], extracted_text)
reference_dates = re.findall(pattern['date_regex'], extracted_text)
reference_titles = [s.replace(' ', '') for s in reference_titles]
reference_dates = [s.replace(' ', '') for s in reference_dates]
print(reference_dates, reference_titles)
if 'remove' in pattern:
for remove_string in pattern['remove']:
reference_titles = [s.replace(remove_string, '') for s in reference_titles]
for title, date in zip(reference_titles, reference_dates):
print(title, date)
try:
date = datetime.strptime(date, pattern['date_format'])
except:
date = datetime(2006, 1, 1)
dates = []
if 'date_range' in pattern:
for i in range(pattern['date_range'] + 1):
dates.append((date + timedelta(days=i)).strftime('%Y-%m-%d'))
dates.append((date - timedelta(days=i)).strftime('%Y-%m-%d'))
dates.append(date.strftime('%Y-%m-%d'))
date = date.strftime('%Y-%m-%d')
if 'split' in pattern:
for split_item in pattern['split']:
if 'exceptional_string' in split_item:
if split_item['string'] in title and isnot_substring(split_item['exceptional_string'], title):
title = re.split(split_item['string'], title)[split_item['index']]
else:
if split_item['string'] in title:
title = title.split(split_item['string'])[split_item['index']]
if len(data[(data['titleCN'].str.contains(title)) & (data['site'] == row['site']) & (data['publishdate'].isin(dates))]) == 0:
print("------------ = 0 ------------")
print(date, repr(title))
elif len(data[(data['titleCN'].str.contains(title)) & (data['site'] == row['site']) & (data['publishdate'].isin(dates))]) > 1:
print("------------ > 1 ------------")
print(date, repr(title))
else:
print("------------ = 1 ------------")
reference_df = data[(data['titleCN'].str.contains(title)) & (data['site'] == row['site']) & (data['publishdate'].isin(dates))]
row['referenceID'] = reference_df.iloc[0]['id']
row['link'] = reference_df.iloc[0]['link']
row['sourceID'] = row['id_x']
row['refID'] = uuid.uuid5(uuid.NAMESPACE_OID, str(row['sourceID'])+str(row['referenceID']))
print(date, repr(title), row['sourceID'],row['referenceID'])
except Exception as error:
print(error)
# update_reference(row)
def translate(text):
return translator.translate(text, dest='en').text
def datemodifier(date_string, date_format):
"""Date Modifier Function"""
try:
to_date = time.strptime(date_string,date_format)
return time.strftime("%Y-%m-%d",to_date)
except:
return False
def fetch_url(url):
response = requests.get(url, timeout = 60)
if response.status_code == 200:
return response.text
else:
return None
def translist(infolist):
"""Translist Function"""
out = list(filter(lambda s: s and
(isinstance (s,str) or len(s.strip()) > 0), [i.strip() for i in infolist]))
return out
def encode(content):
"""Encode Function"""
text = ''
for element in content:
if isinstance(element, etree._Element):
subelement = etree.tostring(element).decode()
subpage = etree.HTML(subelement)
tree = subpage.xpath('//text()')
line = ''.join(translist(tree)).\
replace('\n','').replace('\t','').replace('\r','').replace(' ','').strip()
else:
line = element
text += line
return text
def encode_content(content):
"""Encode Function"""
text = ''
for element in content:
if isinstance(element, etree._Element):
subelement = etree.tostring(element).decode()
subpage = etree.HTML(subelement)
tree = subpage.xpath('//text()')
line = ''.join(translist(tree)).\
replace('\n','').replace('\t','').replace('\r','').replace(' ','').strip()
else:
line = element
if line != '':
line = line + '\n'
text += line
index = text.find('打印本页')
if index != -1:
text = text[:index]
try:
summary = '\n'.join(text.split('\n')[:2])
except:
summary = text
return text, summary
def extract_from_pdf(url):
# Send a GET request to the URL and retrieve the PDF content
response = requests.get(url, timeout=60)
pdf_content = response.content
# Save the PDF content to a local file
with open("downloaded_file.pdf", "wb") as file:
file.write(pdf_content)
# Open the downloaded PDF file and extract the text
with open("downloaded_file.pdf", "rb") as file:
pdf_reader = PdfReader(file)
num_pages = len(pdf_reader.pages)
extracted_text = ""
for page in range(num_pages):
text = pdf_reader.pages[page].extract_text()
if text and text[0].isdigit():
text = text[1:]
# first_newline_index = text.find('。\n')
# text = text[:first_newline_index+1].replace('\n', '') + text[first_newline_index+1:]
text = text.replace('?\n', '?-\n').replace('!\n', '!-\n').replace('。\n', '。-\n').replace('\n','').replace('?-','?\n').replace('!-','!\n').replace('。-','。\n')
print(text)
if text != '':
extracted_text += text
try:
summary = '\n'.join(extracted_text.split('\n')[:2])
except:
summary = text
return extracted_text, summary
def get_db_connection():
"""Get dynamoDB connection"""
dynamodb = boto3.resource(
service_name='dynamodb',
region_name='us-east-1',
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY
)
return dynamodb
def sentiment_computation(content):
label_dict = {
"positive": "+",
"negative": "-",
"neutral": "0",
}
sentiment_score = 0
maximum_value = 0
raw_sentiment = analyzer(content[:511], top_k=None)
sentiment_label = None
for sentiment_dict in raw_sentiment:
value = sentiment_dict["score"]
if value > maximum_value:
sentiment_label = sentiment_dict["label"]
maximum_value = value
if sentiment_dict["label"] == "positive":
sentiment_score = sentiment_score + value
if sentiment_dict["label"] == "negative":
sentiment_score = sentiment_score - value
else:
sentiment_score = sentiment_score + 0
return sentiment_score, label_dict[sentiment_label]
def crawl(url, article):
domain = '.'.join(urlparse(url).netloc.split('.')[1:])
req = urllib.request.urlopen(url)
text = req.read()
html_text = text.decode("utf-8")
page = etree.HTML(html_text)
contentCN, summary = encode_content(page.xpath(xpath_dict[domain]['content']))
article['originSite'] = xpath_dict[domain]['siteCN']
article['site'] = xpath_dict[domain]['site']
article['titleCN'] = encode(page.xpath(xpath_dict[domain]['title']))
article['title'] = translate(article['titleCN'])
if 'author' in xpath_dict[domain]:
article['author'] = translate(encode(page.xpath(xpath_dict[domain]['author'])))
else:
article['author'] = ""
article['contentCN'] = repr(contentCN)[1:-1].strip()
if len(article['contentCN']) < 10:
return None
CONTENT_ENG = ''
for element in contentCN.split("\n"):
CONTENT_ENG += translate(element) + '\n'
article['content'] = repr(CONTENT_ENG)[1:-1].strip()
if 'subtitle' in xpath_dict[domain]:
article['subtitle'] = translate(encode(page.xpath(xpath_dict[domain]['subtitle'])))
else:
article['subtitle'] = translate(summary)
article['publishDate'] = datemodifier(encode(page.xpath(xpath_dict[domain]['publishdate'])), xpath_dict[domain]['datetime_format'])
article['link'] = url
article['attachment'] = ""
article['sentimentScore'], article['sentimentLabel'] = sentiment_computation(CONTENT_ENG.replace("\n",""))
article['id'] = uuid.uuid5(uuid.NAMESPACE_OID, article['titleCN']+article['publishDate'])
print(article['id'], article['site'] )
update_content(article)
def upsert_content(report):
"""Upsert the content records"""
dynamodb = get_db_connection()
table = dynamodb.Table('article_china')
# Define the item data
item = {
'id': str(report['id']),
'site': report['site'],
'title': report['title'],
'titleCN': report['titleCN'],
'contentCN': report['contentCN'],
'category': report['category'],
'author': report['author'],
'content': report['content'],
'subtitle': report['subtitle'],
'publishDate': report['publishDate'],
'link': report['link'],
'attachment': report['attachment'],
# 'authorID': str(report['authorid']),
# 'entityList': report['entitylist'],
'sentimentScore': Decimal(str(report['sentimentScore'])).quantize(Decimal('0.01')),
'sentimentLabel': report['sentimentLabel'],
'LastModifiedDate': datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
}
response = table.put_item(Item=item)
print(response)
# def get_client_connection():
# """Get dynamoDB connection"""
# dynamodb = boto3.client(
# service_name='dynamodb',
# region_name='us-east-1',
# aws_access_key_id=AWS_ACCESS_KEY_ID,
# aws_secret_access_key=AWS_SECRET_ACCESS_KEY
# )
# return dynamodb
def delete_records(item):
dynamodb_client = get_client_connection()
dynamodb_client.delete_item(
TableName="article_test",
Key={
'id': {'S': item['id']},
'site': {'S': item['site']}
}
)
def update_content(report):
dynamodb = get_client_connection()
response = dynamodb.update_item(
TableName="article_china",
Key={
'id': {'S': str(report['id'])},
'site': {'S': report['site']}
},
UpdateExpression='SET title = :title, titleCN = :titleCN, contentCN = :contentCN, category = :category, author = :author, content = :content, subtitle = :subtitle, publishDate = :publishDate, link = :link, attachment = :attachment, sentimentScore = :sentimentScore, sentimentLabel = :sentimentLabel, LastModifiedDate = :LastModifiedDate',
ExpressionAttributeValues={
':title': {'S': report['title']},
':titleCN': {'S': report['titleCN']},
':contentCN': {'S': report['contentCN']},
':category': {'S': report['category']},
':author': {'S': report['author']},
':content': {'S': report['content']},
':subtitle': {'S': report['subtitle']},
':publishDate': {'S': report['publishDate']},
':link': {'S': report['link']},
':attachment': {'S': report['attachment']},
':LastModifiedDate': {'S': datetime.now().strftime("%Y-%m-%dT%H:%M:%S")},
':sentimentScore': {'N': str(Decimal(str(report['sentimentScore'])).quantize(Decimal('0.01')))},
':sentimentLabel': {'S': report['sentimentLabel']}
}
)
print(response)
def update_content_sentiment(report):
dynamodb = get_client_connection()
response = dynamodb.update_item(
TableName="article_test",
Key={
'id': {'S': report['id']},
'site': {'S': report['site']}
},
UpdateExpression='SET sentimentScore = :sentimentScore, sentimentLabel = :sentimentLabel',
ExpressionAttributeValues={
':sentimentScore': {'N': str(Decimal(str(report['sentimentscore'])).quantize(Decimal('0.01')))},
':sentimentLabel': {'S': report['sentimentlabel']}
}
)
print(response)
data = download_files_from_s3('data')